macrame/

connection.rs

use std::path::{Path, PathBuf};
use std::sync::Arc;
use tokio::sync::{mpsc, oneshot};

use crate::error::{classify, BulkInterrupted, BulkResult, DbError, Result, WriteOp};
use crate::graph::edge::EdgeAssertion;
use crate::graph::lineage::{Ancestor, LineageShape, Lineages};
use crate::integrity::{rebuild_current, RebuildReport};
use crate::plan::ReadPlan;
use crate::schema::migrations;
use crate::temporal::archive::{archive, rehydrate, ArchiveReport, RehydrateReport};
use crate::temporal::interval::Interval;
use crate::temporal::snapshot::{self, SnapshotCadence};
use crate::util::clock::FutureStampPolicy;
use crate::util::clock::{Clock, SystemClock};
use crate::util::timestamp;
use crate::vector::ModelName;

/// Rows per chunk on the background write paths (§5.1.5, D-011, D-014, D-058).
///
/// The Write Actor holds the sole write connection, so a single large statement
/// blocks every other writer for its duration. Chunking bounds that stall; the
/// cost is that a bulk import is *not* atomic across chunks, which is why
/// all-or-nothing is [`Database::write_bulk_atomic`] — a separate entry point,
/// with its own command on the actor's protocol — rather than a tuning
/// parameter here.
///
/// # Why these are four constants and not one
///
/// Through 0.5.5 this was a single `CHUNK_ROWS = 1000` for all four bulk paths.
/// The golden rule it was meant to serve is a bound on *duration* — a background
/// chunk must commit fast enough that an interactive write queued behind it is
/// not made to wait — and one row count cannot express one duration across paths
/// whose measured per-row costs differ by 60× (D-058). At 1,000 rows the four
/// paths took 3.5 ms, 24 ms, 89 ms and 143 ms: the same constant, four answers,
/// three of them far outside the bound.
///
/// Each size below is derived from `benches/budgets.rs`'s `chunk_scaling`
/// sweep against [`CHUNK_BUDGET`], then verified by measuring that size directly.
/// They are *measurements of this machine*, not universal constants — D-055's
/// reasoning about reference hardware applies here too, and re-deriving them on
/// materially different storage is a `cargo bench` away.
///
/// # Sized for the tail, not the median
///
/// The first derivation solved `f + c·n = 3 ms` exactly and produced sizes whose
/// *median* commit was 2.93 ms and whose upper estimate was 2.96 — inside the
/// bound as reported and outside it for any chunk slower than typical. A latency
/// bound is a statement about the chunk an unlucky interactive write actually
/// queues behind, so these solve for ≈2.5 ms instead, leaving the remainder as
/// headroom for the tail. That costs a few percent of throughput on the two
/// linear paths and nothing on the two superlinear ones.
///
/// As measured by `chunk_budget`, each at its own size: edges **2.39 ms**,
/// concepts **2.35 ms**, annotations **2.36 ms**, embeddings **2.06 ms**, no
/// upper estimate above 2.42.
///
/// # Known limitation: these are empty-database figures
///
/// `chunk_budget` seeds concepts and starts with **no links and no vectors**,
/// and D-059 established that per-row cost on the edge and embedding paths grows
/// with the size of the structure being written, not with the chunk. The same
/// 90-edge chunk takes **9.06 ms** into an 8,000-edge table. So the bound is met
/// as measured here and *not* met on a populated database.
///
/// That gap was published as 47.7 ms until 0.10.0 and attributed to the schema
/// defect D-059 documents. The defect was fixed by the `v5 → v6` rung and the
/// figure was never updated. 9.08 ms is a 0.10.0 measurement, not D-059's 8.0 ms
/// carried forward: `chunk_budget` gained a seeded arm, because until it did,
/// nothing in the bench suite wrote a chunk into a populated table and this
/// number was unfalsifiable. It agrees with D-059 once the session is accounted
/// for — the empty arm read 2.69 and 2.65 ms beside it against the 2.39 ms
/// published above, so the *ratio* is 3.4× here and 3.35× there.
///
/// **The residual is attributed as of 0.11.0 (D-142).** It is not the missing
/// index, which shipped in 0.5.6; it is the `links_current` write. Dropping the
/// three `links` insert triggers one at a time puts effectively all of the
/// growth in `trg_links_current_sync` — the single-open guard contributes none,
/// the log trigger and the base insert ~0.35 ms of a 4.15 ms rise — and within
/// that trigger, 89% of the growth is maintenance of `idx_lc_traversal_cover`
/// and `idx_lc_open_interval` rather than the upsert itself, which costs 0.49 ms
/// run directly against the same table. Page-cache size, foreign keys and the
/// fixture's key distribution were each tested and are each not the cause.
///
/// Knowing the cause does not by itself change the constant: the expensive index
/// is D-042's covering index for the traversal, so narrowing it moves cost onto
/// the read path it exists to protect. Re-deriving these constants against the
/// D-088 fixture matrix is the named successor.
///
/// # These are ceilings as of 0.12.0, not sizes
///
/// D-143 re-derived all four against the D-088 matrix and the edge path came
/// back **20** against a shipped 90 — and 20 would have been wrong at 80,000
/// edges for the same reason 90 is wrong at 8,000, because per-row cost there
/// grows with `links_current`. The finding was that no row count can bound a
/// duration on such a path.
///
/// So the chunk loop stopped trying to pick one ahead of time. Each chunk is
/// timed by the actor and its measured hold chooses the next size; these
/// constants are the **largest** size that will ever be asked for, and every
/// derivation below still applies to them as such. A path may run well under its
/// constant on a populated database and at exactly it on an empty one, and both
/// are the bound being met rather than a size being missed.
pub mod chunk_rows {
    /// Edge assertions (`bulk_import`).
    ///
    /// Per-row cost on this path rises with the size of `links_current`, not
    /// with the chunk (D-059) — so cutting the chunk buys latency and costs
    /// throughput, ~11% for 1,000 edges. An earlier version of this comment
    /// claimed it was 3.3× *faster*; that came from multiplying eleven copies of
    /// a chunk measured into an empty database.
    ///
    /// **This size does not meet the 3 ms bound on a populated database.** 90
    /// edges into an 8,000-edge table take **9.06 ms** — measured, two sessions
    /// at 9.08 and 9.05, against an empty-table arm of 2.69 and 2.65 beside
    /// them (D-136).
    ///
    /// The reason given here until 0.10.0 — that `trg_links_single_open`'s
    /// `EXISTS` scans the whole out-degree, "a schema defect with a proven fix,
    /// recorded in D-059 and not applied here" — described 0.5.5. The fix *was*
    /// applied, as the `v5 → v6` rung, and took this from 47.7 ms to ~8 ms.
    /// What survives is the miss: the bound is still exceeded ~3×. Its cause is
    /// no longer unknown — D-142 attributes it to `trg_links_current_sync`, and
    /// within that to secondary-index maintenance on `links_current` — and the
    /// guard this comment used to blame contributes **no** growth at all.
    ///
    /// **The constant is unchanged, and that is now a measured decision**
    /// (D-143). Re-derived against all four D-088 shapes at 8,000 edges, they
    /// agree that the largest size meeting the bound is **20**. It stays at 90
    /// because 20 is the same miss at a larger population — per-row cost grows
    /// with `links_current`, so a constant fitted at 8,000 edges is wrong at
    /// 80,000 — while the throughput cost of turning eleven chunks into fifty
    /// is certain and immediate (D-058). The fix is not a row count: it is for
    /// the chunk loop to stop on elapsed time, **delivered in 0.12.0**. This
    /// number is now the ceiling that loop starts from and never exceeds; on a
    /// populated table it converges below it within a chunk or two.
    ///
    /// D-134 retired the growth claim on the neighbouring *single-assertion*
    /// path and did not measure this one; D-136 is why this line now carries a
    /// measurement rather than a figure quoted from 0.5.6.
    pub const EDGES: usize = 90;

    /// Concept upserts (`write_concepts`).
    ///
    /// Linear at ~23 µs per row, so unlike [`EDGES`] this size *is* a genuine
    /// throughput sacrifice: 1,000-row chunks ran at 23.6 µs per row against
    /// ~35 µs here. Paid deliberately — a 1,000-row chunk takes 24 ms, eight
    /// times the bound.
    pub const CONCEPTS: usize = 70;

    /// Analytics annotations (`write_analytics_annotations`).
    ///
    /// The one path where the old constant was nearly right, and the only bulk
    /// table with no triggers at all: ~2.5 µs per row, linear, so the bound buys
    /// a large chunk. 1,000 rows would be 3.5 ms — over, but only just.
    pub const ANNOTATIONS: usize = 600;

    /// Embedding vectors (`upsert_embeddings`).
    ///
    /// The smallest by a wide margin, because DiskANN index maintenance makes an
    /// embedding the most expensive row in the system. That cost grows with the
    /// **corpus**, not the chunk (D-059): a fixed 30-vector chunk costs 49 µs per
    /// vector into an empty corpus and 224 µs into an 8,000-vector one. Graph
    /// insertion getting dearer as the graph grows is what DiskANN is, so unlike
    /// [`EDGES`] there is nothing here to fix — but it does mean this size buys
    /// latency at some throughput, not for free.
    pub const EMBEDDINGS: usize = 30;
}

/// What one chunk transaction cost, reported by the actor to the caller-side
/// chunk loop (0.12.0, W1).
///
/// `held` is measured **inside** the actor, around its own transaction, and
/// therefore excludes the time the command spent queued. That exclusion is the
/// point: queue time is what strict preemption *does*, and a controller fed
/// `send + await` would shrink chunks as punishment for the actor correctly
/// serving an interactive write first.
///
/// Crate-internal, along with the command enums that carry it. It was `pub`
/// through 0.13.32 only because they were (D-206).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct ChunkOutcome {
    /// Rows the transaction actually wrote.
    pub rows: usize,
    /// How long the actor held the write lock for them.
    pub held: std::time::Duration,
}

/// A flag a caller can raise to stop a chunked bulk write (0.13.8, W7.6, D-181).
///
/// Cheap to clone and safe to set from any thread, which is the whole point: the
/// task running the import is the one thing that cannot cancel it. Hand a clone
/// to whatever *can* — a signal handler, a UI thread, a timeout task — and it
/// takes effect at the next chunk boundary.
///
/// **A boundary, not an abort.** Nothing rolls back and no in-flight
/// transaction is interrupted: the loop notices between chunks and stops
/// sending. The chunks that committed stay committed, and
/// [`BulkInterrupted::written`](crate::BulkInterrupted::written) says how many
/// rows those were. That is the same per-chunk boundary
/// [`Database::bulk_import`] already documents, so cancellation adds a reason to
/// stop and no new failure mode.
///
/// Setting it after the last chunk has committed does nothing — a finished
/// write reports success, because it succeeded.
#[derive(Clone, Debug, Default)]
pub struct CancelToken(Arc<std::sync::atomic::AtomicBool>);

impl CancelToken {
    /// A token that has not been cancelled.
    pub fn new() -> Self {
        Self::default()
    }

    /// Ask the bulk write holding a clone of this token to stop at its next
    /// chunk boundary. Idempotent; a token never un-cancels.
    pub fn cancel(&self) {
        // `Relaxed` on both sides is sufficient and deliberate: nothing is
        // published *through* this flag. The rows are ordered by the database
        // and the chunk results by the response channel, so the only thing the
        // reader needs is to observe the store eventually, which every ordering
        // guarantees.
        self.0.store(true, std::sync::atomic::Ordering::Relaxed);
    }

    /// Whether [`Self::cancel`] has been called on this token or any clone.
    pub fn is_cancelled(&self) -> bool {
        self.0.load(std::sync::atomic::Ordering::Relaxed)
    }
}

/// One chunk's worth of progress, handed to the callback on
/// [`BulkControl::on_progress`] (0.13.8, W7.6).
///
/// Reported *after* the chunk has committed, so `written` is a count of rows
/// that are in the database and will stay there even if the next chunk fails.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub struct BulkProgress {
    /// Rows committed so far, across every chunk including this one.
    pub written: usize,
    /// Rows in the batch the caller passed. `written` reaching this means the
    /// last chunk has committed.
    pub total: usize,
    /// Rows this chunk wrote. Not a constant: the loop resizes chunks against
    /// [`CHUNK_BUDGET`] as it measures them (D-058).
    pub rows: usize,
    /// How long the actor held the write lock for this chunk — the same figure
    /// the controller steers on. Measured inside the actor, around its own
    /// transaction, so it excludes the time the command spent queued.
    pub held: std::time::Duration,
}

/// Cancellation and progress for the four chunked bulk paths (0.13.8, W7.6,
/// D-181).
///
/// Default is "neither", which is what [`Database::bulk_import`] and its three
/// siblings pass. The `_with` variants take one of these:
///
/// ```no_run
/// # use macrame::{BulkControl, CancelToken, Database};
/// # async fn f(db: &Database, edges: Vec<macrame::prelude::EdgeAssertion>) {
/// let token = CancelToken::new();
/// let stopper = token.clone();
/// tokio::spawn(async move {
///     tokio::time::sleep(std::time::Duration::from_secs(30)).await;
///     stopper.cancel();
/// });
///
/// let control = BulkControl::new()
///     .cancel_with(token)
///     .on_progress(|p| println!("{}/{} rows", p.written, p.total));
///
/// match db.bulk_import_with(edges, control).await {
///     Ok(n) => println!("imported {n}"),
///     Err(e) => println!("stopped after {}: {}", e.written, e.cause),
/// }
/// # }
/// ```
///
/// **The callback runs on the importing task, between chunks.** It is therefore
/// on the critical path: whatever it does is time the next chunk is not being
/// sent in. Printing or updating a counter is what it is for; a blocking write
/// is not, and neither is anything that calls back into the same `Database`,
/// which would deadlock the loop against a channel it is itself draining.
#[derive(Default, Clone)]
pub struct BulkControl {
    cancel: Option<CancelToken>,
    on_progress: Option<Arc<dyn Fn(BulkProgress) + Send + Sync>>,
}

impl std::fmt::Debug for BulkControl {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("BulkControl")
            .field("cancel", &self.cancel)
            .field("on_progress", &self.on_progress.is_some())
            .finish()
    }
}

impl BulkControl {
    /// Neither cancellation nor progress — what the plain bulk methods pass.
    pub fn new() -> Self {
        Self::default()
    }

    /// Stop at the next chunk boundary when `token` is cancelled.
    pub fn cancel_with(mut self, token: CancelToken) -> Self {
        self.cancel = Some(token);
        self
    }

    /// Call `f` after every chunk commits. See the note on [`BulkControl`]
    /// about what this closure is allowed to do.
    pub fn on_progress(mut self, f: impl Fn(BulkProgress) + Send + Sync + 'static) -> Self {
        self.on_progress = Some(Arc::new(f));
        self
    }

    fn is_cancelled(&self) -> bool {
        self.cancel.as_ref().is_some_and(CancelToken::is_cancelled)
    }

    fn report(&self, progress: BulkProgress) {
        if let Some(f) = &self.on_progress {
            f(progress);
        }
    }
}

/// Smallest chunk the adaptive loop will fall to (0.12.0, W2).
///
/// # A floor is a deliberate, measured violation of [`CHUNK_BUDGET`]
///
/// Feedback alone converges to whatever size meets the budget, and on a
/// populated `links` table that size keeps falling — per-row cost there grows
/// with the table (D-059, D-142), so there is no size at which the *fixed* cost
/// of a transaction stops dominating. Left unbounded the loop reaches chunks of
/// one or two rows, where nearly all the work is `BEGIN`/`COMMIT` and the import
/// no longer finishes.
///
/// 35 is measured, and **re-measured against the loop that uses it** — the
/// difference matters, because the figure this constant shipped with was an
/// extrapolation. `examples/chunk_matrix.rs -- converge` runs a 900-edge
/// `bulk_import` into each of the four D-088 shapes at 8,000 edges and reports
/// the actor's own per-transaction readings. A 35-row chunk costs **3.11–3.43 ms**
/// across the four shapes, two sessions, excluding the run-up. The floor misses
/// the 3 ms bound by 0.1–0.4 ms, not by the ~1.1 ms predicted from the sweep.
///
/// The miss is **steady state** — not a one-chunk transient on the way down —
/// and the defense is the argument [`CHUNK_BUDGET`] is answerable to rather than
/// the number itself: an interactive assertion arriving at the worst moment
/// waits ~3.2 ms for the chunk in flight and then runs its own ≤ 5 ms write, so
/// ~8.2 ms against a 16.7 ms frame.
///
/// What the same measurement says about the *size*: on this path at this
/// population the loop goes `[90, 35, 35, …]` on all four shapes and never picks
/// anything between. The proportional shrink from a 90-row chunk proposes ~31
/// rows, which clamps here — so on the edge path the floor is not a safety net
/// under the controller, it **is** the operating point, and this number is
/// carrying more weight than a backstop normally would. Re-measure it, not the
/// controller, when the edge path's per-row cost changes.
const CHUNK_FLOOR: usize = 35;

/// Size of the next chunk, from what the last one cost (0.12.0, W2).
///
/// Pure on purpose — no clock, no database, no actor — so the control law can be
/// tested for the properties that matter without a fixture. Three regimes:
///
/// | last hold | response | why |
/// |---|---|---|
/// | over `budget` | shrink to `current · budget / held`, × 0.9 | back off *fast* from a bound already being exceeded; the 0.9 undershoots so the correction does not have to be repeated |
/// | under `budget / 2` | grow by a quarter of `current`, at least one row | approach the bound *slowly*; the dead band above it stops a size that is merely comfortable from oscillating |
/// | otherwise | hold | in band, and moving costs more than it buys |
///
/// The asymmetry is the whole design. Proportional shrinking converges from
/// above in one or two steps, which matters because every step over budget is a
/// latency miss a caller can feel; additive growth cannot overshoot by more than
/// 25%, which matters because the ceiling is a throughput preference and not a
/// bound.
///
/// `ceiling` is the path's [`chunk_rows`] constant, which is why those constants
/// keep their values and their derivations: they are no longer the size, they
/// are the largest size this will ever ask for. `floor` is [`CHUNK_FLOOR`] —
/// see there for the budget it knowingly misses.
///
/// Never returns 0, at any input, including `held == 0` or `current == 0`.
fn next_chunk_size(
    current: usize,
    held: std::time::Duration,
    budget: std::time::Duration,
    floor: usize,
    ceiling: usize,
) -> usize {
    let held = held.as_nanos().max(1);
    let budget_ns = budget.as_nanos().max(1);
    let current = current.max(1);

    let next = if held > budget_ns {
        // Integer math, and the `max(1)` matters: a chunk 200× over budget
        // would otherwise propose 0 and the loop would stop making progress.
        let scaled = (current as u128) * budget_ns * 9 / (held * 10);
        (scaled as usize).max(1)
    } else if held * 2 < budget_ns {
        // Saturating because `current` is a `usize` and this is the one branch
        // that adds to it. Nothing sane reaches the boundary; the clamp below
        // makes the answer correct anyway rather than a debug panic.
        current.saturating_add((current / 4).max(1))
    } else {
        current
    };

    // Applied last and unconditionally, so a caller that passes a reversed pair
    // gets the floor rather than a panic — and `max(1)` last of all, because a
    // chunk of zero rows is the single answer no loop can make progress from.
    next.clamp(floor.min(ceiling), ceiling).max(1)
}

/// The latency bound [`chunk_rows`] is derived from (§5.1.5, D-058).
///
/// This is the golden rule's actual content. §9 has carried it as a row count
/// with a duration attached — "chunk commit, 500 rows ≤ 3 ms" — which reads as
/// two requirements and is one: the duration is the requirement, and the row
/// count is whatever satisfies it on a given path and machine.
///
/// 3 ms is §9's number, kept rather than renegotiated. What it buys, end to end:
/// an interactive assertion arriving at the worst possible moment waits for the
/// chunk in flight (≤ 3 ms — the SQLite write lock is not preemptible, so
/// priority buys the *next* turn and not this one) and then runs its own write
/// (≤ 5 ms, §9), so ≤ 8 ms
/// worst case. That fits inside a 60 Hz frame with room, which is the standard
/// this bound is ultimately answerable to.
///
/// # Some operations are exempt, and the exemption is a contract, not an oversight
///
/// This was recorded in three separate rustdoc notes and nowhere near the bound
/// itself, which is where a reader looks for its scope (§8.6). Stated here, with
/// Wave 3's measurements:
///
/// | Path | Bound | Why it cannot be chunked |
/// |---|---|---|
/// | [`Database::write_bulk_atomic`] | none — caller-sized `Vec` | D-014: the batch is *one act* under one stamp. Splitting it is the thing the method exists not to do |
/// | [`Database::archive`] | measured **26.8 ms** for 2,000 archivable edges; see [`Database::archive_windowed`] | D-012: copy-then-delete must be atomic, or a crash between the phases duplicates or loses rows |
/// | `rebuild_current` | measured **24.6 / 104 / 318 ms** at 4K / 16K / 40K rows in `links` (was "~50 s per 10M edges", which nothing had measured) | D-023: the window between `DELETE` and `INSERT` is the whole of current belief; a reader landing in it sees a graph with no edges and no error |
/// | [`Database::rehydrate`] | unmeasured; a function of how many rows the caller named | D-012 backwards: the same copy-then-delete atomicity, in the other direction. **A row here since 0.12.9 only because it was previously invisible** — rehydration reported as `archive` and inherited its exemption without anyone deciding on it (W4.3, D-152) |
/// | [`Database::archive_branch`] | unmeasured; a function of how much one lineage wrote | D-012 again, and D-230's chain: the links, the log entries and the `branches` row leave together or the ledger disagrees with itself about what is currently believed. There is no smaller unit — half a forgotten lineage is a lineage whose reads are answered by its parent |
/// | the swap turn of [`Database::rebuild_current_chunked`], counted as `shadow_swap` | measured **46.8 ms** at the largest fixture (D-082), and it grows with the table | Index names are global and SQLite has no `ALTER INDEX … RENAME`, so the shadow cannot carry `idx_lc_traversal_cover` while the live table still holds it — all three indexes are built here, under the lock. This is the residual T1.2 could not remove, and there is no smaller unit: half a swapped projection is not a projection. **Exempt since 0.14.16** (W12.16, D-233). The *fill* half keeps its own kind and is deliberately absent from this table, which is what makes a violation there a regression rather than a constant |
/// | [`Database::checkpoint`] | a function of the WAL's size, which is a function of how long since the last checkpoint — not of anything the caller passes | It is not a transaction at all. `PRAGMA wal_checkpoint` copies frames back into the main file and there is no unit smaller than the frame it is already working in; the caller asked for exactly this, and the alternative to a long checkpoint is a WAL that keeps growing (0.12.13, W5.2, D-156) |
/// | the drop turn of [`Database::bulk_embeddings`], counted as `drop_embedding_index` (0.16.2, D-276) | µs-scale; one `DROP INDEX IF EXISTS` | One statement, no smaller unit — the same shape as [`Database::checkpoint`] by nature and [`Database::write_bulk_atomic`] by atomicity. Its kind exists for attribution beside `rebuild_embedding_index`, not for cost |
/// | the rebuild turn of [`Database::bulk_embeddings`], counted as `rebuild_embedding_index` (0.16.2, D-276) | measured **2.61 / 19.7 / 39.0 s** for 2,000 vectors at dim 64 / 256 / 512, ~10 ms/vector at dim 256, growing with the corpus | One `CREATE INDEX` over the whole table — the one-pass DiskANN build is indivisible, exactly the criterion `shadow_swap` and `rebuild_current` meet. The difference is schedule: this hold is caller-scheduled and opt-in, so the docstring states the number instead of arguing it. Counted would add a permanent `N(bulk loads)` to every database that ever bulk-embedded — `shadow_swap`'s own argument, unchanged |
/// | the toggle turn of [`Database::bulk_import_deferred`], counted as `links_current_mirror` (0.16.3, D-277) | µs-scale; one DDL statement either direction | Same shape as `drop_embedding_index`: one statement, no smaller unit, and its kind exists for attribution beside the load it wraps. The window's *cost* is the chunked rebuild that follows, which keeps its own counted kinds rather than hiding behind the toggle's exemption — the toggle is not where the time goes |
///
/// The `archive` figure is end-to-end through this method, so it **includes**
/// the re-derivation `archive()` runs inside its transaction — but it does not
/// attribute it, and until D-077 more than half of that re-derivation was an
/// audit comparing `links_current` against the query that had just filled it.
/// Note also which variable that cost scales with: `rebuild_within` reprojects
/// **all of `links`**, so the archive's repair term grows with the *surviving*
/// table and not with the batch being archived. A budget stated per "100K closed
/// intervals" ([§9](../docs/architecture/s6-s10-flows-to-dependencies.md)) is
/// therefore parameterised on the wrong quantity.
///
/// The first four are atomic **by contract**, which is why "cap the batch" and
/// "add a third tier" were both considered and neither was taken: capping breaks the
/// guarantee the operation exists to provide, and a third tier changes which
/// caller waits without changing how long the lock is held. What was wrong was
/// never the exemption — it was that the bound was stated as though it had none.
///
/// A caller who needs the latency bound and not the atomicity has
/// [`Database::bulk_import`], which is the same write chunked at
/// [`chunk_rows::EDGES`] and explicitly *not* atomic overall (D-011).
///
/// # One of them is no longer unbounded (T1.1, D-080)
///
/// `archive` was the worst of them, because its hold is a function of *how long
/// since the last archive* rather than of anything the caller chose.
/// [`Database::archive_windowed`] runs the same work as N sessions, each
/// atomic, each its own actor turn. Measured on an 8,000-key fixture with four
/// generations of superseded history: the longest single hold falls from
/// **3.3 s to 0.77 s** at one-hour windows, for total wall time that is flat
/// within this cycle's noise.
///
/// The same measurement at 2,000 keys goes the other way — the hold falls
/// 260 ms → 117 ms while total time rises 260 ms → 671 ms — so windowing is a
/// trade and not a free improvement. It pays when the backlog is large, which
/// is when the unwindowed hold is a problem in the first place. `archive` is
/// kept, not deprecated, for exactly that reason.
pub const CHUNK_BUDGET: std::time::Duration = std::time::Duration::from_millis(3);

/// Predicted hold above which [`Database::write_bulk_atomic`] warns (T1.3).
///
/// 250 ms is fifteen frames at 60 Hz: not a hitch, a visible freeze. It is well
/// above [`CHUNK_BUDGET`] on purpose — this path is exempt from that bound by
/// contract, so warning at 3 ms would fire on batches that are working exactly
/// as designed and train the reader to filter the message out.
pub const BULK_ATOMIC_WARN_HOLD: std::time::Duration = std::time::Duration::from_millis(250);

/// Roughly how long [`Database::write_bulk_atomic`] will hold the actor for
/// this batch (T1.3, D-081; re-fitted 0.13.6, W7.5, D-179).
///
/// # Two terms, and the batch's shape is no longer one of them
///
/// T1.3 asks for "rows × measured per-row cost". Through 0.13.5 that was wrong
/// in a way worth a paragraph: `write_edges_atomic` opened with a
/// `reject_overlaps_within` that compared **every pair**, and the quadratic
/// term's constant depended on the batch's *shape* rather than its size, so two
/// 20,000-edge batches held the actor for **2.6 s** and **18.1 s** — a size-only
/// model was off by 7× between them, in the under-predicting direction.
///
/// W7.5 sorts and sweeps instead, and the 18.1 s batch now holds for **2.2 s**.
/// The shape term is gone from the code and therefore from here: measured on
/// the same machine, the two shapes are within 15% of each other at every size
/// from 100 to 20,000 rows, which is inside the noise this model claims.
///
/// What is left is not flat either, and the second term is why. Per-row cost
/// rises from ~36 µs at 100 rows to ~111 µs at 20,000, because each insert
/// maintains indexes and two triggers against a table the batch is itself
/// growing:
///
/// ```text
/// hold ≈ rows · (7.4 µs + 7.24 µs · ⌊log₂ rows⌋)
/// ```
///
/// # What this is calibrated against, and where it will be wrong
///
/// libSQL 0.9.30, one machine, best of three, 100–20,000 rows in both shapes;
/// within 15% from 500 rows up. Below that it under-predicts by up to 3×, which
/// is harmless in the same way the old 3× over-prediction was — nothing that
/// small approaches [`BULK_ATOMIC_WARN_HOLD`].
///
/// **The log term reads the batch because the batch is all it has.** It stands
/// for the depth of a structure the batch is loading, and this signature never
/// sees the table. That is exact for the bulk import this warns about, and
/// optimistic for a small batch appended to an already-large table — the same
/// blind spot the flat per-row model had, now visible instead of averaged away.
///
/// It is machine-specific and says nothing about disk. It exists to turn
/// "uncapped" into an order of magnitude a caller can act on, and should not be
/// read more precisely than that. `examples/bulk_atomic_diag.rs` prints
/// predicted against measured, so the model's drift is visible rather than
/// assumed.
pub fn estimated_bulk_hold(edges: &[EdgeAssertion]) -> std::time::Duration {
    let rows = edges.len() as u64;
    if rows == 0 {
        return std::time::Duration::ZERO;
    }

    // Nanoseconds throughout, saturating: a caller who passes a batch large
    // enough to overflow this has a problem the arithmetic cannot express, and
    // saturating to ~584 years still crosses every threshold above.
    let per_row = 7_400u64.saturating_add((rows.ilog2() as u64).saturating_mul(7_240));
    std::time::Duration::from_nanos(rows.saturating_mul(per_row))
}

/// Most sessions [`Database::archive_windowed`] will run for one call (T1.1).
///
/// A limit exists because the session count is a function of *transaction-time
/// span divided by window*, and both come from the caller — a one-second window
/// over a decade of history is ten million actor turns, each opening a
/// transaction and writing a horizon row. That is not a slow archive, it is a
/// caller who meant something else.
///
/// 4,096 is chosen against the operation it bounds rather than against a clock:
/// at the measured 26.8 ms for a session with work in it, a full run of this
/// many is about two minutes of background writing, and the whole point of
/// windowing is that those two minutes are interruptible. It is a refusal
/// rather than a clamp — see [`DbError::ArchiveWindow`] for why.
pub const MAX_ARCHIVE_SESSIONS: usize = 4_096;

/// A concept assertion: the payload of an upsert.
///
/// `#[non_exhaustive]` since 0.14.8 for
/// [`EdgeAssertion`]'s reason: `branch` is the
/// first field added since it was written, and one break is better than a
/// recurring one.
#[derive(Debug, Clone, PartialEq)]
#[non_exhaustive]
pub struct ConceptUpsert {
    pub id: String,
    pub title: String,
    pub content: String,
    pub embedding_model: Option<String>,
    pub valid_from: String,
    pub valid_to: String,
    pub retired: bool,
    /// The lineage this concept is minted on, or `None` for the trunk (§15.2,
    /// D-225).
    ///
    /// **The rule here is narrower than the edge's, and it is the schema's
    /// rather than this crate's.** `concepts` is a current-state projection
    /// keyed by identity — `id` is `NOT NULL UNIQUE` — so two lineages holding
    /// different beliefs about one concept is two rows with one `id`, which the
    /// unique index refuses on its own. `trg_concepts_cross_lineage` turns that
    /// refusal into [`DbError::CrossLineage`] so it says which rule was broken.
    ///
    /// So a branch **inherits** its parent's concepts and cannot restate them;
    /// what this field is for is a concept the branch *mints*, which is the
    /// case the trunk has no row for. A branch that needs to disagree with its
    /// parent about a concept's content is asking for the overlay design, which
    /// is deferred with its reopen trigger named (D-214).
    pub branch: Option<crate::branch::BranchId>,
}

impl ConceptUpsert {
    pub fn new(id: impl Into<String>, title: impl Into<String>) -> Self {
        Self {
            id: id.into(),
            title: title.into(),
            content: String::new(),
            embedding_model: None,
            valid_from: String::new(),
            valid_to: timestamp::OPEN_SENTINEL.to_string(),
            retired: false,
            branch: None,
        }
    }

    pub fn content(mut self, content: impl Into<String>) -> Self {
        self.content = content.into();
        self
    }

    pub fn embedding_model(mut self, model: impl Into<String>) -> Self {
        self.embedding_model = Some(model.into());
        self
    }

    pub fn valid_from(mut self, ts: impl Into<String>) -> Self {
        self.valid_from = ts.into();
        self
    }

    pub fn valid_to(mut self, ts: impl Into<String>) -> Self {
        self.valid_to = ts.into();
        self
    }

    /// Mint this concept on `branch` rather than on the trunk (0.14.8).
    ///
    /// See [`branch`](Self::branch) for why a branch may mint a concept and may
    /// not restate one it inherited.
    pub fn on_branch(mut self, branch: crate::branch::BranchId) -> Self {
        self.branch = Some(branch);
        self
    }

    /// The lineage this upsert names, spelled out. See
    /// [`EdgeAssertion::branch_name`](crate::graph::EdgeAssertion).
    pub(crate) fn branch_name(&self) -> &str {
        self.branch
            .as_ref()
            .map_or(crate::schema::ddl::MAIN_BRANCH, |b| b.as_str())
    }

    pub fn retired(mut self, retired: bool) -> Self {
        self.retired = retired;
        self
    }

    /// Put the timestamps in canonical form (D-029) before they cross the channel.
    pub fn normalized(mut self) -> Result<Self> {
        crate::util::ids::validate_id(&self.id)?;
        self.valid_from = timestamp::normalize(&self.valid_from)?;
        self.valid_to = timestamp::normalize(&self.valid_to)?;
        Ok(self)
    }
}

/// One derived analytics result for one concept (§5.4, D-041).
///
/// Not a `ConceptUpsert`. The distinction is the whole of D-041: a concept
/// upsert is a statement about the world and belongs in the ledger, while an
/// annotation is a function of an algorithm applied to a graph and belongs in
/// `analytics_annotations`, which carries no log trigger. Writing one as the
/// other overwrote the concept's `content` with the label and recorded every
/// analytics rerun as a fresh version of the world.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub struct Annotation {
    pub concept_id: String,
    /// Namespaced by convention, e.g. `louvain.community`, `kcore.shell`.
    pub label: String,
    /// JSON-encoded payload. Opaque to this crate.
    pub value: String,
}

impl Annotation {
    pub fn new(
        concept_id: impl Into<String>,
        label: impl Into<String>,
        value: impl Into<String>,
    ) -> Self {
        Self {
            concept_id: concept_id.into(),
            label: label.into(),
            value: value.into(),
        }
    }
}

