agentplane 0.4.0

//! The embedded journal, on redb.
//!
//! Two of the runtime's three core guarantees are enforced *here*, by the shape
//! of the data rather than by application logic — because application logic can
//! be bypassed by the next caller:
//!
//! * **Exactly-once** — [`EFFECT_ONCE`] is keyed by `(run_id, effect_key)`, so a
//!   second start for one effect is not rejected, it is *inexpressible*. `insert`
//!   returns the prior value, and a prior value means the effect already began.
//!   This is stronger than the partial unique index it replaces: an index is a
//!   declaration a later migration can drop, a key is the table's identity.
//! * **Fencing** — the lease epoch is read and the records written in **one**
//!   write transaction. redb gives multi-table atomicity, so there is no window
//!   between the check and the append for a stale writer to slip into.
//!
//! redb is pure Rust and two crates deep. That is not tidiness: the previous
//! backend linked a cloud sync engine and a vector-search index that nothing
//! here uses, and every downstream consumer paid for them.
//!
//! It is also **synchronous**, so writes go through [`spawn_blocking`]. A commit
//! fsyncs, and an fsync on the async executor stalls every other task in the
//! process.
//!
//! [`spawn_blocking`]: tokio::task::spawn_blocking

use std::path::Path;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use async_trait::async_trait;
use redb::{Database, ReadableDatabase, ReadableTable, TableDefinition};

use crate::core::{Digest, EffectKey, Epoch, RunId, Seq, StoreError};
use crate::journal::{Append, Cancellation, Head, JournalStore, Lease, Record};

/// `(run_id, seq) -> encoded record`. Ordered by construction, so a run's
/// history is a range scan and its head is the last entry of that range.
const JOURNAL: TableDefinition<(&str, u64), &[u8]> = TableDefinition::new("journal");

/// Every journaled record that belongs to a case, in case order.
///
/// `(tenant, case, run, seq) -> ()`. The primary row stays in [`JOURNAL`]; this
/// is only the ordering, so a record has exactly one home and the two cannot
/// disagree about its bytes.
///
/// Why an index rather than a filter: *show me everything about this matter* is
/// the question a regulated deployment asks, and answering it by listing a
/// case's runs and reading each one is a join whose cost grows with the case's
/// life. It also **misses** any record written by a run the case does not own —
/// which is exactly what a sweep is, since one tick may escalate several cases
/// and belongs to none of them.
///
/// Tenant-first, like every other key here, so a scan that forgets the
/// predicate returns nothing rather than somebody else's matter.
///
/// The write costs one B-tree insert inside the transaction that was already
/// open. Measured against uncased appends it is **below the noise floor**: two
/// runs of 2,000 appends disagreed about the sign, because a durable append is
/// dominated by the fsync and this is not. A number quoted from either run
/// would have been noise with a decimal point.
const JOURNAL_BY_CASE: TableDefinition<(&str, &str, &str, u64), ()> =
    TableDefinition::new("journal_by_case");

/// `(run_id, effect_key) -> seq`. Exactly-once. Only `EffectStarted` is written
/// here, which is the `WHERE` clause of the partial index this replaces.
const EFFECT_ONCE: TableDefinition<(&str, &str), u64> = TableDefinition::new("effect_once");

/// `run_id -> (owner, epoch, expires_at)`.
const RUN_LEASE: TableDefinition<&str, (&str, u64, u64)> = TableDefinition::new("run_lease");

/// `run_id -> (outcome, chain_head, sealed_at)`.
const RUN_SEAL: TableDefinition<&str, (&str, &[u8], u64)> = TableDefinition::new("run_seal");

/// `(tenant, outcome, ordinal) -> run_key`: concluded runs by how they ended.
///
/// A **derived** index. The outcome's home is the chain — the executor appends
/// `RunSealed` with the conclusion, so tamper detection covers how a run ended,
/// which it would not if a side table were authoritative. This is a convenience
/// for the one question the chain answers slowly: *what is quarantined right
/// now?*
///
/// Maintained by `append`, in the same transaction as the record it derives
/// from, and **the last conclusion wins**: a run that failed, was resumed and
/// then succeeded moves from `failed` to `succeeded`. Deriving it from the seal
/// instead would freeze the *first* conclusion forever — a resumed run would be
/// listed as failed for the rest of its life, which is worse than no index,
/// because a wrong answer reads exactly like a right one.
///
/// The ordinal is a per-tenant conclusion counter, so a scan is oldest-first
/// without a sort and a bounded reverse scan yields the newest conclusions.
const RUN_BY_OUTCOME: TableDefinition<(&str, &str, u64), &str> =
    TableDefinition::new("run_by_outcome");

/// `(tenant, run_key) -> (outcome, ordinal)`: the reverse of `RUN_BY_OUTCOME`.
///
/// What lets a re-conclusion *replace* its index row rather than accumulate a
/// second one — without it, a failed-then-succeeded run would appear under both
/// outcomes, and the failed listing would never drain.
const RUN_OUTCOME: TableDefinition<(&str, &str), (&str, u64)> = TableDefinition::new("run_outcome");

/// `(tenant, updated_at, run_id) -> ()`, every run by last durable append.
const RUN_ACTIVITY: TableDefinition<(&str, u64, &str), ()> = TableDefinition::new("run_activity");
/// `(tenant, run_id) -> updated_at`, for replacing the activity index row.
const RUN_LAST_ACTIVITY: TableDefinition<(&str, &str), u64> =
    TableDefinition::new("run_last_activity");