/// Commands sent to the Write Actor on the high-priority channel (UI-driven work).
pub(crate) enum HighPriCommand {
    AssertEdge {
        edge: EdgeAssertion,
        responder: oneshot::Sender<Result<()>>,
    },
    RetireEdge {
        source: String,
        target: String,
        edge_type: String,
        valid_from: String,
        valid_to: String,
        /// The lineage doing the retiring, or `None` for the trunk (0.14.8).
        branch: Option<crate::branch::BranchId>,
        responder: oneshot::Sender<Result<()>>,
    },
    UpsertConcept {
        concept: ConceptUpsert,
        responder: oneshot::Sender<Result<()>>,
    },
    WriteBulkAtomic {
        edges: Vec<EdgeAssertion>,
        responder: oneshot::Sender<Result<usize>>,
    },
    RebuildCurrent {
        responder: oneshot::Sender<Result<RebuildReport>>,
    },
    /// Create a model's embedding table and its DiskANN index (D-037, D-048).
    ///
    /// High priority despite being setup work: it is one small transaction, and
    /// every embedding write for the model blocks on it, so queueing it behind a
    /// bulk job would stall the thing it gates.
    RegisterModel {
        model: ModelName,
        dim: usize,
        responder: oneshot::Sender<Result<()>>,
    },
    /// Move WAL frames back into the main database file (§4.5, F-30, D-156).
    ///
    /// High priority, and for once the reason is not latency: a caller asking
    /// for a checkpoint is asking for it *now*, usually at the end of a bulk
    /// load or before taking a copy of the file, and queueing it behind the
    /// background work it was meant to follow inverts the intent. It is also
    /// the only command here that is not a transaction.
    Checkpoint {
        responder: oneshot::Sender<Result<CheckpointReport>>,
    },
    /// Register a lineage (0.14.7, §15.4).
    ///
    /// High priority, and not because it is urgent: it is one insert into a
    /// table with no secondary indices, so it is the cheapest turn the actor
    /// takes. What makes it high priority is that everything the caller does
    /// next is a write *on* this branch, and queueing a fork behind a bulk
    /// import would stall the work it exists to enable — `RegisterModel`'s
    /// argument, for the same reason.
    ///
    /// It goes through the actor rather than the read connection for the
    /// ordinary reason every write does, plus one specific to it: the duplicate
    /// and parent checks are only sound if nothing can register a colliding
    /// name between the check and the insert, and the actor is what makes the
    /// pair one turn.
    Fork {
        name: crate::branch::BranchId,
        parent: crate::branch::BranchId,
        responder: oneshot::Sender<Result<crate::branch::Branch>>,
    },
    Shutdown {
        responder: oneshot::Sender<Result<()>>,
    },
}

/// What `PRAGMA wal_checkpoint` returned (0.12.13, W5.2, D-156).
///
/// The three columns SQLite gives back, named, rather than `()` — a checkpoint
/// that did nothing and a checkpoint that reclaimed a 400 MB WAL are the same
/// `Ok(())`, and the difference is the entire reason a caller asked.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub struct CheckpointReport {
    /// `true` when SQLite could not complete the requested mode because a
    /// reader or writer was in the way.
    ///
    /// **This is not an error, and it is not ignorable.** `TRUNCATE` waits for
    /// readers only as long as `busy_timeout` allows; past that it gives up and
    /// says so, having possibly still copied frames. A caller checkpointing
    /// before copying the file away must read this, because a busy checkpoint
    /// means the main file is not self-contained yet.
    pub busy: bool,
    /// Frames left in the WAL at the end. `0` when the checkpoint completed,
    /// since the mode run is `TRUNCATE`.
    pub log_frames: u64,
    /// Frames moved back into the database file.
    ///
    /// Read from a `FULL` pass rather than from the `TRUNCATE` — see
    /// `run_checkpoint` for why a truncating checkpoint cannot report this
    /// number itself.
    pub checkpointed_frames: u64,
}

impl CheckpointReport {
    /// The WAL was fully reclaimed: nothing blocked, and nothing is left.
    pub fn is_complete(&self) -> bool {
        !self.busy && self.log_frames == 0
    }
}

/// Commands sent to the Write Actor on the low-priority channel (background work).
pub(crate) enum LowPriCommand {
    /// One chunk of **concepts** — a ledger write, logged and versioned.
    WriteConceptsChunk {
        chunk: Vec<ConceptUpsert>,
        responder: oneshot::Sender<Result<ChunkOutcome>>,
    },
    /// One chunk of **derived annotations** — off-ledger, no log trigger (D-041).
    ///
    /// The pair is named apart deliberately: this variant was `WriteAnalyticsChunk`
    /// beside a `WriteAnnotationsChunk` that carried concepts, which is the
    /// crossing D-075 undid.
    WriteAnalyticsChunk {
        chunk: Vec<Annotation>,
        responder: oneshot::Sender<Result<ChunkOutcome>>,
    },
    /// One chunk of vectors for one model (§5.9, D-048).
    ///
    /// Low priority: embedding is bulk derived work and must never preempt an
    /// interactive assertion.
    UpsertEmbeddingChunk {
        model: ModelName,
        chunk: Vec<(String, Vec<f32>)>,
        responder: oneshot::Sender<Result<ChunkOutcome>>,
    },
    /// Drop a model's DiskANN index — bulk-embedding setup (D-276).
    ///
    /// Low priority like the chunk it serves: it is a caller's bulk load that
    /// wants the index gone, and nothing interactive should queue behind it.
    /// One `DROP INDEX IF EXISTS` statement, atomic by nature.
    DropEmbeddingIndex {
        model: ModelName,
        responder: oneshot::Sender<Result<()>>,
    },
    /// One-pass rebuild of a model's DiskANN index — bulk-embedding finish
    /// (D-276).
    ///
    /// One `CREATE INDEX` statement, and the indivisible half of the recipe:
    /// measured at 19.7 s for 2,000 vectors at dim 256 on the reference box.
    /// Budget-exempt by the same criterion as `ShadowSwap` — atomic by
    /// necessity — with the difference that this one is *caller-scheduled*, so
    /// a slow build is a choice the caller made knowingly.
    RebuildEmbeddingIndex {
        model: ModelName,
        responder: oneshot::Sender<Result<()>>,
    },
    BulkImportChunk {
        chunk: Vec<EdgeAssertion>,
        responder: oneshot::Sender<Result<ChunkOutcome>>,
    },
    /// Drop or restore `trg_links_current_sync` — the links_current mirror's
    /// window, opened and closed by [`Database::bulk_import_deferred`]
    /// (D-277).
    ///
    /// Low priority beside the chunks it serves. One DDL statement either way
    /// (`DROP TRIGGER IF EXISTS` down, `CREATE TRIGGER IF NOT EXISTS` up — the
    /// same idempotent pair the v19 rung used on `trg_links_single_open`), so
    /// the toggle is safe to send twice and the restore cannot fail on an
    /// already-present trigger.
    LinksCurrentMirror {
        present: bool,
        responder: oneshot::Sender<Result<()>>,
    },
    Archive {
        cutoff: String,
        archive_path: PathBuf,
        responder: oneshot::Sender<Result<ArchiveReport>>,
    },
    /// Forget one lineage, moving its whole ledger to the cold file (0.14.13,
    /// §15.4, D-230).
    ///
    /// Low priority for `Archive`'s reason and one of its own: it is bulk
    /// physical movement holding the write lock for its whole transaction, and
    /// it is the least urgent write in the crate — the rows it moves belong to
    /// a lineage nobody is reading.
    ArchiveBranch {
        branch: String,
        archive_path: PathBuf,
        responder: oneshot::Sender<Result<ArchiveReport>>,
    },
    /// Move named concepts back out of the cold file (0.9.0, C3).
    ///
    /// Low priority for the same reason `Archive` is: it is bulk physical
    /// movement with no latency bound, and it holds the write lock for its whole
    /// transaction.
    Rehydrate {
        ids: Vec<String>,
        archive_path: PathBuf,
        responder: oneshot::Sender<Result<RehydrateReport>>,
    },
    /// Reconstruct the FTS index from `concepts` (§5.9, D-036, D-051).
    ///
    /// Low priority: it is maintenance on a derivative table, and a search index
    /// that is a few seconds stale is a smaller cost than an interactive write
    /// that waits behind a full reindex.
    RebuildFts {
        responder: oneshot::Sender<Result<()>>,
    },
    /// Refresh or top up the query planner's statistics (0.12.4, D-149).
    ///
    /// Low priority, and not a close call: statistics being a few seconds stale
    /// costs a plan that was already the plan a moment ago, where preempting an
    /// interactive assertion costs a caller their latency bound. It is a write —
    /// it writes `sqlite_stat1` — so it takes the write lock like anything else,
    /// and `PRAGMA analysis_limit` in `configure` is what keeps the hold a
    /// function of the index count instead of the table size.
    Analyze {
        /// `true` runs `PRAGMA optimize`, which re-analyses only what SQLite
        /// believes has gone stale; `false` runs `ANALYZE` unconditionally.
        incremental: bool,
        responder: oneshot::Sender<Result<()>>,
    },
    /// One step of a chunked shadow rebuild (§5.8, T1.2, D-082).
    ///
    /// Low priority, and one command per step rather than one per rebuild: the
    /// whole value of building beside the live table is that the actor returns
    /// here between chunks. See [`Database::rebuild_current_chunked`].
    ShadowRebuild {
        step: crate::integrity::ShadowStep,
        responder: oneshot::Sender<Result<crate::integrity::ShadowOutcome>>,
    },
}

/// What the write actor knows between turns (0.15.6, W14.3, [D-248]).
///
/// [`run_writer_actor`] owned a connection and nothing else. Every command was
/// handed `&conn`, and every fact a command established died with it — so a
/// single-edge assertion asked `branches` how many lineages exist, compiled the
/// overlap guard, and compiled `INSERT_LINK`, on every call, having done all
/// three on the previous one. Measured on the trunk that is 76 µs of a 160 µs
/// write; once the database has forked it is 155 µs of a 343 µs write, because
/// the statement being compiled each time is the guard's resolved form.
///
/// **Everything here is cached for one reason and invalidated by name for the
/// same one: the actor is the only writer** (D-014). `branches` is written by
/// `Fork` and by `ArchiveBranch` and by nothing else in the crate; the
/// statements are bound to a connection this task owns for the process
/// lifetime. A cache whose only writer is holding it cannot go stale behind its
/// own back, which is why this is a plain `&mut` and not an epoch or a lock.
///
/// Built lazily rather than at open. Eager construction pays on a database that
/// never asserts an edge, and it puts fallible work in the spawn path, where
/// there is no caller to hand the error to.
///
/// # What is deliberately not here
///
/// The **hot-log intactness verdict** (review C-5). It is read on `read_conn`
/// by every recorded-time read, not by the actor, so caching it here would put
/// it on the wrong side of the process. It needs a shared cell and an
/// invalidation argument about a *reader* seeing a stale answer, which is a
/// different argument from this one and gets its own release.
///
/// The **batch paths' statements**. `write_edges_atomic` prepares inside its
/// own transaction and drops before it commits, because a live statement is
/// what makes SQLite refuse to end one. It already prepares once per chunk
/// rather than once per row (D-056, §8.8), which is where that path's cost was.
///
/// [D-248]: ../../docs/architecture/s13-decision-register.md#d-248
struct ActorState {
    /// `branches`, as this actor last left it.
    lineages: Option<Lineages>,
    /// `INSERT_LINK`, compiled against the actor's connection.
    insert_link: Option<libsql::Statement>,
    /// The overlap guard, with the shape it was compiled for.
    guard: Option<OverlapGuard>,
}

// `Lineages` moved to `graph::lineage` in 0.15.17 ([D-259]). It held
// `(branch_id, is_root)` here, because the shape was all the actor needed;
// resolving ancestry in Rust needs `parent_id` and `forked_at` as well, on
// both sides of the crate, and two structs answering one question from one
// table is what D-030 is about.
//
// [D-259]: ../docs/architecture/s13-decision-register.md#d-259

impl ActorState {
    fn new() -> Self {
        Self {
            lineages: None,
            insert_link: None,
            guard: None,
        }
    }

    /// Forget `branches`. Called by the two commands that write it.
    fn forget_lineages(&mut self) {
        self.lineages = None;
    }

    /// Drop the compiled statements.
    ///
    /// Called by the commands that `ATTACH`, `DETACH`, or otherwise move the
    /// schema under the connection. SQLite recompiles a statement across a
    /// schema change on its own and this does not rely on that: a statement
    /// dropped here costs one prepare on the next write, against a class of bug
    /// whose symptom would be a stale plan on the archive path in production.
    fn forget_statements(&mut self) {
        self.insert_link = None;
        self.guard = None;
    }

    /// Both of the above, for a command that does both.
    fn forget_everything(&mut self) {
        self.forget_lineages();
        self.forget_statements();
    }

    async fn lineages(&mut self, conn: &libsql::Connection) -> Result<&Lineages> {
        if self.lineages.is_none() {
            self.lineages = Some(Lineages::load(conn).await?);
        }
        Ok(self
            .lineages
            .as_ref()
            .expect("loaded on the line above or already present"))
    }

    /// [`Lineages::shape_of`], against the cache.
    ///
    /// This is what replaced `check_lineages`, and the round trip it replaced
    /// is the one the review counted (C-6): one `SELECT` over `branches` per
    /// write, for an answer that changes when a lineage is forked or forgotten.
    async fn shape_of(
        &mut self,
        conn: &libsql::Connection,
        names: &[&str],
    ) -> Result<LineageShape> {
        self.lineages(conn).await?.shape_of(names)
    }

    /// The overlap guard, compiled at most once per shape.
    ///
    /// Keyed on the shape rather than on the statement text because that is the
    /// thing [`check_prepared`] reads: it binds four parameters for `Trunk` and
    /// five otherwise, so a guard held under one shape and used under another
    /// would bind the wrong row even where the SQL happened to match.
    async fn guard(
        &mut self,
        conn: &libsql::Connection,
        shape: LineageShape,
        branch: &str,
    ) -> Result<&OverlapGuard> {
        // Keyed on the lineage as well as the shape since 0.15.17: the bound
        // ancestry is one reader's answer, so a guard held for `main` cannot
        // serve a branch that happens to share its shape. A caller writing to
        // one lineage — which is every caller this crate has — still prepares
        // once and keeps it across turns.
        if !self
            .guard
            .as_ref()
            .is_some_and(|g| g.answers_for(shape, branch))
        {
            // Resolved against the actor's own cache, then dropped, so the
            // borrow ends before the assignment. `lineages` refreshes it when
            // `Fork` or `ArchiveBranch` forgot it.
            let lineages = self.lineages(conn).await?.clone();
            self.guard = Some(OverlapGuard::prepare(conn, shape, &lineages, branch).await?);
        }
        Ok(self
            .guard
            .as_ref()
            .expect("prepared on the line above or already present"))
    }

    /// `INSERT_LINK`, compiled once.
    ///
    /// The single-edge path ran `conn.execute(INSERT_LINK, …)`, which compiles
    /// the statement on every call — 61 µs of it, because `links` carries the
    /// projection triggers and they are compiled with the insert. This is D-056
    /// and D-057's lesson, which was learned on the batch path and never
    /// carried across to the path a caller actually waits on.
    async fn insert_link(&mut self, conn: &libsql::Connection) -> Result<&libsql::Statement> {
        if self.insert_link.is_none() {
            self.insert_link = Some(conn.prepare(INSERT_LINK).await?);
        }
        Ok(self
            .insert_link
            .as_ref()
            .expect("prepared on the line above or already present"))
    }
}

enum LoopCtl {
    Continue,
    Break,
}

/// Primary database handle for Macrame bitemporal ledger.
///
/// # Why this is not `Clone`, and what a multi-consumer caller uses instead
///
/// **Share it as `Arc<Database>`.** Every method here but one takes `&self`, so
/// an `Arc` is a complete handle and not a workaround: reads run concurrently
/// off `read_conn`, writes queue behind the actor's channel exactly as they do
/// through a `&Database`, and nothing becomes serialised that was not
/// serialised already. The exception is [`Database::close`], which takes `self`,
/// so the last owner closes with
/// `Arc::into_inner(db).expect("last handle").close().await`.
///
/// That exception is the whole reason `Clone` is absent. Cloning would have to
/// duplicate **the right to shut down**, and each field carrying that right
/// breaks differently when duplicated:
///
/// - `writer` is a [`tokio::task::JoinHandle`], which is not `Clone` at all —
///   so a hand-written impl would have to give the copy a `None`, and
///   `close()` on that copy returns `Ok(())` without ever checking the actor's
///   exit status. That status is one of the two reasons [`Drop`] tells callers
///   to prefer `close()`.
/// - `cadence_stop` is a [`tokio::sync::watch::Sender`], which **is** `Clone`,
///   and that is the worse case. Its contract is that *dropping* it stops the
///   snapshot task; a watch channel closes when the last sender goes, so one
///   surviving copy keeps that task running against a database that is going
///   away. Nothing returns an error, which is why this is the argument rather
///   than the `JoinHandle`.
/// - `closed` is per-handle, so two copies disagree about whether the ledger
///   was closed: `Drop` warns about a database that *was* closed, or stays
///   silent about one that was not.
///
/// And the ordering `close()` documents — cadence stopped, actor joined, *then*
/// the final snapshot, so that no write can land between the fold and the file
/// — is only enforceable while one handle can perform it. A second `close()`
/// writes a "final" snapshot with the actor still alive.
///
/// So the missing impl is the type saying shutdown has exactly one owner. The
/// Python binding reached the same shape from the other side and for the same
/// reason: `PyDatabase` holds a `RwLock<Option<Database>>` rather than a copy
/// per caller (0.13.30, W11.1, D-203).
pub struct Database {
    db: libsql::Database,
    /// The file this handle opened, kept so [`Database::diagnostic_conn`] can
    /// open it again under different flags (T5.1, D-091). `archive_path` and
    /// `snapshots_dir` are derived from it and were previously the only trace
    /// of it on the struct.
    path: PathBuf,
    read_conn: libsql::Connection,
    highpri_tx: mpsc::Sender<HighPriCommand>,
    lowpri_tx: mpsc::Sender<LowPriCommand>,
    clock: Arc<dyn Clock>,
    archive_path: PathBuf,
    snapshots_dir: PathBuf,
    schema_version: u32,
    /// Kept so [`Database::diagnostic_conn`] can configure the connections it
    /// mints the same way `open()` configured the internal readers (0.12.16,
    /// W5.5, D-159). Before that split, each one ran with SQLite's defaults.
    reader_cache_size: Option<i32>,
    /// The `SQLITE_OPEN_READ_ONLY` connection behind
    /// [`Database::diagnostic_conn`], opened on first use and dropped with this
    /// handle (0.15.14, W15.4, review C-9, [D-256]).
    ///
    /// **The connection, not the `libsql::Database` handle**, and the
    /// difference is the measurement rather than a preference. The first shape
    /// written here cached the handle and minted a connection per call, on the
    /// argument that `diagnostic_conn` promises a connection the *caller* owns.
    /// `examples/diagnostic_conn_probe.rs` says `Builder::…build()` costs
    /// **0.10 µs and opens nothing** — it succeeds against a path that does not
    /// exist — while `connect()` costs **51.5 µs** and is where
    /// `SQLITE_CANTOPEN` arrives for a missing file. The handle cache removes a
    /// call that does no work.
    ///
    /// **`Mutex<Option<_>>` rather than the `OnceCell` 0.15.14 shipped**
    /// (0.15.15, W15.5, [D-257]). A `OnceCell` can be filled and never
    /// emptied, and this connection is handed to arbitrary SQL, so it acquires
    /// state that the *next* caller must not inherit — a leaked `BEGIN` above
    /// all, which pins a WAL read snapshot and makes both this surface's reads
    /// stale and [`Database::checkpoint`] a no-op. The slot has to be
    /// clearable for the dirty ones to be replaced, and clearable is what a
    /// `OnceCell` is not. `tokio`'s mutex rather than `std`'s because it is
    /// held across the open.
    ///
    /// A failed first attempt leaves the slot empty, so a database whose file
    /// appears later is not poisoned by the call that came too early — the
    /// property the `OnceCell` had, kept.
    ///
    /// [D-256]: ../../docs/architecture/s13-decision-register.md#d-256
    /// [D-257]: ../../docs/architecture/s13-decision-register.md#d-257
    diagnostic_conn: tokio::sync::Mutex<Option<libsql::Connection>>,
    writer: Option<tokio::task::JoinHandle<()>>,
    /// Stops the snapshot cadence. Dropping it stops the task too, which is what
    /// keeps a `Database` that is dropped rather than closed from leaving a task
    /// running against a connection whose database is going away.
    cadence_stop: Option<tokio::sync::watch::Sender<bool>>,
    cadence: Option<tokio::task::JoinHandle<()>>,
    /// Set by [`Database::close`]. Read only by [`Drop`], which warns when it is
    /// still false — see that impl for why the omission is worth a warning.
    closed: bool,
    /// Shared with the actor (T1.4, T1.2). Held here rather than behind
    /// `#[cfg(feature = "metrics")]` so `open_inner` has one shape; with the
    /// feature off the metrics half is a zero-sized type and only
    /// [`Database::metrics`] is gated — which is also why the field is unread in
    /// the default build: the actor holds the other `Arc` and does the writing.
    #[cfg_attr(not(feature = "metrics"), allow(dead_code))]
    shared: Arc<ActorShared>,
}

/// What the snapshot cadence should do, for [`Tuning::cadence`].
///
/// # Why this is not `Option<SnapshotCadence>`
///
/// [`Database::open_with_cadence`] takes `Option<SnapshotCadence>`, where `None`
/// means *no cadence at all*. Carrying that field into [`Tuning`] unchanged
/// would have made it the one field in the struct whose `None` is a request to
/// change the behaviour rather than a request to leave it alone — and since
/// `Tuning` derives `Default`, `open_tuned(path, Tuning::default())` would then
/// have silently disabled snapshots, while `open(path)` runs them. Two calls
/// that read as synonyms, one of which stops writing anchors.
///
/// So the tri-state is written out. `Default` is the default cadence, matching
/// [`Database::open`]; `Disabled` is `open_with_cadence(path, None)`, and has to
/// be asked for by name.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
#[non_exhaustive]
pub enum CadencePolicy {
    /// [`SnapshotCadence::default`], as [`Database::open`] uses.
    #[default]
    Default,
    /// No cadence task. `close()` is then the only thing that writes an anchor
    /// (§5.5, D-053).
    Disabled,
    /// An explicit cadence.
    Every(SnapshotCadence),
}

impl CadencePolicy {
    /// Collapse to the `Option` the open path has always taken.
    fn resolve(self) -> Option<SnapshotCadence> {
        match self {
            Self::Default => Some(SnapshotCadence::default()),
            Self::Disabled => None,
            Self::Every(cadence) => Some(cadence),
        }
    }
}

/// When SQLite should checkpoint the WAL on its own, for
/// [`Tuning::wal_autocheckpoint`] (0.12.14, W5.3, D-157).
///
/// # Why this is not `Option<u32>`
///
/// The same reason [`CadencePolicy`] is not `Option<SnapshotCadence>`, and the
/// plan for this wave specified `Option<u32>` here too. In a struct that derives
/// `Default`, a field whose `None` means *turn the mechanism off* is a field
/// that turns the mechanism off for everyone who did not mention it. Absence
/// means "leave it alone" everywhere in [`Tuning`], and disabling the automatic
/// checkpointer — which is not safe without an explicit
/// [`Database::checkpoint`] to replace it — has to be asked for by name.
///
/// **The default does not change.** 1,000 pages is SQLite's default and stays
/// SQLite's default; F-30 is a control-loop perturbation, not a correctness bug,
/// and changing a default is a behaviour change for every existing caller.
///
/// # What disabling it actually buys, measured (0.12.14, W5.3, D-157)
///
/// F-30 says the automatic checkpointer is an unbudgeted hold *inside* 0.12.0's
/// adaptive chunk controller: a checkpoint firing during a chunk transaction is
/// charged to that chunk, and since D-146 made the measured hold the input to
/// `next_chunk_size`, the controller shrinks in response to work the chunk did
/// not do. Three rounds, 6,000 concepts of 1 KB each through `write_concepts`,
/// release build:
///
/// | | longest chunk hold | mean | chunks | over budget | wall |
/// |---|---|---|---|---|---|
/// | autocheckpoint on (default) | **9.3–10.3 ms** | 2.40–2.44 ms | 125–130 | 24–28 | 304–321 ms |
/// | autocheckpoint off | **4.50 ms** | 2.08–2.20 ms | 142–153 | 18–27 | 298–339 ms |
///
/// **The tail is the finding, and it is real and reproducible.** The longest
/// hold roughly halves, and the >10 ms histogram bucket is populated only with
/// the checkpointer on — that bucket is the checkpoint, landing inside somebody
/// else's transaction and being charged to it. Every round agrees.
///
/// **What it does not buy is a calmer controller.** `over_budget` overlaps
/// between the arms, and total wall time is the same within noise. The
/// controller works near the budget boundary either way, because
/// [D-090](../docs/architecture/s13-decision-register.md)'s ~0.8 ms
/// per-transaction floor and the convergence cost do not go anywhere. So the
/// honest statement is that disabling autocheckpoint removes an outlier, not an
/// oscillation.
///
/// **And the cost is deferred, not removed.** The explicit
/// [`Database::checkpoint`] at the end of the same fixture moved **8,400–9,100
/// frames in 41–45 ms** with the checkpointer off, against **~860 frames in
/// 5.5–6.2 ms** with it on. That is the whole trade in one line: the same work,
/// moved out of the latency-bounded path and into one hold the caller chose the
/// moment for. It is a good trade for a bulk importer and a bad one for an
/// interactive process, which is why this is a knob and not a new default.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
#[non_exhaustive]
pub enum WalCheckpointPolicy {
    /// SQLite's own default: checkpoint once the WAL passes 1,000 pages.
    #[default]
    Default,
    /// No automatic checkpointing.
    ///
    /// **Only correct if you call [`Database::checkpoint`] yourself.** Without
    /// one, the WAL grows for the life of the process and the database file is
    /// never brought up to date.
    Disabled,
    /// Checkpoint once the WAL passes this many pages.
    ///
    /// `0` is not special-cased to [`Self::Disabled`] even though SQLite treats
    /// it that way, because a caller who computed a threshold and got zero has
    /// a bug, and inheriting SQLite's overload would turn it into a silently
    /// unbounded WAL.
    EveryPages(u32),
}

impl WalCheckpointPolicy {
    /// The pragma to run, or `None` to leave the connection at SQLite's
    /// default.
    fn pragma(self) -> Option<String> {
        match self {
            Self::Default => None,
            Self::Disabled => Some("PRAGMA wal_autocheckpoint = 0".to_string()),
            Self::EveryPages(pages) => Some(format!("PRAGMA wal_autocheckpoint = {pages}")),
        }
    }
}

/// Everything [`Database::open_tuned`] can be told, in one growable struct
/// (0.12.12, W5.1, D-155).
///
/// # Why a struct rather than a fourth constructor
///
/// There were three — [`Database::open`], [`Database::open_with_cadence`],
/// [`Database::open_with_clock`] — and each new knob added one more, with the
/// combinatorics of the ones before it. 0.13.0 alone wanted three knobs
/// (`wal_autocheckpoint`, and a page cache each for the writer and the
/// readers), which is the point at which the naming stops being possible.
///
/// **Setters plus `#[non_exhaustive]` are the whole design** (0.15.13, W15.3,
/// [C-11], [D-255]). They make a new knob an additive change: callers write
/// `Tuning::default().cadence(..)`, and the fields that arrive after them are
/// the ones they did not ask about. That is not a hypothetical — W5.1 shipped
/// this struct with two fields, and W5.3/W5.4 added three more without
/// touching a caller.
///
/// # Why the attribute needed the setters, and why 0.5.1's answer was the
/// other one
///
/// D-155 specified `#[non_exhaustive]` for this struct and then could not
/// ship it, on a fact that is still true: a `#[non_exhaustive]` **struct**
/// cannot be built with literal syntax outside its own crate *at all*, and
/// the functional-update form is literal syntax, so
/// `Tuning { cadence, ..Default::default() }` is `E0639` for every external
/// caller — the exact expression the attribute was there to protect. (The rule
/// differs from `#[non_exhaustive]` on an enum, which only forces a wildcard
/// arm; [`CadencePolicy`] has kept it for that reason since W4.2.) D-155 named
/// the two ways to have both — *a builder with setters, or plain `Default`* —
/// and chose `Default`, because the field-literal form is the legible one.
///
/// **What changed is not the argument but the deadline.** `Default` alone
/// leaves the growth additive only for callers who wrote
/// `..Default::default()`; a caller who wrote the exhaustive literal breaks on
/// the next field. Before 1.0 that is a compile error with an obvious fix.
/// After it, it is a major version — and this struct is the one in the crate
/// whose whole documented purpose is to keep acquiring fields. So the release
/// that is still allowed to break callers pays D-155's other price and writes
/// the setters, which is the half of its own analysis it declined at the time.
///
/// **The fields stay `pub` and stay readable**, and on a value you own they
/// stay assignable: `let mut t = Tuning::default(); t.cadence = ..;` compiles
/// outside this crate exactly as it did. What the attribute forbids is the
/// *literal*, which is the one form that enumerates every field and therefore
/// the one form a new field can break.
///
/// [C-11]: ../../docs/Macrame%20Update%20Plan%20v0.16.0.md
/// [D-255]: ../../docs/architecture/s13-decision-register.md#d-255
///
/// # The three constructors stay
///
/// They delegate here and are not deprecated. `open(path)` is the right call for
/// most callers and should not acquire a warning for being the common case; the
/// consolidation is about where the *next* knob goes, not about moving anyone.
///
/// ```no_run
/// # use macrame::prelude::*;
/// # async fn f() -> macrame::Result<()> {
/// let db = Database::open_tuned(
///     "graph.db",
///     Tuning::default().cadence(CadencePolicy::Disabled),
/// )
/// .await?;
/// # Ok(()) }
/// ```
#[derive(Clone, Default)]
#[non_exhaustive]
pub struct Tuning {
    /// What the snapshot cadence should do. Defaults to
    /// [`SnapshotCadence::default`], as [`Database::open`] does.
    pub cadence: CadencePolicy,
    /// A clock to stamp `recorded_at` with, for tests (§5.1.2, D-062). `None`
    /// is [`SystemClock`]. Floored against the database exactly as
    /// [`Database::open_with_clock`] describes — read that before injecting
    /// one against a non-empty file.
    pub clock: Option<Arc<dyn Clock>>,
    /// When SQLite checkpoints the WAL on its own (0.12.14, W5.3, F-30).
    ///
    /// Applied to the **write connection**, which is the only connection in
    /// this crate that commits, and therefore the only one whose autocheckpoint
    /// setting can ever fire. Pair [`WalCheckpointPolicy::Disabled`] with an
    /// explicit [`Database::checkpoint`] or the WAL grows without bound.
    pub wal_autocheckpoint: WalCheckpointPolicy,
    /// Page cache for the **write** connection, as SQLite's `cache_size`
    /// (0.12.15, W5.4).
    ///
    /// `None` leaves SQLite's default of −2000, which is −2000 *kibibytes*, or
    /// 2 MB. **Negative values are KiB and positive values are pages** — that
    /// is SQLite's convention and it is preserved rather than smoothed over,
    /// because a caller who knows the pragma should not have to discover that
    /// this crate redefined it. `Some(-64_000)` is 64 MB; `Some(64_000)` is
    /// 64,000 pages, which at the 4 KiB page size this crate gets is 256 MB.
    ///
    /// The writer wants a large cache: it is one connection, it holds the write
    /// lock while it works, and every page it has to re-read from disk is time
    /// no other writer can use.
    ///
    /// # Unlike the two above, `None` here is not a policy enum
    ///
    /// Because SQLite's default is a *value* rather than a mechanism. Absence
    /// still means "leave it alone" — it just happens that leaving this alone
    /// is expressible as not running a pragma, where leaving the automatic
    /// checkpointer alone required saying which of two things "alone" meant.
    pub writer_cache_size: Option<i32>,
    /// Page cache for every **read-only** connection: the shared
    /// [`Database::read_conn`], the snapshot cadence's own connection, and
    /// (since W5.5) each [`Database::diagnostic_conn`] (0.12.15, W5.4).
    ///
    /// Same units as [`Self::writer_cache_size`], and the same `None`.
    ///
    /// Split from the writer's because the profiles are opposite and one number
    /// cannot serve both. There is exactly one writer and it is long-lived, so
    /// its cache is a fixed cost paid once. Read-only connections are plural —
    /// the shared reader and the cadence's — so a large value here is
    /// multiplied by however many exist, which is the wrong size for the one
    /// connection that holds the write lock.
    ///
    /// **The multiplier used to be unbounded** and is not since 0.15.14
    /// (W15.4, [D-256]): `diagnostic_conn` minted a connection per call, so a
    /// caller in a loop multiplied this number by their own call count. There
    /// is one such connection per `Database` now, so the count is three.
    ///
    /// [D-256]: ../../docs/architecture/s13-decision-register.md#d-256
    pub reader_cache_size: Option<i32>,
    /// What to do about a stored `recorded_at` in the future (0.13.5, W7.4,
    /// §3.4).
    ///
    /// The clock floors itself at `MAX(recorded_at)` so stamps stay strictly
    /// increasing across restarts, which means one row from the future becomes
    /// this process's floor and every stamp it issues inherits it — into rows
    /// the next open reads back. Defaults to refusing beyond
    /// [`crate::DEFAULT_FUTURE_STAMP_TOLERANCE`], a day.
    ///
    /// Like [`Self::wal_autocheckpoint`] and unlike the two cache sizes, this
    /// is a policy enum rather than an `Option`, for
    /// [D-155](../../docs/architecture/s13-decision-register.md)'s reason: it
    /// guards an invariant, and a `None` that switches it off would switch it
    /// off for every caller who never heard of it.
    pub future_stamps: FutureStampPolicy,
}

impl Tuning {
    // The setters below are what make `#[non_exhaustive]` payable (0.15.13,
    // W15.3, D-255). One per field, named after it, taking `self` — so
    // `Tuning::default().cadence(x).writer_cache_size(y)` is an expression, and
    // a field added later is a method added later rather than a break. They are
    // deliberately not clever: no `Into`, no grouping of two knobs under one
    // name, nothing that would have to be redesigned the first time a field
    // does not fit the pattern.

    /// What the snapshot cadence should do — the [`cadence`](Self::cadence)
    /// field.
    pub fn cadence(mut self, cadence: CadencePolicy) -> Self {
        self.cadence = cadence;
        self
    }

    /// Inject a clock — the [`clock`](Self::clock) field.
    ///
    /// Takes the clock rather than an `Option`, because `None` is what
    /// [`Tuning::default`] already holds and a setter whose argument can undo
    /// itself invites `clock(None)` as a way of saying nothing.
    /// [`Database::open_with_clock`] documents the flooring this is subject to;
    /// read it before injecting one against a non-empty file.
    pub fn clock(mut self, clock: Arc<dyn Clock>) -> Self {
        self.clock = Some(clock);
        self
    }

    /// When SQLite checkpoints the WAL on its own — the
    /// [`wal_autocheckpoint`](Self::wal_autocheckpoint) field.
    pub fn wal_autocheckpoint(mut self, policy: WalCheckpointPolicy) -> Self {
        self.wal_autocheckpoint = policy;
        self
    }

    /// Page cache for the write connection, in SQLite's units — the
    /// [`writer_cache_size`](Self::writer_cache_size) field, which documents
    /// why negative means KiB and positive means pages.
    pub fn writer_cache_size(mut self, size: i32) -> Self {
        self.writer_cache_size = Some(size);
        self
    }

    /// Page cache for every read-only connection — the
    /// [`reader_cache_size`](Self::reader_cache_size) field, which documents
    /// why this is not the same number as the writer's.
    pub fn reader_cache_size(mut self, size: i32) -> Self {
        self.reader_cache_size = Some(size);
        self
    }

    /// What to do about a stored `recorded_at` in the future — the
    /// [`future_stamps`](Self::future_stamps) field.
    pub fn future_stamps(mut self, policy: FutureStampPolicy) -> Self {
        self.future_stamps = policy;
        self
    }

    /// The `Option<SnapshotCadence>` the three older constructors take, mapped
    /// onto the tri-state. `None` there means *disabled*, which is why
    /// [`CadencePolicy`] exists — see its docs.
    fn from_legacy(cadence: Option<SnapshotCadence>, clock: Option<Arc<dyn Clock>>) -> Self {
        Self {
            cadence: match cadence {
                Some(cadence) => CadencePolicy::Every(cadence),
                None => CadencePolicy::Disabled,
            },
            clock,
            wal_autocheckpoint: WalCheckpointPolicy::default(),
            writer_cache_size: None,
            reader_cache_size: None,
            future_stamps: FutureStampPolicy::default(),
        }
    }
}

// `Clock` is not `Debug` — it is a behavioural trait with two methods and
// requiring `Debug` of every implementor to print a handle here would be the
// tail wagging the dog. So the field is reported as present-or-absent, which is
// the only part of it a reader of a `Tuning` dump can act on.
impl std::fmt::Debug for Tuning {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Tuning")
            .field("cadence", &self.cadence)
            .field("clock", &self.clock.as_ref().map(|_| "<injected>"))
            .field("wal_autocheckpoint", &self.wal_autocheckpoint)
            .field("writer_cache_size", &self.writer_cache_size)
            .field("reader_cache_size", &self.reader_cache_size)
            .finish()
    }
}

impl Database {
    /// Open a database file at `path`, configuring pragmas, running migrations, and spawning the Write Actor.
    ///
    /// The snapshot cadence runs with [`SnapshotCadence::default`]. Use
    /// [`Database::open_with_cadence`] to tune or disable it.
    pub async fn open(path: impl AsRef<Path>) -> Result<Self> {
        Self::open_with_cadence(path, Some(SnapshotCadence::default())).await
    }

    /// Open with an explicit snapshot cadence, or `None` to run without one
    /// (§5.5, D-053).
    ///
    /// `None` restores the pre-0.5.5 behaviour, where `close()` is the only
    /// thing that ever writes an anchor. That is the right setting for a
    /// short-lived process that will not accumulate a delta worth bounding, and
    /// for tests that assert on the contents of the snapshot directory.
    pub async fn open_with_cadence(
        path: impl AsRef<Path>,
        cadence: Option<SnapshotCadence>,
    ) -> Result<Self> {
        Self::open_inner(path.as_ref(), Tuning::from_legacy(cadence, None)).await
    }

    /// Open with an injected clock (§5.1.2, **defect K**, D-062).
    ///
    /// The reason this exists is testing: `recorded_at` is the transaction-time
    /// axis, and until now every test that wanted to assert on one had to either
    /// avoid it or drive a raw connection, because `open()` hardcoded
    /// [`SystemClock`]. `FakeClock` has been public and constructed in the test
    /// harness since 0.5.2 with nothing to inject it into — the compiler warned
    /// about the dead field on every build for three releases.
    ///
    /// **The clock is floored against the database before the actor starts.**
    /// [`Clock::raise_floor`] is called with the newest `recorded_at` in the
    /// ledger, so an injected clock cannot issue a stamp below what is already
    /// stored — which would abort the next concept write on
    /// `trg_concepts_monotonic_ra` rather than merely being odd. This is the
    /// step whose absence kept the defect open: the obvious implementation
    /// (take an `Arc<dyn Clock>`, use it) produces a `Database` that fails on
    /// its first write against any non-empty file.
    ///
    /// On a fresh database there is no floor, so an injected `FakeClock` issues
    /// exactly the stamps it was given.
    pub async fn open_with_clock(
        path: impl AsRef<Path>,
        cadence: Option<SnapshotCadence>,
        clock: Arc<dyn Clock>,
    ) -> Result<Self> {
        Self::open_inner(path.as_ref(), Tuning::from_legacy(cadence, Some(clock))).await
    }

    /// Open with an explicit [`Tuning`] (0.12.12, W5.1, D-155).
    ///
    /// The consolidated form of the three constructors above, and the one that
    /// grows: every knob 0.13.0 adds arrives as a field here rather than as a
    /// fourth `open_*`. See [`Tuning`] for why the struct is
    /// `#[non_exhaustive]` and why that makes the growth additive.
    ///
    /// That sentence was written at 0.12.12 and was false until 0.15.13: the
    /// struct was *not* `#[non_exhaustive]`, and [`Tuning`]'s own docs carried
    /// a section arguing at length that it should not be. Two documents in one
    /// file contradicting each other for eleven releases, which is the shape
    /// C-16 is about; W15.3 resolved it by making this one true.
    pub async fn open_tuned(path: impl AsRef<Path>, tuning: Tuning) -> Result<Self> {
        Self::open_inner(path.as_ref(), tuning).await
    }

    async fn open_inner(path: &Path, tuning: Tuning) -> Result<Self> {
        let Tuning {
            cadence,
            clock: injected,
            wal_autocheckpoint,
            writer_cache_size,
            reader_cache_size,
            future_stamps,
        } = tuning;
        let cadence = cadence.resolve();
        let db = libsql::Builder::new_local(path).build().await?;
        let write_conn = configure(db.connect()?, writer_cache_size).await?;
        // The writer is the only connection that commits, so it is the only one
        // whose `wal_autocheckpoint` can ever fire. Setting it on the readers
        // would be a pragma with no path to running (0.12.14, W5.3, D-157).
        if let Some(pragma) = wal_autocheckpoint.pragma() {
            let _ = write_conn.query(&pragma, ()).await?;
        }
        let read_conn = configure(db.connect()?, reader_cache_size).await?;

        // PRAGMA query_only = ON on reader connection (§5.1.2)
        read_conn.execute("PRAGMA query_only = ON", ()).await?;

        let migration = migrations::run(&write_conn).await?;

        let (highpri_tx, highpri_rx) = mpsc::channel(256);
        let (lowpri_tx, lowpri_rx) = mpsc::channel(64);

        // Floored after `migrations::run`, so the tables the floor is read from
        // are guaranteed to exist.
        let clock: Arc<dyn Clock> = match injected {
            Some(clock) => {
                if let Some(floor) =
                    crate::util::clock::recorded_at_floor(&read_conn, future_stamps).await?
                {
                    clock.raise_floor(floor);
                }
                clock
            }
            None => Arc::new(SystemClock::new(&read_conn, future_stamps).await?),
        };
        let shared = Arc::new(ActorShared::default());
        let writer = tokio::spawn(run_writer_actor(
            write_conn,
            Arc::clone(&clock),
            highpri_rx,
            lowpri_rx,
            Arc::clone(&shared),
        ));

        let archive_path = derive_archive_path(path);
        let snapshots_dir = derive_snapshots_dir(path);

        // **The cadence gets its own connection (Wave 4.1).** It used to share
        // `read_conn`, on the reasoning that `libsql::Connection` is an
        // Arc-backed handle and R15 makes every extra local connection a cost worth
        // not paying for nothing. The cost it was not paying for turned out to be
        // real: `reconstruct` brackets a fold with `ATTACH cold … DETACH cold`,
        // that region is per-connection state, and it is not synchronised. Two
        // folds on one connection can therefore interleave so that one DETACHes
        // the handle the other is mid-fold on.
        //
        // Recorded in §8.5 as a hazard rather than a defect because it **did not
        // reproduce**: 200 concurrent reconstructions against a 1 ms cadence with
        // an archive present produced zero errors, since the cadence anchors at
        // `MAX(recorded_at)` and so almost always takes the hot path. Narrow, and
        // real — a write landing between `log_head` and the fold opens it.
        //
        // Separate connections remove the interleaving rather than ordering it,
        // which is why this is preferred to a mutex around the region: there is
        // no shared state left to race on, and nothing to remember to hold. The
        // R15 objection does not apply — that fault is about *concurrent* opens,
        // and this is one more sequential open during `open()`.
        let (cadence_stop, cadence) = match cadence {
            Some(cadence) => {
                let cadence_conn = configure(db.connect()?, reader_cache_size).await?;
                cadence_conn.execute("PRAGMA query_only = ON", ()).await?;
                let (tx, rx) = tokio::sync::watch::channel(false);
                let handle = tokio::spawn(snapshot::run_cadence(
                    cadence_conn,
                    snapshots_dir.clone(),
                    archive_path.clone(),
                    cadence,
                    rx,
                    Arc::clone(&shared) as Arc<dyn snapshot::CommittedTurns>,
                ));
                (Some(tx), Some(handle))
            }
            None => (None, None),
        };

        let handle = Self {
            db,
            path: path.to_path_buf(),
            read_conn,
            highpri_tx,
            lowpri_tx,
            clock,
            archive_path,
            snapshots_dir,
            schema_version: migrations::current_version(),
            reader_cache_size,
            diagnostic_conn: tokio::sync::Mutex::new(None),
            writer: Some(writer),
            cadence_stop,
            cadence,
            closed: false,
            shared,
        };

        // **Re-anchor after a migration (Wave 4.4).**
        //
        // D-043 makes a `SCHEMA_VERSION` bump invalidate every snapshot on disk,
        // which is correct — a snapshot is a serialised `MaterializedState` and a
        // schema change can change what that means. What was missing is the other
        // half: nothing wrote a replacement, so the first `reconstruct` after an
        // upgrade skipped every file as incompatible and folded from genesis. On
        // a database with a large log that is the difference between reading one
        // snapshot and folding the whole history, and the only trace was a
        // `warn!` per skipped file.
        //
        // Written here rather than left to the cadence because the cadence fires
        // on log *growth* (D-053): an upgraded database that is then read but not
        // written would never re-anchor at all.
        //
        // Failure is logged, not returned. A missing anchor costs time and no
        // information — snapshots are derivative under Doctrine VI — so refusing
        // to open a database because its optimisation could not be rebuilt would
        // trade a real capability for a performance one.
        //
        // Gated on the cadence being enabled, as well as on an actual upgrade:
        // `open_with_cadence(None)` means *this handle writes no snapshots except
        // at close()*, and a one-off write at open would contradict that for a
        // caller who asked for the quiet mode precisely to control when files
        // appear. They still get an anchor from `close()`.
        if migration.upgraded() && handle.cadence.is_some() {
            let ts = handle.clock.now();
            let archive = crate::temporal::archive::archive_present(&handle.archive_path)
                .then_some(handle.archive_path.as_path());
            match snapshot::write_final(&handle.read_conn, &handle.snapshots_dir, &ts, archive)
                .await
            {
                Ok(path) => tracing::info!(
                    "schema moved v{} -> v{}; re-anchored snapshots at {:?}",
                    migration.from,
                    migration.to,
                    path
                ),
                Err(e) => tracing::warn!(
                    "schema moved v{} -> v{} but the re-anchor failed: {e}. \
                     Reconstruction stays correct and folds from genesis until the \
                     cadence writes one.",
                    migration.from,
                    migration.to
                ),
            }
        }

        Ok(handle)
    }

    /// Read connection handle for queries, traversals, and folds.
    pub fn read_conn(&self) -> &libsql::Connection {
        &self.read_conn
    }

    /// The file this handle opened.
    pub fn path(&self) -> &Path {
        &self.path
    }

    /// The **OS-level read-only** connection to this database, for diagnostics
    /// (§4.7, T5.1, D-091).
    ///
    /// # Why this exists when `read_conn()` already does
    ///
    /// Two different things, and the difference is the point:
    ///
    /// * `read_conn()` returns a shared `&Connection` carrying
    ///   `PRAGMA query_only = ON`. That pragma is **per-connection and
    ///   reversible by its holder in one statement**, so it is a guardrail
    ///   against accident, not a capability boundary. And because it is the
    ///   connection the crate's own traversals and folds run on, a caller who
    ///   runs a long reporting query there is competing with all of them.
    /// * This returns a connection opened with `SQLITE_OPEN_READ_ONLY`, which is
    ///   enforced by the engine below the pragma layer, and which nothing inside
    ///   the crate runs on.
    ///
    /// # One connection per `Database`, shared between callers (0.15.14, W15.4)
    ///
    /// Through 0.15.13 this minted a connection per call and the sentence above
    /// read *"a **new, independently owned** … connection"*. Review item C-9
    /// asked for the file to be opened once per handle instead, and the
    /// measurement behind it (`examples/diagnostic_conn_probe.rs`) was sharper
    /// than the ask: `connect()` is **51.5 µs of an 82.7 µs call**, and
    /// `Builder::…build()` — which every document in this crate called *the
    /// open* — is **0.10 µs and opens nothing**, succeeding against a path that
    /// does not exist. Caching the handle would have removed 0.10 µs. What
    /// ships caches the connection: **82.7 µs → 19.9 µs**, and what remains is
    /// the `stat` below, not the connection.
    ///
    /// **So per-connection state is shared between diagnostic callers.** An
    /// `ATTACH`, a `PRAGMA`, a temp table one caller creates is visible to the
    /// next — which matters here and nowhere else in this API, because
    /// `diagnostic_query` is the one arbitrary-SQL surface the crate exposes.
    /// Asserted in `diagnostic_callers_share_one_connection_and_its_state`
    /// rather than left to this paragraph.
    ///
    /// What that costs is isolation between *diagnostic* callers. What it does
    /// not cost is the thing D-091 was for: this is still not `read_conn()`, a
    /// reporting query here still does not compete with the crate's traversals
    /// and folds, and the read-only boundary below is untouched.
    ///
    /// # What is scrubbed before you get it, and what is not (0.15.15, W15.5)
    ///
    /// 0.15.14 shipped the sharing and documented it. [D-257] measured what it
    /// actually admits, and one of the four is not an isolation nuisance but a
    /// correctness hole that leaves this surface entirely: **a leaked `BEGIN`
    /// pins a WAL read snapshot.** Measured — 200 writes through the typed
    /// surface while a diagnostic caller held an unclosed read transaction —
    /// later diagnostic reads answered **1 row instead of 201**, silently, and
    /// [`Database::checkpoint`] became a no-op with the WAL stuck at 8.5 MB
    /// until the transaction was rolled back. Stale answers on the surface a
    /// caller reaches for when they already distrust the typed one, and an
    /// unbounded WAL on a method that has nothing to do with diagnostics.
    ///
    /// So this method scrubs on entry rather than trusting the caller, and the
    /// prices are from `examples/diagnostic_hygiene_probe.rs`:
    ///
    /// | left behind | how it is found | what happens |
    /// |---|---|---|
    /// | an open transaction | `is_autocommit()`, **0.04 µs** | `ROLLBACK`, 2.4 µs |
    /// | a temp table or view | `PRAGMA temp.schema_version` | connection dropped, re-minted at 56.6 µs |
    /// | an `ATTACH` | `PRAGMA database_list` | connection dropped, re-minted |
    /// | `busy_timeout`, `cache_size` | not detected — restated, 1.0 µs | reset to the crate's values |
    /// | any other pragma | **not detected** | **inherited by the next caller** |
    ///
    /// The two dirt questions are asked as pragmas because the same two asked
    /// over `temp.sqlite_master` and `pragma_database_list` cost **7.8 µs**
    /// against **2.4 µs**, on a call whose entire warm cost is the `stat`.
    ///
    /// The last row is the honest residue. SQLite has no cheap enumeration of
    /// connection-scoped pragma state, so the crate restates the two pragmas
    /// *it* set (D-159's `busy_timeout` above all — a caller who sets it to 0
    /// would otherwise remove the 5 s margin from every later diagnostic call)
    /// and leaves the rest. A caller who sets `case_sensitive_like` or
    /// `recursive_triggers` changes what **later diagnostic queries on this
    /// handle** see, and nothing else: the crate's own readers and writer are
    /// different connections, so no typed answer can move. That is a smaller
    /// blast radius than 0.15.14 had and a larger one than zero, and it is
    /// written down rather than rounded off.
    ///
    /// That paragraph is about pragmas that belong to a *connection*, which is
    /// what "residue" means and what the scrub is for. **Not every pragma
    /// reachable here is one**, and the section below is the exception —
    /// measured after D-257 claimed this one "cannot change any typed answer"
    /// without checking (0.15.16, [D-258]).
    ///
    /// **Scrubbed on entry, not on exit**, because there is no exit: this
    /// returns a `Connection` clone the caller keeps for as long as it likes,
    /// and the crate is never told they are done. Entry is the one place that
    /// covers both this method and `diagnostic_query`. The consequence is that
    /// a caller who leaks a transaction and never calls again holds the pin
    /// until the handle drops — so the Python binding, which *does* know when
    /// a query is over, scrubs on exit as well.
    ///
    /// **Measured on libSQL 0.9.30 rather than assumed**
    /// (`examples/readonly_open_probe.rs`), against a live WAL database with the
    /// write actor running:
    ///
    /// | | `read_conn()` | `diagnostic_conn()` |
    /// |---|---|---|
    /// | `SELECT`, `EXPLAIN QUERY PLAN` | allowed | allowed |
    /// | `INSERT` | refused | refused |
    /// | `PRAGMA query_only = OFF` | **allowed** | allowed |
    /// | `INSERT` after that | **allowed** | **refused** |
    /// | `ATTACH` an existing file | allowed | allowed |
    /// | `INSERT` into the attachment | refused¹ | **refused** |
    /// | `ATTACH` a path that does not exist | — | refused (`SQLITE_CANTOPEN`) |
    ///
    /// The third and fourth rows are the whole difference: turning the pragma
    /// off restores writes on `read_conn()` and does not here. That is what
    /// "boundary rather than guardrail" means, and it is now a number rather
    /// than a claim.
    ///
    /// ¹ On `read_conn()` that refusal is `query_only` — the same reversible
    /// thing as row 2. On `diagnostic_conn()` it is the open flags, and the
    /// probe runs it *after* `query_only = OFF` so that the pragma cannot be
    /// what is doing the work.
    ///
    /// # `ATTACH` is permitted, and does not widen the write boundary
    ///
    /// Checked because `diagnostic_query` (Python) is the only arbitrary-SQL
    /// surface this crate exposes, and an attachment is a second `open` whose
    /// flags it does not obviously inherit. It does inherit them: the
    /// attachment is read-only, and a nonexistent path is `SQLITE_CANTOPEN`
    /// rather than a new file, because `SQLITE_OPEN_CREATE` is dropped for the
    /// attachment as it is for `main`. So `SQLITE_OPEN_READ_ONLY` bounds the
    /// **connection**, not just the one file it names (0.10.0, W4.3).
    ///
    /// What it does widen is *reading*: an `ATTACH` can name any file the
    /// process can open, so this connection is a read surface over the
    /// filesystem, not over this database. That is a property of arbitrary SQL
    /// rather than of the flags, and it is unchanged by them.
    ///
    /// # One pragma here can end the process, and sharing is not why
    ///
    /// `PRAGMA hard_heap_limit = 1` through this connection leaves the whole
    /// **process** unable to use SQLite. Not this connection, not this handle:
    /// measured (`tests_py/probes/diagnostic_global_pragmas.py`), the next
    /// ordinary write, the next read, `checkpoint()`, `close()`, and opening a
    /// *different* database file all fail with `out of memory`, permanently.
    ///
    /// `SQLITE_OPEN_READ_ONLY` does not stand in the way because setting it is
    /// not a write to the database file, and the scrub above does not help
    /// because there is nothing left on the connection to scrub: the limit
    /// lives in the SQLite library, one per process. **Re-measured with a
    /// connection minted per call — the 0.15.13 shape, before any of the
    /// sharing this method now does — the outcome is identical.** So this is
    /// not a cost of [D-256]'s shared connection and no amount of hygiene
    /// addresses it.
    ///
    /// Six other candidates were measured and are harmless: `soft_heap_limit`
    /// (a hint, not a wall), `locking_mode = EXCLUSIVE` (accepted; the writer
    /// kept working), `temp_store_directory`, `max_page_count` (clamped, and
    /// per-connection), `case_sensitive_like` (the control), and
    /// `wal_checkpoint`, which is refused outright because this connection is
    /// read-only.
    ///
    /// It belongs with the `ATTACH` note above rather than with the scrub: both
    /// are properties of handing a caller **arbitrary SQL**, not of the flags
    /// the connection was opened with. The practical form is one sentence —
    /// *`diagnostic_query` is not a safe place to put a string that came from
    /// somewhere else* — which `ATTACH` already made true and this makes
    /// sharper. Not blocked by refusing statements that look like this one,
    /// because matching SQL text is guesswork wearing the costume of a
    /// guarantee, and it would do nothing for a Rust caller holding the
    /// connection directly ([D-258]).
    ///
    /// [D-258]: ../../docs/architecture/s13-decision-register.md#d-258
    ///
    /// # One way this is *more* permissive, which is worth knowing
    ///
    /// `CREATE TEMP TABLE` **succeeds** here and is refused by `read_conn()`.
    /// Temp tables live in a separate temporary database that is writable
    /// regardless of how the main one was opened, whereas `query_only` refuses
    /// them outright — which is the mechanism [D-050] measured when it removed
    /// `TwoPhaseTempTable` for returning `SQLITE_READONLY (8)` on the read
    /// connection. So the stronger boundary is not uniformly stronger, and a
    /// strategy that needs a temp table has a connection it could run on. That
    /// is recorded, not acted on: D-050 removed the strategy for two reasons and
    /// this addresses one of them.
    ///
    /// # Calling this concurrently was R15's shape, and 0.15.14 is why it is not
    ///
    /// Through 0.15.13 this section said *"this is the one method on `Database`
    /// that opens the file … each call is a fresh `libsql::Builder::…build()`,
    /// so *N* threads calling it at once are *N* concurrent opens"*. The first
    /// half was true and the second was wrong about which call does it:
    /// `build()` opens nothing, and `connect()` is the open. The conclusion
    /// happened to be right for the wrong reason, which is why it took a
    /// measurement to move.
    ///
    /// **Measured through the unlocked Python binding**, 48 threads on a
    /// barrier, 30 runs per arm (`tests_py/probes/r15_diagnostic_path.py`):
    ///
    /// | arm | bad runs |
    /// |---|---|
    /// | a connection per call, as before 0.15.14 | **3 / 30** |
    /// | the `libsql::Database` handle cached, a connection per call | 2 / 30 |
    /// | the `connect()` serialised behind a mutex | 1 / 18 |
    /// | **one connection, as shipped** | **0 / 30** |
    ///
    /// Rows two and three are why the shape that preserved the old contract was
    /// not taken: caching the handle leaves the crash because it leaves the
    /// `connect()`, and serialising the `connect()` alone does not reach zero —
    /// the race is between minting a connection and the *use* of the others,
    /// not between two mintings.
    ///
    /// **There is nothing left on this path to bound**, because after the first
    /// call it no longer opens anything. `Database::open` from many threads is
    /// still R15's shape and `examples/r15_soak.rs` still reproduces it; this
    /// method is no longer a way to reach it. The Python binding keeps its
    /// mutex as margin rather than as a measured necessity — see
    /// `PyDatabase::diagnostic_rows`.
    ///
    /// # Errors
    ///
    /// The file must already exist. `SQLITE_OPEN_READ_ONLY` drops
    /// `SQLITE_OPEN_CREATE` with it, so a missing file is `SQLITE_CANTOPEN`
    /// rather than a fresh empty database — which is the right failure, and is
    /// surfaced as a typed error rather than as libSQL's error 14.
    ///
    /// The check is a `stat` on **every** call, not only the first, and it is
    /// still most of what a warm call costs (18.6 µs of the 22 µs a clean one
    /// takes since 0.15.15). It is kept at that price because the alternative
    /// is the worst failure a *diagnostic* surface can have: a cached
    /// connection whose file has been deleted and replaced answers from the old
    /// inode, silently, on the one method a caller reaches for when they
    /// already doubt the typed answer.
    pub async fn diagnostic_conn(&self) -> Result<libsql::Connection> {
        let fail = |reason: String| DbError::DiagnosticConn {
            path: self.path.display().to_string(),
            reason,
        };
        // Checked per call rather than once, because it is the documented
        // error of *this* method and costs a `stat`. The handle below is
        // opened once; the question "is the file there" is asked every time,
        // so a caller who deletes the file still gets the typed refusal on the
        // next call rather than a connection to an inode nothing can name.
        if !self.path.exists() {
            return Err(fail(
                "the file does not exist, and a read-only open cannot create it".to_string(),
            ));
        }
        let mut slot = self.diagnostic_conn.lock().await;

        // Scrub what the last caller left, before this one can inherit it
        // (0.15.15, W15.5, D-257). Free on a clean connection: the transaction
        // check is a C call at 0.04 us and the two dirt pragmas are 2.4 us,
        // against a `stat` of 18.6 that has already happened above.
        scrub(&mut slot).await;

        match slot.as_ref() {
            // Restated per call since 0.15.15 rather than once at mint. The
            // pragmas are per-connection, and there is one connection now, so
            // 0.15.14 set them once — which was right until the shared
            // connection was also something a caller could move them on. No
            // dirt check can see a pragma, so the crate restores its own
            // instead of detecting that they went: 1.0 us, against a
            // `busy_timeout` of 0 outliving the call that set it.
            Some(conn) => configure_common(conn, self.reader_cache_size).await?,
            None => {
                *slot = Some(self.open_diagnostic_conn(&fail).await?);
            }
        }

        Ok(slot
            .as_ref()
            .expect("the slot was filled above or the open returned Err")
            .clone())
    }

    /// Roll back and discard whatever the last diagnostic caller left behind,
    /// without handing a connection out (0.15.15, W15.5, [D-257]).
    ///
    /// [`Database::diagnostic_conn`] does this on the way *in*, which is the
    /// only place the crate can do it: the method returns a `Connection` clone
    /// and is never told the caller is finished with it. That covers every
    /// caller and leaves one gap — somebody who leaks a transaction and then
    /// never calls again holds the WAL read snapshot until the handle drops,
    /// which makes [`Database::checkpoint`] a no-op for that whole time.
    ///
    /// This is for the callers that *do* know when they are done. It costs the
    /// scrub and not the `stat` — around 3.5 µs on a clean connection — because
    /// it hands nothing back and so has nothing to promise about the file still
    /// being there.
    ///
    /// **The Python binding does not call it**, though the first draft did. A
    /// mutation deleting that call left the whole suite green, and the reason is
    /// that the gap is not reachable from there: `diagnostic_query` runs one
    /// statement, a bare `BEGIN` pins nothing — the snapshot is taken by the
    /// first *read* inside the transaction — and any statement that would take
    /// it arrives through the same method, whose entry scrub has already rolled
    /// the transaction back. The gap is real for a Rust caller holding a clone
    /// across both, which is what this method and
    /// `scrubbing_releases_the_pin_without_handing_out_a_connection` are for.
    ///
    /// Infallible by construction: everything it might have reported is
    /// something it responds to by discarding the connection, and the next
    /// [`Database::diagnostic_conn`] opens a fresh one.
    ///
    /// [D-257]: ../../docs/architecture/s13-decision-register.md#d-257
    pub async fn scrub_diagnostic_conn(&self) {
        let mut slot = self.diagnostic_conn.lock().await;
        scrub(&mut slot).await;
    }

    /// The cold half of [`Database::diagnostic_conn`]: the actual open.
    ///
    /// Reached on the first call and on any call whose predecessor left the
    /// connection dirty enough to discard. It costs 56.5 µs against the warm
    /// path's scrub, and it is split out for reading rather than for speed:
    /// `Box::pin`ning it here — on the theory that carrying `Builder::build()`'s
    /// state machine inside the warm path's was what made a clean call 29.8 µs
    /// rather than the 21.8 its parts measure — changed nothing at all
    /// (0.15.15, W15.5, [D-257]).
    ///
    /// [D-257]: ../../docs/architecture/s13-decision-register.md#d-257
    async fn open_diagnostic_conn(
        &self,
        fail: &dyn Fn(String) -> DbError,
    ) -> Result<libsql::Connection> {
        let db = libsql::Builder::new_local(&self.path)
            .flags(libsql::OpenFlags::SQLITE_OPEN_READ_ONLY)
            .build()
            .await
            .map_err(|e| fail(e.to_string()))?;
        let conn = db.connect().map_err(|e| fail(e.to_string()))?;
        // Configured since 0.12.16 (W5.5, D-159). Until then this connection
        // ran with SQLite's defaults while every other connection in the
        // process ran with the crate's — most consequentially a `busy_timeout`
        // of 0 against everyone else's 5 s, on the one surface whose job is to
        // answer questions when the typed path is already suspect. Only the
        // common half: `SQLITE_OPEN_READ_ONLY` cannot set `journal_mode`, and
        // the rest govern writes this connection cannot make.
        configure_common(&conn, self.reader_cache_size).await?;
        // The `libsql::Database` is dropped here and the connection outlives
        // it, which is the ownership libSQL's own API implies: `connect()`
        // returns a `Connection` that does not borrow the builder's handle.
        Ok(conn)
    }

    /// Cross-check the snapshot chain against a fold from genesis (§5.5, T5.3,
    /// D-092).
    ///
    /// `write_final` composes onto the previous snapshot, so snapshot *n* is
    /// derived from snapshot *n−1* and nothing in the chain ever folds the whole
    /// log. An error at any link propagates forward forever and every read
    /// agrees with it, because every read descends from it. This is the check
    /// that would notice.
    ///
    /// # When to run it
    ///
    /// **Not on a schedule this crate chooses.** A genesis fold is precisely the
    /// cost snapshots exist to avoid, so running it periodically by default
    /// would give every application the bill snapshots were bought to remove —
    /// on a database whose log is large enough for snapshots to matter, which is
    /// the only kind where this is worth doing. The plan calls it a scheduling
    /// problem and it is the caller's schedule: an idle period, a nightly job,
    /// or once per *N* anchors, chosen against a log size this crate cannot see.
    ///
    /// The cadence is deliberately left alone for the same reason — it runs on a
    /// connection shared with nothing and a fold there would compete with
    /// interactive reads at a moment nobody chose.
    ///
    /// # It reports; it does not repair
    ///
    /// A divergence means the snapshots are a wrong **cache**, not that the
    /// ledger is corrupt: [Doctrine VI] makes them disposable, so deleting
    /// [`Self::snapshots_dir`] restores correctness and costs only speed.
    /// Rewriting the file here would destroy the evidence that composition has a
    /// defect, which is the only thing this can tell you that you did not
    /// already know.
    ///
    /// Pair it with the actor counters ([`Self::metrics`], D-079) so a
    /// divergence found by a scheduled run is visible beside the write latency
    /// of the period that produced it.
    ///
    /// [Doctrine VI]: ../../docs/architecture/s0-s3-foundations.md#doctrine-vi
    pub async fn verify_snapshot_chain(&self, ts: &str) -> Result<crate::temporal::ChainCheck> {
        let archive = crate::temporal::archive::archive_present(&self.archive_path)
            .then_some(self.archive_path.as_path());
        crate::temporal::verify_snapshot_chain(&self.read_conn, ts, archive, &self.snapshots_dir)
            .await
    }

    /// Check the newest link of the snapshot chain (0.15.19, review C-18).
    ///
    /// The affordable half of [`Self::verify_snapshot_chain`]: re-derive the
    /// newest snapshot from the one before it and compare, which is one
    /// anchored delta rather than a fold from genesis. `Ok(None)` when there
    /// are not two snapshots yet.
    ///
    /// The snapshot cadence already runs this after every anchor it writes and
    /// logs a divergence at `warn`, so a caller reaching for it directly is
    /// usually one that wants the [`crate::temporal::ChainCheck`] itself — the
    /// disagreeing ids — rather than a yes or no.
    ///
    /// **It reports; it does not repair.** A snapshot is derivative
    /// (Doctrine VI), so the repair is to delete the snapshot directory, which
    /// is the caller's call and one line. What this cannot tell you is whether
    /// the chain went wrong further back than one link; that is what
    /// [`Self::verify_snapshot_chain`] is for, and why it stays.
    pub async fn verify_last_link(&self) -> Result<Option<crate::temporal::ChainCheck>> {
        let archive = crate::temporal::archive::archive_present(&self.archive_path)
            .then_some(self.archive_path.as_path());
        crate::temporal::verify_last_link(&self.read_conn, archive, &self.snapshots_dir).await
    }

    /// The clock every write is stamped with (§5.1.1).
    pub fn clock(&self) -> &Arc<dyn Clock> {
        &self.clock
    }

    /// Schema version this handle opened against.
    pub fn schema_version(&self) -> u32 {
        self.schema_version
    }

    /// Cold database path, derived by convention from the main file.
    pub fn archive_path(&self) -> &Path {
        &self.archive_path
    }

    /// Snapshot directory, derived by convention from the main file.
    pub fn snapshots_dir(&self) -> &Path {
        &self.snapshots_dir
    }

    /// What the write actor has done since this handle was opened (T1.4, D-079).
    ///
    /// Requires the `metrics` feature. The counters are per-handle and start at
    /// zero on `open()` — they are not read from the database, because the thing
    /// being measured is *this process's* actor and merging two processes'
    /// histograms would produce a number about neither.
    ///
    /// The intended first question is [`crate::metrics::MetricsSnapshot::budget_violations`]:
    ///
    /// ```no_run
    /// # async fn f(db: &macrame::Database) {
    /// # #[cfg(feature = "metrics")] {
    /// for k in db.metrics().budget_violations() {
    ///     eprintln!("{} broke the 3 ms bound {} times", k.kind, k.over_budget);
    /// }
    /// # }
    /// # }
    /// ```
    ///
    /// Reading this does not stop the actor — see
    /// [`crate::metrics::ActorMetrics::snapshot`] for what that costs in
    /// consistency, and why the trade goes that way.
    #[cfg(feature = "metrics")]
    pub fn metrics(&self) -> crate::metrics::MetricsSnapshot {
        self.shared.metrics.snapshot()
    }