/// `log_index -> run_id`, the plane's Merkle log in seal order.
///
/// Positions are never reissued, so this table keeps gaps when a run is removed
/// — deliberately, so a freed position cannot be silently reused by a different
/// run. The *tree* is built by iterating in key order, which yields dense
/// positions; that is what makes the `ROW_NUMBER() OVER (ORDER BY log_index)`
/// the SQL backend needed unnecessary here, and with it the rank-vs-index bug
/// that cost an afternoon.
const SEAL_LOG: TableDefinition<(&str, u64), &str> = TableDefinition::new("seal_log");

/// `run_id -> (actor, reason, requested_at)`. An operator's stop request.
///
/// Keyed by run, so the request is idempotent and a retry cannot overwrite who
/// intervened. Deliberately not fenced: whoever wants a run stopped is not its
/// owner, holds no epoch, and is usually asking because the owner is busy.
const RUN_CANCEL: TableDefinition<&str, (&str, &str, u64)> = TableDefinition::new("run_cancel");

/// The next log position to hand out. `MAX + 1` over a table that keeps its gaps
/// would reuse a removed run's slot.
const COUNTERS: TableDefinition<&str, u64> = TableDefinition::new("counters");

const NEXT_LOG_INDEX: &str = "next_log_index";

/// The next conclusion ordinal to hand out, per tenant.
///
/// Separate from `NEXT_LOG_INDEX` because conclusions are not seals: a failed
/// run concludes without entering the Merkle log, and interleaving the two
/// sequences would leave the log with gaps that mean nothing.
const NEXT_CONCLUSION: &str = "next_conclusion";

/// An upper bound for a `&str` range end.
///
/// redb orders by encoded bytes, and `\u{10FFFF}` encodes to the highest
/// sequence UTF-8 admits, so no valid key can sort above it.
pub(super) const MAX_STR: &str = "\u{10FFFF}";

/// A single-node journal store on redb.
#[derive(Debug, Clone)]
pub struct RedbStore {
    db: Arc<Database>,
    /// Attached to the store rather than passed per append, because only the
    /// store knows a record's chain hash — it assigns `seq` and `prev_hash`
    /// inside the same transaction that writes. Signing anywhere else would be
    /// signing a guess about what the hash will be.
    signer: Option<Arc<dyn crate::core::Signer>>,
    /// Names this plane's Merkle log in every checkpoint.
    origin: String,
    /// Which tenant's keyspace this handle addresses.
    ///
    /// Part of the *key*, not a filter applied after reading. A store handle for
    /// tenant A cannot name tenant B's run: the key it builds does not exist, so
    /// a cross-tenant read is a miss rather than a row that some `WHERE` clause
    /// was supposed to remove. A filter that can be forgotten is not isolation,
    /// and the query that forgets it looks exactly like working software.
    tenant: crate::core::TenantId,
}

impl RedbStore {
    /// Open (or create) a database file.
    ///
    /// # Errors
    ///
    /// If the file cannot be opened or the tables cannot be created.
    pub fn open(path: impl AsRef<Path>) -> Result<Self, StoreError> {
        let db = Database::create(path).map_err(|e| be(&e))?;
        Self::init(db)
    }

    /// An ephemeral database — for tests and for the simulator.
    ///
    /// redb has no in-memory backend, so this is a file in a temporary
    /// directory that is removed when the process exits. Named with a counter
    /// rather than a random suffix because ambient randomness is denied
    /// crate-wide; two stores in one process must still not collide.
    ///
    /// # Errors
    ///
    /// If the temporary file cannot be created.
    pub fn open_in_memory() -> Result<Self, StoreError> {
        use std::sync::atomic::{AtomicU64, Ordering};
        static N: AtomicU64 = AtomicU64::new(0);
        let n = N.fetch_add(1, Ordering::Relaxed);
        let pid = std::process::id();
        let path = std::env::temp_dir().join(format!("agentplane-{pid}-{n}.redb"));
        let _ = std::fs::remove_file(&path);
        let db = Database::create(&path).map_err(|e| be(&e))?;
        // Unlinked immediately: the handle stays valid, so the file is reachable
        // by this process and by nothing else, and it cannot outlive a crash.
        let _ = std::fs::remove_file(&path);
        Self::init(db)
    }

    /// Create every table, so a read on a fresh database is not a missing-table
    /// error the read paths would each have to special-case.
    fn init(db: Database) -> Result<Self, StoreError> {
        let w = begin_write(&db)?;
        {
            w.open_table(JOURNAL).map_err(|e| be(&e))?;
            w.open_table(JOURNAL_BY_CASE).map_err(|e| be(&e))?;
            w.open_table(EFFECT_ONCE).map_err(|e| be(&e))?;
            w.open_table(RUN_LEASE).map_err(|e| be(&e))?;
            w.open_table(RUN_SEAL).map_err(|e| be(&e))?;
            w.open_table(RUN_BY_OUTCOME).map_err(|e| be(&e))?;
            w.open_table(RUN_OUTCOME).map_err(|e| be(&e))?;
            w.open_table(SEAL_LOG).map_err(|e| be(&e))?;
            w.open_table(RUN_CANCEL).map_err(|e| be(&e))?;
            w.open_table(COUNTERS).map_err(|e| be(&e))?;
            super::redb_cases::create_tables(&w)?;
            super::redb_events::create_tables(&w)?;
            super::redb_tasks::create_tables(&w)?;
            super::redb_timers::create_tables(&w)?;
            super::redb_batches::create_tables(&w)?;
        }
        w.commit().map_err(|e| be(&e))?;
        Ok(Self {
            db: Arc::new(db),
            signer: None,
            origin: "agentplane".to_owned(),
            tenant: crate::core::TenantId::default(),
        })
    }