    /// The underlying libSQL database, for callers that need their own connection.
    ///
    /// # Actor containment is a convention above this line, not a guarantee
    ///
    /// **Kept public, and the honest statement of what that costs (Wave 4.3).**
    /// §5.1 says the write actor is the sole writer, and two mechanisms make that
    /// true of the handle: every write method goes through a channel, and
    /// [`Self::read_conn`] carries `PRAGMA query_only = ON`. **Nothing protects a
    /// connection obtained from here.** A caller can open one, write to `links`
    /// directly, and the actor will not know — the triggers still fire and the
    /// ledger stays internally consistent, but the single-writer property that
    /// [`crate::CHUNK_BUDGET`]'s latency argument rests on is gone, and so is the
    /// serialisation the overlap guard (D-060) relies on.
    ///
    /// This is the same shape as the limit stated in §4.2 for that guard, and it
    /// is one fact rather than two: **the storage layer permits what this API
    /// refuses.** Making it private would not change that — the database file is
    /// reachable by any SQLite client on the machine — it would only remove the
    /// supported way to do the thing, which is how escape hatches become
    /// `unsafe`-adjacent folklore.
    ///
    /// The free functions [`crate::register_model`] and
    /// [`crate::upsert_embedding`] take a bare connection for the same reason and
    /// carry the same caveat; prefer [`Self::register_model`] and
    /// [`Self::upsert_embeddings`], which go through the actor.
    ///
    /// # The legitimate-use list is now one item long (T5.1, D-091)
    ///
    /// It used to read: `EXPLAIN QUERY PLAN` and other diagnostics, read-only
    /// reporting queries wanting their own connection rather than sharing the
    /// reader, and provoking a guard in a test. The first two are exactly what
    /// [`Self::diagnostic_conn`] now does, and it does them behind an OS-level
    /// read-only open rather than on a handle that can write. **Use that.**
    ///
    /// What is left is the one use that genuinely requires write access through
    /// a connection the actor does not own: *provoking a guard* — writing the
    /// state §4.7 says the storage layer permits and this API refuses, so a test
    /// can assert the gap is still where the document says it is. That is the
    /// only thing this crate's own suite uses it for.
    ///
    /// # Why `#[doc(hidden)]` and not a `raw-access` feature
    ///
    /// T5.1 offers either. The feature is the stronger declaration — it shows up
    /// in the consumer's `Cargo.toml`, where a reviewer sees it — and it was
    /// **not** taken, for a reason specific to what uses this:
    ///
    /// Cargo features are additive and cannot be *required* by a test target
    /// except through `required-features`, which makes a plain `cargo test`
    /// **skip** that binary silently. The binaries that call this are
    /// `storage_boundary_tests` and `wave1_regression_tests` — the §4.7
    /// tripwires, whose entire job is to fail when a documented gap moves. Gating
    /// them behind a feature would mean the ordinary `cargo test` stopped running
    /// the tests that enforce the section this item is about, to make a
    /// declaration about a hatch. That trade is the wrong way round, and it is
    /// the same failure the project already names: a suite that quietly does less
    /// than it appears to.
    ///
    /// So the hatch stays reachable and stops being *discoverable*: it is absent
    /// from the docs, and the documented path for every non-write use is
    /// [`Self::diagnostic_conn`]. [D-068] is unchanged — removing it would buy
    /// the appearance of a guarantee, since the file is reachable by any SQLite
    /// client on the machine.
    ///
    /// [D-068]: ../../docs/architecture/s13-decision-register.md#d-068
    // convention (D-068/D-091): `raw()` is #[doc(hidden)] and is NOT exposed by
    // any binding. Everything above this line is invisible on docs.rs and
    // invisible to a contributor reading the Python surface list, which is where
    // the decision to expose it would actually be taken — hence this sentinel and
    // its twin in `bindings/python/src/lib.rs` (0.10.0, W4.10). The documented
    // path for every non-write use is `diagnostic_conn`.
    #[doc(hidden)]
    pub fn raw(&self) -> &libsql::Database {
        &self.db
    }

    // -- write surface (§5.1, Appendix A) --
    //
    // Every method here validates and canonicalises before the value crosses the
    // channel, so a bad edge type or a second-precision timestamp is a typed
    // error at the call site rather than an engine `CHECK` failure surfacing
    // from the far side of an actor with no context attached.
    //
    // NOTE (§5.1.8, D-028): awaiting one of these waits on a Rust channel, not
    // in SQLite, so `busy_timeout` does not bound it. During an in-flight
    // `rebuild_current` or `archive` the caller stalls for that transaction's
    // duration. Wrap in `tokio::time::timeout` if you need a bound — but a
    // timeout is not a cancellation: the command stays queued and commits when
    // the actor reaches it.

    /// Assert an edge (Doctrine III: a new row, never an update).
    ///
    /// # One row costs a transaction, so N rows cost N transactions
    ///
    /// This is the correct method for a caller who genuinely has one edge, and
    /// it is the wrong one in a loop. Each call is its own transaction and pays
    /// the ~0.8 ms per-transaction floor (D-090) whole, so a thousand edges
    /// asserted one at a time spend roughly **0.8 s in transaction overhead
    /// alone** — before any of the work — and mint a thousand distinct
    /// `recorded_at` stamps for what the caller probably means as one act.
    ///
    /// There are two bulk forms and the difference between them is the one to
    /// get right:
    ///
    /// - [`Self::bulk_import`] is **chunked** against [`CHUNK_BUDGET`] and
    ///   atomic per chunk. It amortises the transaction floor across the batch
    ///   while still yielding to interactive work at every chunk boundary. This
    ///   is the one a loop should almost always become.
    /// - [`Self::write_bulk_atomic`] is one transaction under one stamp and is
    ///   **the one write with no latency bound** — the hold is a function of
    ///   `edges.len()`, tabulated in its own docs, and is time every other
    ///   writer spends waiting. Reach for it when the batch is genuinely one
    ///   act that must not be observable half-applied, not for speed.
    ///
    /// The choice is the caller's and neither form is deprecated. Doctrine III
    /// makes "one act, one stamp" a semantic claim rather than a performance
    /// one, and only the caller knows whether their thousand edges are one act.
    pub async fn assert_edge(&self, edge: EdgeAssertion) -> Result<()> {
        let edge = edge.normalized()?;
        self.high(|responder| HighPriCommand::AssertEdge { edge, responder })
            .await
    }

    /// Close an open interval by asserting its replacement (Doctrine III).
    pub async fn retire_edge(
        &self,
        source: impl Into<String>,
        target: impl Into<String>,
        edge_type: impl Into<String>,
        valid_from: &str,
        valid_to: &str,
    ) -> Result<()> {
        let edge_type = edge_type.into();
        crate::graph::edge::validate_edge_type(&edge_type)?;
        let valid_from = timestamp::normalize(valid_from)?;
        let valid_to = timestamp::normalize(valid_to)?;
        let (source, target) = (source.into(), target.into());

        self.high(|responder| HighPriCommand::RetireEdge {
            source,
            target,
            edge_type,
            valid_from,
            valid_to,
            branch: None,
            responder,
        })
        .await
    }

    /// Retire an edge **on a lineage**, which is a different write (0.14.8).
    ///
    /// The `_on` suffix is the crate's established spelling for the
    /// branch-taking variant of a call whose trunk form predates branching —
    /// [`query_as_of_edges_on`](crate::temporal::query_as_of_edges_on) is the
    /// other one. A sixth positional `Option<BranchId>` on
    /// [`Self::retire_edge`] would have made every existing call site read as
    /// though it had made a lineage decision it never made.
    ///
    /// # This closes a row; it does not close *the* row
    ///
    /// Retiring an edge the branch **inherited** writes the branch's own row at
    /// the ancestor's key, carrying the closed interval and this lineage's id.
    /// The ancestor's row is untouched, and the read prefers the nearer one, so
    /// the edge is gone from this lineage's view and unchanged in its parent's.
    /// That is **shadow retirement**, and it is the only retirement across
    /// lineages that does not commit the parent corruption
    /// [Doctrine III](../../docs/architecture/s0-s3-foundations.md#doctrine-iii)
    /// forbids — which is not a rule this method obeys but a shape the ledger
    /// cannot express: `links` is append-only and no statement in this crate
    /// closes a row in place.
    ///
    /// `weight` and `properties` are carried over from the visible row rather
    /// than restated, which is what makes this a retirement rather than a new
    /// assertion that happens to be closed.
    ///
    /// # Errors
    ///
    /// - [`DbError::UnknownBranch`] when `branch` is not registered.
    /// - [`DbError::NotFound`] when this lineage can see no open row at that
    ///   `valid_from`. On a branch that includes *never inherited it* and
    ///   *inherited it and already shadowed it*, which are one answer here
    ///   because they are one answer to the question asked: there is nothing
    ///   at that key to retire.
    pub async fn retire_edge_on(
        &self,
        source: impl Into<String>,
        target: impl Into<String>,
        edge_type: impl Into<String>,
        valid_from: &str,
        valid_to: &str,
        branch: crate::branch::BranchId,
    ) -> Result<()> {
        let edge_type = edge_type.into();
        crate::graph::edge::validate_edge_type(&edge_type)?;
        let valid_from = timestamp::normalize(valid_from)?;
        let valid_to = timestamp::normalize(valid_to)?;
        let (source, target) = (source.into(), target.into());

        self.high(|responder| HighPriCommand::RetireEdge {
            source,
            target,
            edge_type,
            valid_from,
            valid_to,
            branch: Some(branch),
            responder,
        })
        .await
    }

    /// Insert or update a concept.
    ///
    /// # One row costs a transaction
    ///
    /// The same trade [`Self::assert_edge`] describes, for the same reason and
    /// with the same ~0.8 ms floor (D-090): correct for one concept, wrong in a
    /// loop. [`Self::write_concepts`] takes a `Vec` and commits it as one
    /// transaction under one stamp.
    ///
    /// There is no atomic-across-chunks concept path and none is needed to make
    /// the choice: `write_concepts` is chunked against [`CHUNK_BUDGET`] and
    /// atomic per chunk, so a large `Vec` is cooperative rather than a stall.
    /// The responsiveness argument for writing one row at a time therefore does
    /// not apply — the bulk form already yields at every chunk boundary.
    pub async fn upsert_concept(&self, concept: ConceptUpsert) -> Result<()> {
        let concept = concept.normalized()?;
        self.high(|responder| HighPriCommand::UpsertConcept { concept, responder })
            .await
    }

    /// A handle on one lineage (§15.4, 0.14.9, [D-226]).
    ///
    /// Takes `&Arc<Self>` rather than `&self` because the view holds the handle
    /// and must not be able to end it: `close` takes `self` by value and an
    /// `Arc` cannot surrender that while a clone survives, so the restriction
    /// is structural rather than documented. Sharing the handle is already
    /// `Arc<Database>` (§5.1.11), so this asks for nothing a caller did not
    /// have.
    ///
    /// Does no I/O and cannot fail. Whether the lineage is *registered* is
    /// asked by every operation on the view, which is where
    /// [`DbError::UnknownBranch`] names it.
    ///
    /// [D-226]: ../../docs/architecture/s13-decision-register.md#d-226
    pub fn view(
        self: &std::sync::Arc<Self>,
        branch: crate::branch::BranchId,
    ) -> crate::branch::BranchView {
        crate::branch::BranchView::new(std::sync::Arc::clone(self), branch)
    }

    /// Cut a new lineage from an existing one (§15.2, §15.4).
    ///
    /// # A fork is O(1) in rows written
    ///
    /// One row in `branches`, and nothing else. No ledger table is read, copied
    /// or touched: a branch inherits its parent's history by *resolution at
    /// read* rather than by owning a copy of it, which is what
    /// [`TraversalBuilder::on_branch`](crate::graph::TraversalBuilder::on_branch)
    /// resolves and 0.14.6 bounds by the fork point. The cost of that choice is
    /// on the read side and is measured — [D-220] for the resolution, [D-223]
    /// for the cutoff — and the cost of the alternative would be here, as an
    /// O(rows) fork and storage multiplied by branch count (§15.3, option 3).
    ///
    /// # The fork point is *now*, and that is a bound on this release rather
    /// than on the design
    ///
    /// `forked_at` is stamped from the same clock as every other write, so the
    /// new lineage sees its parent's history up to this instant. Forking from a
    /// *past* instant is a coherent thing to want and the schema has always
    /// allowed it — `branches` carries `forked_at` and `created_at` as separate
    /// columns under `CHECK (forked_at <= created_at)` — but it is not in this
    /// release and is additive when it is.
    ///
    /// # What this lineage can do
    ///
    /// It can be **read**: every traversal entry point takes a branch, and on a
    /// forked ledger the read resolves along the ancestry and stops at the fork
    /// point. Since 0.14.8 it can also be **written** — [`EdgeAssertion`] and
    /// [`ConceptUpsert`] carry a lineage, and [`Self::retire_edge_on`] shadows
    /// an inherited edge (D-225). Through 0.14.7 they did not, and a caller who
    /// forked and then called `assert_edge` got a successful write **on the
    /// trunk**; that is fixed rather than documented now.
    ///
    /// What a branch still may not do is **restate an inherited concept**.
    /// `concepts` is keyed by identity, so that is refused as
    /// [`DbError::CrossLineage`] — see [`ConceptUpsert::branch`]. Edges are the
    /// thing a lineage may hold its own belief about, and superseding one is a
    /// row written *beside* the ancestor's rather than over it.
    ///
    /// # Errors
    ///
    /// - [`DbError::UnknownBranch`] when `from` is not registered. Named rather
    ///   than left to the foreign key, because the caller asked about a branch.
    /// - [`DbError::BranchExists`] when `name` is taken — including `"main"`,
    ///   which every database has from its first migration.
    /// - [`DbError::ForkPrecedesParent`] when the clock would place this fork
    ///   point before the parent's *own* — not before the parent's
    ///   `created_at`, which is what the schema comment promised until 0.14.7
    ///   and is not checkable: the trunk's `created_at` is stamped during
    ///   migration from the wall clock, before an injected clock exists, so
    ///   that rule refuses every fork on every `FakeClock` database (D-224).
    ///   Reachable with [`FakeClock`](crate::util::FakeClock), and the one
    ///   refusal here that no `CHECK` could have made — it is cross-row, and a
    ///   `CHECK` sees one row.
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use macrame::prelude::*;
    /// # async fn f(db: &Database) -> Result<()> {
    /// let alt = db.fork(BranchId::new("turn/17/alt/1")?, BranchId::main()).await?;
    /// let seen = TraversalBuilder::new("socrates")
    ///     .on_branch(alt.id.clone())
    ///     .execute_ids(db.read_conn(), "2026-08-29T00:00:00.000000Z")
    ///     .await?;
    /// # let _ = seen;
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// [D-220]: ../../docs/architecture/s13-decision-register.md#d-220
    /// [D-223]: ../../docs/architecture/s13-decision-register.md#d-223
    pub async fn fork(
        &self,
        name: crate::branch::BranchId,
        from: crate::branch::BranchId,
    ) -> Result<crate::branch::Branch> {
        self.high(|responder| HighPriCommand::Fork {
            name,
            parent: from,
            responder,
        })
        .await
    }

    /// Every lineage the ledger knows about, trunk first (§15.4).
    ///
    /// Read through [`Self::read_conn`] rather than the write actor, which is
    /// the difference between this and [`Self::fork`] and is deliberate:
    /// `branches` is append-only, so the only way this listing can be stale is
    /// by missing a branch created after it was taken, and a caller who wanted
    /// to know about that branch would have had to create it. Queueing a read
    /// behind the write actor would make listing branches wait on a bulk import
    /// for no answer it could change.
    ///
    /// A database that has never forked returns exactly one row: the trunk,
    /// with no parent and no fork point.
    pub async fn branches(&self) -> Result<Vec<crate::branch::Branch>> {
        crate::branch::list(self.read_conn()).await
    }

    /// The beliefs `a` holds that `b` does not (§15.4, 0.14.11, D-228).
    ///
    /// One [`Divergence`](crate::branch::Divergence) per edge key the two
    /// lineages disagree about, in key order: `b` holds no belief about it, or
    /// holds one with a different interval or weight. Not symmetric —
    /// `diff(b, a)` is the other half, and composing the two is *two* snapshots
    /// even though each is one.
    ///
    /// Read through the read connection rather than the actor, like
    /// [`Self::branches`], and taken at one snapshot rather than two: see
    /// `graph::lineage::diff_sql` for why that decides the shape of the query.
    ///
    /// There is no instant parameter. A diff filtered to a valid-time instant
    /// cannot report the one divergence that is *about* an instant having
    /// passed — a branch that retired an edge its parent still holds open — so
    /// this compares the whole of both views.
    ///
    /// # Errors
    ///
    /// [`DbError::UnknownBranch`], naming whichever of the two is not
    /// registered, and `a` first when neither is.
    pub async fn diff(
        &self,
        a: &crate::branch::BranchId,
        b: &crate::branch::BranchId,
    ) -> Result<Vec<crate::branch::Divergence>> {
        crate::branch::diff(self.read_conn(), a, b).await
    }

    /// Assert many edges in one transaction under one stamp (D-014).
    ///
    /// # This is the one write with no latency bound, and here is what it costs
    ///
    /// The batch is one act under one `recorded_at`, so it cannot be chunked —
    /// splitting it is the thing this method exists not to do. That makes the
    /// actor's hold a function of `edges.len()`, and until now the only
    /// statement of that anywhere was the prose "uncapped" in
    /// [`CHUNK_BUDGET`]'s table. A caller who stalls every other writer for
    /// eight seconds should have been able to predict it from the signature.
    ///
    /// Measured on libSQL 0.9.30 (T1.3, D-081), holding the actor for:
    ///
    /// | rows | hold |
    /// |---|---|
    /// | 500 | ~34 ms |
    /// | 2,000 | ~155 ms |
    /// | 10,000 | ~1.0 s |
    /// | 20,000 | ~2.6 s |
    ///
    /// [`estimated_bulk_hold`] is that curve as a function, and this method
    /// emits a `tracing::warn!` when it predicts more than
    /// [`BULK_ATOMIC_WARN_HOLD`]. **The estimate is a shape, not a promise** —
    /// see [`estimated_bulk_hold`] for what it is calibrated against and where
    /// it will be wrong.
    ///
    /// A caller who needs the latency bound and not the atomicity wants
    /// [`Self::bulk_import`], which is the same write chunked and explicitly not
    /// atomic overall (D-011).
    pub async fn write_bulk_atomic(&self, edges: Vec<EdgeAssertion>) -> Result<usize> {
        let estimate = estimated_bulk_hold(&edges);
        if estimate > BULK_ATOMIC_WARN_HOLD {
            // Warned here rather than in the actor, and before the send: this is
            // the caller's own task, so the log line lands with their span
            // attached and names the call site that chose the batch size. By the
            // time the actor has it, the only context left is "a large batch".
            tracing::warn!(
                rows = edges.len(),
                estimated_hold_ms = estimate.as_millis() as u64,
                "write_bulk_atomic will hold the write actor for roughly \
                 {estimate:?} — it is atomic by contract (D-014) and cannot be \
                 chunked. Every other writer waits that long. Use bulk_import \
                 if the batch does not need to be all-or-nothing."
            );
        }

        let edges = normalize_all(edges)?;
        self.high(|responder| HighPriCommand::WriteBulkAtomic { edges, responder })
            .await
    }

    /// Move the WAL back into the main database file (§4.5, F-30, 0.12.13,
    /// W5.2, D-156).
    ///
    /// Runs `PRAGMA wal_checkpoint(FULL)` and then `(TRUNCATE)` on the write
    /// connection, as one actor turn, and returns what SQLite reported. **Read
    /// [`CheckpointReport::busy`]** — a checkpoint that could not run is an
    /// `Ok` whose WAL is still there.
    ///
    /// Two passes rather than one because **a truncating checkpoint cannot
    /// report its own work**: the counts describe the WAL *after* the
    /// operation, and after a truncation there is nothing left to describe, so
    /// `TRUNCATE` alone answers `busy=0, log=0, checkpointed=0` on success —
    /// indistinguishable from having done nothing. `FULL` supplies the frame
    /// count and `TRUNCATE` resets the file; `busy` is the union of the two.
    ///
    /// # When a caller needs this
    ///
    /// Three cases, and only three:
    ///
    /// - **Before copying the database file elsewhere.** In WAL mode the `.db`
    ///   file alone is not the database; recent commits live in the `-wal`. A
    ///   complete checkpoint is what makes the main file self-contained.
    /// - **At the end of a bulk load that turned the automatic checkpointer
    ///   off.** That is the pairing this method exists for — see
    ///   [`Tuning::wal_autocheckpoint`]. Disabling autocheckpoint without
    ///   calling this leaves a WAL that grows for the life of the process.
    /// - **Before a long idle period**, to give back the disk.
    ///
    /// Nobody else should call it on a timer. SQLite checkpoints automatically
    /// every 1,000 pages and that default is not changed by this method
    /// existing; a periodic explicit checkpoint on top of it buys nothing and
    /// takes the write lock to do so.
    ///
    /// # It takes the write lock, and it is budget-exempt
    ///
    /// The hold is a function of how many frames have accumulated, which is a
    /// function of how long since the last checkpoint — not of anything passed
    /// in. It is on [`CHUNK_BUDGET`]'s exemption table for that reason, and it
    /// is the one entry there that is not a transaction: there is no smaller
    /// unit to chunk into, because the operation *is* the copy.
    pub async fn checkpoint(&self) -> Result<CheckpointReport> {
        self.high(|responder| HighPriCommand::Checkpoint { responder })
            .await
    }

    /// Rebuild `links_current` from `links` and verify zero drift (§5.8).
    ///
    /// One transaction holding the write lock for its whole duration, because
    /// [D-023] will not let the `DELETE` and the `INSERT` be split: a reader
    /// landing between them would see a graph with no edges and no error.
    /// [`Self::rebuild_current_chunked`] is the same result with a different
    /// latency profile, and is what a populated database wants.
    ///
    /// The report's `drift_after` is the audit run inside the same transaction,
    /// so a repair that did not converge is reported by the call that made it
    /// rather than by the next one to look.
    ///
    /// [D-023]: ../docs/architecture/s13-decision-register.md#d-023
    pub async fn rebuild_current(&self) -> Result<RebuildReport> {
        self.high(|responder| HighPriCommand::RebuildCurrent { responder })
            .await
    }

    /// Rebuild `links_current` beside itself, in chunks (§5.8, T1.2, D-082).
    ///
    /// Same result as [`Self::rebuild_current`], different latency profile.
    /// `rebuild_current` is one transaction holding the write lock for its whole
    /// duration, because D-023 will not let the `DELETE` and the `INSERT` be
    /// split: a reader landing between them sees a graph with no edges and no
    /// error. This builds the replacement in a shadow table instead — the live
    /// table stays live and trigger-maintained throughout — and swaps it in at
    /// the end.
    ///
    /// Each step is its own actor turn, so an interactive assertion can jump the
    /// queue between chunks. That is the whole of the improvement, and it is why
    /// the loop is here rather than inside the actor's arm (the same reasoning
    /// as [`Self::archive_windowed`] and [`Self::bulk_import`]).
    ///
    /// # What the swap still costs
    ///
    /// Not microseconds. Index names are global and SQLite has no `ALTER INDEX
    /// … RENAME`, so the shadow cannot be built carrying `links_current`'s index
    /// names while `links_current` still holds them — and building it under
    /// other names would leave the table permanently indexed under names absent
    /// from [`CREATE_INDICES`](crate::schema::ddl::CREATE_INDICES), so the next
    /// migration would create a second copy of each.
    /// `DROP TABLE` frees the names, so the swap transaction is where
    /// the three indexes get built. What the chunking moves off the lock is the
    /// **projection** — the window function over all of `links` — which is the
    /// O(E log E) term.
    ///
    /// # When this returns an error rather than a repair
    ///
    /// [`DbError::RebuildInterrupted`] means an archive committed while the
    /// shadow was being built. Its deletions are invisible to a catch-up pass
    /// keyed on `recorded_at` — a deleted row has no `recorded_at` left to find
    /// it by — so the work is discarded rather than swapped in. `links_current`
    /// is untouched and the call can simply be retried.
    ///
    /// Use [`Self::rebuild_current`] when the repair must be one atomic act, or
    /// when nothing else is contending for the actor and the extra turns are
    /// pure overhead.
    pub async fn rebuild_current_chunked(&self) -> Result<RebuildReport> {
        use crate::integrity::{ShadowOutcome, ShadowStep};

        // Each `else` arm is unreachable: the actor maps each step to its own
        // outcome variant. Written as a refutable pattern rather than an
        // `unwrap` so that adding a step cannot turn a mismatch into a panic on
        // the write path — and `WriterDroppedResponder` is the honest name for
        // "the actor answered with something this cannot use".
        let ShadowOutcome::Started { build_start, epoch } =
            self.shadow_step(ShadowStep::Begin).await?
        else {
            return Err(DbError::WriterDroppedResponder);
        };

        let mut after: Option<String> = None;
        loop {
            let ShadowOutcome::Filled { last } = self
                .shadow_step(ShadowStep::Fill {
                    after: after.take(),
                })
                .await?
            else {
                return Err(DbError::WriterDroppedResponder);
            };
            match last {
                Some(last) => after = Some(last),
                None => break,
            }
        }

        let ShadowOutcome::Swapped { rows } = self
            .shadow_step(ShadowStep::Swap { build_start, epoch })
            .await?
        else {
            return Err(DbError::WriterDroppedResponder);
        };

        Ok(RebuildReport {
            rows_rebuilt: rows,
            // Not audited. The chunked path's whole argument is that the
            // expensive work happens off the lock, and `audit_current` is two
            // `EXCEPT` passes over the projection — the cost D-077 removed from
            // the archive for the same reason. A caller who wants the check has
            // `audit_current` on the read connection, where it costs nobody the
            // write lock.
            drift_after: 0,
        })
    }

    /// Run one step of a chunked rebuild, for a caller doing its own scheduling.
    ///
    /// [`Self::rebuild_current_chunked`] is this in a loop and is what almost
    /// everyone wants. This exists because that loop offers no seam: it drives
    /// `Begin`, then `Fill` to exhaustion, then `Swap`, and a caller who needs to
    /// do something *between* steps — pace them against a frame budget, abandon
    /// a rebuild that has run long enough, or provoke the archive interlock in a
    /// test — cannot get in.
    ///
    /// The obligation that comes with it: `epoch` from
    /// [`ShadowOutcome::Started`](crate::integrity::ShadowOutcome) must be handed
    /// back to [`ShadowStep::Swap`](crate::integrity::ShadowStep), or the
    /// archive interlock is defeated and a stale projection can be swapped in.
    /// The looping version cannot get that wrong; this one can.
    pub async fn shadow_step(
        &self,
        step: crate::integrity::ShadowStep,
    ) -> Result<crate::integrity::ShadowOutcome> {
        self.low(|responder| LowPriCommand::ShadowRebuild { step, responder })
            .await
    }

    /// Import edges on the background channel, chunked (D-011).
    ///
    /// Atomic *per chunk*, not overall: a failure partway leaves earlier chunks
    /// committed. That is the tradeoff [`chunk_rows`] documents — use
    /// [`Database::write_bulk_atomic`] when the batch must be all-or-nothing.
    ///
    /// Chunked adaptively, at most [`chunk_rows::EDGES`] rows at a time: that
    /// constant is where the loop starts and the largest chunk it will send, and
    /// each chunk's measured hold sizes the next against [`CHUNK_BUDGET`]. It is
    /// also faster in total than the larger chunks this used through 0.5.5
    /// (D-058).
    ///
    /// A consequence worth planning for: the chunk boundaries — and so the
    /// `recorded_at` stamps this import writes — depend on how fast the machine
    /// was, not only on how many edges were passed (§5.1.6).
    ///
    /// # The WAL during a bulk (measured, 0.16.1 — see D-274's diagnostics)
    ///
    /// Every chunk's commit runs WAL pages back into the database file when
    /// SQLite's own autocheckpoint threshold (1,000 pages) is reached, and on a
    /// bulk that fires every few chunks. Measured on the 16,000-edge random-pair
    /// ladder (medians of 3, `benchmarks/diagnostics/`): 5.0 s with the default,
    /// 3.7 s with the threshold at 10,000 pages (`wal_autocheckpoint = 10_000`
    /// at open), 3.6 s with it disabled outright — and the checkpoint the bulk
    /// was paying is 240 ms once at the end instead. The WAL is the price of the
    /// third: ~554 MB for that fixture against 43 MB at the 10,000-page
    /// threshold and ~6 MB at the default, so the 10,000-page setting is the
    /// recipe unless the disk is known to be large. On an ordered (chain-shaped)
    /// import the sweep is a wash — the recipe is for the random-order shape a
    /// real importer produces. The knob is per-handle and set at
    /// [`Database::open`](crate::Database::open); call
    /// [`Self::checkpoint`] once after a disabled-checkpoint bulk.
    ///
    /// Returns [`BulkInterrupted`] rather than [`DbError`] on failure, because
    /// a path that is not all-or-nothing owes its caller the count of what
    /// landed (0.13.8, W7.6). `?` into a `Result<_, DbError>` still compiles
    /// and drops the count, which is the caller's decision to take.
    ///
    /// [`Self::bulk_import_with`] adds cancellation and per-chunk progress.
    pub async fn bulk_import(&self, edges: Vec<EdgeAssertion>) -> BulkResult<usize> {
        self.bulk_import_with(edges, BulkControl::new()).await
    }

    /// [`Self::bulk_import`] with cancellation and progress (0.13.8, W7.6,
    /// D-181).
    ///
    /// The chunk boundaries this path already has are what make both possible:
    /// the loop is between transactions several times a second, which is where
    /// a token can be read and a callback run without holding anything.
    pub async fn bulk_import_with(
        &self,
        edges: Vec<EdgeAssertion>,
        control: BulkControl,
    ) -> BulkResult<usize> {
        let edges = normalize_all(edges).map_err(before_any_chunk)?;
        self.low_chunked(edges, chunk_rows::EDGES, control, |chunk, responder| {
            LowPriCommand::BulkImportChunk { chunk, responder }
        })
        .await
    }

    /// Load edges **without the maintained projection in the way**, then
    /// re-derive it in one chunked rebuild (D-277, plan §8.1 / F1).
    ///
    /// Three phases through the actor: drop `trg_links_current_sync` (the
    /// per-row upsert into `links_current`), load every edge through the same
    /// chunked path [`Self::bulk_import`] uses, restore the trigger, then
    /// [`Self::rebuild_current_chunked`] — the re-derivation D-082 made
    /// chunked, whose fill costs ~104 ms per 16k rows and whose swap is the
    /// exempt 46.8 ms turn. Measured on the reference box, 16,000 random-pair
    /// edges (medians of three):
    ///
    /// | arm | bulk | rebuild + restore | total |
    /// |---|---|---|---|
    /// | shipped (`bulk_import`) | 5.08 s | — | 5.08 s |
    /// | this | 2.29 s | 0.19 s | **2.48 s (2.05×)** |
    ///
    /// The mirror is what the ladder showed growing with the graph on random
    /// pairs (plan §9.2): the near-chain shape pays it too, just flatter, so
    /// this is the lever for both — and the rebuild's cost is a function of
    /// the table, not the shape, which is why the skip arm's total is flat
    /// where the shipped arm's is not.
    ///
    /// # What the window actually touches, stated precisely
    ///
    /// **The ledger is complete at every instant of the window.** The log
    /// mirror and the single-open guard stay up: `links` and
    /// `transaction_log` gain every row this call loads, versioned and
    /// guarded exactly as the shipped path does. What lags is
    /// `links_current` — Doctrine VI's derivative state, which the crate
    /// has always maintained as *the projection you could fold from the
    /// ledger*. Current-time reads during the window (`traverse`,
    /// `query_as_of_edges` without an instant) see a partial projection and
    /// are not wrong about the past — they are stale about the present, the
    /// one state Doctrine VI calls disposable.
    ///
    /// The window **ends when this method returns**, success, failure, or
    /// cancellation: the restore runs before the rebuild, so any write that
    /// lands after it mirrors again, and the rebuild then sweeps the backlog.
    /// If the future is dropped or the process unwinds mid-load, the window
    /// stays open — the projection stays stale but the ledger stays complete,
    /// `audit_current` reports the drift as [`DbError::CurrentDrift`], and
    /// [`Self::rebuild_current_chunked`] closes it. That is the honest doc
    /// entry plan §8.1 asked for, and why this is a signature rather than a
    /// default: the shipped path never has a window at all.
    ///
    /// # What a failure does
    ///
    /// The load's failure — [`BulkInterrupted`] with its `written` count —
    /// propagates only **after** the trigger is restored and the projection
    /// rebuilt from what did commit, so a failed deferred bulk is in exactly
    /// the state a caller expects: the prefix is committed, the projection is
    /// true, the mirror is on. An empty `edges` is a no-op: no DDL, no
    /// rebuild, nothing attributed.
    pub async fn bulk_import_deferred(
        &self,
        edges: Vec<EdgeAssertion>,
    ) -> BulkResult<usize> {
        self.bulk_import_deferred_with(edges, BulkControl::new())
            .await
    }

    /// [`Self::bulk_import_deferred`] with cancellation and progress, on the
    /// load half — the same chunked loop, the same token, the same callbacks
    /// [`Self::bulk_import_with`] takes. The toggle and the rebuild are
    /// single turns around it.
    pub async fn bulk_import_deferred_with(
        &self,
        edges: Vec<EdgeAssertion>,
        control: BulkControl,
    ) -> BulkResult<usize> {
        let edges = normalize_all(edges).map_err(before_any_chunk)?;
        if edges.is_empty() {
            return Ok(0);
        }

        if let Err(cause) = self
            .low(|responder| LowPriCommand::LinksCurrentMirror {
                present: false,
                responder,
            })
            .await
        {
            // A toggle that failed has left the mirror exactly where it was —
            // on. Stopping here is safe: nothing was loaded, nothing is stale.
            return Err(BulkInterrupted { written: 0, cause });
        }

        let loaded = self
            .low_chunked(edges, chunk_rows::EDGES, control, |chunk, responder| {
                LowPriCommand::BulkImportChunk { chunk, responder }
            })
            .await;

        // Restore before reporting, and before rebuilding: any write that
        // lands after the restore maintains the projection itself, and the
        // rebuild then sweeps whatever the window left behind.
        let restored = self
            .low(|responder| LowPriCommand::LinksCurrentMirror {
                present: true,
                responder,
            })
            .await;
        let rebuilt = self.rebuild_current_chunked().await;

        // The mirror's absence out-ranks the rebuild's failure out-ranks the
        // load's: a caller must not read success beside a mirror that is off,
        // and a load error beside a restored-and-rebuilt projection is just
        // the load's answer.
        let written = match &loaded {
            Ok(n) => *n,
            Err(e) => e.written,
        };
        match (loaded, restored, rebuilt) {
            (Ok(n), Ok(()), Ok(_)) => Ok(n),
            // A failed load with the mirror restored and the projection
            // rebuilt is the load's own answer — the prefix committed, the
            // projection is true, the mirror is on.
            (Err(e), Ok(()), Ok(_)) => Err(e),
            (_, Err(cause), _) => Err(BulkInterrupted { written, cause }),
            (Err(e), Ok(()), Err(report)) => Err(BulkInterrupted {
                written: e.written,
                cause: report,
            }),
            (Ok(n), Ok(()), Err(report)) => Err(BulkInterrupted {
                written: n,
                cause: report,
            }),
        }
    }

    /// Upsert many **concepts** on the background channel, chunked (D-011).
    ///
    /// This is the bulk concept path, and every row it writes is a ledger write:
    /// it versions the concept and lands in `transaction_log`. Derived analytics
    /// output does not belong here — see
    /// [`Database::write_analytics_annotations`] and D-041.
    ///
    /// Called `write_annotations` through 0.5.6, from when the two writes were
    /// one call. D-041 split them and the name stayed on the wrong one for three
    /// releases, so the crate had a `write_annotations` that wrote concepts
    /// sitting beside a `write_analytics_annotations` that wrote annotations
    /// (D-075).
    ///
    /// Chunked, so it returns [`BulkInterrupted`] and its `written` count on
    /// failure (0.13.8, W7.6); [`Self::write_concepts_with`] adds cancellation
    /// and progress.
    pub async fn write_concepts(&self, concepts: Vec<ConceptUpsert>) -> BulkResult<usize> {
        self.write_concepts_with(concepts, BulkControl::new()).await
    }

    /// [`Self::write_concepts`] with cancellation and progress (0.13.8, W7.6).
    pub async fn write_concepts_with(
        &self,
        concepts: Vec<ConceptUpsert>,
        control: BulkControl,
    ) -> BulkResult<usize> {
        let concepts: Vec<ConceptUpsert> = concepts
            .into_iter()
            .map(ConceptUpsert::normalized)
            .collect::<Result<_>>()
            .map_err(before_any_chunk)?;
        self.low_chunked(
            concepts,
            chunk_rows::CONCEPTS,
            control,
            |chunk, responder| LowPriCommand::WriteConceptsChunk { chunk, responder },
        )
        .await
    }

    /// State as believed at `ts` (§5.5, D-026, D-049).
    ///
    /// A read: it runs on `read_conn` and never touches the Write Actor, so a
    /// reconstruction and a full-speed write-back do not slow each other.
    ///
    /// Prefer this to calling [`crate::temporal::reconstruct`] directly. The
    /// free function takes the archive path and the snapshot directory as
    /// arguments, and a caller who passes `None` for the second gets a correct
    /// answer that folds the whole log every time — the composition is opt-in
    /// at that layer and easy to leave off by accident. Here both come from the
    /// handle, so the fast path is the default one.
    pub async fn reconstruct(&self, ts: &str) -> Result<crate::temporal::MaterializedState> {
        let ts = timestamp::normalize(ts)?;
        crate::temporal::reconstruct(
            &self.read_conn,
            &ts,
            Some(&self.archive_path),
            Some(&self.snapshots_dir),
        )
        .await
    }

    /// State at `ts` as `branch` saw it (0.15.17, [D-259], review C-10).
    ///
    /// [`Self::reconstruct`] with the ancestry resolved: each ancestor bounded
    /// at its fork point, one belief per edge key from the nearest lineage
    /// holding it. See [`crate::temporal::reconstruct_on`] for how it is
    /// assembled, what it costs, and the two things it does **not** do —
    /// concepts are not resolved by lineage, and the result must not be saved
    /// as a snapshot.
    ///
    /// A read, on `read_conn`, like [`Self::reconstruct`]. The archive path and
    /// the snapshot directory come from the handle, so snapshot composition is
    /// on by default for each of the folds this runs.
    ///
    /// [D-259]: ../../docs/architecture/s13-decision-register.md#d-259
    pub async fn reconstruct_on(
        &self,
        ts: &str,
        branch: &str,
    ) -> Result<crate::temporal::MaterializedState> {
        let ts = timestamp::normalize(ts)?;
        crate::temporal::reconstruct_on(
            &self.read_conn,
            &ts,
            branch,
            Some(&self.archive_path),
            Some(&self.snapshots_dir),
        )
        .await
    }

    /// `branch`'s ancestry, nearest first, each with its fork-point cutoff.
    ///
    /// The input [`crate::temporal::resolve_beliefs`] takes. Resolved from
    /// `branches` in Rust since 0.15.17 ([D-259]) — the walk is a few
    /// microseconds and the table is tiny and append-only, so this is a read
    /// like any other rather than something to cache.
    ///
    /// The trunk of an unforked database answers with one row and no cutoff,
    /// which is its true ancestry. A lineage that is not registered is refused
    /// by name with [`DbError::UnknownBranch`].
    ///
    /// [D-259]: ../../docs/architecture/s13-decision-register.md#d-259
    pub async fn ancestry(&self, branch: &str) -> Result<Vec<crate::branch::Ancestor>> {
        let lineages = crate::graph::lineage::Lineages::load(&self.read_conn).await?;
        // Checked before it is walked: `resolve` answers for a name it has never
        // seen with a one-row ancestry, which is the right answer for a root and
        // the D-069 wrong-looking-right answer for a typo.
        lineages.shape(branch)?;
        Ok(lineages.ancestry(branch))
    }

    /// Every edge one [`ReadPlan`] names (0.15.9, W13.4, [D-251]).
    ///
    /// The whole projection filtered to the plan's instants and lineage —
    /// topology only, no start node, and no budget on the answer. On a large
    /// ledger that is a large `Vec`; [`Self::load_subgraph`] is the bounded
    /// neighbourhood read and [`crate::graph::TraversalBuilder`] is the
    /// anchored one.
    ///
    /// # What this can express that nothing else could
    ///
    /// [`crate::temporal::query_as_of_edges_on`] is the same read at a
    /// valid-time instant, and it takes no transaction-time one: before this
    /// release, *"which edges did we believe existed, as of March, as they
    /// stood in January"* had exactly two answers available — walk it from a
    /// start node, or fold the entire log with [`Self::reconstruct`] and filter
    /// the result. The first needs an anchor the question does not have and the
    /// second is a different order of work. The fold this uses is the
    /// traversal's own, so the bitemporal cell is now readable whole at the
    /// cost of reading it.
    ///
    /// The two functions share one statement, which is why neither can drift
    /// from the other; `query_as_of_edges_on` is this with `recorded` unset and
    /// the lineage dropped from each row.
    ///
    /// # Errors
    ///
    /// [`DbError::UnknownBranch`] naming a
    /// lineage that was never registered — refused rather than answered for the
    /// trunk, for `graph::lineage::Lineages::shape`'s reason.
    /// [`DbError::RecordedInstantUnreachable`]
    /// when [`ReadPlan::recorded`] is below what the hot log still covers
    /// ([D-247](../../docs/architecture/s13-decision-register.md#d-247)).
    /// [`DbError::InvalidTimestamp`] for a
    /// stamp that is not canonical, from the same normaliser every other read
    /// uses — a plan is inert and validates nothing, so this is where a
    /// malformed instant is noticed.
    ///
    /// [D-251]: ../../docs/architecture/s13-decision-register.md#d-251
    pub async fn edges(&self, plan: ReadPlan) -> Result<Vec<crate::temporal::EdgeBelief>> {
        // `None` is now, and now is this handle's clock rather than the
        // system's: a database opened on a `FakeClock` reads at the instant it
        // is writing at, which is the whole reason the clock is a handle
        // property (§5.1.1).
        let valid = match plan.valid.as_deref() {
            Some(ts) => timestamp::normalize(ts)?,
            None => self.clock.now(),
        };
        let recorded = plan
            .recorded
            .as_deref()
            .map(timestamp::normalize)
            .transpose()?;
        crate::plan::edges_at(
            &self.read_conn,
            &valid,
            recorded.as_deref(),
            plan.branch_name(),
            plan.limit,
        )
        .await
    }

    /// Create a model's embedding table and DiskANN index (§5.9, D-048).
    ///
    /// Idempotent: registering a model that already exists at the same
    /// dimension succeeds, and at a different dimension fails with
    /// [`DbError::DimMismatch`] naming both, rather than no-opping through
    /// `IF NOT EXISTS` and leaving the caller believing the dimension they
    /// asked for is the one in force.
    ///
    /// This issues DDL, which everywhere else in the crate is the migration
    /// runner's exclusive business (D-032). The exception is bounded and
    /// deliberate: a model's table is created once, by an explicit call, and
    /// the alternative — a caller-supplied write connection — is the very thing
    /// the Write Actor exists to make impossible.
    ///
    /// # Latency
    ///
    /// One small transaction, but it queues like any other write: see §5.1.8.
    pub async fn register_model(&self, model: &ModelName, dim: usize) -> Result<()> {
        let model = model.clone();
        self.high(|responder| HighPriCommand::RegisterModel {
            model,
            dim,
            responder,
        })
        .await
    }

    /// Store or replace vectors for `model`, chunked (§5.9, D-011, D-048).
    ///
    /// The write path for embeddings. Before 0.5.4 there was none:
    /// [`crate::vector::upsert_embedding`] takes a raw connection, `read_conn`
    /// is `query_only`, and the write connection lives inside the actor — so an
    /// application could search vectors it had no way to store.
    ///
    /// Low priority and chunked at [`chunk_rows::EMBEDDINGS`], because embedding
    /// is bulk derived work: a 50,000-vector backfill must yield to an
    /// interactive assertion at every chunk boundary. That constant is the
    /// smallest of the four by a wide margin — DiskANN index maintenance makes an
    /// embedding the most expensive row in the system (D-058). Atomic per chunk, not overall, which
    /// is the same trade [`Database::bulk_import`] makes and is safer here than
    /// there — an embedding is derived (Doctrine VII), so a partially written
    /// batch is recoverable by re-embedding.
    ///
    /// Fails with [`DbError::ModelNotRegistered`] if `model` has no table, and
    /// [`DbError::DimMismatch`] if a vector's length is not the declared
    /// dimension. The dimension is read from the schema once per chunk (D-037):
    /// the crate keeps no registry of its own to fall out of date.
    ///
    /// Chunked, so it returns [`BulkInterrupted`] and its `written` count on
    /// failure (0.13.8, W7.6). A 50,000-vector backfill is the longest-running
    /// write the crate has, which makes it the one most likely to be cancelled
    /// — [`Self::upsert_embeddings_with`] is how.
    pub async fn upsert_embeddings(
        &self,
        model: &ModelName,
        rows: Vec<(String, Vec<f32>)>,
    ) -> BulkResult<usize> {
        self.upsert_embeddings_with(model, rows, BulkControl::new())
            .await
    }

    /// [`Self::upsert_embeddings`] with cancellation and progress (0.13.8,
    /// W7.6).
    pub async fn upsert_embeddings_with(
        &self,
        model: &ModelName,
        rows: Vec<(String, Vec<f32>)>,
        control: BulkControl,
    ) -> BulkResult<usize> {
        self.low_chunked(rows, chunk_rows::EMBEDDINGS, control, |chunk, responder| {
            LowPriCommand::UpsertEmbeddingChunk {
                model: model.clone(),
                chunk,
                responder,
            }
        })
        .await
    }

    /// Load embeddings **without the DiskANN index in the way**, then rebuild
    /// it in one pass (D-276, plan §9.1).
    ///
    /// Three actor turns in sequence: drop the index (`DROP INDEX IF EXISTS`,
    /// µs-scale), load every row through the same chunked path
    /// [`Self::upsert_embeddings`] uses, rebuild the index in one statement.
    /// Measured on the reference box, medians of three, 2,000 vectors:
    ///
    /// | dim | indexed (`upsert_embeddings`) | `bulk_embeddings` | one-pass build alone |
    /// |---|---|---|---|
    /// | 64 | 3.78 s | 2.62 s | 2.61 s |
    /// | 256 | 31.0 s | 19.7 s | 19.7 s |
    /// | 512 | 56.0 s | 39.0 s | 39.0 s |
    ///
    /// At 5,000 × 256: 89.2 s indexed, 48.6 s here — **1.8×**. The build, not
    /// the blob writes, is the cost: inserts without the index are flat at
    /// ~5 µs/row at every dimension measured. The rest of the gap against an
    /// HNSW writer is DiskANN-vs-HNSW build economics inside libSQL, engine-
    /// side, and out of scope for a crate that does not fork its engine.
    ///
    /// # The trade, stated
    ///
    /// Between the drop and the rebuild, the model's vectors are not
    /// searchable (`vector_top_k` reports the missing index) and **not
    /// dimension-checked at the storage layer** — the index is that check
    /// (D-037, `ddl::create_embeddings_index`). The crate-side check
    /// ([`crate::vector::EmbeddingCodec::encode`]) still applies to everything
    /// this method loads, and the rebuild restores the storage check at the
    /// end; what is given up is the check on rows a *different* client inserts
    /// during the window. That is the measured decision, opt-in by signature:
    /// the plain path keeps the storage backstop at every instant.
    ///
    /// **A failed or cancelled load still rebuilds.** The load's own failure —
    /// [`BulkInterrupted`] with its `written` count — propagates only after
    /// the rebuild has run, so a bulk load can never leave the file in the
    /// disarmed state. An empty `rows` is a no-op: the index is not touched
    /// for a load of nothing.
    ///
    /// The rebuild turn is budget-exempt (atomic by necessity, same criterion
    /// as `shadow_swap` — [`crate::CHUNK_BUDGET`]'s table), and its hold grows
    /// with the corpus: ~10 ms/vector at dim 256, ~20 ms at 512. For a very
    /// large backfill that hold is the price of the recipe; the alternative —
    /// keep the index and pay the same total spread across rows — is what
    /// `upsert_embeddings` already is.
    pub async fn bulk_embeddings(
        &self,
        model: &ModelName,
        rows: Vec<(String, Vec<f32>)>,
    ) -> BulkResult<usize> {
        self.bulk_embeddings_with(model, rows, BulkControl::new())
            .await
    }

    /// [`Self::bulk_embeddings`] with cancellation and progress, on the load
    /// half. The drop and the rebuild are single statements; `progress` and
    /// `cancel` bound the chunked load between them, exactly as they do for
    /// [`Self::upsert_embeddings_with`].
    pub async fn bulk_embeddings_with(
        &self,
        model: &ModelName,
        rows: Vec<(String, Vec<f32>)>,
        control: BulkControl,
    ) -> BulkResult<usize> {
        // The table must exist before anything drops its index, and the first
        // row's width is checked *before* the drop so a wholly wrong batch
        // fails without ever disarming the storage check.
        let dim = match crate::vector::declared_dimension(self.read_conn(), model).await {
            Ok(dim) => dim,
            Err(cause) => return Err(BulkInterrupted { written: 0, cause }),
        };
        if let Some((_, first)) = rows.first() {
            if first.len() != dim {
                return Err(BulkInterrupted {
                    written: 0,
                    cause: DbError::DimMismatch {
                        got: first.len(),
                        expected: dim,
                        model: model.to_string(),
                    },
                });
            }
        } else {
            return Ok(0);
        }

        if let Err(cause) = self
            .low(move |responder| LowPriCommand::DropEmbeddingIndex {
                model: model.clone(),
                responder,
            })
            .await
        {
            // A drop that failed has left the index in place — the armed,
            // searchable state — so stopping here is safe: the load never
            // ran, and the file is exactly what it was.
            return Err(BulkInterrupted { written: 0, cause });
        }

        let loaded = self
            .low_chunked(rows, chunk_rows::EMBEDDINGS, control, |chunk, responder| {
                LowPriCommand::UpsertEmbeddingChunk {
                    model: model.clone(),
                    chunk,
                    responder,
                }
            })
            .await;

        // Whether the load committed or was interrupted, the index goes back:
        // the one state this method must never leave the file in is the state
        // `drop_embedding_index` documents as unsearchable and unchecked.
        let rebuilt = self
            .low(move |responder| LowPriCommand::RebuildEmbeddingIndex {
                model: model.clone(),
                responder,
            })
            .await;

        // The load's failure is the caller's answer; the rebuild's failure is
        // worse and takes precedence, because a load that reports success
        // beside a missing index is the silent half of the same defect. In the
        // double-failure case the rebuild becomes the cause and the load's
        // `written` count survives as the count.
        match (loaded, rebuilt) {
            (Ok(count), Ok(())) => Ok(count),
            (Err(e), Ok(())) => Err(e),
            (Ok(count), Err(e)) => Err(BulkInterrupted {
                written: count,
                cause: e,
            }),
            (Err(load), Err(rebuild)) => Err(BulkInterrupted {
                written: load.written,
                cause: rebuild,
            }),
        }
    }

    /// Reconstruct the concept-text search index from the ledger (§5.9, D-036).
    ///
    /// The FTS index is derivative: D-036 promises every derivative table can be
    /// rebuilt from the ledger tables, and this is that promise made callable
    /// for `concepts_fts`. Needed after a restore that skipped the shadow
    /// tables, or if the index is ever suspected of drifting from the text —
    /// and, as a matter of policy, cheaper to run than to reason about.
    ///
    /// The work is `INSERT INTO concepts_fts(concepts_fts) VALUES('rebuild')`,
    /// which is FTS5's own operation over the content table, so this is not a
    /// second implementation of the sync triggers that could disagree with them.
    pub async fn rebuild_fts(&self) -> Result<()> {
        self.low(|responder| LowPriCommand::RebuildFts { responder })
            .await
    }

    /// Refresh the query planner's statistics (0.12.4, [D-149]).
    ///
    /// Runs `ANALYZE`, which writes `sqlite_stat1`. **Before 0.12.4 nothing in
    /// this crate ever did**, so the planner costed every query against SQLite's
    /// built-in defaults — assume ~1M rows, assume each bound equality column
    /// divides by ten. That estimate is structural: it depends on how many
    /// columns a query binds, not on what the table contains.
    ///
    /// Which is this schema's own worst defect restated. D-042, D-059 and D-064
    /// are three occasions where *a covering index captured a query because it
    /// contained the columns, not because it discriminated*, and two of the four
    /// declared indices lead on the same column. Statistics are what let the
    /// planner tell them apart by measurement instead of by shape.
    ///
    /// # Cost, and why it is bounded
    ///
    /// This is a write and it takes the write lock. `PRAGMA analysis_limit`
    /// (set per connection, see [`ddl::ANALYSIS_LIMIT`]) caps the rows examined
    /// per index. It is scheduled as low-priority work and will not preempt an
    /// interactive assertion.
    ///
    /// **The bound is a constant factor, not an independence** (0.12.23,
    /// D-166). This rustdoc said the hold "scales with the number of indices —
    /// four — and not with the size of `links_current`", which is measurably
    /// wrong: the pragma is worth 3–4× and what remains still grows with the
    /// table. Measured, `examples/analyze_hold.rs`: **5.26 ms at 10,000 edges,
    /// 19.1 ms at 40,000**, against a 3 ms [`crate::CHUNK_BUDGET`].
    ///
    /// So this call **misses the budget by ~6× on a moderately sized ledger**,
    /// and [`crate::metrics::CommandKind::Analyze`] is deliberately not among
    /// the budget-exempt kinds — `metrics().budget_violations()` names it. That
    /// is the honest position: the work is low priority and preemptible between
    /// commands, but it is one indivisible statement and cannot be chunked, so
    /// the hold is what it is. Prefer [`optimize`], which does nothing when
    /// nothing has moved.
    ///
    /// **Since 0.13.24 the counter is this call and not also [`optimize`]**
    /// (W10.5, [D-197]). The two shared `CommandKind::Analyze` until then, which
    /// is why an `analyze` row in `budget_violations()` used to be unreadable:
    /// it could have been an explicit call or a handle close.
    ///
    /// # When to call it
    ///
    /// After a bulk import, and after anything that changes a table's shape by
    /// an order of magnitude. Prefer [`optimize`] for routine upkeep: it does
    /// nothing when nothing has moved, and this does the work unconditionally.
    ///
    /// Statistics are derived state in the sense Doctrine VI means it — deleting
    /// `sqlite_stat1` costs plan quality and no information, and this call
    /// rebuilds it.
    ///
    /// [D-149]: ../docs/architecture/s13-decision-register.md#d-149
    /// [`ddl::ANALYSIS_LIMIT`]: crate::schema::ddl::ANALYSIS_LIMIT
    /// [`optimize`]: Database::optimize
    pub async fn analyze(&self) -> Result<()> {
        self.low(|responder| LowPriCommand::Analyze {
            incremental: false,
            responder,
        })
        .await
    }

    /// Re-analyse only what has gone stale (0.12.4, [D-149]).
    ///
    /// `PRAGMA optimize`. SQLite tracks how far each table has drifted since its
    /// last analysis and re-analyses only where it believes the statistics no
    /// longer hold — so this is a no-op on an idle database and the full cost of
    /// [`analyze`] on one that has changed completely.
    ///
    /// That property is the whole point: it is safe to call on a schedule, where
    /// [`analyze`] is not. `close()` runs it, so a process that opens, works and
    /// closes keeps its statistics current without anybody arranging it.
    ///
    /// # What it costs, measured, and the threshold it applies rather than takes
    /// (0.13.24, W10.5, [D-197])
    ///
    /// `examples/optimize_hold.rs`, on a 40,000-edge ledger: **10.7 ms the
    /// first time on a database that has never been analysed** — there is
    /// nothing incremental about the first call — and **90–220 µs every time
    /// after**, well inside [`crate::CHUNK_BUDGET`].
    ///
    /// **The staleness test is SQLite's and it is a ratio, not a row count.**
    /// Measured by reading `sqlite_stat1` across the call rather than by timing
    /// it: growth of 2× and 5× both left the statistics **untouched**, and
    /// only at 25× did it re-analyse — for a 460 ms hold. So this is not a
    /// cheaper `analyze()` and calling it after a bulk load is not a way to
    /// refresh statistics the load invalidated: below the ratio it declines,
    /// and above it it costs what [`analyze`] costs. It reports as
    /// [`crate::metrics::CommandKind::Optimize`] since 0.13.24, which is what
    /// makes those two outcomes distinguishable in the metrics at all.
    ///
    /// [D-197]: ../docs/architecture/s13-decision-register.md#d-197
    ///
    /// [D-149]: ../docs/architecture/s13-decision-register.md#d-149
    /// [`analyze`]: Database::analyze
    pub async fn optimize(&self) -> Result<()> {
        self.low(|responder| LowPriCommand::Analyze {
            incremental: true,
            responder,
        })
        .await
    }

    // **There is deliberately no `verify_fts()` (§5.9, D-071).**
    //
    // `rebuild_fts` is the repair with no way to ask whether it is needed, and
    // Wave 5 set out to add the missing half. FTS5 offers `'integrity-check'`,
    // which looked like exactly the engine-provided answer this crate prefers.
    // It is not: on libSQL 0.9.30 it verifies the index's *internal* consistency
    // and not its agreement with the content table. Measured — after
    // `'delete-all'` the index matches nothing where it matched ten rows, and
    // both `'integrity-check'` and `'integrity-check', 0` still report success.
    //
    // A `verify_fts()` on that footing would answer "healthy" for an empty
    // index, which is worse than having no method at all: it is the shape of
    // defect AC, a function that looks like it checks something and does not.
    // `an_emptied_fts_index_still_passes_integrity_check` pins the limitation so
    // that if a later libSQL fixes it, the test fails and says so.

    /// Write derived analytics results on the background channel, chunked
    /// (§5.4, D-041).
    ///
    /// Rows go to `analytics_annotations`, which has no log trigger, so nothing
    /// written here reaches `transaction_log` and nothing here versions a
    /// concept. Rerunning an algorithm replaces the previous pass rather than
    /// recording that the world changed.
    ///
    /// Low priority and chunked at up to [`chunk_rows::ANNOTATIONS`] — the
    /// largest ceiling of the four, because this is the only bulk table carrying
    /// no triggers at all
    /// and its rows are correspondingly cheap (D-058) — so a 50,000-label Louvain
    /// save yields to interactive writes at every chunk boundary and carries the
    /// per-chunk fidelity boundary of §5.1.6 — a partially written pass is
    /// recoverable by rerunning, which is the property that makes derived state
    /// safe to write this way and assertions not.
    ///
    /// Chunked, so it returns [`BulkInterrupted`] and its `written` count on
    /// failure (0.13.8, W7.6); [`Self::write_analytics_annotations_with`] adds
    /// cancellation and progress.
    pub async fn write_analytics_annotations(
        &self,
        annotations: Vec<Annotation>,
    ) -> BulkResult<usize> {
        self.write_analytics_annotations_with(annotations, BulkControl::new())
            .await
    }

    /// [`Self::write_analytics_annotations`] with cancellation and progress
    /// (0.13.8, W7.6).
    pub async fn write_analytics_annotations_with(
        &self,
        annotations: Vec<Annotation>,
        control: BulkControl,
    ) -> BulkResult<usize> {
        self.low_chunked(
            annotations,
            chunk_rows::ANNOTATIONS,
            control,
            |chunk, responder| LowPriCommand::WriteAnalyticsChunk { chunk, responder },
        )
        .await
    }

    /// Move closed intervals and superseded log rows older than `cutoff` to the
    /// cold database (§5.7, D-012).
    pub async fn archive(&self, cutoff: &str) -> Result<ArchiveReport> {
        let cutoff = timestamp::normalize(cutoff)?;
        let archive_path = self.archive_path.clone();
        self.low(|responder| LowPriCommand::Archive {
            cutoff,
            archive_path,
            responder,
        })
        .await
    }

    /// Forget one lineage: move its whole ledger to the cold database and
    /// remove the lineage record (0.14.13, §15.4, D-230).
    ///
    /// The abandonment arm. A conversation tree discards most of what it grows,
    /// and [`Self::archive`] cannot reclaim it: that arm is indexed by *time*,
    /// so archiving an abandoned branch's recent history means archiving the
    /// trunk's recent history with it.
    ///
    /// **Everything the lineage holds moves in one transaction** — its `links`,
    /// its `concepts`, its `transaction_log` entries and its `branches` row —
    /// and afterwards the name is unknown: every read and write naming it
    /// raises [`DbError::UnknownBranch`]. That is the design's whole shape, and
    /// `temporal::archive::archive_branch` records why it has no smaller
    /// version.
    ///
    /// # It refuses more than it accepts, on purpose
    ///
    /// - The trunk, and a name that is not registered
    ///   ([`DbError::UnknownBranch`]).
    /// - A branch with **descendants**: they read through it, so archiving it
    ///   would delete rows they still believe.
    /// - A branch whose **concepts another lineage's hot link names**. The road
    ///   map assumed an abandoned branch's rows were "a contiguous archivable
    ///   set by construction"; a concept is keyed by identity across the whole
    ///   ledger (D-214), so they are not, and this refusal is what makes them
    ///   contiguous in the cases it accepts.
    ///
    /// All but the first return [`DbError::BranchNotArchivable`] with a reason.
    ///
    /// The lineage record lands in `cold.branches` with an `archived_at`, so a
    /// cold row's `branch_id` still resolves to something — in the cold file,
    /// which is now the only place it does.
    pub async fn archive_branch(&self, branch: crate::branch::BranchId) -> Result<ArchiveReport> {
        let branch = branch.as_str().to_string();
        let archive_path = self.archive_path.clone();
        self.low(|responder| LowPriCommand::ArchiveBranch {
            branch,
            archive_path,
            responder,
        })
        .await
    }

    /// Move the named concepts back from the cold database into the hot tables
    /// (§2.3, C3).
    ///
    /// Rehydration is a **physical move back, not a write**: it mints no
    /// transaction-time facts and is invisible to both clocks. An id that is not
    /// in the cold file is skipped rather than being an error — the caller
    /// generally has a list from a cold-side query, and a partially-stale list is
    /// the normal case rather than a mistake. The report says how many actually
    /// moved.
    ///
    /// See [`RehydrateReport::rowids_reassigned`] for the one way a rehydrated
    /// row can differ from the row that was archived.
    pub async fn rehydrate(&self, ids: &[&str]) -> Result<RehydrateReport> {
        let ids: Vec<String> = ids.iter().map(|s| (*s).to_string()).collect();
        let archive_path = self.archive_path.clone();
        self.low(|responder| LowPriCommand::Rehydrate {
            ids,
            archive_path,
            responder,
        })
        .await
    }

    /// Archive up to `cutoff` as a sequence of sessions, each covering at most
    /// `window` of **transaction** time (T1.1, D-080).
    ///
    /// `archive(cutoff)` is one transaction whose size is set by how long it has
    /// been since the last one, which makes it the least bounded of the three
    /// operations exempt from [`CHUNK_BUDGET`] — its hold is a function of
    /// operational history rather than of anything a caller chose. This runs the
    /// same work as *N* complete sessions, each with its own marker, horizon row
    /// and rebuild, and returns one [`ArchiveReport`] per session in order.
    ///
    /// # D-012 is satisfied per session, and that is what it requires
    ///
    /// The atomicity D-012 demands is that copy-then-delete never be split — a
    /// crash between the phases duplicates or loses rows. *N* small sessions
    /// satisfy that exactly as one large one does. The obligation windowing adds
    /// is that a partial run leave a coherent intermediate state, which it does:
    /// each session commits a valid horizon, so a failure at window *k* leaves a
    /// database archived up to boundary *k−1* and nothing in between. **The
    /// sequence is not atomic and does not claim to be** — on error, the reports
    /// for the sessions that did commit are lost with it, but their effect is
    /// not, and re-running with the same `cutoff` completes the job.
    ///
    /// # Each session is its own actor turn, and that is the entire point
    ///
    /// This loop lives here, on the handle, rather than inside the actor's
    /// `Archive` arm. Putting it there would have produced *N* small
    /// transactions inside **one** hold, which shrinks the transaction and
    /// changes the latency not at all: the actor is single-threaded, so nothing
    /// else writes until its turn returns regardless of how many `COMMIT`s the
    /// turn contains. Sending *N* commands returns the actor to its `select!`
    /// between sessions, which is where an interactive assertion gets to jump
    /// the queue — and it is high-priority, so it does.
    ///
    /// The same reasoning is why [`Self::bulk_import`] chunks here and not
    /// there, and it is the trap T1.2 names for `CREATE TABLE … AS SELECT`.
    ///
    /// # Choosing a window
    ///
    /// The bound is on *transaction* time, so the session count is set by how
    /// far back the hot file goes, not by how much it holds. A window is
    /// rejected rather than clamped if it would need more than
    /// [`MAX_ARCHIVE_SESSIONS`] sessions — see [`DbError::ArchiveWindow`].
    ///
    /// Windows containing nothing archivable are cheap but not free: each still
    /// opens a transaction and writes a horizon row. What they no longer do is
    /// re-project `links_current`, which `archive_session` now skips when its
    /// `DELETE` removed no rows — without that, windowing costs *more* in total
    /// than not windowing, because the repair term scales with the surviving
    /// table and not with the batch (D-077).
    pub async fn archive_windowed(
        &self,
        cutoff: &str,
        window: std::time::Duration,
    ) -> Result<Vec<ArchiveReport>> {
        let cutoff = timestamp::normalize(cutoff)?;
        let boundaries = self.archive_boundaries(&cutoff, window).await?;

        let mut reports = Vec::with_capacity(boundaries.len());
        for boundary in boundaries {
            let archive_path = self.archive_path.clone();
            reports.push(
                self.low(|responder| LowPriCommand::Archive {
                    cutoff: boundary,
                    archive_path,
                    responder,
                })
                .await?,
            );
        }
        Ok(reports)
    }

    /// The cutoffs [`Self::archive_windowed`] will run, ascending, ending at
    /// `cutoff` exactly.
    ///
    /// Read on `read_conn`, not on the actor: this is two `MIN`s and the actor
    /// has no reason to hold its lock for them.
    ///
    /// The lower end comes from the data rather than from the clock. Stepping
    /// from some fixed epoch would make the session count a function of the
    /// calendar — a database opened yesterday would still be asked to archive
    /// 1970 — whereas the oldest `recorded_at` actually present is the earliest
    /// boundary that can contain anything.
    async fn archive_boundaries(
        &self,
        cutoff: &str,
        window: std::time::Duration,
    ) -> Result<Vec<String>> {
        // A single session at `cutoff` is exactly `archive(cutoff)`, and it is
        // the right answer for an empty hot file: it still writes the horizon
        // row, so windowed and unwindowed runs leave the same observable state.
        let Some(oldest) = self.oldest_hot_stamp(cutoff).await? else {
            return Ok(vec![cutoff.to_string()]);
        };

        let start = timestamp::parse(&oldest)?;
        let end = timestamp::parse(cutoff)?;
        let Ok(span) = end.duration_since(start) else {
            // Everything in the hot file is at or after the cutoff, so there is
            // nothing in range to divide.
            return Ok(vec![cutoff.to_string()]);
        };

        if window.is_zero() {
            return Err(DbError::ArchiveWindow {
                window,
                reason: "a zero-length window never advances past the first boundary".into(),
            });
        }

        // `div_ceil` on nanos: a span of 90 minutes in 60-minute windows is two
        // sessions, not one. `as_nanos` is u128, so neither the division nor the
        // span can overflow for any timestamp this crate can store.
        let sessions = span.as_nanos().div_ceil(window.as_nanos());
        if sessions > MAX_ARCHIVE_SESSIONS as u128 {
            return Err(DbError::ArchiveWindow {
                window,
                reason: format!(
                    "a span of {span:?} would need {sessions} sessions (limit \
                     {MAX_ARCHIVE_SESSIONS}); widen the window"
                ),
            });
        }

        let mut boundaries = Vec::with_capacity(sessions as usize);
        for k in 1..sessions {
            boundaries.push(timestamp::format(start + window * k as u32));
        }
        // The last boundary is `cutoff` itself and not `start + n*window`, which
        // would overshoot and archive rows the caller excluded.
        boundaries.push(cutoff.to_string());
        Ok(boundaries)
    }

    /// Oldest `recorded_at` below `cutoff` in either hot table, or `None`.
    async fn oldest_hot_stamp(&self, cutoff: &str) -> Result<Option<String>> {
        let mut oldest: Option<String> = None;
        for table in ["links", "transaction_log"] {
            let found: Option<String> = self
                .read_conn
                .query(
                    &format!("SELECT MIN(recorded_at) FROM {table} WHERE recorded_at < ?1"),
                    libsql::params![cutoff],
                )
                .await?
                .next()
                .await?
                .and_then(|row| row.get(0).ok());
            if let Some(found) = found {
                if oldest.as_ref().is_none_or(|o| found < *o) {
                    oldest = Some(found);
                }
            }
        }
        Ok(oldest)
    }