    /// A handle addressing one tenant's keyspace.
    ///
    /// Every key this store builds is prefixed, so two tenants may share one
    /// physical database and still not reach each other: a read for a run that
    /// belongs to another tenant finds nothing, because the key it looks under
    /// was never written.
    ///
    /// The origin moves with it. Two tenants sharing a database would otherwise
    /// publish checkpoints under one origin name, and a witness could not tell
    /// whose history it was cosigning.
    #[must_use]
    pub fn for_tenant(mut self, tenant: crate::core::TenantId) -> Self {
        self.origin = format!("{}/{}", self.origin, tenant);
        self.tenant = tenant;
        self
    }

    /// The storage key for a run, and the only place one is derived.
    ///
    /// One derivation because the alternative is eleven, and eleven places
    /// building the same string is how one of them ends up building a slightly
    /// different one.
    pub(super) fn run_key(&self, run: RunId) -> String {
        format!("{}/{run}", self.tenant)
    }

    /// This handle's tenant, for a key or a range bound.
    ///
    /// Owned rather than borrowed because every caller moves it into a closure
    /// that outlives the borrow.
    pub(super) fn tenant_name(&self) -> String {
        self.tenant.to_string()
    }

    /// Name this plane's Merkle log.
    ///
    /// Goes into every checkpoint, so two planes' checkpoints cannot be
    /// confused for one another — which matters the moment they are published
    /// to a shared witness.
    #[must_use]
    pub fn origin(mut self, origin: impl Into<String>) -> Self {
        self.origin = origin.into();
        self
    }

    /// Write records signed as this identity.
    ///
    /// Off unless asked, and the default is *unsigned rather than self-signed*:
    /// a plane that quietly minted its own key would produce records that look
    /// attested and prove nothing.
    #[must_use]
    pub fn signing_as(mut self, signer: Arc<dyn crate::core::Signer>) -> Self {
        self.signer = Some(signer);
        self
    }

    /// Run a closure against the database on the blocking pool.
    ///
    /// redb is synchronous and a commit fsyncs; doing that on the async
    /// executor stalls every other task in the process.
    pub(super) async fn with_db<T, F>(&self, f: F) -> Result<T, StoreError>
    where
        T: Send + 'static,
        F: FnOnce(&Database) -> Result<T, StoreError> + Send + 'static,
    {
        let db = Arc::clone(&self.db);
        tokio::task::spawn_blocking(move || f(&db))
            .await
            .map_err(|e| StoreError::Backend(format!("blocking pool: {e}")))?
    }

    /// The log's leaves, in seal order.
    async fn log_leaves(&self) -> Result<Vec<Digest>, StoreError> {
        let tenant = self.tenant.clone();
        self.with_db(move |db| {
            let r = db.begin_read().map_err(|e| be(&e))?;
            let log = r.open_table(SEAL_LOG).map_err(|e| be(&e))?;
            let seals = r.open_table(RUN_SEAL).map_err(|e| be(&e))?;
            let mut out = Vec::new();
            // This tenant's range only. Iterating the whole table would build a
            // Merkle log covering other tenants' runs, so a checkpoint would
            // commit to history its holder may not see.
            for entry in log
                .range((tenant.as_str(), 0)..=(tenant.as_str(), u64::MAX))
                .map_err(|e| be(&e))?
            {
                let (_, run) = entry.map_err(|e| be(&e))?;
                let Some(seal) = seals.get(run.value()).map_err(|e| be(&e))? else {
                    continue;
                };
                let (_, head, _) = seal.value();
                out.push(crate::core::merkle::leaf_hash(&digest(head)?));
            }
            Ok(out)
        })
        .await
    }

    /// Where a run sits in the log, and its leaf value.
    ///
    /// The position is the run's index in seal order, counted while iterating —
    /// dense by construction, because the tree is built from the same walk.
    async fn log_position(&self, run: RunId) -> Result<Option<(usize, Digest)>, StoreError> {
        let key = self.run_key(run);
        let tenant = self.tenant.clone();
        self.with_db(move |db| {
            let r = db.begin_read().map_err(|e| be(&e))?;
            let log = r.open_table(SEAL_LOG).map_err(|e| be(&e))?;
            let seals = r.open_table(RUN_SEAL).map_err(|e| be(&e))?;
            let mut rank = 0usize;
            for entry in log
                .range((tenant.as_str(), 0)..=(tenant.as_str(), u64::MAX))
                .map_err(|e| be(&e))?
            {
                let (_, run_id) = entry.map_err(|e| be(&e))?;
                let Some(seal) = seals.get(run_id.value()).map_err(|e| be(&e))? else {
                    continue;
                };
                if run_id.value() == key {
                    let (_, head, _) = seal.value();
                    return Ok(Some((rank, digest(head)?)));
                }
                rank += 1;
            }
            Ok(None)
        })
        .await
    }

    /// Overwrite a record's stored bytes while leaving its hash untouched,
    /// simulating an after-the-fact edit.
    ///
    /// Exists so the test suite can prove that tampering is *detected* rather
    /// than assumed impossible. Not part of the supported surface.
    ///
    /// # Errors
    ///
    /// If the write fails.
    #[doc(hidden)]
    pub async fn tamper_for_test(
        &self,
        run: RunId,
        seq: Seq,
        body: Vec<u8>,
    ) -> Result<(), StoreError> {
        let key = self.run_key(run);
        self.with_db(move |db| {
            let w = begin_write(db)?;
            {
                let mut t = w.open_table(JOURNAL).map_err(|e| be(&e))?;
                let existing = t
                    .get((key.as_str(), seq))
                    .map_err(|e| be(&e))?
                    .map(|v| v.value().to_vec());
                if let Some(bytes) = existing {
                    let row = Row::decode(&bytes)?;
                    let tampered = Row { body, ..row };
                    t.insert((key.as_str(), seq), tampered.encode().as_slice())
                        .map_err(|e| be(&e))?;
                }
            }
            w.commit().map_err(|e| be(&e))?;
            Ok(())
        })
        .await
    }