    /// Send a high-priority command and wait for its answer.
    ///
    /// The two error mappings here are the whole reason this helper exists.
    /// `send` failing means the actor is gone — `WriterUnavailable`. The
    /// responder being dropped without an answer means the actor took the
    /// command and never replied — `WriterDroppedResponder`, which is a bug in
    /// the actor rather than a condition the caller can retry. Both variants
    /// existed in `error.rs` from 0.4.5 and neither was ever constructed, so a
    /// dead actor and a hung one were both just a caller waiting forever.
    async fn high<T>(
        &self,
        make: impl FnOnce(oneshot::Sender<Result<T>>) -> HighPriCommand,
    ) -> Result<T> {
        let (tx, rx) = oneshot::channel();
        self.highpri_tx
            .send(make(tx))
            .await
            .map_err(|_| DbError::WriterUnavailable)?;
        rx.await.map_err(|_| DbError::WriterDroppedResponder)?
    }

    /// Send each chunk in turn and sum the counts — the shape all four bulk
    /// paths share (T3.4, D-086).
    ///
    /// # This is sequential on purpose, and the purpose is a measurement
    ///
    /// T3.4 proposed pipelining: send *k* chunks ahead so the actor never finds
    /// an empty queue. The reasoning is that awaiting each chunk before building
    /// the next leaves the actor idle for a channel round trip every time, which
    /// on a 1M-edge import is ~11,000 idle gaps.
    ///
    /// Both halves of that are true and the conclusion does not follow. The gaps
    /// are real; they are also **four orders of magnitude smaller than the work
    /// they interrupt**. A tokio mpsc hop is sub-microsecond and a chunk takes
    /// 13–21 ms. Implemented and swept at depths 1, 2, 4, 8 and 16 over 20K and
    /// 100K edges: every cell landed within 1% of sequential, in both directions
    /// — see `examples/pipeline_diag.rs`, which is kept precisely so this is not
    /// re-proposed from the same reasoning.
    ///
    /// So the pipelining was removed and the deduplication kept. It was not free
    /// to hold: with chunks in flight, a failure at chunk `i` no longer leaves a
    /// **prefix** committed, because `i+1 ..= i+k-1` were already sent and commit
    /// anyway. D-011 promises "earlier chunks committed", and paying for that
    /// with a weaker recovery story in exchange for nothing measurable is the
    /// wrong trade.
    ///
    /// Sending stops at the first error, so what commits is exactly the prefix
    /// before the failure.
    /// # The size is now measured, not assumed (0.12.0, W3)
    ///
    /// Until 0.11.0 the caller pre-split into `chunks(chunk_rows::WHATEVER)` and
    /// this loop sent what it was given. That made the constant *the* size, and
    /// D-143 is the record of a constant fitted at one population being wrong at
    /// another: all four D-088 shapes agreed the largest in-budget edge chunk was
    /// **20** against a shipped 90, and 20 would itself have been wrong at 80,000
    /// edges, because per-row cost on that path grows with `links_current`.
    ///
    /// No row count can bound a duration on such a path, so the loop stopped
    /// trying to pick one ahead of time. `ceiling` — still the path's
    /// [`chunk_rows`] constant, with its derivation intact — is now the largest
    /// size this will ever ask for, and each chunk's measured hold chooses the
    /// next through `next_chunk_size`.
    ///
    /// **Feedback, not preemption.** The chunk in flight always commits in full;
    /// the SQLite write lock is not preemptible, so nothing here can shorten a
    /// transaction already running. A batch of one chunk gets no protection at
    /// all, and convergence costs one or two chunks — which is the price of the
    /// bound being a duration rather than a promise.
    ///
    /// The last chunk's outcome is discarded, there being no next chunk to size.
    /// The chunk loop behind all four bulk paths.
    ///
    /// **Every exit carries `written`** (0.13.8, W7.6, D-181). It used to
    /// carry it only out of the success arm: the three error paths were `?` on
    /// a [`DbError`], which discards the local, so a caller whose 20,000-row
    /// import failed in the last chunk learned that it failed and not that
    /// 19,000 rows were already in the database. The count was never expensive
    /// to keep — it is right there, and the loop needs it anyway to size the
    /// next chunk.
    async fn low_chunked<T>(
        &self,
        items: Vec<T>,
        ceiling: usize,
        control: BulkControl,
        make: impl Fn(Vec<T>, oneshot::Sender<Result<ChunkOutcome>>) -> LowPriCommand,
    ) -> BulkResult<usize> {
        let total = items.len();
        let mut items = items.into_iter();
        let mut size = ceiling.max(1);
        let mut written = 0usize;
        loop {
            let chunk: Vec<T> = items.by_ref().take(size).collect();
            if chunk.is_empty() {
                // Emptiness is checked before cancellation on purpose: a token
                // raised after the last chunk committed is asking to stop work
                // that is already done, and reporting that as a failure would
                // make a race between the caller's two threads decide whether a
                // complete import counts as one.
                return Ok(written);
            }
            // Between chunks, never inside one. Nothing is rolled back and no
            // transaction is interrupted -- the loop simply stops sending, and
            // the prefix that committed is the same kind of prefix a failure
            // would have left.
            if control.is_cancelled() {
                return Err(BulkInterrupted {
                    written,
                    cause: DbError::BulkCancelled,
                });
            }
            let stop = |cause: DbError| BulkInterrupted { written, cause };
            let (tx, rx) = oneshot::channel();
            self.lowpri_tx
                .send(make(chunk, tx))
                .await
                .map_err(|_| stop(DbError::WriterUnavailable))?;
            let outcome = match rx.await {
                Err(_) => return Err(stop(DbError::WriterDroppedResponder)),
                Ok(Err(e)) => return Err(stop(e)),
                Ok(Ok(outcome)) => outcome,
            };
            written += outcome.rows;
            control.report(BulkProgress {
                written,
                total,
                rows: outcome.rows,
                held: outcome.held,
            });
            size = next_chunk_size(size, outcome.held, CHUNK_BUDGET, CHUNK_FLOOR, ceiling);
        }
    }

    async fn low<T>(
        &self,
        make: impl FnOnce(oneshot::Sender<Result<T>>) -> LowPriCommand,
    ) -> Result<T> {
        let (tx, rx) = oneshot::channel();
        self.lowpri_tx
            .send(make(tx))
            .await
            .map_err(|_| DbError::WriterUnavailable)?;
        rx.await.map_err(|_| DbError::WriterDroppedResponder)?
    }

    /// Clean shutdown: stop the Write Actor, then write the final snapshot (§5.1.7).
    ///
    /// Order matters. The snapshot is taken *after* the actor has stopped and
    /// been joined, so no write can land between the fold and the file — the
    /// anchor it records is the last thing that happened, not the last thing
    /// that happened to be visible.
    ///
    /// A failed snapshot is reported rather than swallowed. It is not a
    /// durability loss — the ledger is in the WAL and the log replays without
    /// it — but it means the next open starts from an older anchor, and a caller
    /// that never hears about it cannot know why startup got slower.
    ///
    /// **The cadence stops first (§5.5, D-053).** Both it and `write_final` end
    /// by running retention over the snapshot directory, and retention deletes
    /// files. Letting them overlap would mean one pass enumerating the directory
    /// while the other removes from it — not a correctness problem for the
    /// ledger, which is why the ordering is stated rather than locked, but a
    /// source of spurious warnings and of a final anchor that could be deleted
    /// by a cleanup that started before it existed. Stopping the cadence, then
    /// the actor, then taking the snapshot leaves exactly one writer at each
    /// step.
    pub async fn close(mut self) -> Result<()> {
        if let Some(stop) = self.cadence_stop.take() {
            let _ = stop.send(true);
        }
        if let Some(handle) = self.cadence.take() {
            let _ = handle.await;
        }

        // Top up the planner's statistics while the actor is still alive to do
        // it (0.12.4, D-149). `PRAGMA optimize` re-analyses only what SQLite
        // believes has gone stale, so on a database that did nothing this costs
        // nothing, and on one that was just bulk-loaded it is the difference
        // between the next process planning on measurements and planning on
        // built-in guesses.
        //
        // **Deliberately not fatal.** A failure here costs plan quality on the
        // next open and nothing else — no ledger state depends on it — and
        // `close()` is where a caller learns whether their *writes* survived.
        // Turning a stale-statistics problem into a failed close would bury that
        // answer under a much less important one.
        if let Err(e) = self.optimize().await {
            tracing::warn!(
                "PRAGMA optimize failed during close(): {e}. Statistics may be \
                 stale for the next process; call analyze() to rebuild them. \
                 Nothing else is affected."
            );
        }

        let (tx, rx) = oneshot::channel();
        let _ = self
            .highpri_tx
            .send(HighPriCommand::Shutdown { responder: tx })
            .await;
        let _ = rx.await;

        // **The writer's exit status is propagated, not discarded (Wave 4.2).**
        // It used to be `let _ = handle.await`, so an actor that had died closed
        // "successfully" and the caller's last chance to learn that the write
        // path was gone was spent silently.
        //
        // Through 0.13.3 this awaited a `JoinHandle<Result<()>>` and did
        // `Ok(res) => res?`, which looked like two failure paths and was one:
        // the actor's `Result` could not be `Err` (W7.3, D-177). What remains is
        // the branch that can fire — the actor panicked or was aborted — mapped
        // by `writer_exit`, which is tested against a real `JoinError`.
        //
        // Ordered before the final snapshot on purpose: a snapshot written after
        // a dead writer records a state the caller has no reason to trust, and
        // returning the error while also having written that file is worse than
        // not writing it.
        if let Some(handle) = self.writer.take() {
            writer_exit(handle.await)?;
        }

        let ts = self.clock.now();
        let archive = crate::temporal::archive::archive_present(&self.archive_path)
            .then_some(self.archive_path.as_path());
        snapshot::write_final(&self.read_conn, &self.snapshots_dir, &ts, archive).await?;

        // Marks the handle closed so `Drop` knows not to complain.
        self.closed = true;
        Ok(())
    }
}

/// Notes a missed `close()` at `warn!`, and deliberately does **not** assert.
///
/// **§7.3 offered option B — document `close()` as mandatory and `debug_assert`
/// in `Drop` — and Wave 4.2 implemented it, measured the consequence, and
/// reduced it to a warning.** The assert fired on roughly thirty tests on its
/// first run. That is the signal it was built to produce, and the right reading
/// of it was not "thirty tests are wrong".
///
/// What dropping actually costs is one final snapshot. Nothing else: every
/// public write method awaits its responder, so by the time a caller *can* drop
/// the handle, every write it issued has already committed; and the cadence stops
/// on its own, because `cadence_stop` is a `watch::Sender` whose drop signals the
/// task. A snapshot is derivative state under Doctrine VI — disposable,
/// reconstructible, and never the only copy of anything. Losing one makes the
/// next `reconstruct` fold from an older anchor, which is **slower, not wrong**.
///
/// A `debug_assert` aborts a test run. Spending that on a performance loss, in a
/// project whose own notes say a suite that fails for reasons unrelated to the
/// code under test trains people to ignore red, is the wrong trade — and paying
/// it in thirty places would have made `close()` look mandatory by ceremony
/// rather than by consequence. `close()` remains the right thing to call, and
/// the two reasons to call it are now stated where they can be acted on: the
/// snapshot, and the writer's `Result`, which only `close()` can return.
///
/// Option A ("abort the actor and log") stays rejected, for the reason it was
/// rejected twice before: `Drop` cannot await, so it cannot drain, and cleanup
/// that cannot clean up is worse than none — it looks like cleanup.
impl Drop for Database {
    fn drop(&mut self) {
        if !self.closed {
            tracing::warn!(
                "Database dropped without close(): the final snapshot was not written, \
                 so the next reconstruct folds from an older anchor, and the write \
                 actor's exit status was not checked. Prefer close().await."
            );
        }
    }
}

/// A failure before the first chunk was sent, which committed nothing.
///
/// Normalisation runs over the whole batch up front, so its errors are the one
/// class the chunk loop never sees — and they are still [`BulkInterrupted`],
/// because a caller matching on one error type should not have to match on two
/// to find out that nothing landed (0.13.8, W7.6).
fn before_any_chunk(cause: DbError) -> BulkInterrupted {
    BulkInterrupted { written: 0, cause }
}

fn normalize_all(edges: Vec<EdgeAssertion>) -> Result<Vec<EdgeAssertion>> {
    edges.into_iter().map(EdgeAssertion::normalized).collect()
}

/// One `FULL` checkpoint for the numbers, then a `TRUNCATE` for the file.
///
/// # Why `TRUNCATE` and not a mode parameter
///
/// The four SQLite modes are not four things a caller of *this* crate wants.
/// `PASSIVE` is what the automatic checkpointer already runs on its own, so an
/// explicit `PASSIVE` asks for something that was going to happen anyway;
/// `RESTART` and `FULL` differ from `TRUNCATE` only in whether the WAL file is
/// left at its high-water size. The reason W5.2 exists is
/// [`Tuning::wal_autocheckpoint`] — a bulk importer turns the automatic
/// checkpointer off and calls this once at the end — and what that caller wants
/// is the WAL *gone*, not smaller than it was. So the mode is fixed and decided
/// here rather than pushed to the caller as a choice they would have to read
/// SQLite's documentation to make. If a mode ever needs selecting, that is an
/// additive method, not a change to this one.
///
/// # Why it is two pragmas, which is not the obvious implementation
///
/// **A successful `TRUNCATE` reports `busy=0, log=0, checkpointed=0`** — the
/// counts describe the WAL *after* the operation, and after a truncation there
/// is no WAL to describe. Measured, not inferred: on a 387-frame WAL, `PASSIVE`
/// returns `0, 387, 387` and `TRUNCATE` on the same file returns `0, 0, 0`. So
/// the single-pragma implementation returns a [`CheckpointReport`] whose two
/// counts are structurally zero on success, which makes the whole struct a
/// less useful `bool`.
///
/// `FULL` copies every frame back and reports what it moved; the `TRUNCATE`
/// that follows finds nothing left to copy and resets the file. The second pass
/// is close to free for exactly that reason — it is a file operation, not a
/// second copy. `busy` is the **union**: a checkpoint that was blocked in
/// either phase did not fully happen, and a caller about to copy the database
/// file elsewhere needs the pessimistic answer.
///
/// # They return rows, so they go through `query()`
///
/// The same libsql constraint the pragmas in `configure` document: `execute()`
/// rejects any statement that yields rows, and these yield the row that is the
/// entire point.
async fn run_checkpoint(conn: &libsql::Connection) -> Result<CheckpointReport> {
    // The columns are `busy, log, checkpointed`. SQLite reports -1 for the two
    // counts when the checkpoint could not run; clamped to 0 rather than
    // surfaced as a signed count, because `busy` already carries "this did not
    // happen" and a negative frame count is not a quantity anyone can use.
    //
    // A database not in WAL mode returns no row at all. `configure` puts every
    // connection this crate opens into WAL, so that is unreachable here — but a
    // zeroed report is a better failure than a panic if it stops being.
    async fn one(conn: &libsql::Connection, sql: &str) -> Result<(bool, u64, u64)> {
        let mut rows = conn.query(sql, ()).await?;
        let Some(row) = rows.next().await? else {
            return Ok((false, 0, 0));
        };
        let field = |i: i32| -> u64 { row.get::<i64>(i).unwrap_or(0).max(0) as u64 };
        Ok((row.get::<i64>(0).unwrap_or(0) != 0, field(1), field(2)))
    }

    let (full_busy, _, moved) = one(conn, "PRAGMA wal_checkpoint(FULL)").await?;
    let (trunc_busy, log_frames, _) = one(conn, "PRAGMA wal_checkpoint(TRUNCATE)").await?;

    Ok(CheckpointReport {
        busy: full_busy || trunc_busy,
        log_frames,
        checkpointed_frames: moved,
    })
}

/// Pragmas that mean something on **any** connection, including one opened
/// `SQLITE_OPEN_READ_ONLY` (0.12.16, W5.5, D-159).
///
/// Both of these are per-connection state that a reader is subject to just as a
/// writer is. `busy_timeout` is the one that made this a finding:
/// [`Database::diagnostic_conn`] ran with SQLite's default of **0** — return
/// `SQLITE_BUSY` immediately — while every other connection in the process
/// waited 5 s, so the one surface whose job is to answer questions when the
/// typed path is already suspect was also the one most likely to fail with
/// "database is locked" under exactly the contention that prompted the
/// question.
/// Empty the slot unless what is in it is fit to hand to the next caller
/// (0.15.15, W15.5, [D-257]).
///
/// Shared by [`Database::diagnostic_conn`], which runs it on the way in, and
/// [`Database::scrub_diagnostic_conn`], which the Python binding runs on the way
/// out. One implementation because the two must agree: a connection the entry
/// path would have discarded is one the exit path must not leave sitting there.
async fn scrub(slot: &mut Option<libsql::Connection>) {
    let Some(conn) = slot.as_ref() else {
        return;
    };
    // A leaked `BEGIN` is the one that leaves this surface: it pins a WAL read
    // snapshot, so later diagnostic reads answer from it and `checkpoint()`
    // cannot truncate past it. Rolled back rather than reported, because the
    // caller who would read the report is the one who did not do it.
    let recovered = conn.is_autocommit() || conn.execute("ROLLBACK", ()).await.is_ok();
    // A connection that will not roll back is not one to hand on, and neither
    // is one whose own state cannot be read: either way the answer is a fresh
    // connection, which costs 56.5 us on the call that dirtied it and nothing
    // on any other.
    if !recovered || diagnostic_is_dirty(conn).await.unwrap_or(true) {
        *slot = None;
    }
}

/// Is this cached diagnostic connection carrying state from an earlier caller
/// (0.15.15, W15.5, [D-257])?
///
/// Two questions — are there temp objects, is anything attached beyond `main`
/// and `temp` — asked as pragmas rather than as a query over
/// `temp.sqlite_master` and `pragma_database_list`. Same answers, and
/// `examples/diagnostic_hygiene_probe.rs` measures the pragma form at
/// **2.4 µs** against the query form's **7.8 µs**, on a call whose whole warm
/// cost is the `stat` in front of it. Detection is measured rather than
/// assumed: `temp.schema_version` goes 0 → 1 on `CREATE TEMP TABLE`, and
/// `database_list` 2 → 3 on `ATTACH`.
///
/// **What it cannot see is a `PRAGMA`**, and neither can any other cheap check:
/// SQLite does not enumerate connection-scoped pragma state. The crate restates
/// the two it sets instead — see the call to `configure_common` in
/// [`Database::diagnostic_conn`].
///
/// An `Err` here is not propagated by the caller: a connection whose own state
/// cannot be read is replaced rather than reported on.
///
/// [D-257]: ../../docs/architecture/s13-decision-register.md#d-257
async fn diagnostic_is_dirty(conn: &libsql::Connection) -> Result<bool> {
    let mut rows = conn.query("PRAGMA temp.schema_version", ()).await?;
    let temp_schema: i64 = match rows.next().await? {
        Some(row) => row.get(0)?,
        // No row at all is not a clean connection, it is an answer this
        // function did not understand.
        None => return Ok(true),
    };
    if temp_schema != 0 {
        return Ok(true);
    }
    let mut rows = conn.query("PRAGMA database_list", ()).await?;
    let mut databases = 0_usize;
    while rows.next().await?.is_some() {
        databases += 1;
    }
    // `main` and `temp`, always both, on a connection nobody has attached to.
    Ok(databases > 2)
}

async fn configure_common(conn: &libsql::Connection, cache_size: Option<i32>) -> Result<()> {
    // NOTE: `busy_timeout` returns its resulting value as a row, and libsql's
    // `execute()` rejects any statement that yields rows ("Execute returned
    // rows"). It must be issued through `query()`.
    let _ = conn.query("PRAGMA busy_timeout = 5000", ()).await?;
    // Per-connection, and split writer from reader since 0.12.15 (W5.4,
    // D-158). `None` runs no pragma at all rather than restating SQLite's
    // default, so the default remains SQLite's to change.
    if let Some(pages) = cache_size {
        conn.execute(&format!("PRAGMA cache_size = {pages}"), ())
            .await?;
    }
    Ok(())
}

/// Pragmas that only mean anything where writes can happen (0.12.16, W5.5).
///
/// Not run on [`Database::diagnostic_conn`], and the reason is not tidiness:
/// `journal_mode = WAL` is a change to the *database file*, which a connection
/// opened `SQLITE_OPEN_READ_ONLY` cannot make. The rest —
/// `synchronous`, `foreign_keys`, `recursive_triggers`, and the `ANALYZE`
/// bound — govern how writes behave, and a connection that cannot write is not
/// governed by them.
///
/// The write connection and the two internal readers all still get these. The
/// internal readers are opened from the same read-write `libsql::Database`, so
/// the pragmas apply; leaving them out would be a behaviour change made for
/// symmetry, which is not a reason.
async fn configure_writable(conn: &libsql::Connection) -> Result<()> {
    // Returns its resulting value as a row — see the note in `configure_common`.
    let _ = conn.query("PRAGMA journal_mode = WAL", ()).await?;
    conn.execute("PRAGMA synchronous = NORMAL", ()).await?;
    conn.execute("PRAGMA foreign_keys = ON", ()).await?;
    conn.execute("PRAGMA recursive_triggers = OFF", ()).await?;
    // Bounds every `ANALYZE` this connection will ever run, explicit or
    // triggered by `PRAGMA optimize` (D-149). Set here rather than around the
    // call sites so the scheduled path is bounded too — that is the half that
    // runs with nobody watching. Returns the previous limit as a row, so it goes
    // through `query()` for the reason the note above gives.
    let _ = conn.query(crate::schema::ddl::ANALYSIS_LIMIT, ()).await?;
    Ok(())
}

/// Full pragma configuration, for a connection that can write.
async fn configure(
    conn: libsql::Connection,
    cache_size: Option<i32>,
) -> Result<libsql::Connection> {
    configure_writable(&conn).await?;
    configure_common(&conn, cache_size).await?;
    Ok(conn)
}

/// Helper to derive the snapshot directory by convention: foo.db -> foo_snapshots/
fn derive_snapshots_dir(path: &Path) -> PathBuf {
    let mut dir = path.to_path_buf();
    let stem = path
        .file_stem()
        .and_then(|s| s.to_str())
        .unwrap_or("macrame");
    dir.set_file_name(format!("{stem}_snapshots"));
    dir
}

/// Helper to derive archive database path by convention: foo.db -> foo_archive.db
fn derive_archive_path(path: &Path) -> PathBuf {
    let mut archive = path.to_path_buf();
    if let Some(stem) = path.file_stem().and_then(|s| s.to_str()) {
        let ext = path.extension().and_then(|e| e.to_str()).unwrap_or("db");
        archive.set_file_name(format!("{stem}_archive.{ext}"));
    } else {
        archive.set_extension("archive.db");
    }
    archive
}

/// Dedicated Write Actor event loop prioritizing high-priority UI requests over low-priority background work.
///
/// # The turn is the unit, not the statement (T1.4)
///
/// One iteration of this loop is one *hold*: the actor is single-threaded and
/// the SQLite write lock is not preemptible, so from the moment a command starts
/// executing until it returns, nothing else writes. That is the quantity
/// [`CHUNK_BUDGET`] bounds, and so it is the quantity
/// [`crate::metrics::ActorMetrics`] measures — deliberately around the whole
/// `execute` call rather than inside it. Timing the SQL alone would have
/// reported a bound that held while callers waited.
///
/// Queue depth is sampled *before* the `select!`, so it is the backlog the turn
/// found on arrival rather than the one it left behind.
///
/// # `biased` has no floor, and since 0.12.10 that is measured (W4.4, D-153)
///
/// `biased` makes the arms poll in declaration order, so high-priority work is
/// taken whenever any is ready. Nothing bounds how long that can continue:
/// sustained interactive traffic can hold the low tier off indefinitely, and
/// through 0.12.9 nothing in the crate could say whether it ever did.
/// `record_priority_choice` counts the turns where the choice went against
/// queued low-priority work, and the longest unbroken run of them, which is the
/// half that distinguishes "prioritised" from "starved".
///
/// **No forced yield is added here.** Whether one is needed is the question the
/// counter answers, and adding a policy now would be fixing a bound nobody has
/// observed being hit — the same mistake D-124 was retracted for.
///
/// # It returns nothing, and used to return a `Result` it could not fail
/// (0.13.4, W7.3, §3.5, [D-177])
///
/// The two exits are `LoopCtl::Break` from [`HighPriCommand::Shutdown`] and the
/// `else` arm when both channels are closed. Neither can fail, and neither
/// could before: every command's error goes back on that command's own
/// responder, where the caller who issued it can act on it. There was no third
/// thing for an actor-level `Err` to carry, and none was ever constructed.
///
/// A `Result` that is structurally always `Ok` is not free. It reads as a
/// failure path under review, so `close()`'s `res?` looked like it was doing
/// something, and the branch that actually fires — a **panicked** actor,
/// reported as a `JoinError` — sat beside it untested. That is the swap this
/// change makes: the unfireable branch is gone and the real one is pinned, in
/// [`writer_exit`].
async fn run_writer_actor(
    conn: libsql::Connection,
    clock: Arc<dyn Clock>,
    mut highpri_rx: mpsc::Receiver<HighPriCommand>,
    mut lowpri_rx: mpsc::Receiver<LowPriCommand>,
    shared: Arc<ActorShared>,
) {
    // Owned by the loop and lent to each command, which is the whole of A-3:
    // the actor had a connection and no memory, so every turn re-established
    // what the turn before it had just established (0.15.6, W14.3, D-248).
    let mut state = ActorState::new();
    loop {
        // Read once and reused by both the depth sample and the starvation
        // counter, so the two cannot disagree about what was queued when this
        // turn went looking (W4.4, D-153).
        let low_queued = lowpri_rx.len();
        shared.metrics.record_turn(highpri_rx.len(), low_queued);

        let ctl = tokio::select! {
            biased;
            Some(cmd) = highpri_rx.recv() => {
                shared.metrics.record_priority_choice(true, low_queued);
                let turn = Turn::start(cmd.kind(), &shared);
                cmd.execute(&conn, &*clock, &turn, &mut state).await
            }
            Some(cmd) = lowpri_rx.recv() => {
                shared.metrics.record_priority_choice(false, low_queued);
                let turn = Turn::start(cmd.kind(), &shared);
                cmd.execute(&conn, &*clock, &turn, &mut state).await
            }
            else => LoopCtl::Break,
        };
        if matches!(ctl, LoopCtl::Break) {
            break;
        }
    }
}

/// Turn the write actor's join status into the error `close()` reports.
///
/// One line of mapping, given a name so it can be tested against a real
/// [`tokio::task::JoinError`]. Before 0.13.4 this was inline beside a `res?` on
/// an actor `Result` that could only ever be `Ok`, and the arrangement had the
/// coverage exactly backwards: the branch that cannot fire was plumbed through
/// two signatures, and the branch that does fire — the actor panicked, and the
/// caller's writes are going nowhere — had no test at all (W7.3, D-177).
///
/// Cancellation is folded in with panics deliberately. `JoinError` distinguishes
/// them, and nothing in the crate ever aborts this task, so a cancelled writer
/// means something outside the crate reached in and stopped it. That is not a
/// gentler condition than a panic and must not read as one.
fn writer_exit(joined: std::result::Result<(), tokio::task::JoinError>) -> Result<()> {
    joined.map_err(|e| DbError::WriterStopped(format!("the write actor did not exit cleanly: {e}")))
}

/// One command's hold: the timer, its label, and the counters it reports to.
///
/// # The hold is recorded *before* the caller is answered, and it has to be
///
/// The obvious placement — time the whole `execute` call from the loop — is
/// wrong in a way that only shows up under test. Every arm of `execute` ends by
/// sending on a `oneshot`, which wakes the waiting caller; the actor then
/// returns to the loop and records. Those are two tasks, so a caller that awaits
/// its own write and immediately reads [`Database::metrics`] can be scheduled
/// first and see a turn count that does not include the write it just did.
///
/// Not a correctness bug in the ledger, and it would never have been noticed in
/// production — a dashboard sampling every few seconds cannot see the window.
/// It makes every test and diagnostic of the counters flaky, which is worse: the
/// instrumentation would have been *believed* while being wrong exactly when
/// someone tried to check it. `examples/bulk_atomic_diag.rs` was the thing that
/// caught it, reporting a 20,000-row batch as a 0 ms hold.
///
/// So `answer` records and then sends, in that order, and the ordering is the
/// method's whole reason to exist. What it costs is that the `oneshot::send`
/// itself falls outside the measurement, which is a few nanoseconds against a
/// turn measured in microseconds at best.
struct Turn<'a> {
    kind: crate::metrics::CommandKind,
    timer: crate::metrics::HoldTimer,
    shared: &'a ActorShared,
}

/// State the actor owns and a `Turn` needs to reach.
///
/// `archive_epoch` is here rather than in [`crate::metrics::ActorMetrics`]
/// because it is **not** a metric: T1.2's shadow rebuild reads it to decide
/// whether its work is still valid, so it has to be present in every build, not
/// only under the `metrics` feature. Counting archives happens to be what both
/// want; only one of them is allowed to be compiled out.
///
/// `turns` is here for the same reason and serves the snapshot cadence — see
/// its own note.
#[derive(Default)]
struct ActorShared {
    metrics: crate::metrics::ActorMetrics,
    archive_epoch: std::sync::atomic::AtomicU64,
    /// Commands this actor has answered `Ok` to (0.15.19, review C-19).
    ///
    /// # What it is for, and why it is not a `seq_id`
    ///
    /// `snapshot::run_cadence` used to run `SELECT MAX(seq_id), MAX(recorded_at)`
    /// on **every tick**, five seconds apart by default, whether or not
    /// anything had been written. Two aggregates on an idle database, for a
    /// fact the actor already had: *nothing has happened*.
    ///
    /// The review asked for a `watch<u64>` of the last committed `seq_id`. This
    /// is the same idea one step cheaper, and the difference matters. The actor
    /// does not currently know the `seq_id` its writes produced — the log rows
    /// are written by triggers — so publishing one would mean adding a query to
    /// **every write** in order to remove a query from an idle timer, which is
    /// the wrong direction. A turn count needs no query at all: one relaxed
    /// `fetch_add` on a path already doing a database round trip.
    ///
    /// # Why this cannot change when a snapshot is written
    ///
    /// The cadence skips its tick when the count has not moved since the last
    /// one. That is sound because the implication runs the right way: if
    /// `MAX(seq_id)` grew, some command committed, so some turn answered `Ok`,
    /// so the count moved. The converse is not claimed and does not need to be
    /// — a turn that answered `Ok` without writing a log row makes the cadence
    /// do exactly the query it used to do every time. It over-counts, never
    /// under-counts, and the tick it protects had nothing to do anyway.
    ///
    /// `Relaxed` because nothing is ordered against it. The cadence reads a
    /// number to compare with a number it read before; a value one tick stale
    /// costs one deferred tick and no correctness, and the same is true of the
    /// `archive_epoch` beside it.
    turns: std::sync::atomic::AtomicU64,
}

impl crate::temporal::snapshot::CommittedTurns for ActorShared {
    fn committed_turns(&self) -> u64 {
        self.turns.load(std::sync::atomic::Ordering::Relaxed)
    }
}

impl<'a> Turn<'a> {
    fn start(kind: crate::metrics::CommandKind, shared: &'a ActorShared) -> Self {
        Self {
            kind,
            timer: crate::metrics::HoldTimer::start(),
            shared,
        }
    }

    fn epoch(&self) -> u64 {
        self.shared
            .archive_epoch
            .load(std::sync::atomic::Ordering::Relaxed)
    }

    /// Record that an archive session committed.
    ///
    /// Bumped on **success only**: a failed archive rolls back, so it deletes
    /// nothing and invalidates no shadow build.
    fn archive_committed(&self) {
        self.shared
            .archive_epoch
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    }

    /// Close the hold and hand the result back. Never the other way round.
    ///
    /// The `let _ =` on the send is deliberate and predates this: a caller that
    /// dropped its receiver — `tokio::time::timeout` around a write, which
    /// [`Database`]'s write surface explicitly documents — is not an actor
    /// error, and the command committed regardless.
    fn answer<T>(&self, responder: oneshot::Sender<Result<T>>, res: Result<T>) {
        self.shared
            .metrics
            .record_hold(self.kind, self.timer.elapsed());
        if res.is_ok() {
            self.turn_committed();
        }
        let _ = responder.send(res);
    }

    /// [`answer`](Self::answer) for a chunk: the same reading, handed back to the
    /// caller as well as recorded (0.12.0, W1).
    ///
    /// One `elapsed()` serves both, so the duration the chunk loop sizes against
    /// is *the same number* the histogram shows — a controller and a dashboard
    /// disagreeing about what a chunk cost would be a bad way to spend a
    /// debugging session.
    ///
    /// The record-then-send ordering documented on [`Turn`] is preserved, and
    /// matters here for the same reason: the send wakes the caller, which may be
    /// scheduled before this method returns.
    fn answer_chunk(&self, responder: oneshot::Sender<Result<ChunkOutcome>>, res: Result<usize>) {
        let held = self.timer.elapsed();
        self.shared.metrics.record_hold(self.kind, held);
        if res.is_ok() {
            self.turn_committed();
        }
        let _ = responder.send(res.map(|rows| ChunkOutcome { rows, held }));
    }

    /// Record that a turn answered `Ok`, for [`ActorShared::turns`].
    ///
    /// Called from both answer paths rather than from the loop, because those
    /// are the two places that know the result. A command that returns an error
    /// does not bump it: a failed write rolls back and the log is where it was.
    fn turn_committed(&self) {
        self.shared
            .turns
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    }
}

const INSERT_LINK: &str = "INSERT INTO links \
     (source_id, target_id, edge_type, valid_from, valid_to, weight, properties, \
      recorded_at, branch_id) \
     VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)";

/// The parameter row for [`INSERT_LINK`], in one place since 0.14.8.
///
/// The single-edge path and the chunk path spelled these out separately, which
/// was survivable at eight and is not at nine: `branch_id` is the one parameter
/// whose omission is *silent* — the column defaults to `'main'`, so a path that
/// forgot it would write to the trunk and pass every test that did not fork.
/// [`concept_params`] has existed for this reason since D-056.
fn edge_params<'a>(edge: &'a EdgeAssertion, stamp: &'a str) -> [libsql::Value; 9] {
    [
        edge.source.as_str().into(),
        edge.target.as_str().into(),
        edge.edge_type.as_str().into(),
        edge.valid_from.as_str().into(),
        edge.valid_to.as_str().into(),
        edge.weight.into(),
        edge.properties.as_str().into(),
        stamp.into(),
        edge.branch_name().into(),
    ]
}

/// Shared by the single-concept write and the chunked one, so the two paths
/// cannot drift into upserting different column sets — and so the chunk has a
/// statement text it can prepare once (D-056).
const UPSERT_CONCEPT: &str = "INSERT INTO concepts \
     (id, title, content, embedding_model, valid_from, valid_to, recorded_at, retired, \
      branch_id) \
     VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9) \
     ON CONFLICT(id) DO UPDATE SET \
         title = excluded.title, \
         content = excluded.content, \
         embedding_model = excluded.embedding_model, \
         valid_from = excluded.valid_from, \
         valid_to = excluded.valid_to, \
         recorded_at = excluded.recorded_at, \
         retired = excluded.retired";
// `branch_id` is deliberately **not** in that `DO UPDATE` list. The column is
// provenance and minting happened once (D-214), and
// `trg_concepts_branch_immutable` would abort an update that moved it — so
// listing it would turn every re-upsert of an inherited concept into a guard
// abort instead of the no-op it is. The insert arm carries it; the update arm
// leaves the row where it was minted.

/// The parameter row for [`UPSERT_CONCEPT`], in one place for the same reason.
fn concept_params<'a>(concept: &'a ConceptUpsert, stamp: &'a str) -> [libsql::Value; 9] {
    [
        concept.id.as_str().into(),
        concept.title.as_str().into(),
        concept.content.as_str().into(),
        concept
            .embedding_model
            .as_deref()
            .map_or(libsql::Value::Null, Into::into),
        concept.valid_from.as_str().into(),
        concept.valid_to.as_str().into(),
        stamp.into(),
        (concept.retired as i64).into(),
        concept.branch_name().into(),
    ]
}

/// Check every lineage a write names, and decide which shape its guard takes.
///
/// **One function, two answers, one query per distinct lineage** — and it is
/// [`Lineages::shape`](crate::graph::lineage::Lineages::shape), the same function
/// the read path calls, for the same reason it calls it. A write naming a
/// branch that is not in `branches` has asked about something that does not
/// exist, and answering it by writing to the trunk is [D-069]'s failure in its
/// most expensive form: not a right-looking answer to a question that was not
/// asked, but a *durable* one.
///
/// Relying on the foreign key instead would refuse the write — `branch_id`
/// `REFERENCES branches(branch_id)` and the key is enforced — but it would
/// refuse it as an unqualified "FOREIGN KEY constraint failed" from inside a
/// rolled-back transaction, naming neither the column nor the branch. The same
/// argument [`classify`](crate::error::classify) makes for annotations and
/// edges, one table further along.
///
/// # Why a trunk write pays for it too
///
/// `None` resolves to `'main'` here rather than skipping the check, and that is
/// not tidiness. Once a second lineage can write, the *trunk's* overlap guard
/// is wrong in the other direction — it would be refused for overlapping a
/// branch's belief it cannot see — so the shape decision is one every write
/// needs, not one that branched writes need. On a database that has never
/// forked the answer is [`LineageShape::Trunk`] and the guard is the statement
/// it has always been.
///
/// # This was a query per name until 0.15.6
///
/// It ran [`crate::graph::lineage::Lineages::shape`] once per name and kept the
/// last answer — one round trip per write, for a table only this task writes.
/// [`ActorState`] holds `branches` instead, and [`Lineages::shape_of`] carries
/// what is left of this function's reasoning, including the part about the last
/// answer that stopped being true at 0.15.2.
///
/// [D-069]: ../../docs/architecture/s13-decision-register.md
async fn check_lineages(
    state: &mut ActorState,
    conn: &libsql::Connection,
    names: &[&str],
) -> Result<LineageShape> {
    state.shape_of(conn, names).await
}

/// The shape and the rows, for a caller that also has to resolve an ancestry.
///
/// [`check_lineages`] with the table it read handed back rather than dropped
/// (0.15.17). Both callers need it: the guard compiles a statement per lineage,
/// and the retirement binds one.
async fn check_lineages_with<'a>(
    state: &'a mut ActorState,
    conn: &libsql::Connection,
    names: &[&str],
) -> Result<(LineageShape, &'a Lineages)> {
    let lineages = state.lineages(conn).await?;
    Ok((lineages.shape_of(names)?, lineages))
}

/// The distinct lineages a batch names, in first-seen order.
///
/// A `Vec` and a linear scan rather than a set: batches name one lineage in
/// every case this crate has, the bound is the number of *branches* and not the
/// number of rows, and a `BTreeSet` would allocate per batch to deduplicate a
/// list of length one.
fn distinct_branches(edges: &[EdgeAssertion]) -> Vec<&str> {
    let mut out: Vec<&str> = Vec::with_capacity(1);
    for edge in edges {
        let name = edge.branch_name();
        if !out.contains(&name) {
            out.push(name);
        }
    }
    if out.is_empty() {
        out.push(crate::schema::ddl::MAIN_BRANCH);
    }
    out
}

/// The overlap guard's prepared statement, and which question it asks.
///
/// The two statements take different parameter counts and mean different things
/// by the rows they return, so pairing them with the shape here is what stops
/// [`check_prepared`] from having to be told twice.
struct OverlapGuard {
    stmt: libsql::Statement,
    shape: LineageShape,
    /// The lineage this statement was compiled for. See [`Self::prepare`].
    branch: String,
    /// That lineage's ancestry, bound after the statement's own parameters.
    /// Empty under both trunk shapes, which emit no `lineage` relation.
    ancestry: Vec<Ancestor>,
}

impl OverlapGuard {
    /// Prepare once per turn or per chunk, never per row (D-056, §8.8).
    ///
    /// # One statement per *lineage* since 0.15.17 ([D-259])
    ///
    /// The statement used to be a function of the shape alone: the recursive
    /// `lineage` CTE derived the ancestry from the branch bound at `?5`, so one
    /// compiled form answered for every lineage that shared a shape. A bound
    /// ancestry is not derived from anything — it *is* the answer for one
    /// reader — so the guard now carries the lineage it was compiled for and
    /// the values that lineage binds.
    ///
    /// The cost is bounded by [`distinct_branches`], which is a `Vec` because
    /// every batch this crate has seen names one lineage. A chunk that names
    /// two prepares two, which is the price of the resolution being correct for
    /// both; the shape it would otherwise share is `Resolved`, since the two
    /// trunk shapes each describe a database with exactly one lineage to name.
    ///
    /// [D-259]: ../../docs/architecture/s13-decision-register.md#d-259
    async fn prepare(
        conn: &libsql::Connection,
        shape: LineageShape,
        lineages: &Lineages,
        branch: &str,
    ) -> Result<Self> {
        // Three shapes, three statements, one spelling (0.15.8, W13.3,
        // D-250). Until that release the trunk had a hand-written constant and
        // the other two shared the resolved form, which was exact for a root
        // only because a root's ancestry is itself — so `Lineages::shape_of`
        // could return either of them and nothing observable changed. It
        // cannot now: the root gets a two-predicate lookup on the projection
        // and a branch gets the four-CTE resolution, and D-248's C-24 repair
        // is what decides which.
        let ancestry = match shape {
            LineageShape::Resolved => lineages.ancestry(branch),
            _ => Vec::new(),
        };
        let sql = crate::graph::lineage::overlap_candidates_resolved(shape, &ancestry);
        Ok(Self {
            stmt: conn.prepare(&sql).await?,
            shape,
            branch: branch.to_string(),
            ancestry,
        })
    }

    /// Whether this guard answers for `branch` under `shape`.
    fn answers_for(&self, shape: LineageShape, branch: &str) -> bool {
        self.shape == shape && self.branch == branch
    }
}

impl HighPriCommand {
    /// The metrics label for this variant (T1.4).
    ///
    /// Exhaustive for the same reason `execute` is: a new variant that silently
    /// borrowed another's label would attribute its holds to the wrong command,
    /// and the one question the counters exist to answer is *which* command
    /// broke the budget.
    fn kind(&self) -> crate::metrics::CommandKind {
        use crate::metrics::CommandKind as K;
        match self {
            HighPriCommand::AssertEdge { .. } => K::AssertEdge,
            HighPriCommand::RetireEdge { .. } => K::RetireEdge,
            HighPriCommand::UpsertConcept { .. } => K::UpsertConcept,
            HighPriCommand::WriteBulkAtomic { .. } => K::WriteBulkAtomic,
            HighPriCommand::RebuildCurrent { .. } => K::RebuildCurrent,
            HighPriCommand::RegisterModel { .. } => K::RegisterModel,
            HighPriCommand::Fork { .. } => K::Fork,
            HighPriCommand::Checkpoint { .. } => K::Checkpoint,
            HighPriCommand::Shutdown { .. } => K::Shutdown,
        }
    }

    /// Run one command and answer its caller.
    ///
    /// Deliberately exhaustive — there is no `_` arm. The 0.4.5–0.5.4 actor
    /// matched `Shutdown` and `AssertEdge` and sent everything else to
    /// `_ => LoopCtl::Continue`, which **dropped the responder**: the caller's
    /// `rx.await` resolved to a `RecvError` that no code mapped, so four of six
    /// commands were indistinguishable from a hung database. An exhaustive match
    /// makes that failure a compile error instead of a runtime silence, which is
    /// why adding a variant should break this function.
    async fn execute(
        self,
        conn: &libsql::Connection,
        clock: &dyn Clock,
        turn: &Turn<'_>,
        state: &mut ActorState,
    ) -> LoopCtl {
        match self {
            HighPriCommand::Shutdown { responder } => {
                turn.answer(responder, Ok(()));
                return LoopCtl::Break;
            }
            HighPriCommand::Checkpoint { responder } => {
                let res = run_checkpoint(conn).await;
                turn.answer(responder, res);
            }
            HighPriCommand::AssertEdge { edge, responder } => {
                let stamp = clock.now();
                // Before the guard, because a write naming an unregistered
                // lineage should be refused by name rather than by whatever the
                // guard happens to find when it looks in the wrong place.
                let shape = match check_lineages(state, conn, &[edge.branch_name()]).await {
                    Ok(shape) => shape,
                    Err(e) => {
                        turn.answer(responder, Err(e));
                        return LoopCtl::Continue;
                    }
                };
                if let Err(e) = reject_overlapping_interval(state, conn, &edge, shape).await {
                    turn.answer(responder, Err(e));
                    return LoopCtl::Continue;
                }
                // The statement is held across turns, so it is reset before it
                // is bound rather than after it was stepped — `check_prepared`
                // makes the same argument at more length.
                let res = match state.insert_link(conn).await {
                    Err(e) => Err(e),
                    Ok(stmt) => {
                        stmt.reset();
                        match stmt.execute(edge_params(&edge, &stamp)).await {
                            Ok(_) => Ok(()),
                            Err(e) => Err(classify(
                                conn,
                                e,
                                WriteOp::Edge {
                                    source_id: &edge.source,
                                    target_id: &edge.target,
                                    edge_type: &edge.edge_type,
                                },
                            )
                            .await),
                        }
                    }
                };
                turn.answer(responder, res);
            }
            HighPriCommand::RetireEdge {
                source,
                target,
                edge_type,
                valid_from,
                valid_to,
                branch,
                responder,
            } => {
                let stamp = clock.now();
                let name = branch
                    .as_ref()
                    .map_or(crate::schema::ddl::MAIN_BRANCH, |b| b.as_str());
                let resolved = check_lineages_with(state, conn, &[name])
                    .await
                    .map(|(shape, l)| (shape, l.ancestry(name)));
                let res = match resolved {
                    Ok((shape, ancestry)) => {
                        retire_edge(
                            conn,
                            &source,
                            &target,
                            &edge_type,
                            &valid_from,
                            &valid_to,
                            &stamp,
                            name,
                            shape,
                            &ancestry,
                        )
                        .await
                    }
                    Err(e) => Err(e),
                };
                turn.answer(responder, res);
            }
            HighPriCommand::UpsertConcept { concept, responder } => {
                let stamp = clock.now();
                let res = match check_lineages(state, conn, &[concept.branch_name()]).await {
                    Ok(_) => upsert_concept(conn, &concept, &stamp).await,
                    Err(e) => Err(e),
                };
                turn.answer(responder, res);
            }
            HighPriCommand::WriteBulkAtomic { edges, responder } => {
                // One stamp for the whole batch (D-014): the rows were asserted
                // by one act, and giving them different transaction times would
                // invent an ordering the caller never expressed.
                let stamp = clock.now();
                let res = write_edges_atomic(state, conn, &edges, &stamp).await;
                turn.answer(responder, res);
            }
            HighPriCommand::RebuildCurrent { responder } => {
                turn.answer(responder, rebuild_current(conn).await);
            }
            HighPriCommand::RegisterModel {
                model,
                dim,
                responder,
            } => {
                turn.answer(
                    responder,
                    crate::vector::register_model(conn, &model, dim).await,
                );
            }
            HighPriCommand::Fork {
                name,
                parent,
                responder,
            } => {
                // The same clock as every other write, and the same instant in
                // both columns: `forked_at` is a transaction-time point in the
                // parent's history, and the point this release can fork from is
                // now. See `branch::Branch::created_at` for why they are two
                // columns anyway.
                let stamp = clock.now();
                let res = crate::branch::fork(conn, &name, &parent, &stamp).await;
                // `branches` has a row it did not have. Unconditional rather
                // than `if res.is_ok()`: a fork that failed leaves the table
                // as it was, so forgetting costs one query and asserting that
                // it failed cleanly costs an argument (0.15.6, D-248).
                state.forget_lineages();
                turn.answer(responder, res);
            }
        }
        LoopCtl::Continue
    }
}

impl LowPriCommand {
    /// The metrics label for this variant (T1.4). See [`HighPriCommand::kind`].
    fn kind(&self) -> crate::metrics::CommandKind {
        use crate::metrics::CommandKind as K;
        match self {
            LowPriCommand::WriteConceptsChunk { .. } => K::WriteConceptsChunk,
            LowPriCommand::WriteAnalyticsChunk { .. } => K::WriteAnalyticsChunk,
            LowPriCommand::UpsertEmbeddingChunk { .. } => K::UpsertEmbeddingChunk,
            LowPriCommand::DropEmbeddingIndex { .. } => K::DropEmbeddingIndex,
            LowPriCommand::RebuildEmbeddingIndex { .. } => K::RebuildEmbeddingIndex,
            LowPriCommand::BulkImportChunk { .. } => K::BulkImportChunk,
            LowPriCommand::LinksCurrentMirror { .. } => K::LinksCurrentMirror,
            LowPriCommand::Archive { .. } => K::Archive,
            // Its own counter since 0.12.9 (W4.3, D-152). It reported as
            // `K::Archive` from 0.9.0 to 0.12.8 — the budget really is shared,
            // but attribution is not budget, and an operator reading a long
            // `archive` hold could not tell whether anything had been archived.
            // What kept it folded was that a `CommandKind` variant was a
            // breaking addition; `#[non_exhaustive]` (W4.2) removed that.
            LowPriCommand::Rehydrate { .. } => K::Rehydrate,
            // Its own counter from the day it shipped, which is the whole point
            // of the paragraph above: `Rehydrate` spent four releases folded
            // into `Archive` for a reason that was never good, and the cost of
            // unfolding it was a rung's worth of care about declaration order.
            LowPriCommand::ArchiveBranch { .. } => K::ArchiveBranch,
            LowPriCommand::RebuildFts { .. } => K::RebuildFts,
            // Two kinds out of one variant since 0.13.24 (W10.5, D-197). The
            // command carries the flag; the counter has to carry it too, or the
            // budget exemption for either half is decided about both (D-168).
            LowPriCommand::Analyze { incremental, .. } => {
                if *incremental {
                    K::Optimize
                } else {
                    K::Analyze
                }
            }
            // Two kinds out of one variant since 0.14.16 (W12.16, D-233),
            // and for D-197's reason one line up: the command carries the step,
            // so the counter has to carry it too, or the budget exemption for
            // either half is decided about both. Here that is not hypothetical
            // — the halves want opposite answers. The swap is over budget by
            // construction and the fill chunks are meant to fit, so a merged
            // kind's `over_budget` read `N(rebuilds) + regressions` and could
            // not be decomposed.
            LowPriCommand::ShadowRebuild { step, .. } => match step {
                crate::integrity::ShadowStep::Swap { .. } => K::ShadowSwap,
                crate::integrity::ShadowStep::Begin | crate::integrity::ShadowStep::Fill { .. } => {
                    K::ShadowRebuild
                }
            },
        }
    }

    /// Run one background command and answer its caller.
    ///
    /// Also exhaustive. The pre-0.5.4 version was a single `LoopCtl::Continue`
    /// for *every* variant — every background write silently discarded, its
    /// caller waiting forever.
    async fn execute(
        self,
        conn: &libsql::Connection,
        clock: &dyn Clock,
        turn: &Turn<'_>,
        state: &mut ActorState,
    ) -> LoopCtl {
        match self {
            LowPriCommand::BulkImportChunk { chunk, responder } => {
                // A stamp per chunk, not per batch: the chunks commit
                // separately, so a shared stamp would claim a simultaneity the
                // storage does not have.
                let stamp = clock.now();
                turn.answer_chunk(
                    responder,
                    write_edges_atomic(state, conn, &chunk, &stamp).await,
                );
            }
            LowPriCommand::LinksCurrentMirror { present, responder } => {
                // Schema work on the actor's own connection, like
                // RegisterModel and the embedding-index pair: the
                // single-writer invariant is what keeps the mirror's window
                // on one timeline with the chunks it wraps.
                let sql = if present {
                    crate::schema::ddl::CREATE_LINKS_CURRENT_SYNC.to_string()
                } else {
                    "DROP TRIGGER IF EXISTS trg_links_current_sync".to_string()
                };
                turn.answer(
                    responder,
                    conn.execute(&sql, ()).await.map(|_| ()).map_err(Into::into),
                );
            }
            LowPriCommand::WriteConceptsChunk { chunk, responder } => {
                let stamp = clock.now();
                turn.answer_chunk(
                    responder,
                    write_concepts_atomic(state, conn, &chunk, &stamp).await,
                );
            }
            LowPriCommand::WriteAnalyticsChunk { chunk, responder } => {
                let stamp = clock.now();
                turn.answer_chunk(
                    responder,
                    write_annotations_atomic(conn, &chunk, &stamp).await,
                );
            }
            LowPriCommand::UpsertEmbeddingChunk {
                model,
                chunk,
                responder,
            } => {
                // No clock reading: an embedding carries no timestamp on either
                // axis. It is a derived artifact of a model applied to content
                // (Doctrine VII), and the ledger already records when the
                // content changed.
                turn.answer_chunk(
                    responder,
                    crate::vector::search::upsert_embedding_chunk(conn, &model, &chunk).await,
                );
            }
            LowPriCommand::DropEmbeddingIndex { model, responder } => {
                // Schema work on the actor's own connection, like
                // RegisterModel: the single-writer invariant is what keeps the
                // drop and the load on one timeline.
                turn.answer(
                    responder,
                    crate::vector::registry::drop_embedding_index(conn, &model).await,
                );
            }
            LowPriCommand::RebuildEmbeddingIndex { model, responder } => {
                // One statement, one turn, no smaller unit. Its hold is the
                // one-pass DiskANN build's and it is budget-exempt by the same
                // criterion as `ShadowSwap` — see the CommandKind doc.
                turn.answer(
                    responder,
                    crate::vector::registry::rebuild_embedding_index(conn, &model).await,
                );
            }
            LowPriCommand::Archive {
                cutoff,
                archive_path,
                responder,
            } => {
                // The archive *time*, not the cutoff. `archive_horizon` records
                // both and they are different facts — see `archive()` (Wave 4.5).
                let archived_at = clock.now();
                let res = archive(conn, &cutoff, &archived_at, &archive_path).await;
                // The session attached and detached a second database. The
                // statements are dropped rather than trusted to recompile —
                // see [`ActorState::forget_statements`].
                state.forget_statements();
                // Before the answer, so a shadow rebuild that reads the epoch on
                // its next turn cannot miss an archive that has already deleted
                // rows out from under it (T1.2).
                if res.is_ok() {
                    turn.archive_committed();
                }
                turn.answer(responder, res);
            }
            LowPriCommand::ArchiveBranch {
                branch,
                archive_path,
                responder,
            } => {
                // The wall clock, recorded in `cold.branches.archived_at`: when
                // the ledger stopped knowing about the lineage. Not a ledger
                // fact and not on either of Doctrine II's timelines — nothing
                // was asserted or retired here.
                let archived_at = clock.now();
                let res = crate::temporal::archive::archive_branch(
                    conn,
                    &branch,
                    &archived_at,
                    &archive_path,
                )
                .await;
                // Both: this session attaches a second database *and* deletes
                // the lineage's row from `branches`.
                state.forget_everything();
                // `Archive`'s reason exactly: a shadow rebuild reading the epoch
                // on its next turn must not miss a session that has already
                // deleted rows out from under it (T1.2).
                if res.is_ok() {
                    turn.archive_committed();
                }
                turn.answer(responder, res);
            }
            LowPriCommand::Rehydrate {
                ids,
                archive_path,
                responder,
            } => {
                let refs: Vec<&str> = ids.iter().map(String::as_str).collect();
                let res = rehydrate(conn, &refs, &archive_path).await;
                // Attaches, like the two archive sessions. It restores concepts
                // and log rows, never a lineage — `cold.branches` is read by
                // `archive_hint` and not written back — so the lineages stand.
                state.forget_statements();
                // Same reason as `Archive`: rehydration moves rows into `links`'
                // parent table, so a shadow rebuild in flight must see the epoch
                // move before the caller is answered (T1.2).
                if res.is_ok() {
                    turn.archive_committed();
                }
                turn.answer(responder, res);
            }
            LowPriCommand::ShadowRebuild { step, responder } => {
                use crate::integrity::{shadow, ShadowOutcome, ShadowStep};
                let res = match step {
                    ShadowStep::Begin => {
                        shadow::begin(conn)
                            .await
                            .map(|build_start| ShadowOutcome::Started {
                                build_start,
                                epoch: turn.epoch(),
                            })
                    }
                    ShadowStep::Fill { after } => shadow::fill_chunk(conn, after.as_deref())
                        .await
                        .map(|last| ShadowOutcome::Filled { last }),
                    ShadowStep::Swap { build_start, epoch } => {
                        shadow::swap(conn, &build_start, epoch, turn.epoch())
                            .await
                            .map(|rows| ShadowOutcome::Swapped { rows })
                    }
                };
                turn.answer(responder, res);
            }
            LowPriCommand::RebuildFts { responder } => {
                let res = conn
                    .execute(crate::schema::ddl::REBUILD_CONCEPTS_FTS, ())
                    .await
                    .map(|_| ())
                    .map_err(Into::into);
                turn.answer(responder, res);
            }
            LowPriCommand::Analyze {
                incremental,
                responder,
            } => {
                // Both go through `query()`, not `execute()`. `PRAGMA optimize`
                // yields rows, and libsql's `execute()` rejects any statement
                // that does ("Execute returned rows") — the same trap
                // `configure` documents. `ANALYZE` does not yield rows, but is
                // issued the same way so the two arms cannot drift into needing
                // different call shapes for no visible reason.
                let sql = if incremental {
                    crate::schema::ddl::OPTIMIZE
                } else {
                    crate::schema::ddl::ANALYZE
                };
                let res = conn.query(sql, ()).await.map(|_| ()).map_err(Into::into);
                turn.answer(responder, res);
            }
        }
        LoopCtl::Continue
    }
}

/// Close an open interval by asserting its successor (Doctrine III).
///
/// Never an `UPDATE`. The replacement row copies weight and properties from
/// current belief and differs only in `valid_to` and `recorded_at`, so the
/// original assertion survives intact and `reconstruct` at an earlier instant
/// still sees the interval open — which is the entire point of a bitemporal
/// ledger.
// The first of these in the crate proper (0.14.8). All nine are the edge key,
// two stamps and the lineage — a struct to carry them would exist for one call
// site and would put a name between the caller and parameters it already spells
// out positionally at the only place it calls this.
#[allow(clippy::too_many_arguments)]
async fn retire_edge(
    conn: &libsql::Connection,
    source: &str,
    target: &str,
    edge_type: &str,
    valid_from: &str,
    valid_to: &str,
    stamp: &str,
    branch: &str,
    shape: LineageShape,
    ancestry: &[Ancestor],
) -> Result<()> {
    // Shadow retirement: the row being closed may belong to an ancestor, and
    // the row written carries *this* lineage's id. See
    // `lineage::retire_from_resolved`.
    //
    // One statement for all three shapes since 0.15.8 (W13.3, D-250). The
    // trunk had its own until then, kept apart on [`LineageShape`]'s ground —
    // the resolved form was opaque to the planner and cost 3.0x where there
    // was nothing to resolve (D-220). That ground is gone rather than
    // overruled: a keyed `Trunk` resolution lowers to no CTEs at all, so the
    // statement the lowering emits *is* the one this arm used to hold, with
    // the lineage stamped rather than defaulted.
    // The ancestry follows the seven, at `RETIRE_ANCESTRY_SLOT`.
    let mut params: Vec<libsql::Value> = vec![
        source.into(),
        target.into(),
        edge_type.into(),
        valid_from.into(),
        branch.into(),
        valid_to.into(),
        stamp.into(),
    ];
    params.extend(crate::graph::lineage::ancestry_params(ancestry));
    let affected = conn
        .execute(
            &crate::graph::lineage::retire_from_resolved(shape, ancestry),
            params,
        )
        .await
        .map_err(DbError::Engine)?;

    if affected == 0 {
        return Err(DbError::NotFound(format!(
            "{source} -> {target} ({edge_type}) at {valid_from}"
        )));
    }
    Ok(())
}

async fn upsert_concept(
    conn: &libsql::Connection,
    concept: &ConceptUpsert,
    stamp: &str,
) -> Result<()> {
    let res = conn
        .execute(UPSERT_CONCEPT, concept_params(concept, stamp))
        .await;

    match res {
        Ok(_) => Ok(()),
        Err(e) => Err(classify(
            conn,
            e,
            WriteOp::Concept {
                id: &concept.id,
                recorded_at: stamp,
                branch: concept.branch_name(),
            },
        )
        .await),
    }
}

/// Whether this pair is the storage layer's case rather than this guard's.
///
/// Two **open** intervals overlap — they share every instant from the later
/// start onwards — so a naive overlap check reports them, and reporting them
/// here would leave `DbError::SingleOpenViolation` constructible by nothing.
/// That variant is the more specific error, it is enforced by
/// `trg_links_single_open` rather than by this function, and its field names
/// were ratified in §1.2. Shadowing it with a general one would be defect Q's
/// shape reintroduced by a fix: a typed error that no code path can produce.
///
/// So the two guards partition the space rather than overlapping it. Both open
/// belongs to the trigger. Everything else — open against closed, closed against
/// closed — is unguarded at the storage layer and belongs here. That the split
/// is exactly the trigger's `WHEN` clause is not a coincidence; it is the
/// definition of what was missing.
fn defer_to_single_open(proposed: &Interval, existing: &Interval) -> bool {
    proposed.is_open() && existing.is_open()
}

/// Refuse an assertion whose valid-time interval overlaps one already recorded
/// for the same `(source, target, edge_type)` — **defect AA, D-060**.
///
/// `trg_links_single_open` fires only `WHEN NEW.valid_to = '9999-…'`, so it
/// guards the open sentinel and nothing else. Two *closed* intervals that
/// overlap were accepted without complaint, and `query_as_of_edges` at an
/// instant inside both returned one relationship as two edges.
///
/// **This runs in the write actor, which is what makes it sound.** The obvious
/// place is `EdgeAssertion::normalized`, and it cannot go there — `normalized`
/// is a pure function with no connection, and doing the read at the API boundary
/// instead would leave a check-then-write race between the read and the actor's
/// insert. Inside the actor there is one writer by construction (D-014), and for
/// the batch paths this runs inside the same transaction as the insert, so the
/// window does not exist rather than being small.
///
/// **What it does not cover, and §4.2 now says so:** raw SQL against the same
/// file. The storage layer permits what this API refuses, which is the honest
/// cost of not putting the check in a trigger. The alternative was a second
/// index probe inside `trg_links_single_open` on every insert — on the path
/// D-059 has just finished making fast — for a guarantee that only holds against
/// callers who were going through the actor anyway.
///
/// `valid_from <> ?4` excludes the row being re-asserted. Re-assertion at the
/// same `valid_from` is Doctrine III's ordinary case — a new belief about the
/// same interval — and is settled by the primary key and the single-open
/// trigger, not here.
/// The single-assertion path holds one statement across turns (0.15.6, D-248);
/// the batch path prepares one inside its own transaction and calls
/// [`check_prepared`] per row.
async fn reject_overlapping_interval(
    state: &mut ActorState,
    conn: &libsql::Connection,
    edge: &EdgeAssertion,
    shape: LineageShape,
) -> Result<()> {
    let branch = edge.branch_name().to_string();
    check_prepared(state.guard(conn, shape, &branch).await?, edge).await
}

/// The guard's body, against a statement the caller has already prepared.
///
/// **Split out because preparing per row was worth 10.4 ms on a 90-edge chunk**
/// (§8.8) — the same defect D-056 and D-057 diagnosed and fixed for
/// `INSERT_LINK`, reintroduced by the Wave 2 guard that was written beside it.
/// Measured with and without the guard, on a 2,000-edge hub: 8.65 ms → 19.25 ms,
/// and *identical* with and without `idx_lc_open_interval`, which is what
/// identified preparation rather than a scan as the cost. A guard that reads an
/// index correctly and prepares its statement 90 times is indistinguishable, at
/// the call site, from one that scans.
///
/// `reset()` between rows is not optional: libsql binds and steps without
/// resetting, so a reused statement must be returned to its initial state.
///
/// # And once more on the way out (0.15.6, W14.3)
///
/// The reset used to be enough at the top, because the guard was compiled per
/// call or per chunk and dropped where it was made — the drop finalized it, and
/// SQLite's objection to a live statement never came up. [`ActorState`] holds
/// this one across turns, and the loop below can leave a cursor open on it: the
/// overlap arm returns from inside the `while`. A statement left mid-scan is
/// what makes SQLite refuse to end a transaction, so the next `Archive` — not
/// the next assertion — would be the thing that failed, a command and a
/// diagnosis apart from the code that caused it. So the scan is a function of
/// its own, and the statement is reset on both ways out of it.
async fn check_prepared(guard: &OverlapGuard, edge: &EdgeAssertion) -> Result<()> {
    let proposed = Interval::new(edge.valid_from.clone(), edge.valid_to.clone());

    guard.stmt.reset();
    // The resolved form takes a fifth parameter, the writing lineage, and
    // returns what that lineage can see; the trunk form takes four and returns
    // the table. Binding five to the trunk statement would be an error from
    // libsql rather than a wrong answer, which is the failure mode to prefer.
    let mut rows = match guard.shape {
        LineageShape::Trunk => {
            guard
                .stmt
                .query(libsql::params![
                    edge.source.as_str(),
                    edge.target.as_str(),
                    edge.edge_type.as_str(),
                    edge.valid_from.as_str()
                ])
                .await?
        }
        LineageShape::Resolved | LineageShape::TrunkOnForked => {
            // The ancestry follows the five, at `GUARD_ANCESTRY_SLOT`, and is
            // empty for `TrunkOnForked` — a root emits no `lineage` relation.
            let mut params: Vec<libsql::Value> = vec![
                edge.source.as_str().into(),
                edge.target.as_str().into(),
                edge.edge_type.as_str().into(),
                edge.valid_from.as_str().into(),
                edge.branch_name().into(),
            ];
            params.extend(crate::graph::lineage::ancestry_params(&guard.ancestry));
            guard.stmt.query(params).await?
        }
    };

    let verdict = scan_candidates(&mut rows, &proposed, edge).await;
    drop(rows);
    guard.stmt.reset();
    verdict
}

/// The guard's loop, over candidates the statement has already produced.
///
/// Split from [`check_prepared`] so that the statement is reset on both exits
/// from it, including the one that returns an overlap.
async fn scan_candidates(
    rows: &mut libsql::Rows,
    proposed: &Interval,
    edge: &EdgeAssertion,
) -> Result<()> {
    while let Some(row) = rows.next().await? {
        let existing = Interval::new(row.get::<String>(0)?, row.get::<String>(1)?);
        if defer_to_single_open(proposed, &existing) {
            continue;
        }
        if proposed.overlaps(&existing) {
            return Err(DbError::OverlappingInterval {
                overlap: Box::new(crate::error::Overlap {
                    source_id: edge.source.clone(),
                    target_id: edge.target.clone(),
                    edge_type: edge.edge_type.clone(),
                    valid_from: edge.valid_from.clone(),
                    valid_to: edge.valid_to.clone(),
                    existing_from: existing.valid_from,
                    existing_to: existing.valid_to,
                    // This guard reads committed rows, so the interval it names
                    // is one the caller can go and look at (D-180).
                    within_batch: false,
                }),
            });
        }
    }

    Ok(())
}