    /// Remove a run entirely — journal rows and seal alike.
    ///
    /// The one tampering the per-run chain structurally cannot see, so a suite
    /// that could not perform it could not test the mechanism that does. Not
    /// part of the supported surface.
    ///
    /// # Errors
    ///
    /// If the write fails.
    #[doc(hidden)]
    pub async fn delete_run_for_test(&self, run: RunId) -> Result<(), StoreError> {
        let key = self.run_key(run);
        self.with_db(move |db| {
            let w = begin_write(db)?;
            {
                let mut j = w.open_table(JOURNAL).map_err(|e| be(&e))?;
                let seqs: Vec<u64> = j
                    .range((key.as_str(), 0u64)..=(key.as_str(), u64::MAX))
                    .map_err(|e| be(&e))?
                    .filter_map(Result::ok)
                    .map(|(k, _)| k.value().1)
                    .collect();
                for s in seqs {
                    j.remove((key.as_str(), s)).map_err(|e| be(&e))?;
                }
                w.open_table(RUN_SEAL)
                    .map_err(|e| be(&e))?
                    .remove(key.as_str())
                    .map_err(|e| be(&e))?;
                // The log position is left in place, holding its gap: a freed
                // slot must never be reissued to a different run.
            }
            w.commit().map_err(|e| be(&e))?;
            Ok(())
        })
        .await
    }
}

/// One journal row.
///
/// Encoded by hand rather than through serde: the body is already bytes, and a
/// JSON envelope would base64 it on every append and every read.
struct Row {
    body: Vec<u8>,
    prev_hash: [u8; 32],
    hash: [u8; 32],
    key_id: Option<String>,
    signature: Option<Vec<u8>>,
}

impl Row {
    fn encode(&self) -> Vec<u8> {
        let mut out = Vec::with_capacity(self.body.len() + 96);
        out.extend_from_slice(&self.prev_hash);
        out.extend_from_slice(&self.hash);
        // A key id without a signature (or the reverse) is a half-written row,
        // not an unsigned record. Written as one optional pair so the two
        // cannot drift apart.
        match (&self.key_id, &self.signature) {
            (Some(k), Some(s)) => {
                out.push(1);
                push_bytes(&mut out, k.as_bytes());
                push_bytes(&mut out, s);
            }
            _ => out.push(0),
        }
        push_bytes(&mut out, &self.body);
        out
    }

    fn decode(raw: &[u8]) -> Result<Self, StoreError> {
        let corrupt = |what: &str| StoreError::Corrupt {
            seq: 0,
            detail: format!("journal row truncated in {what}"),
        };
        if raw.len() < 65 {
            return Err(corrupt("header"));
        }
        let prev_hash: [u8; 32] = raw[0..32].try_into().map_err(|_| corrupt("prev_hash"))?;
        let hash: [u8; 32] = raw[32..64].try_into().map_err(|_| corrupt("hash"))?;
        let mut at = 64;
        let attested = raw[at] == 1;
        at += 1;
        let (key_id, signature) = if attested {
            let (k, n) = take_bytes(raw, at).ok_or_else(|| corrupt("key_id"))?;
            at = n;
            let (s, n) = take_bytes(raw, at).ok_or_else(|| corrupt("signature"))?;
            at = n;
            (
                Some(String::from_utf8(k).map_err(|_| corrupt("key_id utf-8"))?),
                Some(s),
            )
        } else {
            (None, None)
        };
        let (body, _) = take_bytes(raw, at).ok_or_else(|| corrupt("body"))?;
        Ok(Self {
            body,
            prev_hash,
            hash,
            key_id,
            signature,
        })
    }

    fn into_record(self) -> Result<Record, StoreError> {
        let attestation = self
            .key_id
            .zip(self.signature)
            .map(|(key_id, signature)| crate::core::Attestation { key_id, signature });
        Record::from_stored_attested(
            self.body,
            Digest::from_bytes(self.prev_hash),
            Digest::from_bytes(self.hash),
            attestation,
        )
    }
}

fn push_bytes(out: &mut Vec<u8>, bytes: &[u8]) {
    // A record is refused above `MAX_RECORD_BYTES`, so no field can reach 4 GiB
    // — but the cast is checked anyway rather than assumed, because a silent
    // wrap here would write a length prefix that disagrees with the payload and
    // corrupt every record after it.
    let len = u32::try_from(bytes.len()).unwrap_or(u32::MAX);
    out.extend_from_slice(&len.to_le_bytes());
    out.extend_from_slice(bytes);
}

fn take_bytes(raw: &[u8], at: usize) -> Option<(Vec<u8>, usize)> {
    let end = at.checked_add(4)?;
    let len = u32::from_le_bytes(raw.get(at..end)?.try_into().ok()?) as usize;
    let stop = end.checked_add(len)?;
    Some((raw.get(end..stop)?.to_vec(), stop))
}

fn digest(bytes: &[u8]) -> Result<Digest, StoreError> {
    let b: [u8; 32] = bytes.try_into().map_err(|_| StoreError::Corrupt {
        seq: 0,
        detail: "a stored hash is not 32 bytes".into(),
    })?;
    Ok(Digest::from_bytes(b))
}