/// The same guard applied *within* a batch, before any of it is written.
///
/// The database check cannot see rows that are not in the database yet, so a
/// batch carrying two overlapping intervals for one relationship would pass
/// every per-row check and commit the overlap in one transaction.
///
/// # Sorted and swept rather than compared pairwise (0.13.6, W7.5, D-179)
///
/// This used to compare every pair. At [`chunk_rows::EDGES`] = 90 that is
/// nothing, and the chunked paths are the only ones where 90 is the bound —
/// [`Database::write_bulk_atomic`] is exempt from [`CHUNK_BUDGET`] by contract,
/// so its batch is whatever the caller passed, and the quadratic term is what
/// made 20,000 corrections to one relationship's history cost seconds rather
/// than milliseconds. Sorting by `(source, target, edge_type, valid_from)` and
/// sweeping costs `n log n` and changes nothing a caller can observe except the
/// wait.
///
/// **Adjacent pairs are not sufficient, and that is the whole difficulty.** For
/// plain intervals they would be: sort by start, and if any two overlap then
/// some neighbouring two overlap. That proof needs every pair to be *eligible*,
/// and here two are not — identical `valid_from` is re-assertion rather than
/// overlap, and two open intervals belong to `trg_links_single_open`. Skip an
/// adjacent pair for either reason and a real overlap can hide behind it:
/// `[5,20)`, `[5,6)`, `[7,8)` has the first pair skipped for equal `valid_from`
/// and the second not overlapping, while `[5,20)` and `[7,8)` overlap plainly.
/// So the sweep carries the widest `valid_to` reached so far instead of looking
/// only backwards one step, and carries a second one restricted to closed
/// intervals — because an open predecessor is excluded for an open candidate
/// and eligible for a closed one, which are different questions with different
/// answers.
///
/// Equal `valid_from` is handled by advancing in runs: everything with the same
/// start is checked against the maxima, and only then folded into them, so the
/// members of a run never see each other.
///
/// The report names the *earlier* interval as the existing one, which is the
/// pairwise version's input order only by accident. Within a batch neither is
/// older in transaction time — they arrive under one stamp — so valid-time order
/// is the only ordering that means anything, and it is the one a reader will
/// assume the words carry.
fn reject_overlaps_within(edges: &[EdgeAssertion]) -> Result<()> {
    // Indices, not the edges. The batch is borrowed and its order is the order
    // the rows are written in; sorting it would either clone it or reorder the
    // caller's data, which is `estimated_bulk_hold`'s reason for grouping too.
    let mut order: Vec<u32> = (0..edges.len() as u32).collect();
    order.sort_unstable_by(|&i, &j| {
        let a = &edges[i as usize];
        let b = &edges[j as usize];
        (
            &a.source,
            &a.target,
            &a.edge_type,
            a.branch_name(),
            &a.valid_from,
        )
            .cmp(&(
                &b.source,
                &b.target,
                &b.edge_type,
                b.branch_name(),
                &b.valid_from,
            ))
    });

    fn key(e: &EdgeAssertion) -> (&str, &str, &str, &str) {
        (
            e.source.as_str(),
            e.target.as_str(),
            e.edge_type.as_str(),
            e.branch_name(),
        )
    }
    let at = |k: usize| &edges[order[k] as usize];

    let mut group = 0;
    while group < order.len() {
        let mut group_end = group + 1;
        while group_end < order.len() && key(at(group_end)) == key(at(group)) {
            group_end += 1;
        }

        // The furthest `valid_to` reached by anything already swept in this key
        // group, and the edge it came from so the error can name it. The second
        // one ignores open intervals: an open candidate may not be compared
        // against an open predecessor, and the sentinel would otherwise win the
        // maximum every time and make every such pair look like an overlap.
        let mut widest: Option<&EdgeAssertion> = None;
        let mut widest_closed: Option<&EdgeAssertion> = None;

        let mut run = group;
        while run < group_end {
            let mut run_end = run + 1;
            while run_end < group_end && at(run_end).valid_from == at(run).valid_from {
                run_end += 1;
            }

            for k in run..run_end {
                let e = at(k);
                let existing = if e.valid_to == timestamp::OPEN_SENTINEL {
                    widest_closed
                } else {
                    widest
                };
                let Some(p) = existing else { continue };
                // `Interval::overlaps` is `max(from) < min(to)`, and the sort
                // has already settled the max: `p.valid_from <= e.valid_from`.
                // What is left is the same predicate with the maximum resolved,
                // and it is written out rather than allocating two `Interval`s
                // per row to ask the same question.
                if e.valid_from < p.valid_to && e.valid_from < e.valid_to {
                    return Err(DbError::OverlappingInterval {
                        overlap: Box::new(crate::error::Overlap {
                            source_id: e.source.clone(),
                            target_id: e.target.clone(),
                            edge_type: e.edge_type.clone(),
                            valid_from: e.valid_from.clone(),
                            valid_to: e.valid_to.clone(),
                            existing_from: p.valid_from.clone(),
                            existing_to: p.valid_to.clone(),
                            // Nothing here is in the database, and the batch is
                            // refused whole, so nothing here ever will be. The
                            // message has to say so (D-180).
                            within_batch: true,
                        }),
                    });
                }
            }

            for k in run..run_end {
                let e = at(k);
                if widest.is_none_or(|w| e.valid_to > w.valid_to) {
                    widest = Some(e);
                }
                if e.valid_to != timestamp::OPEN_SENTINEL
                    && widest_closed.is_none_or(|w| e.valid_to > w.valid_to)
                {
                    widest_closed = Some(e);
                }
            }

            run = run_end;
        }

        group = group_end;
    }

    Ok(())
}

/// Write every edge or none, under a single stamp.
///
/// **The statement is prepared once for the whole chunk (§9, D-056).** It used to
/// be `tx.execute(INSERT_LINK, …)` per row, which re-prepares on every call — and
/// `links` carries two triggers, so each preparation compiles their bodies along
/// with the insert.
///
/// Measured at 500 rows: **≈62 ms → ≈37 ms, a 41% saving.** Preparation was a
/// large cost and *not* the dominant one, which the first guess had it as. The
/// residual is the triggers themselves: the same 500 rows with
/// `trg_links_log_insert` and `trg_links_current_sync` dropped commit in **2.96
/// ms**, so trigger amplification is ~92% of what remains. There is no further
/// win available here without changing what the ledger records, and Doctrine IV
/// is what says it must be recorded. See D-056 for what that implies about §9's
/// ≤ 3 ms budget — briefly, 2.96 ms *is* the un-amplified figure, so the budget
/// appears to have been set without the amplification its own preamble says is
/// included.
///
/// `reset()` between rows is not optional: libsql's `execute` binds and steps
/// without resetting, so a reused statement must be returned to its initial state
/// or the second row steps a completed statement.
async fn write_edges_atomic(
    state: &mut ActorState,
    conn: &libsql::Connection,
    edges: &[EdgeAssertion],
    stamp: &str,
) -> Result<usize> {
    if edges.is_empty() {
        return Ok(0);
    }

    // Before the transaction opens: a batch that contradicts itself is refused
    // without taking the write lock at all (D-060), and a batch naming a
    // lineage that does not exist is refused before it can take the lock at all
    // (0.14.8).
    reject_overlaps_within(edges)?;
    let branches = distinct_branches(edges);
    let (shape, lineages) = check_lineages_with(state, conn, &branches).await?;
    let lineages = lineages.clone();

    let tx = conn
        .transaction_with_behavior(libsql::TransactionBehavior::Immediate)
        .await?;

    // Inside the transaction, so the rows this checks against cannot change
    // between the check and the insert.
    // One preparation for the whole chunk, not one per row — see
    // `check_prepared`, and D-056 for the same lesson learned on `INSERT_LINK`.
    // One per lineage the chunk names, not one per row — see
    // `OverlapGuard::prepare` for why the shape alone stopped being enough, and
    // `distinct_branches` for why this is a `Vec` of length one in every batch
    // this crate has seen.
    let mut guards = Vec::with_capacity(branches.len());
    for name in &branches {
        guards.push(OverlapGuard::prepare(&tx, shape, &lineages, name).await?);
    }
    for edge in edges {
        let guard = guards
            .iter()
            .find(|g| g.answers_for(shape, edge.branch_name()))
            .expect("a guard per distinct branch, and the row names one of them");
        if let Err(e) = check_prepared(guard, edge).await {
            // Released before the rollback: a live statement on the connection
            // is what makes SQLite refuse to end a transaction.
            drop(guards);
            let _ = tx.rollback().await;
            return Err(e);
        }
    }
    drop(guards);

    let stmt = tx.prepare(INSERT_LINK).await?;

    for edge in edges {
        stmt.reset();
        let res = stmt.execute(edge_params(edge, stamp)).await;

        if let Err(e) = res {
            let typed = classify(
                &tx,
                e,
                WriteOp::Edge {
                    source_id: &edge.source,
                    target_id: &edge.target,
                    edge_type: &edge.edge_type,
                },
            )
            .await;
            // Released before the rollback: a live statement on the connection
            // is exactly what makes SQLite refuse to end a transaction.
            drop(stmt);
            let _ = tx.rollback().await;
            return Err(typed);
        }
    }

    drop(stmt);
    tx.commit().await?;
    Ok(edges.len())
}

/// Write every concept or none, under a single stamp.
/// Upsert one chunk of derived annotations in a single transaction (D-041).
///
/// `stamp` is the actor's clock reading, exactly as for every other chunk — but
/// it lands in `computed_at`, not in a `recorded_at`, and the difference is not
/// cosmetic. `recorded_at` is the transaction-time axis and is subject to
/// Doctrine II and the monotonicity guard; `computed_at` is a note about when a
/// derivation last ran, on a table the ledger does not see. Rerunning an
/// algorithm therefore replaces the row and advances the note, rather than
/// versioning a concept the world did not change.
///
/// # Failures name the concept (0.13.3, W7.2, D-176)
///
/// This was the one write path in the crate that returned
/// [`DbError::Engine`] raw, and the omission looked harmless: the table
/// carries no triggers, so none of [`crate::error::AbortKind`]'s guards can
/// fire on it and [`classify`] would have returned the same raw error it was
/// given. What that reasoning missed is the foreign key onto `concepts`, which
/// the engine enforces itself. Annotating a concept that does not exist is the
/// one failure a caller can cause here, and it reported as
/// `FOREIGN KEY constraint failed` with no row named — out of a chunk of up to
/// [`chunk_rows::ANNOTATIONS`].
///
/// It now goes through [`classify`] with [`WriteOp::Annotation`] like every
/// other write, and a missing concept returns [`DbError::NotFound`] carrying
/// its id.
async fn write_annotations_atomic(
    conn: &libsql::Connection,
    annotations: &[Annotation],
    stamp: &str,
) -> Result<usize> {
    if annotations.is_empty() {
        return Ok(0);
    }

    let tx = conn
        .transaction_with_behavior(libsql::TransactionBehavior::Immediate)
        .await?;

    let stmt = tx
        .prepare(
            "INSERT INTO analytics_annotations (concept_id, label, value, computed_at) \
             VALUES (?1, ?2, ?3, ?4) \
             ON CONFLICT(concept_id, label) DO UPDATE SET \
                 value = excluded.value, computed_at = excluded.computed_at",
        )
        .await?;

    for a in annotations {
        stmt.reset();
        let res = stmt
            .execute(libsql::params![
                a.concept_id.as_str(),
                a.label.as_str(),
                a.value.as_str(),
                stamp
            ])
            .await;
        if let Err(e) = res {
            let typed = classify(
                &tx,
                e,
                WriteOp::Annotation {
                    concept_id: &a.concept_id,
                },
            )
            .await;
            drop(stmt);
            let _ = tx.rollback().await;
            return Err(typed);
        }
    }

    drop(stmt);
    tx.commit().await?;
    Ok(annotations.len())
}

async fn write_concepts_atomic(
    state: &mut ActorState,
    conn: &libsql::Connection,
    concepts: &[ConceptUpsert],
    stamp: &str,
) -> Result<usize> {
    if concepts.is_empty() {
        return Ok(0);
    }

    // Named lineages, before the write lock — `check_lineages`' reason, and the
    // shape it also returns is unused here because `concepts` is keyed by
    // identity and has no resolution to do (see `ConceptUpsert::branch`).
    let mut named: Vec<&str> = Vec::with_capacity(1);
    for concept in concepts {
        let name = concept.branch_name();
        if !named.contains(&name) {
            named.push(name);
        }
    }
    check_lineages(state, conn, &named).await?;

    let tx = conn
        .transaction_with_behavior(libsql::TransactionBehavior::Immediate)
        .await?;

    // Prepared once, like the edge chunk (D-056). This no longer routes through
    // [`upsert_concept`] — that function prepares per call by construction — but
    // it shares that function's statement text and parameter row, so the two
    // cannot upsert different columns.
    let stmt = tx.prepare(UPSERT_CONCEPT).await?;

    for concept in concepts {
        stmt.reset();
        let res = stmt.execute(concept_params(concept, stamp)).await;

        if let Err(e) = res {
            let typed = classify(
                &tx,
                e,
                WriteOp::Concept {
                    id: &concept.id,
                    recorded_at: stamp,
                    branch: concept.branch_name(),
                },
            )
            .await;
            drop(stmt);
            let _ = tx.rollback().await;
            return Err(typed);
        }
    }

    drop(stmt);
    tx.commit().await?;
    Ok(concepts.len())
}

#[cfg(test)]
mod lineage_cache {
    //! [`Lineages::shape_of`] against every shape combination (0.15.6, W14.3).
    //!
    //! Unit tests rather than a write through the actor, because the case this
    //! function exists for **cannot be observed from outside**: where a batch
    //! names lineages of different shapes, both of them currently compile the
    //! same overlap statement, so a wrong choice between them is invisible
    //! until W13.3 gives the guard a third lowering. A behavioural test would
    //! pass on the code this replaces and on the code that replaces it, and
    //! would go on passing through the release that made it matter.

    use super::*;

    /// `(name, is_root)`, kept as the fixture's spelling because that is what
    /// these tests are about — the shape, not the ancestry. A root is a row
    /// with no parent and no fork point, which is the pairing the `branches`
    /// CHECK enforces; a non-root is given both, since a row with one and not
    /// the other is not a state the schema permits.
    fn lineages(rows: &[(&str, bool)]) -> Lineages {
        Lineages {
            rows: rows
                .iter()
                .map(|(id, root)| crate::graph::lineage::BranchRow {
                    id: (*id).into(),
                    parent: (!root).then(|| "main".to_string()),
                    forked_at: (!root).then(|| "2026-01-01T00:00:00.000000Z".to_string()),
                })
                .collect(),
        }
    }

    fn trunk_only() -> Lineages {
        lineages(&[("main", true)])
    }

    fn forked() -> Lineages {
        lineages(&[("main", true), ("alt", false), ("other", false)])
    }

    /// One lineage is the trunk, whoever asks.
    #[test]
    fn one_lineage_is_the_trunk() {
        assert_eq!(
            trunk_only().shape_of(&["main"]).unwrap(),
            LineageShape::Trunk
        );
    }

    /// A root on a forked database is `TrunkOnForked`; anything else resolves.
    #[test]
    fn the_shape_reads_the_name_and_not_only_the_count() {
        let l = forked();
        assert_eq!(
            l.shape_of(&["main"]).unwrap(),
            LineageShape::TrunkOnForked,
            "a root has no ancestors and D-244 emits that reduction directly"
        );
        assert_eq!(l.shape_of(&["alt"]).unwrap(), LineageShape::Resolved);
    }

    /// The case review C-24 is about: names of two different shapes.
    ///
    /// Pinned in **both orders**, which is the whole content of the finding.
    /// The loop this replaces returned whichever came last, so a batch naming
    /// `[main, alt]` and one naming `[alt, main]` disagreed about their own
    /// shape while describing the same set of lineages.
    #[test]
    fn a_batch_of_mixed_shapes_resolves_rather_than_taking_the_last_name() {
        let l = forked();
        assert_eq!(
            l.shape_of(&["main", "alt"]).unwrap(),
            LineageShape::Resolved,
            "the resolved form is exact for a root as well, which is the \
             argument OverlapGuard::prepare already makes"
        );
        assert_eq!(
            l.shape_of(&["alt", "main"]).unwrap(),
            LineageShape::Resolved,
            "and it must not depend on the order the batch happened to name them"
        );
    }

    /// Agreement is kept rather than widened: two forks are still `Resolved`,
    /// and two mentions of one root are still `TrunkOnForked`.
    #[test]
    fn names_that_agree_keep_their_shape() {
        let l = forked();
        assert_eq!(
            l.shape_of(&["alt", "other"]).unwrap(),
            LineageShape::Resolved
        );
        assert_eq!(
            l.shape_of(&["main", "main"]).unwrap(),
            LineageShape::TrunkOnForked
        );
    }

    /// Existence is checked for **every** name, not only the one that decides.
    ///
    /// The check is why the loop existed at all, and the shape was the thing it
    /// returned on the way past. A version that stopped at the first name would
    /// let a batch name a lineage that does not exist and be refused later by
    /// the foreign key, from inside a rolled-back transaction, naming neither
    /// the column nor the branch — `check_lineages`' opening paragraph.
    #[test]
    fn an_unknown_name_is_refused_wherever_it_sits() {
        let l = forked();
        for names in [
            ["ghost", "main"].as_slice(),
            ["main", "ghost"].as_slice(),
            ["main", "ghost", "alt"].as_slice(),
        ] {
            match l.shape_of(names) {
                Err(DbError::UnknownBranch(name)) => assert_eq!(name, "ghost"),
                other => panic!("expected UnknownBranch for {names:?}, got {other:?}"),
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn edge(target: &str, micros: usize) -> EdgeAssertion {
        EdgeAssertion::new("src", target, "LINKS")
            .valid_from(format!("2026-01-01T00:00:00.{micros:06}Z"))
            .valid_to(format!("2026-01-01T00:00:00.{:06}Z", micros + 1))
    }

    /// The estimate must **no longer** depend on the batch's shape (0.13.6).
    ///
    /// Its dependence on shape was correct for as long as the guard was
    /// quadratic and the constant differed 16× between the two paths through
    /// its inner loop. W7.5 removed that term, and measurement agrees: 1.94 s
    /// and 2.22 s for the two 20,000-edge batches that used to differ by 7×.
    /// A model that kept predicting a 7× spread would now be wrong in the
    /// *expensive* direction — warning loudly about a batch that is fine.
    #[test]
    fn two_batches_of_one_size_are_predicted_alike() {
        const N: usize = 20_000;
        let fanout: Vec<_> = (0..N).map(|i| edge(&format!("t{i:07}"), i)).collect();
        let history: Vec<_> = (0..N).map(|i| edge("t0", i)).collect();

        assert_eq!(
            estimated_bulk_hold(&fanout),
            estimated_bulk_hold(&history),
            "the guard no longer reads the batch's shape, so neither may this"
        );
    }

    /// Measured on libSQL 0.9.30 after W7.5: 1.94 s and 2.22 s for those two
    /// batches, against 2.6 s and 18.1 s before it. This pins that the model
    /// still tracks them — a coefficient edited without re-measuring fails here.
    #[test]
    fn the_estimate_matches_what_was_measured() {
        const N: usize = 20_000;
        let fanout: Vec<_> = (0..N).map(|i| edge(&format!("t{i:07}"), i)).collect();
        let history: Vec<_> = (0..N).map(|i| edge("t0", i)).collect();

        for (batch, measured_ms, label) in
            [(fanout, 1_936u128, "fanout"), (history, 2_220, "history")]
        {
            let predicted = estimated_bulk_hold(&batch).as_millis();
            let ratio = predicted as f64 / measured_ms as f64;
            assert!(
                (0.8..1.25).contains(&ratio),
                "{label}: predicted {predicted} ms against a measured \
                 {measured_ms} ms ({ratio:.2}x). Re-run \
                 examples/bulk_atomic_diag.rs before changing the coefficients."
            );
        }
    }

    /// `ilog2` panics on zero, and an empty batch is the caller asking whether
    /// a batch they have not built yet would be slow.
    #[test]
    fn an_empty_batch_estimates_nothing_rather_than_panicking() {
        assert_eq!(estimated_bulk_hold(&[]), std::time::Duration::ZERO);
        let one = [edge("t0", 0)];
        assert_eq!(
            estimated_bulk_hold(&one),
            std::time::Duration::from_nanos(7_400)
        );
    }

    /// The model is used as a threshold test, so it must not go backwards.
    #[test]
    fn a_bigger_batch_never_predicts_a_shorter_hold() {
        let mut last = std::time::Duration::ZERO;
        for n in [1usize, 2, 3, 7, 8, 100, 511, 512, 513, 5_000, 20_000] {
            let batch: Vec<_> = (0..n).map(|i| edge(&format!("t{i:07}"), i)).collect();
            let now = estimated_bulk_hold(&batch);
            assert!(now >= last, "{n} rows predicts {now:?} after {last:?}");
            last = now;
        }
    }

    /// The warning threshold sits well above the bound this path is exempt from.
    ///
    /// Warning at `CHUNK_BUDGET` would fire on batches working exactly as
    /// designed — the exemption is a contract (D-014), not a failure — and a
    /// warning that fires on correct behaviour gets filtered out, taking the
    /// 18-second case with it.
    #[test]
    fn the_warning_threshold_is_not_the_chunk_budget() {
        assert!(BULK_ATOMIC_WARN_HOLD > CHUNK_BUDGET * 10);
    }

    // -----------------------------------------------------------------------
    // reject_overlaps_within — sorted and swept (0.13.6, W7.5, D-179)
    //
    // The pairwise version was obviously correct and too slow; this one is
    // neither, so what follows pins the cases where the obvious fix is wrong
    // rather than only the cases the guard already caught.
    // -----------------------------------------------------------------------

    /// An edge over an explicit interval, all four key columns spelled out.
    fn span(target: &str, edge_type: &str, from: usize, to: Option<usize>) -> EdgeAssertion {
        let stamp = |n: usize| format!("2026-01-01T00:00:00.{n:06}Z");
        EdgeAssertion::new("src", target, edge_type)
            .valid_from(stamp(from))
            .valid_to(to.map_or_else(|| timestamp::OPEN_SENTINEL.to_string(), stamp))
    }

    fn closed(from: usize, to: usize) -> EdgeAssertion {
        span("t0", "LINKS", from, Some(to))
    }

    fn open_at(from: usize) -> EdgeAssertion {
        span("t0", "LINKS", from, None)
    }

    /// The case that makes adjacent pairs insufficient.
    ///
    /// Sort by start and any overlap shows up between neighbours — but only if
    /// every neighbouring pair is eligible to be checked. `[5,20)` and `[5,6)`
    /// are not: identical `valid_from` is re-assertion. Skip them, and `[5,6)`
    /// against `[7,8)` is a clean gap, and the plain overlap between `[5,20)`
    /// and `[7,8)` never gets looked at.
    #[test]
    fn an_overlap_hidden_behind_an_equal_valid_from_is_still_found() {
        let batch = vec![closed(5, 20), closed(5, 6), closed(7, 8)];
        assert!(matches!(
            reject_overlaps_within(&batch),
            Err(DbError::OverlappingInterval { .. })
        ));
    }

    /// The same trap in the other direction: skipped for being open.
    ///
    /// Two open intervals are `trg_links_single_open`'s case and are passed
    /// over here. A running maximum that counted them would take the sentinel
    /// as the widest reach and report every later open interval as overlapping
    /// it — inventing an error rather than missing one, which is why the sweep
    /// carries a second maximum restricted to closed intervals.
    #[test]
    fn two_open_intervals_are_left_to_the_trigger() {
        let batch = vec![closed(1, 5), open_at(10), open_at(20)];
        assert!(reject_overlaps_within(&batch).is_ok());
    }

    /// An open interval still overlaps a closed one that reaches past its start.
    #[test]
    fn an_open_interval_over_a_closed_one_is_an_overlap() {
        let batch = vec![closed(1, 50), open_at(10)];
        assert!(matches!(
            reject_overlaps_within(&batch),
            Err(DbError::OverlappingInterval { .. })
        ));
    }

    /// Same `valid_from`, different `valid_to`: a batch correcting itself.
    ///
    /// Last writer wins by `seq_id`, exactly as it does across batches. The
    /// guard has no opinion.
    #[test]
    fn equal_valid_from_is_re_assertion_not_overlap() {
        let batch = vec![closed(5, 20), closed(5, 6), closed(5, 900)];
        assert!(reject_overlaps_within(&batch).is_ok());
    }

    /// Grouping is what makes the sweep sound, so it is pinned rather than read.
    #[test]
    fn edges_with_different_keys_do_not_see_each_other() {
        let batch = vec![
            span("t0", "LINKS", 1, Some(50)),
            span("t1", "LINKS", 10, Some(60)),
            span("t0", "CITES", 10, Some(60)),
            span("t0", "LINKS", 50, Some(60)),
        ];
        assert!(reject_overlaps_within(&batch).is_ok());
    }

    /// Which of the two the report calls *existing* (0.13.6).
    ///
    /// Neither is older in transaction time — a batch lands under one stamp —
    /// so the pairwise version's answer was its input order, which means
    /// nothing. Valid-time order is the only ordering the two intervals have.
    #[test]
    fn the_report_names_the_earlier_interval_as_the_existing_one() {
        let batch = vec![closed(7, 8), closed(5, 20)];
        let Err(DbError::OverlappingInterval { overlap }) = reject_overlaps_within(&batch) else {
            panic!("the batch overlaps itself");
        };
        assert!(overlap.valid_from.ends_with(".000007Z"), "{overlap:?}");
        assert!(overlap.existing_from.ends_with(".000005Z"), "{overlap:?}");
    }

    /// A guard whose answer depended on the caller's ordering would be a worse
    /// guard than the one it replaced, and sorting is exactly the change that
    /// could introduce that.
    #[test]
    fn the_answer_does_not_depend_on_the_order_the_caller_passed() {
        let mut batch = vec![closed(5, 20), closed(5, 6), closed(7, 8)];
        batch.reverse();
        assert!(reject_overlaps_within(&batch).is_err());

        let mut clean = vec![closed(1, 5), closed(5, 6), closed(7, 8), open_at(8)];
        clean.reverse();
        assert!(reject_overlaps_within(&clean).is_ok());
    }

    /// §2.6, and the reason the rewrite happened rather than the doc alone.
    ///
    /// Every edge shares a key, so the old loop reached `Interval::overlaps`
    /// on all n(n−1)/2 pairs — 50 million of them here, which is seconds even
    /// in release and considerably worse in the debug profile this runs under.
    /// The bound is loose on purpose: it is an order of magnitude, not a
    /// benchmark, and the only thing it can fail on is the quadratic term
    /// coming back.
    #[test]
    fn one_relationships_whole_history_is_no_longer_quadratic() {
        const N: usize = 10_000;
        let batch: Vec<_> = (0..N).map(|i| closed(i * 2, i * 2 + 1)).collect();

        let started = std::time::Instant::now();
        assert!(reject_overlaps_within(&batch).is_ok());
        let took = started.elapsed();

        assert!(
            took < std::time::Duration::from_secs(2),
            "{N} same-key edges took {took:?} in the guard"
        );
    }

    // -----------------------------------------------------------------------
    // next_chunk_size — the control law (0.12.0, W2)
    //
    // All of these run without a database, a clock or an actor, which is why
    // W2 comes before W3: the loop that will use this function can only be
    // tested against a real write, and the properties below cannot be observed
    // there without also observing the machine.
    // -----------------------------------------------------------------------

    use std::time::Duration;

    /// `next_chunk_size` with the shipped budget and floor.
    fn step_to(current: usize, held_ms: f64, ceiling: usize) -> usize {
        next_chunk_size(
            current,
            Duration::from_nanos((held_ms * 1_000_000.0) as u64),
            CHUNK_BUDGET,
            CHUNK_FLOOR,
            ceiling,
        )
    }

    /// The edge path, which is every test here that does not say otherwise.
    fn step(current: usize, held_ms: f64) -> usize {
        step_to(current, held_ms, chunk_rows::EDGES)
    }

    /// Iterate the law against a machine that costs `per_row_us` per row plus a
    /// fixed `overhead_ms` per transaction — the two-term model D-142 measured.
    fn converge(
        start: usize,
        per_row_us: f64,
        overhead_ms: f64,
        ceiling: usize,
        steps: usize,
    ) -> Vec<usize> {
        let mut size = start;
        (0..steps)
            .map(|_| {
                let held = overhead_ms + per_row_us * size as f64 / 1000.0;
                size = step_to(size, held, ceiling);
                size
            })
            .collect()
    }

    /// The reason the shrink is proportional rather than a halving: at 4× over
    /// budget, halving needs three steps and every one of them is a latency
    /// miss a caller can feel.
    ///
    /// Run on the annotations path, because it is the only one whose ceiling
    /// leaves room to start far above a size that is reachable — on the edge
    /// path a 4× miss lands under [`CHUNK_FLOOR`], which is a different test.
    #[test]
    fn a_chunk_far_over_budget_converges_from_above_in_at_most_two_steps() {
        const CEILING: usize = chunk_rows::ANNOTATIONS;
        let (per_row_us, overhead_ms) = (20.0, 0.05);
        let held = |n: usize| overhead_ms + per_row_us * n as f64 / 1000.0;
        assert!(
            held(CEILING) > 4.0 * 3.0,
            "the start is not far over budget"
        );

        let trace = converge(CEILING, per_row_us, overhead_ms, CEILING, 4);
        let first_in_budget = trace
            .iter()
            .position(|&n| held(n) <= 3.0)
            .expect("never reached the budget");
        assert!(
            first_in_budget <= 1,
            "took {} steps to get under budget: {trace:?}",
            first_in_budget + 1
        );
    }

    /// Growth is additive, so a size that is merely comfortable cannot leap the
    /// ceiling — and cannot overshoot the budget by more than a quarter.
    #[test]
    fn growth_is_slow_and_shrinking_is_fast() {
        let grown = step(40, 1.0);
        assert!(
            (41..=50).contains(&grown),
            "40 rows at 1 ms should grow by about a quarter, got {grown}"
        );
        let shrunk = step(90, 9.0);
        assert!(
            shrunk <= 40,
            "90 rows at 3x the budget should shrink proportionally, got {shrunk}"
        );
    }

    /// The dead band. Between `budget / 2` and `budget` the size is right and
    /// moving it only costs a re-measurement; without this the law oscillates
    /// across the bound forever.
    #[test]
    fn a_chunk_inside_the_band_is_left_alone() {
        for held_ms in [1.6, 2.0, 2.5, 2.9, 3.0] {
            assert_eq!(step(60, held_ms), 60, "moved at {held_ms} ms");
        }
        assert_ne!(step(60, 1.4), 60, "did not grow at well under half budget");
    }

    /// Both clamps, and the floor's violation stated as a test rather than only
    /// as a comment: a populated table drives this to `CHUNK_FLOOR` and holds it
    /// there **over budget**, which is [`CHUNK_FLOOR`]'s documented trade.
    #[test]
    fn the_floor_and_the_ceiling_both_hold() {
        // 118 µs/row + 0.03 ms fixed — the populated arm, where 35 rows is
        // ~4.1 ms and no size in range meets the bound.
        let trace = converge(chunk_rows::EDGES, 118.0, 0.03, chunk_rows::EDGES, 8);
        assert!(
            trace.iter().all(|&n| n >= CHUNK_FLOOR),
            "fell through the floor: {trace:?}"
        );
        assert_eq!(*trace.last().unwrap(), CHUNK_FLOOR, "settled off the floor");

        // A free machine cannot grow past the path's constant.
        let fast = converge(CHUNK_FLOOR, 1.0, 0.01, chunk_rows::EDGES, 40);
        assert_eq!(*fast.last().unwrap(), chunk_rows::EDGES);
        assert!(fast.iter().all(|&n| n <= chunk_rows::EDGES));
    }

    /// Zero is the one answer that cannot be recovered from: a loop asked for
    /// chunks of no rows makes no progress and never finishes. Degenerate
    /// inputs included, since `held` is a measurement and measurements arrive
    /// from a machine under load.
    #[test]
    fn the_law_never_returns_zero() {
        let cases = [
            (0usize, Duration::ZERO),
            (0, Duration::from_secs(60)),
            (1, Duration::from_secs(60)),
            (90, Duration::from_secs(3600)),
            (usize::MAX, Duration::from_nanos(1)),
            (1, Duration::ZERO),
        ];
        for (current, held) in cases {
            for (floor, ceiling) in [(35, 90), (1, 1), (0, 0), (90, 35)] {
                let n = next_chunk_size(current, held, CHUNK_BUDGET, floor, ceiling);
                assert!(
                    n > 0,
                    "returned 0 for current={current}, held={held:?}, \
                     floor={floor}, ceiling={ceiling}"
                );
            }
        }
    }

    /// A zero budget is not a configuration anyone should reach, but it is one
    /// division away from a panic, so it is pinned.
    #[test]
    fn a_zero_budget_shrinks_to_the_floor_rather_than_dividing_by_it() {
        assert_eq!(
            next_chunk_size(90, Duration::from_millis(1), Duration::ZERO, 35, 90),
            35
        );
    }

    /// A panicked write actor is reported, and the report says so (W7.3, D-177).
    ///
    /// This is the branch `close()` actually has. Through 0.13.3 it sat beside
    /// `Ok(res) => res?` on an actor `Result` that could never be `Err`, and the
    /// pair looked like two failure paths under review — so the one that cannot
    /// fire was carried through two signatures and the one that can had no test.
    ///
    /// The `JoinError` is real rather than mocked: `JoinError` has no public
    /// constructor, and one built by hand would pin the mapping against a value
    /// tokio does not produce.
    #[tokio::test]
    async fn a_writer_that_panicked_is_reported_by_close() {
        // Swallow the panic's own output. The task is *meant* to panic, and a
        // backtrace in a green suite trains people to skim it.
        let prev = std::panic::take_hook();
        std::panic::set_hook(Box::new(|_| {}));
        let handle = tokio::spawn(async { panic!("the write connection is gone") });
        let joined = handle.await;
        std::panic::set_hook(prev);

        assert!(
            joined.is_err(),
            "the task must have panicked for this to test anything"
        );

        match writer_exit(joined) {
            Err(DbError::WriterStopped(reason)) => {
                assert!(
                    reason.contains("did not exit cleanly"),
                    "the message must say what happened: {reason}"
                );
            }
            other => panic!("a panicked actor must be WriterStopped, got {other:?}"),
        }
    }

    /// An actor that ran to completion closes clean.
    ///
    /// The other half, and the one that must not acquire a failure mode by
    /// accident: `run_writer_actor` returns `()`, so the only way this can start
    /// reporting an error is if someone gives the actor a `Result` again.
    #[tokio::test]
    async fn a_writer_that_finished_normally_closes_clean() {
        let handle = tokio::spawn(async {});
        assert!(writer_exit(handle.await).is_ok());
    }

    /// A token is a handle to one flag, not a value that is copied (0.13.8,
    /// W7.6). The clone the caller keeps and the clone the import holds have to
    /// be the same flag, or `cancel()` reaches nothing.
    #[test]
    fn a_cloned_token_cancels_the_original() {
        let token = CancelToken::new();
        let held_by_the_import = token.clone();
        assert!(!held_by_the_import.is_cancelled());
        token.cancel();
        assert!(held_by_the_import.is_cancelled());
        // And it stays cancelled: there is no un-cancel, deliberately, because
        // a token that could be reset would let a second import inherit a
        // decision made about the first.
        token.cancel();
        assert!(held_by_the_import.is_cancelled());
    }

    /// The default control is the one the plain bulk methods pass, and it must
    /// never stop a write.
    #[test]
    fn the_default_control_neither_cancels_nor_reports() {
        let control = BulkControl::new();
        assert!(!control.is_cancelled());
        // No callback, so this is a no-op rather than a panic on an `unwrap`.
        control.report(BulkProgress {
            written: 1,
            total: 1,
            rows: 1,
            held: std::time::Duration::ZERO,
        });
    }

    /// The callback receives what it was promised, once per call to `report`.
    #[test]
    fn progress_reaches_the_callback_unchanged() {
        let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
        let control = BulkControl::new().on_progress({
            let seen = Arc::clone(&seen);
            move |p| seen.lock().unwrap().push(p)
        });
        let sample = BulkProgress {
            written: 180,
            total: 900,
            rows: 90,
            held: std::time::Duration::from_millis(12),
        };
        control.report(sample);
        assert_eq!(*seen.lock().unwrap(), vec![sample]);
    }
}