/// Begin a write transaction with its durability stated rather than inherited.
///
/// redb already defaults to [`Durability::Immediate`] — a commit is persistent
/// by the time it returns — which is exactly what the old backend spelled as
/// `PRAGMA synchronous = FULL`. Stating it anyway, in one place every write path
/// goes through, because "a committed record survives the process" is this
/// crate's central promise and a promise resting on a dependency's default can
/// be weakened by an upgrade with nothing here changing.
///
/// What this does *not* buy is a test: no suite here kills a process mid-commit,
/// so a regression would not be caught. It is a belt on a guarantee the tests
/// cannot reach, not a checked invariant.
pub(super) fn begin_write(db: &Database) -> Result<redb::WriteTransaction, StoreError> {
    let mut w = db.begin_write().map_err(|e| be(&e))?;
    w.set_durability(redb::Durability::Immediate)
        .map_err(|e| be(&e))?;
    Ok(w)
}

pub(super) fn be<E: std::fmt::Display>(e: &E) -> StoreError {
    StoreError::Backend(e.to_string())
}

/// Wall-clock read for lease expiry.
///
/// Lease timing is infrastructure, not run logic: it never enters the journal's
/// content and therefore cannot affect replay. Run-visible time goes through
/// `StepCtx::now`, which journals the instant.
#[allow(clippy::disallowed_methods)]
fn now_secs() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_or(0, |d| d.as_secs())
}

/// The last record of a run, or genesis.
fn head_of(
    t: &impl ReadableTable<(&'static str, u64), &'static [u8]>,
    run: &str,
) -> Result<Head, StoreError>
where
{
    let last = t
        .range((run, 0u64)..=(run, u64::MAX))
        .map_err(|e| be(&e))?
        .next_back();
    match last {
        None => Ok(Head::genesis()),
        Some(entry) => {
            let (k, v) = entry.map_err(|e| be(&e))?;
            let row = Row::decode(v.value())?;
            Ok(Head {
                seq: k.value().1,
                hash: Digest::from_bytes(row.hash),
            })
        }
    }
}

#[allow(clippy::too_many_lines)]
#[async_trait]
impl JournalStore for RedbStore {
    fn tenant(&self) -> &str {
        self.tenant.as_str()
    }

    async fn append(&self, epoch: Epoch, batch: Vec<Append>) -> Result<Vec<Record>, StoreError> {
        if batch.is_empty() {
            return Ok(Vec::new());
        }
        let run = batch[0].run;
        if let Some(a) = batch.iter().find(|a| a.run != run) {
            return Err(StoreError::Backend(format!(
                "batch spans runs {run} and {} — a batch is one run's atomic unit",
                a.run
            )));
        }
        let signer = self.signer.clone();
        let key = self.run_key(run);
        let tenant = self.tenant_name();

        self.with_db(move |db| {
            let w = begin_write(db)?;
            let sealed = {
                let mut journal = w.open_table(JOURNAL).map_err(|e| be(&e))?;
                let mut by_case = w.open_table(JOURNAL_BY_CASE).map_err(|e| be(&e))?;
                let mut once = w.open_table(EFFECT_ONCE).map_err(|e| be(&e))?;

                // Fence inside the write transaction: no window between the
                // check and the append for a stale writer to use.
                {
                    let leases = w.open_table(RUN_LEASE).map_err(|e| be(&e))?;
                    if let Some(l) = leases.get(key.as_str()).map_err(|e| be(&e))? {
                        let (_, current, _) = l.value();
                        if epoch != current {
                            return Err(StoreError::Fenced {
                                run: key.clone(),
                                held: epoch,
                                current,
                            });
                        }
                    }
                }

                // Sealed is frozen. The Merkle leaf is the chain head at seal
                // time, so an append past it — even by the epoch's rightful
                // holder — advances the true head past what every checkpoint
                // attests. Checked inside the write transaction like the fence,
                // because the executor's refusal to resume a closed run is
                // logic a future caller can bypass, and this is the constraint
                // that cannot be. The sealing appends themselves precede the
                // seal row, so they never meet this check.
                {
                    let seals = w.open_table(RUN_SEAL).map_err(|e| be(&e))?;
                    if let Some(seal) = seals.get(key.as_str()).map_err(|e| be(&e))? {
                        let (outcome, _, _) = seal.value();
                        return Err(StoreError::RunSealed {
                            run: key.clone(),
                            outcome: outcome.to_owned(),
                        });
                    }
                }

                let mut head = head_of(&journal, &key)?;
                let mut sealed = Vec::with_capacity(batch.len());

                for append in batch {
                    let effect_key = append.effect_key.map(EffectKey::to_hex);
                    let body = append.into_body(head.seq + 1, epoch);
                    let is_start =
                        matches!(body.kind, crate::journal::RecordKind::EffectStarted { .. });
                    let conclusion = match &body.kind {
                        crate::journal::RecordKind::RunSealed { outcome, .. } => {
                            Some(outcome.clone())
                        }
                        _ => None,
                    };
                    let record = Record::seal_signed(body, head.hash, signer.as_deref())?;
                    let seq = record.seq();

                    // Exactly-once. The key is the constraint: a prior value
                    // means this effect already started in this run.
                    if is_start && let Some(ek) = &effect_key {
                        let prior = once
                            .insert((key.as_str(), ek.as_str()), seq)
                            .map_err(|e| be(&e))?;
                        if prior.is_some() {
                            return Err(record.effect_key().map_or_else(
                                || StoreError::Backend("duplicate effect".into()),
                                StoreError::DuplicateEffect,
                            ));
                        }
                    }

                    let row = Row {
                        body: record.raw().to_vec(),
                        prev_hash: *record.prev_hash.as_bytes(),
                        hash: *record.hash.as_bytes(),
                        key_id: record.attestation.as_ref().map(|a| a.key_id.clone()),
                        signature: record.attestation.as_ref().map(|a| a.signature.clone()),
                    };
                    journal
                        .insert((key.as_str(), seq), row.encode().as_slice())
                        .map_err(|e| be(&e))?;
                    // Written in the same transaction as the row it points at,
                    // so the index cannot outlive or precede its record.
                    if let Some(case) = record.body.case {
                        by_case
                            .insert(
                                (
                                    tenant.as_str(),
                                    case.to_string().as_str(),
                                    key.as_str(),
                                    seq,
                                ),
                                (),
                            )
                            .map_err(|e| be(&e))?;
                    }

                    // The outcome index derives from the chain, here, in the
                    // same transaction as the record it derives from — and the
                    // last conclusion wins. A failed run that is resumed and
                    // succeeds moves between listings; an index fed by the seal
                    // would keep saying "failed" forever, and a wrong answer
                    // reads exactly like a right one.
                    if let Some(outcome) = conclusion {
                        let mut by_outcome = w.open_table(RUN_BY_OUTCOME).map_err(|e| be(&e))?;
                        let mut outcomes = w.open_table(RUN_OUTCOME).map_err(|e| be(&e))?;
                        let mut counters = w.open_table(COUNTERS).map_err(|e| be(&e))?;
                        if let Some(prior) = outcomes
                            .get((tenant.as_str(), key.as_str()))
                            .map_err(|e| be(&e))?
                            .map(|v| {
                                let (o, ord) = v.value();
                                (o.to_owned(), ord)
                            })
                        {
                            by_outcome
                                .remove((tenant.as_str(), prior.0.as_str(), prior.1))
                                .map_err(|e| be(&e))?;
                        }
                        let counter = format!("{NEXT_CONCLUSION}/{tenant}");
                        let next = counters
                            .get(counter.as_str())
                            .map_err(|e| be(&e))?
                            .map_or(0, |v| v.value());
                        by_outcome
                            .insert((tenant.as_str(), outcome.as_str(), next), key.as_str())
                            .map_err(|e| be(&e))?;
                        outcomes
                            .insert((tenant.as_str(), key.as_str()), (outcome.as_str(), next))
                            .map_err(|e| be(&e))?;
                        counters
                            .insert(counter.as_str(), next + 1)
                            .map_err(|e| be(&e))?;
                    }

                    head = Head {
                        seq,
                        hash: record.hash,
                    };
                    sealed.push(record);
                }
                let updated = now_secs();
                let mut activity = w.open_table(RUN_ACTIVITY).map_err(|e| be(&e))?;
                let mut last = w.open_table(RUN_LAST_ACTIVITY).map_err(|e| be(&e))?;
                if let Some(previous) = last
                    .get((tenant.as_str(), key.as_str()))
                    .map_err(|e| be(&e))?
                    .map(|value| value.value())
                {
                    activity
                        .remove((tenant.as_str(), previous, key.as_str()))
                        .map_err(|e| be(&e))?;
                }
                activity
                    .insert((tenant.as_str(), updated, key.as_str()), ())
                    .map_err(|e| be(&e))?;
                last.insert((tenant.as_str(), key.as_str()), updated)
                    .map_err(|e| be(&e))?;
                sealed
            };
            w.commit().map_err(|e| be(&e))?;
            Ok(sealed)
        })
        .await
    }

    async fn read(&self, run: RunId, from: Seq) -> Result<Vec<Record>, StoreError> {
        let key = self.run_key(run);
        self.with_db(move |db| {
            let r = db.begin_read().map_err(|e| be(&e))?;
            let t = r.open_table(JOURNAL).map_err(|e| be(&e))?;
            let mut out = Vec::new();
            for entry in t
                .range((key.as_str(), from)..=(key.as_str(), u64::MAX))
                .map_err(|e| be(&e))?
            {
                let (_, v) = entry.map_err(|e| be(&e))?;
                out.push(Row::decode(v.value())?.into_record()?);
            }
            Ok(out)
        })
        .await
    }

    async fn case_history(
        &self,
        case: crate::core::CaseId,
        limit: usize,
    ) -> Result<Vec<Record>, StoreError> {
        let tenant = self.tenant_name();
        let case = case.to_string();
        self.with_db(move |db| {
            let r = db.begin_read().map_err(|e| be(&e))?;
            let idx = r.open_table(JOURNAL_BY_CASE).map_err(|e| be(&e))?;
            let j = r.open_table(JOURNAL).map_err(|e| be(&e))?;
            let mut out = Vec::new();
            for entry in idx
                .range(
                    (tenant.as_str(), case.as_str(), "", 0)
                        ..=(tenant.as_str(), case.as_str(), MAX_STR, u64::MAX),
                )
                .map_err(|e| be(&e))?
            {
                if out.len() >= limit {
                    break;
                }
                let (k, _) = entry.map_err(|e| be(&e))?;
                let (_, _, run, seq) = k.value();
                // The index carries the ordering; the row carries the bytes. A
                // record has one home, so the two cannot disagree about it.
                if let Some(v) = j.get((run, seq)).map_err(|e| be(&e))? {
                    out.push(Row::decode(v.value())?.into_record()?);
                }
            }
            Ok(out)
        })
        .await
    }

    async fn head(&self, run: RunId) -> Result<Head, StoreError> {
        let key = self.run_key(run);
        self.with_db(move |db| {
            let r = db.begin_read().map_err(|e| be(&e))?;
            let t = r.open_table(JOURNAL).map_err(|e| be(&e))?;
            head_of(&t, &key)
        })
        .await
    }

    async fn acquire(&self, run: RunId, owner: &str, ttl: Duration) -> Result<Lease, StoreError> {
        let key = self.run_key(run);
        let owner = owner.to_owned();
        let owner_out = owner.clone();
        let epoch = self
            .with_db(move |db| {
                let w = begin_write(db)?;
                let epoch = {
                    let mut leases = w.open_table(RUN_LEASE).map_err(|e| be(&e))?;
                    let now = now_secs();
                    let expires = now + ttl.as_secs().max(1);

                    let existing = leases.get(key.as_str()).map_err(|e| be(&e))?.map(|v| {
                        let (o, e, x) = v.value();
                        (o.to_owned(), e, x)
                    });

                    let epoch = match existing {
                        // Fresh run.
                        None => 1,
                        // Expired or released: take over and fence whoever held
                        // it, **including this caller**. Checked before the
                        // ownership test on purpose — a lease that has lapsed is
                        // not yours to renew, because you cannot know whether
                        // somebody took over in the gap. Assuming you were
                        // fenced is the only safe reading.
                        Some((_, epoch, expires_at)) if expires_at <= now => epoch + 1,
                        // Ours and still live: renew without bumping, or the
                        // owner fences its own in-flight writes.
                        Some((held_by, epoch, _)) if held_by == owner => epoch,
                        // Someone else's, still live. Not a fencing situation:
                        // this caller is not stale, it is simply not the owner.
                        Some((held_by, epoch, expires_at)) => {
                            return Err(StoreError::LeaseHeld {
                                run: key.clone(),
                                owner: held_by,
                                epoch,
                                remaining_secs: expires_at.saturating_sub(now),
                            });
                        }
                    };

                    leases
                        .insert(key.as_str(), (owner.as_str(), epoch, expires))
                        .map_err(|e| be(&e))?;
                    epoch
                };
                w.commit().map_err(|e| be(&e))?;
                Ok(epoch)
            })
            .await?;

        Ok(Lease {
            run,
            owner: owner_out,
            epoch,
        })
    }

    async fn release_lease(&self, run: RunId, epoch: Epoch) -> Result<(), StoreError> {
        let key = self.run_key(run);
        self.with_db(move |db| {
            let w = begin_write(db)?;
            {
                let mut leases = w.open_table(RUN_LEASE).map_err(|e| be(&e))?;
                let held = leases
                    .get(key.as_str())
                    .map_err(|e| be(&e))?
                    .map(|v| v.value().1);
                // Only the epoch that holds it may free it. A fenced caller
                // shutting down must not hand the run to a third party while
                // the instance that took over is mid-write.
                //
                // Marked expired rather than **removed**, and that distinction
                // is the whole safety of this operation. The epoch lives in
                // this row: delete it and `append` has nothing to fence
                // against, while the next `acquire` restarts at 1 — so a writer
                // already fenced at 2 outranks the new owner. Releasing must
                // free the lease without forgetting what has happened to it.
                if held == Some(epoch) {
                    leases
                        .insert(key.as_str(), ("", epoch, 0))
                        .map_err(|e| be(&e))?;
                }
            }
            w.commit().map_err(|e| be(&e))?;
            Ok(())
        })
        .await
    }

    async fn seal(&self, run: RunId, epoch: Epoch, outcome: &str) -> Result<Digest, StoreError> {
        let key = self.run_key(run);
        let tenant = self.tenant.clone();
        let outcome = outcome.to_owned();
        self.with_db(move |db| {
            let w = begin_write(db)?;
            let head_hash = {
                {
                    let leases = w.open_table(RUN_LEASE).map_err(|e| be(&e))?;
                    if let Some(l) = leases.get(key.as_str()).map_err(|e| be(&e))? {
                        let (_, current, _) = l.value();
                        if epoch != current {
                            return Err(StoreError::Fenced {
                                run: key.clone(),
                                held: epoch,
                                current,
                            });
                        }
                    }
                }

                let head = {
                    let journal = w.open_table(JOURNAL).map_err(|e| be(&e))?;
                    head_of(&journal, &key)?
                };

                let mut seals = w.open_table(RUN_SEAL).map_err(|e| be(&e))?;
                // First seal wins; a re-seal must not rewrite the outcome.
                if seals.get(key.as_str()).map_err(|e| be(&e))?.is_none() {
                    seals
                        .insert(
                            key.as_str(),
                            (
                                outcome.as_str(),
                                head.hash.as_bytes().as_slice(),
                                now_secs(),
                            ),
                        )
                        .map_err(|e| be(&e))?;

                    // Enter the log in the same transaction that seals: a seal
                    // the checkpoint does not commit to is the hole this closes.
                    // The counter is per tenant too. A shared counter would
                    // interleave two tenants' indices in one sequence, and the
                    // log would no longer be dense for either.
                    let counter = format!("{NEXT_LOG_INDEX}/{tenant}");
                    let mut counters = w.open_table(COUNTERS).map_err(|e| be(&e))?;
                    let next = counters
                        .get(counter.as_str())
                        .map_err(|e| be(&e))?
                        .map_or(0, |v| v.value());
                    w.open_table(SEAL_LOG)
                        .map_err(|e| be(&e))?
                        .insert((tenant.as_str(), next), key.as_str())
                        .map_err(|e| be(&e))?;
                    // The outcome index is *not* written here: it derives from
                    // the `RunSealed` record inside `append`, where the last
                    // conclusion wins. The seal freezes the journal and enters
                    // the log — it is not the outcome's home.
                    counters
                        .insert(counter.as_str(), next + 1)
                        .map_err(|e| be(&e))?;
                }
                head.hash
            };
            w.commit().map_err(|e| be(&e))?;
            Ok(head_hash)
        })
        .await
    }

    async fn runs_by_outcome(&self, outcome: &str, limit: usize) -> Result<Vec<RunId>, StoreError> {
        let tenant = self.tenant_name();
        let outcome = outcome.to_owned();
        let prefix = format!("{tenant}/");
        self.with_db(move |db| {
            let r = db.begin_read().map_err(|e| be(&e))?;
            let idx = r.open_table(RUN_BY_OUTCOME).map_err(|e| be(&e))?;
            let mut out = Vec::new();
            // **Newest first**, which is what makes the bound survivable. See
            // `JournalStore::runs_by_outcome`: ascending order plus a page
            // limit means a plane whose backlog already exceeds one page
            // returns the *same* runs forever, and the quarantine that just
            // happened is the one that never appears.
            for entry in idx
                .range(
                    (tenant.as_str(), outcome.as_str(), 0)
                        ..=(tenant.as_str(), outcome.as_str(), u64::MAX),
                )
                .map_err(|e| be(&e))?
                .rev()
            {
                if out.len() >= limit {
                    break;
                }
                let (_, v) = entry.map_err(|e| be(&e))?;
                // Keys are `tenant/run`; the caller asked for runs.
                let key = v.value();
                if let Some(id) = key.strip_prefix(prefix.as_str())
                    && let Ok(run) = RunId::parse(id)
                {
                    out.push(run);
                }
            }
            Ok(out)
        })
        .await
    }

    async fn recent_runs(&self) -> Result<Vec<(RunId, u64)>, StoreError> {
        let tenant = self.tenant_name();
        let prefix = format!("{tenant}/");
        self.with_db(move |db| {
            let r = db.begin_read().map_err(|e| be(&e))?;
            let Ok(activity) = r.open_table(RUN_ACTIVITY) else {
                return Ok(Vec::new());
            };
            activity
                .range((tenant.as_str(), 0, "")..=(tenant.as_str(), u64::MAX, MAX_STR))
                .map_err(|e| be(&e))?
                .rev()
                .filter_map(|entry| match entry {
                    Ok((key, _)) => {
                        let (_, updated, stored) = key.value();
                        stored
                            .strip_prefix(prefix.as_str())
                            .and_then(|id| RunId::parse(id).ok())
                            .map(|run| Ok((run, updated)))
                    }
                    Err(error) => Some(Err(be(&error))),
                })
                .collect()
        })
        .await
    }

    async fn checkpoint(&self) -> Result<crate::journal::Checkpoint, StoreError> {
        let leaves = self.log_leaves().await?;
        Ok(crate::journal::Checkpoint {
            origin: self.origin.clone(),
            size: leaves.len() as u64,
            root: crate::core::merkle::root(&leaves),
        })
    }

    async fn consistency_proof(&self, old_size: u64) -> Result<Vec<Digest>, StoreError> {
        let leaves = self.log_leaves().await?;
        let old = usize::try_from(old_size).unwrap_or(usize::MAX);
        if old > leaves.len() {
            // Refused rather than answered with an empty proof: an empty proof
            // is what a *consistent* log of unchanged size returns, so handing
            // one back here would let "your checkpoint is from the future" read
            // as "everything is fine".
            return Err(StoreError::Backend(format!(
                "a checkpoint of size {old_size} is larger than this log ({}) — \
                 either it belongs to another plane, or runs were removed",
                leaves.len()
            )));
        }
        Ok(crate::core::merkle::consistency_proof(&leaves, old))
    }

    async fn inclusion_proof(
        &self,
        run: RunId,
    ) -> Result<Option<crate::journal::Inclusion>, StoreError> {
        let leaves = self.log_leaves().await?;
        let Some((index, seal)) = self.log_position(run).await? else {
            return Ok(None);
        };
        Ok(Some(crate::journal::Inclusion {
            index: index as u64,
            size: leaves.len() as u64,
            seal,
            proof: crate::core::merkle::inclusion_proof(&leaves, index),
        }))
    }

    async fn request_cancel(
        &self,
        run: RunId,
        actor: &str,
        reason: &str,
    ) -> Result<bool, StoreError> {
        let key = self.run_key(run);
        let (actor, reason) = (actor.to_owned(), reason.to_owned());
        self.with_db(move |db| {
            let w = begin_write(db)?;
            let first = {
                let mut t = w.open_table(RUN_CANCEL).map_err(|e| be(&e))?;
                // The first asker stays on the record; a retry must not rewrite
                // who intervened.
                if t.get(key.as_str()).map_err(|e| be(&e))?.is_some() {
                    false
                } else {
                    t.insert(key.as_str(), (actor.as_str(), reason.as_str(), now_secs()))
                        .map_err(|e| be(&e))?;
                    true
                }
            };
            w.commit().map_err(|e| be(&e))?;
            Ok(first)
        })
        .await
    }

    async fn cancellation(&self, run: RunId) -> Result<Option<Cancellation>, StoreError> {
        let key = self.run_key(run);
        self.with_db(move |db| {
            let r = db.begin_read().map_err(|e| be(&e))?;
            let t = r.open_table(RUN_CANCEL).map_err(|e| be(&e))?;
            Ok(t.get(key.as_str()).map_err(|e| be(&e))?.map(|v| {
                let (actor, reason, _) = v.value();
                Cancellation {
                    actor: actor.to_owned(),
                    reason: reason.to_owned(),
                }
            }))
        })
        .await
    }
}