agentplane 0.45.0

//! The sweeper: what turns a passing instant into something that happened.
//!
//! Until something runs on a clock, a deadline is a number in a table and an
//! unclaimed event is a row nobody reads. That is the failure mode this whole
//! runtime is built against — not a crash, but a silence. A breached regulatory
//! window that nothing announced is indistinguishable, from the outside, from
//! one that was met.
//!
//! One tick does six things:
//!
//! | Finding | What happens |
//! |---|---|
//! | An instance died holding a run | The run is taken over at the next epoch and resumed |
//! | An obligation is approaching | `DeadlineTransition` to `Warned` |
//! | An obligation has passed unmet | `DeadlineTransition` to `Breached`; the case is escalated |
//! | A human task's window closed | The declared `on_expiry` is applied — never decided in the moment |
//! | An event nobody claimed aged out | Dead-lettered with a reason |
//! | A sleeping run's instant arrived | The wake-up is journaled and the run resumes |
//!
//! The middle four are loud. The first and the last are the system working — a
//! recovered run and a fired timer are reported so a quiet plane is
//! distinguishable from a stalled one, not so somebody is paged. A recovery
//! still means an *instance* died, which is worth a look for a different
//! reason than any run.

use std::sync::Arc;

use crate::core::{
    CaseId, DeadlineState, Decision, InboundEvent, OnExpiry, RunId, RuntimeError, StoreError,
    SweptAction, TaskState, Timestamp,
};
use crate::journal::{Append, JournalStore, RecordKind};

use super::executor::{LEASE_TTL, RunStatus, Runtime};
use super::metrics::{self, Census};

/// The source a human decision arrives under.
///
/// This plane's own worklist, not an outside party — and named so a counterparty
/// cannot mint an event that deduplicates against a real decision or that a
/// policy would mistake for one.
pub const SOURCE_WORKLIST: &str = "agentplane://worklist";

/// The epoch a sweep's own record is written under.
///
/// A sweep run is created here and never taken over, so there is no ownership
/// to arbitrate and the epoch never moves. It is named rather than spelled `1`
/// three times, because a literal repeated across a fence is how two of them
/// come to disagree.
const SWEEP_EPOCH: crate::core::Epoch = 1;

/// The outcome a sweep run is sealed with.
///
/// Not `Succeeded`: a sweep does not succeed or fail at a goal, it reports what
/// it found. Reusing a run status would make a tick that breached forty
/// obligations indistinguishable from a plan that completed.
///
/// `pub(crate)` so [`SEALED_OUTCOMES`](super::SEALED_OUTCOMES) can name it
/// rather than restate it — the list and the sealer must agree byte for byte.
pub(crate) const SWEEP_OUTCOME: &str = "swept";

/// What one tick did, opening a run only if there is anything to say.
///
/// Lazy on purpose. A quiet plane sweeps constantly and should leave nothing
/// behind: opening a run per tick would fill the Merkle log with evidence that
/// nothing happened, and a log of nothings is where the somethings hide.
///
/// **No decision is ever off the record.** A note buffered until the tick
/// ends orphans its record permanently on a crash: the state changes a sweep
/// applies (a breach, an escalation) are idempotent transitions, so the next
/// tick finds the transition already applied and re-decides nothing. So `note`
/// appends immediately, and `seal` only closes the run the notes already live
/// in. Most decisions are noted before they act. A breach acts first — the
/// store marks it as owing its account in the same write — and is noted
/// after, so a breach that lost its race writes nothing and one that applied
/// is found again by the next tick until its notes land.
struct SweepLedger {
    run: Option<RunId>,
    /// Whether at least one note has durably landed — the difference between
    /// "the tick decided nothing" and "the tick's first note failed before
    /// anything was applied", both of which leave nothing to seal.
    wrote: bool,
}

impl SweepLedger {
    const fn new() -> Self {
        Self {
            run: None,
            wrote: false,
        }
    }

    /// Durably note one decision, opening the tick's run the first time.
    ///
    /// A note that cannot be written fails the tick loudly. Before the state
    /// change it describes, that leaves the decision unapplied; after a
    /// breach, it leaves the breach owed its account for the next tick.
    ///
    /// `case` stamps the record so `JournalStore::case_history` finds it. That
    /// is the whole point of writing this down: the question is *why is this
    /// case escalated*, and a sweep run belongs to no case, so a walk over the
    /// case's own runs would never reach it.
    async fn note(
        &mut self,
        store: &Arc<dyn JournalStore>,
        case: Option<CaseId>,
        subject: String,
        action: SweptAction,
        detail: Option<String>,
    ) -> Result<(), RuntimeError> {
        let run = *self.run.get_or_insert_with(RunId::generate);
        let mut entry = crate::journal::Append::new(
            run,
            RecordKind::Swept {
                subject,
                action,
                detail,
            },
        );
        if let Some(case) = case {
            entry = entry.case(case);
        }
        store
            .append(SWEEP_EPOCH, vec![entry])
            .await
            .map_err(RuntimeError::from_store)?;
        self.wrote = true;
        Ok(())
    }

    /// Close the tick's record.
    ///
    /// Sealed, so it enters the Merkle log like any other run and the external
    /// audit tool checks it without being taught what a sweep is. Best-effort
    /// in the same sense settlement is: the notes are already durable and the
    /// state already changed, so failing the tick because its *closure* would
    /// not write would turn a bookkeeping problem into an operational one — but
    /// it is loud, because a sweep whose evidence is missing is the case this
    /// whole mechanism exists to prevent.
    ///
    /// The three outcomes are distinct on purpose: one answer for both a quiet
    /// tick and a tick whose evidence write *failed* leaves a report unable to
    /// tell "nothing happened" from "something happened and its record did not"
    /// — the exact detection-without-delivery failure I13 rules out, applied to
    /// the one run whose whole purpose is to make the sweeper's decisions
    /// answerable from the journal.
    async fn seal(self, store: &Arc<dyn JournalStore>) -> SweepRecord {
        let Some(run) = self.run else {
            return SweepRecord::Quiet;
        };
        if !self.wrote {
            // The first note failed before anything was applied: no decision
            // stands, so there is nothing to seal — and the phase error that
            // stopped the tick is already on its way to the caller.
            return SweepRecord::Quiet;
        }

        // The conclusion goes *in* the chain before the chain closes over it,
        // exactly as `conclude` does it for an ordinary run: tamper evidence
        // has to cover how the record ended, and the Merkle log commits to a
        // chain head that already includes its own sealing.
        let head = match store.head(run).await {
            Ok(head) => head,
            Err(e) => {
                tracing::error!(%run, error = %e, "a sweep's record could not be read back");
                return SweepRecord::EvidenceLost;
            }
        };
        let sealed = crate::journal::Append::new(
            run,
            RecordKind::RunConcluded {
                outcome: SWEEP_OUTCOME.to_owned(),
                reason: None,
                exhaustion: None,
                live_spend: crate::core::Spend::default(),
                chain_head: head.hash,
            },
        );
        if let Err(e) = store.append(SWEEP_EPOCH, vec![sealed]).await {
            tracing::error!(%run, error = %e, "a sweep's record could not be closed");
            return SweepRecord::EvidenceLost;
        }
        if let Err(e) = store.seal(run, SWEEP_EPOCH, SWEEP_OUTCOME).await {
            tracing::error!(%run, error = %e, "a sweep's record could not be sealed");
            return SweepRecord::EvidenceLost;
        }
        SweepRecord::Recorded(run)
    }
}

/// The fate of a tick's own evidence.
///
/// Three outcomes rather than an `Option<RunId>`, because a quiet tick and a
/// tick whose evidence could not be written are the two states an operator most
/// needs to tell apart: the first is the plane resting, the second is a decision
/// with no durable account of who made it.
enum SweepRecord {
    /// The tick decided nothing, so there was nothing to record.
    Quiet,
    /// The tick's decisions are in this sealed run.
    Recorded(RunId),
    /// The tick decided something and its record could not be written. The state
    /// changed and the journal did not.
    EvidenceLost,
}

/// What one timer pass did: wakes delivered, and wakes that died trying.
///
/// Two numbers rather than one, because they call for opposite responses. A
/// fired timer is the system working; a failed one is a run whose wake is now
/// late, healing through the recovery pass rather than through this one. A
/// single count would let the second hide inside the first.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct WokenRuns {
    /// Sleeping runs resumed because their instant arrived.
    pub fired: usize,
    /// Due timers whose wake or resume failed; retried by claim expiry, and a
    /// wake that was already recorded reaches the recovery pass.
    pub failed: usize,
}

/// What one redelivery pass finished, and how many parked waits it saw.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct Redelivered {
    /// Deliveries this pass completed.
    pub finished: usize,
    /// Parked waits this pass examined.
    ///
    /// The cap is judged on this rather than on `finished`, for the reason the
    /// timer pass counts failed wakes: a tick whose whole batch was still
    /// live-owned has seen as little of the backlog as one that finished
    /// every delivery in it.
    pub examined: usize,
}

/// Which sweeps hit their cap this tick.
///
/// A bounded query returns a list shaped exactly like a complete one, so a tick
/// that handled its cap and a tick that handled everything are indistinguishable
/// from the counters alone — and those are the two states an operator most needs
/// to tell apart, because the first means the backlog is growing while the
/// report looks normal.
///
/// Named flags rather than a list of strings: a caller that wants to alert on
/// deadlines specifically should not be matching on a message, and a field
/// added here is a field the compiler makes every reader consider.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
// A bool per capped sweep is what this struct *is*: one independent yes/no
// each, meaning "that backlog may be deeper than this tick saw". They are not
// a state machine wearing flags — no combination is invalid — and an enum over
// every combination would be the lint satisfied at the reader's expense.
#[allow(clippy::struct_excessive_bools)]
pub struct Saturation {
    /// Timers fired up to the cap; more may have been due.
    pub timers: bool,
    /// Obligations handled up to the cap; more may have been outstanding.
    pub deadlines: bool,
    /// Expired tasks handled up to the cap; more may have been overdue.
    pub tasks: bool,
    /// Abandoned runs recovered up to the cap; more may have been stranded.
    pub recovery: bool,
    /// Claimed-but-undelivered events examined up to the cap; more may have
    /// been parked.
    ///
    /// The pass with the least tolerance for going unreported: a delivery that
    /// died between the claim and the resume blocks every deduplicated retry of
    /// itself, so this sweep is the only driver left for a message that arrived
    /// in time.
    pub redeliveries: bool,
}

impl Saturation {
    /// Whether any sweep was capped.
    ///
    /// A saturated tick means *at least* the cap was waiting, never that the
    /// cap was all there was.
    #[must_use]
    pub const fn any(self) -> bool {
        self.timers || self.deadlines || self.tasks || self.recovery || self.redeliveries
    }
}

/// How much one sweep will take on, per kind of work.
///
/// Bounded so a tick is bounded: a sweep that tried to drain an unbounded
/// backlog would hold a worker for as long as the backlog is deep, and a plane
/// whose sweeper is busy is a plane whose *next* obligation is not being
/// noticed either.
///
/// Each cap is paired with a saturation check that reaches
/// [`SweepReport::saturated`], because a bounded result is shaped exactly like
/// a complete one and the difference is the whole signal.
const TIMER_BATCH: usize = 128;
const DEADLINE_BATCH: usize = 512;
const TASK_BATCH: usize = 512;
/// The smallest cap of the four, deliberately. Recovering a run is not a row
/// update: it replays the run's journal and then executes **live** from the
/// frontier, which may dispatch a model call. One tick therefore takes on a
/// bounded number of resurrections, and a mass failure — a whole instance's
/// worth of runs orphaned at once — drains over several ticks with the
/// saturation flag up rather than holding one tick hostage for the duration.
const RECOVERY_BATCH: usize = 32;
/// Redelivery walks the waiting list and touches the store per subscription,
/// so it takes the same bounded bite the timer pass does.
const EVENT_BATCH: usize = 128;

/// What one tick found and did.
///
/// Every field is a number worth alerting on. `breached` above zero means a
/// window was missed; `dead_lettered` above zero means a correlation key is
/// wrong somewhere.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct SweepReport {
    pub warned: usize,
    pub breached: usize,
    pub tasks_expired: usize,
    pub tasks_escalated: usize,
    pub dead_lettered: usize,
    /// Claimed-but-undelivered events whose delivery this tick finished.
    ///
    /// Each is a message that arrived in time and whose delivery died between
    /// the claim and the resume — a crash in that window, or an owner that
    /// outlived the delivery's bounded retry. The redelivery is the system
    /// healing; a persistent count here means deliveries keep dying, which is
    /// worth asking why.
    pub events_redelivered: usize,
    /// Sleeping runs woken because their instant arrived.
    ///
    /// Not a number to alert on — a fired timer is the system working. It is
    /// reported so a quiet plane is distinguishable from a stalled one.
    pub timers_fired: usize,
    /// Wake-ups whose resume failed after the wake was durably recorded.
    ///
    /// The timer is gone but the run has not moved. It is not lost: the lease
    /// the wake acquired lapses unreleased, and the recovery pass picks the
    /// run up on a later tick — but a failure on this path means wakes are
    /// arriving late, and a human should know why the first attempt died.
    pub wake_failures: usize,
    /// Runs taken over and resumed because their owner's lease lapsed without
    /// release — an instance died holding them.
    ///
    /// Each is a crash the plane healed. The healing is routine; the dying is
    /// not, so [`needs_attention`](Self::needs_attention) stays quiet about
    /// this count only when it is zero **and** nothing failed — a steady
    /// recovery rate with healthy instances is a contradiction worth seeing.
    pub runs_recovered: usize,
    /// Abandoned runs the recovery pass tried to resume and could not.
    ///
    /// The run stays listed and is retried next tick, so a persistent count
    /// here is one stuck run rather than many — but it is stuck, nothing else
    /// will unstick it, and the reason is in the log of the tick that failed.
    pub recovery_failures: usize,
    /// Which sweeps came back **full**, and therefore may not have seen
    /// everything that was waiting.
    pub saturated: Saturation,
    /// The sealed run holding this tick's own record, when it did anything.
    ///
    /// `None` for a quiet tick, which writes nothing — **and** `None` when the
    /// evidence write failed, which is a different fact entirely and is carried
    /// separately in [`evidence_lost`](Self::evidence_lost). Present means the
    /// sweeper's decisions are answerable from the journal rather than only
    /// from the state they produced.
    pub record: Option<RunId>,
    /// The tick decided something and its own record could not be written.
    ///
    /// The most serious thing this report can carry: the state changed —
    /// obligations breached, cases escalated — and the durable, tamper-evident
    /// account of *who decided that and when* did not. Distinct from a quiet
    /// tick, which also leaves `record` empty but changed nothing, and it makes
    /// [`needs_attention`](Self::needs_attention) true so it cannot pass as one.
    pub evidence_lost: bool,
    /// What the plane is holding, as of this sweep.
    ///
    /// Carried on the report as well as emitted, so an embedder that wants the
    /// numbers does not have to stand up a metrics subscriber to see them.
    pub census: Census,
    /// Cosignatures gathered over this plane's checkpoint this tick.
    ///
    /// Zero on a tick that did not submit — no witnesses configured, or a log
    /// that has not grown since the last round that met its quorum. A number
    /// to see rather than to alert on: evidence accumulating is the system
    /// working.
    pub cosignatures: usize,
    /// Cosignatures the declared quorum still wants.
    ///
    /// The plane is running and its history is anchored by fewer independent
    /// parties than the deployment said it required, which is a state nothing
    /// else will clear.
    pub witness_shortfall: usize,
    /// Witnesses that refused on integrity grounds: a log that shrank, forked,
    /// or whose claimed growth would not verify.
    ///
    /// Reported even when the quorum was **met**. Two honest cosigners do not
    /// answer a third that remembers a different history — they may simply
    /// never have seen the history it remembers.
    pub witness_integrity: usize,
    /// The gauges could not be read this tick.
    ///
    /// The counters above are still real — the tick's work happened — but the
    /// census is a default, not a reading. Carried as a flag rather than an
    /// error because a report that arrives degraded beats one that was thrown
    /// away over a gauge: the breaches and recoveries it accounts for already
    /// happened.
    pub census_unavailable: bool,
}

impl SweepReport {
    /// Whether this tick found anything a human should see.
    #[must_use]
    pub fn needs_attention(&self) -> bool {
        self.breached > 0
            || self.tasks_expired > 0
            || self.dead_lettered > 0
            // A saturated sweep is the case a human most needs to see: the
            // plane is keeping up with its cap rather than with its work.
            || self.saturated.any()
            // A decision whose evidence never landed is the I13 failure this
            // report exists to surface: the state moved and the record did not.
            || self.evidence_lost
            // A recovery that keeps failing is a run nothing else will
            // unstick; a wake whose resume died is a run arriving late.
            // Recoveries that *succeeded* stay off this list — they are the
            // plane healing, findable in the report and the sweep's own run.
            || self.recovery_failures > 0
            || self.wake_failures > 0
            // Gauges that could not be read are a blind spot wearing a
            // default; somebody should know the numbers are missing, not
            // merely zero.
            || self.census_unavailable
            // A plane anchored by fewer parties than it declared, and a
            // witness that saw this history move. The second is loud even
            // beside a met quorum, which is why it is its own count.
            || self.witness_shortfall > 0
            || self.witness_integrity > 0
    }

    #[must_use]
    pub fn is_quiet(&self) -> bool {
        // Enumerated, not `*self == Self::default()`. That idiom ties the
        // meaning of "quiet" to the struct's field list, so adding a field
        // silently redefines it — which is exactly what happened when the
        // census arrived: a plane holding one open case stopped being quiet,
        // though the sweep had done nothing at all.
        //
        // The census is *state*, not activity. A plane with five hundred open
        // cases and nothing due is quiet; that is the whole distinction between
        // a gauge and a counter, and this predicate is about the counters.
        !self.saturated.any()
            && self.warned == 0
            && self.breached == 0
            && self.tasks_expired == 0
            && self.tasks_escalated == 0
            && self.dead_lettered == 0
            && self.events_redelivered == 0
            && self.timers_fired == 0
            && self.wake_failures == 0
            && self.runs_recovered == 0
            && self.recovery_failures == 0
            // Gathering evidence is activity, and a shortfall or an integrity
            // refusal most of all. A tick that submitted nothing — no
            // witnesses, or a log that has not grown — leaves all three at
            // zero and stays quiet.
            && self.cosignatures == 0
            && self.witness_shortfall == 0
            && self.witness_integrity == 0
    }
}

impl Runtime {
    /// Submit this plane's current checkpoint to its witnesses.
    ///
    /// Off the run path and after sealing, which is what witnessing is: the
    /// evidence is read after the fact, and making a run wait on a third party
    /// would trade the plane's availability for it.
    ///
    /// Nothing happens without witnesses configured. With them, a round is
    /// skipped when the log has not grown since the last round that met the
    /// declared quorum — the alternative is re-submitting an unchanged
    /// checkpoint on every tick, which every witness answers and none of which
    /// tells anybody anything.
    ///
    /// A witness failing is never an error here. It is what the counts on
    /// [`SweepReport`] carry, so a shortfall reaches
    /// [`needs_attention`](SweepReport::needs_attention) rather than aborting
    /// the sweep that already breached obligations and recovered runs.
    ///
    /// # Errors
    ///
    /// Only if the **store** cannot answer with a checkpoint or build a
    /// consistency proof.
    async fn cosign_checkpoint(&self, report: &mut SweepReport) -> Result<(), RuntimeError> {
        use std::sync::atomic::Ordering;

        let Some(witnessing) = self.witnessing() else {
            return Ok(());
        };
        let checkpoint = self.store().checkpoint().await?;
        // Equality only: an unchanged log needs no new evidence, while a log
        // reporting a *smaller* size than what was already cosigned is
        // precisely the submission a witness must be given the chance to
        // refuse — so it is not skipped.
        if checkpoint.size == witnessing.submitted.load(Ordering::Relaxed) {
            return Ok(());
        }
        let outcome = crate::journal::cosign_quorum(
            self.store().as_ref(),
            &checkpoint,
            &witnessing.witnesses,
            witnessing.quorum,
        )
        .await?;
        report.cosignatures = outcome.cosignatures.len();
        report.witness_shortfall = outcome.shortfall();
        report.witness_integrity = outcome.integrity.len();
        for (index, refusal) in &outcome.integrity {
            // The audience for this is never only the operator who runs the
            // plane: a witness reporting that this log moved is the one event
            // the party being audited has an interest in nobody hearing.
            tracing::error!(
                target: crate::runtime::telemetry::WITNESS_INTEGRITY,
                witness = index,
                origin = %checkpoint.origin,
                size = checkpoint.size,
                refusal = %refusal,
                "a witness refused this plane's checkpoint on integrity grounds"
            );
        }
        if outcome.met() {
            witnessing
                .submitted
                .store(checkpoint.size, Ordering::Relaxed);
        }
        Ok(())
    }

    /// Run one sweep.
    ///
    /// Idempotent: a transition already applied is not applied twice, so calling
    /// this on a timer — or twice by accident, or from two instances — is safe.
    ///
    /// `now` is passed in rather than read so the caller controls the clock.
    /// That keeps the sweeper testable at all, and lets a simulation drive it
    /// through a year in milliseconds.
    pub async fn sweep(
        &self,
        now: Timestamp,
        event_grace: std::time::Duration,
    ) -> Result<SweepReport, RuntimeError> {
        let mut report = SweepReport::default();
        let mut ledger = SweepLedger::new();

        // The fallible phases, folded into one result so the ledger outlives
        // whichever of them fails. The deadline pass may have breached an
        // obligation *into state* before a later phase errored, and evidence
        // already earned must not leave with the error — a `?` past an
        // unsealed ledger is the sweep silently dropping the account of
        // decisions it already applied, which is the exact failure the ledger
        // exists to prevent, reintroduced by control flow.
        let phases: Result<(), RuntimeError> = async {
            // Recovery first: an abandoned run may be one step from meeting a
            // deadline the pass below would otherwise breach.
            self.recover_abandoned(&mut report, &mut ledger).await?;
            report.warned += self.sweep_deadlines(now, &mut report, &mut ledger).await?;
            self.sweep_tasks(now, &mut report, &mut ledger).await?;

            if self.events().is_some() {
                // Deliveries that died between the claim and the resume — a
                // crash in that window, or an owner that outlived the
                // delivery's bounded retry. The claimed event blocks every
                // dedup'd retry of itself, so this pass is the only driver
                // left; without it the message that arrived in time is parked
                // forever behind its own claim.
                let again = self.redeliver_claimed(EVENT_BATCH).await?;
                report.events_redelivered = again.finished;
                report.saturated.redeliveries = again.examined >= EVENT_BATCH;
                report.dead_lettered = self.sweep_events(event_grace).await?;
            }
            if self.timers().is_some() {
                let woken = self.fire_timers(now).await?;
                report.timers_fired = woken.fired;
                report.wake_failures = woken.failed;
                // Failed wakes consumed claims too, so the cap is judged on
                // the whole batch: a tick that failed its way through the cap
                // has seen as little of the backlog as one that fired it.
                if woken.fired + woken.failed >= TIMER_BATCH {
                    report.saturated.timers = true;
                }
            }
            // Last of the phases, and deliberately: it makes no state change
            // any other phase depends on, and it is the one phase whose
            // counterparty is somebody else's server. A witness that is slow
            // must not delay a breach or a recovery.
            self.cosign_checkpoint(&mut report).await?;
            Ok(())
        }
        .await;

        // Written before the census so the record covers the decisions, not the
        // reading. A tick that decided nothing writes nothing — and a tick that
        // decided something and then *errored* still writes, which is why this
        // sits outside the block above rather than after a `?`.
        match ledger.seal(self.store()).await {
            SweepRecord::Quiet => {}
            SweepRecord::Recorded(run) => report.record = Some(run),
            SweepRecord::EvidenceLost => report.evidence_lost = true,
        }
        phases?;

        // Observed last, so the reading reflects what this sweep just resolved
        // rather than the backlog it was about to clear. A failed reading
        // **degrades** the report rather than discarding it: everything above
        // — the breaches, the recoveries, the sealed evidence run — already
        // happened, and a `?` here threw the whole account away because one
        // gauge could not be read at the end. The flag keeps the absence
        // loud; the counters keep what the tick actually did.
        match self.census(now).await {
            Ok(census) => {
                report.census = census;
                self.meter().census(&report.census);
            }
            Err(error) => {
                report.census_unavailable = true;
                tracing::error!(
                    %error,
                    "the census could not be read; this report carries the tick's \
                     counters and no gauges"
                );
            }
        }

        Ok(report)
    }

    /// Take over and resume the runs an instance died holding.
    ///
    /// The candidate set is exact, not heuristic: every clean exit — sealed,
    /// failed, suspended — hands its lease back, so a lease that **expired
    /// without release** names a run somebody was executing when their process
    /// stopped. Fencing makes this takeover safe (the resume bumps the epoch,
    /// and the store refuses the dead owner's next append), replay makes it
    /// correct (completed effects are read back, never redone), and this pass
    /// is what makes it *happen* — without it, a crashed run with no pending
    /// timer and no inbound event has no driver, appears in no backlog, and
    /// waits forever while looking exactly like work in progress.
    ///
    /// Per-run failures are contained: one unresumable run is counted, logged
    /// and retried once the lease this pass left on it lapses, rather than
    /// blocking the runs behind it. A lease held by someone else is not a
    /// failure at all — another instance got there first, which is the
    /// mechanism working.
    async fn recover_abandoned(
        &self,
        report: &mut SweepReport,
        ledger: &mut SweepLedger,
    ) -> Result<(), RuntimeError> {
        let stranded = self
            .store()
            .abandoned_runs(RECOVERY_BATCH)
            .await
            .map_err(RuntimeError::from_store)?;
        if stranded.len() >= RECOVERY_BATCH {
            report.saturated.recovery = true;
        }
        for run in stranded {
            // The note lands **before** the takeover — the ledger's rule,
            // and this pass is where getting it backwards loses the record
            // permanently rather than briefly: a resume that concludes
            // releases the lease, the run leaves the driving query, and no
            // later tick can re-select it to write the account. The note
            // carries the lapse and the takeover attempt, deliberately not
            // the outcome — the recovered run's own journal answers that.
            ledger
                .note(
                    self.store(),
                    None,
                    run.to_string(),
                    SweptAction::RunRecovered,
                    Some(
                        "its owner's lease lapsed without release; taking the run \
                         over and resuming it"
                            .to_owned(),
                    ),
                )
                .await?;
            match self.recover_abandoned_run(run).await {
                Ok(outcome) => {
                    // A resume that concluded released its lease on the way
                    // out. The one path that does not is the closed-run
                    // no-op — a crash landed between `seal` and the release —
                    // and without this the same sealed run would be "recovered"
                    // every tick forever. Acquire-then-release is how a party
                    // with no epoch clears a lease it does not hold; a live
                    // lease refuses the acquire, which is the guard working.
                    if let Ok(lease) = self.store().acquire(run, self.owner_id(), LEASE_TTL).await {
                        let _ = self.store().release_lease(run, lease.epoch).await;
                    }
                    let status = outcome.status.as_str().to_owned();
                    tracing::info!(
                        target: super::telemetry::RUN_RECOVERED,
                        %run,
                        outcome = %status,
                    );
                    self.meter().count(metrics::RUNS_RECOVERED, &status);
                    report.runs_recovered += 1;
                }
                // Someone else holds it live: a second instance's recovery —
                // or the owner back from a stall, about to be fenced. Either
                // way it is being handled, and counting it as a failure would
                // page an operator about a race the design already settles.
                // The note above stands, and honestly: this instance observed
                // the lapse and moved to take the run over; the run's own
                // journal shows who won. `Fenced` is the same race lost
                // later: the other instance claimed the run mid-resume.
                Err(RuntimeError::LeaseHeld { .. } | RuntimeError::Fenced { .. }) => {}
                // A lease over an **empty** journal: admission acquired and
                // died before its first append landed. There is no run — the
                // atomic admission batch never committed, so nothing was
                // declared, authorized or performed. Clearing the lease is the
                // whole recovery; leaving this to the failure arm below would
                // retry a resume that cannot exist, every tick, forever. The
                // explanation is its own note, written before the clear it
                // describes.
                Err(RuntimeError::Store(StoreError::NotFound(_))) => {
                    ledger
                        .note(
                            self.store(),
                            None,
                            run.to_string(),
                            SweptAction::RunRecovered,
                            Some(
                                "its owner died between acquiring the lease and the \
                                 admission append; no run exists, so clearing the \
                                 lease is the whole recovery"
                                    .to_owned(),
                            ),
                        )
                        .await?;
                    if let Ok(lease) = self.store().acquire(run, self.owner_id(), LEASE_TTL).await {
                        if let Err(error) = self.release_empty_quota_reservation(run).await {
                            tracing::warn!(%run, %error, "could not release the quota slot of an empty abandoned admission");
                        }
                        let _ = self.store().release_lease(run, lease.epoch).await;
                    }
                    report.runs_recovered += 1;
                }
                Err(error) => {
                    tracing::error!(%run, %error, "an abandoned run could not be resumed");
                    self.meter().count(metrics::RECOVERY_FAILURES, "");
                    report.recovery_failures += 1;
                }
            }
        }
        Ok(())
    }

    /// Read the gauges: how much is open right now.
    ///
    /// Queried from the stores rather than accumulated, because a counter
    /// incremented on open and decremented on close drifts permanently the first
    /// time a process dies between the state change and the emission — and it
    /// drifts *plausibly*, which is worse than drifting obviously.
    ///
    /// Stores that were never configured contribute zero rather than an error: a
    /// plane with no human tasks has no open tasks, and refusing to report the
    /// gauges it *does* have would be the silent failure this exists to prevent.
    pub async fn census(&self, now: Timestamp) -> Result<Census, StoreError> {
        let mut c = Census::default();
        if let Some(cases) = self.cases() {
            let cc = cases.census(now).await?;
            c.open_cases = cc.open;
            c.oldest_case_age_secs = cc.oldest_age_secs;
            c.due_deadlines = cc.due;
            c.unaccounted_breaches = cc.breached;
        }
        if let Some(timers) = self.timers() {
            c.pending_timers = timers.pending_count().await?;
        }
        if let Some(tasks) = self.tasks() {
            c.open_tasks = tasks.open_count().await?;
        }
        // Not behind an `if let`: the journal is the one store every plane has,
        // so a quarantine backlog is never the gauge that quietly reports zero
        // because something was not wired.
        c.quarantined_runs = self
            .journal()
            .count_by_outcome(RunStatus::Quarantined(String::new()).as_str())
            .await?;
        Ok(c)
    }

    /// Wake the runs whose instant has arrived.
    ///
    /// The shape mirrors event delivery exactly, and for the same reasons:
    /// claim atomically so two sweepers cannot wake one run twice, record the
    /// wake-up as the sleeping effect's result, then let replay do the rest.
    /// The resumed run reads the timer back like any other completed effect,
    /// so none of the suspension machinery exists twice.
    ///
    /// One run's failure does not block the runs behind it: a batch that stops
    /// at the first error hands a single unresumable run a veto over every
    /// later wake in the tick — and the failed run heals anyway: its wake was
    /// recorded under a lease this pass acquired and never released, so the
    /// recovery pass finds it once that lease lapses. The failure is counted
    /// rather than returned, because a caller that got a count and an error
    /// would have neither.
    pub async fn fire_timers(&self, now: Timestamp) -> Result<WokenRuns, RuntimeError> {
        let timers = self.timers().ok_or_else(|| {
            RuntimeError::PlanContract(
                "this runtime has no timer store — build it with `.timers(store)`".into(),
            )
        })?;

        let due = timers
            .claim_due(now, TIMER_BATCH)
            .await
            .map_err(RuntimeError::from_store)?;
        let mut woken = WokenRuns::default();
        for timer in due {
            match self.fire_one(&timer).await {
                Ok(()) => woken.fired += 1,
                Err(error) => {
                    tracing::error!(
                        run = %timer.run,
                        step = %timer.step,
                        %error,
                        "a due timer's wake failed; the claim lease retries it, \
                         and a wake recorded before the failure reaches the \
                         recovery pass",
                    );
                    woken.failed += 1;
                }
            }
        }

        Ok(woken)
    }

    /// Record one timer's wake and resume its run.
    async fn fire_one(&self, timer: &crate::core::Timer) -> Result<(), RuntimeError> {
        let lease = self
            .store()
            .acquire(timer.run, self.owner_id(), LEASE_TTL)
            .await
            .map_err(RuntimeError::from_store)?;

        // A crash between last tick's append and its disarm leaves the wake
        // recorded and the timer armed, so this claim fires again. The check
        // makes that a second *resume*, not a second record: an append here
        // would put a duplicate wake in the chain, and the journal is the one
        // place a retry must never show up twice.
        let already_recorded = self
            .store()
            .read(timer.run, 1)
            .await
            .map_err(RuntimeError::from_store)?
            .iter()
            .any(|record| {
                record.effect_key() == Some(timer.effect)
                    && matches!(record.kind(), RecordKind::EffectDone { .. })
            });
        if !already_recorded {
            let mut record = Append::new(
                timer.run,
                RecordKind::EffectDone {
                    // The instant it *was due*, not the instant the sweep
                    // noticed. A sweep that ran late must not make the run
                    // believe it slept longer than it was told to, or a
                    // replayed run would compute different downstream deadlines
                    // than the original.
                    output: serde_json::json!({ "fired_at": timer.fire_at.unix_timestamp() }),
                    // A timer has no sender and nobody minted it.
                    source: None,
                    by: None,
                    spend: crate::core::Spend::default(),
                    // The runtime's own instant, chosen and journaled here —
                    // one of the few values that crosses no trust boundary.
                    declared: crate::core::DeclaredOutput::trusted(),
                },
            )
            .effect(timer.effect)
            .step(timer.step)
            .phase(timer.phase);
            if let Some(c) = timer.case {
                record = record.case(c);
            }

            self.store()
                .append(lease.epoch, vec![record])
                .await
                .map_err(RuntimeError::from_store)?;
        }

        self.timers()
            .ok_or_else(|| RuntimeError::PlanContract("timer store vanished mid-sweep".into()))?
            .disarm(timer.run, timer.effect)
            .await
            .map_err(RuntimeError::from_store)?;

        tracing::info!(
            target: super::telemetry::TIMER_FIRED,
            run = %timer.run,
            step = %timer.step,
            due = %timer.fire_at,
        );
        self.meter().count(metrics::TIMERS_FIRED, "");

        // The wake is recorded and the timer disarmed; the resume continues
        // under the *same* lease rather than releasing and re-acquiring. The
        // old release-then-acquire choreography opened a window this pass
        // could not see again: a crash between the release and the resume's
        // acquire left a released lease over a run whose timer was already
        // disarmed — no driver, and absent from the abandonment queue, which
        // lists only leases that expired while still naming an owner. Handing
        // the lease over keeps the run owned continuously from wake to
        // conclusion, so a crash anywhere in the resume leaves an owned lease
        // the sweep drains. A `Fenced` surfaces if this lease lapses
        // mid-resume and somebody else claims the run — they read the recorded
        // wake like any other completed effect, so the wake still happened.
        match self.resume_holding(timer.run, lease).await {
            Ok(_) | Err(RuntimeError::Fenced { .. }) => Ok(()),
            Err(e) => Err(e),
        }
    }

    /// Warn on approaching obligations; escalate breached ones.
    async fn sweep_deadlines(
        &self,
        now: Timestamp,
        report: &mut SweepReport,
        ledger: &mut SweepLedger,
    ) -> Result<usize, RuntimeError> {
        let Some(cases) = self.cases() else {
            return Ok(0);
        };

        // Breaches a crashed tick applied and never accounted for, first: the
        // obligation has left `due`, so this is the only pass that finds it.
        let owed = cases
            .breaches_to_note(DEADLINE_BATCH)
            .await
            .map_err(RuntimeError::from_store)?;
        if owed.len() >= DEADLINE_BATCH {
            report.saturated.deadlines = true;
        }
        for deadline in owed {
            self.account_for_breach(&deadline, report, ledger).await?;
        }

        let mut warned = 0;
        let due = cases
            .due(now, DEADLINE_BATCH)
            .await
            .map_err(RuntimeError::from_store)?;
        if due.len() >= DEADLINE_BATCH {
            report.saturated.deadlines = true;
        }
        for deadline in due {
            if deadline.is_due(now) {
                // Past the instant with the obligation unmet. This is the event
                // the whole deadline machinery exists to produce.
                //
                // **The act first, then its account.** One conditional write:
                // `due` is a read that is already stale — a run may have met
                // the obligation and closed the case since — and the store
                // breaches only what is still outstanding and due, escalating
                // only a case that is not closed, in one transaction. A breach
                // that did not apply wrote nothing, so there is nothing on the
                // record to correct. One that did is marked in the same write
                // as owing its account, and the notes below retire the mark:
                // a crash between the two leaves an obligation the next tick's
                // drain notes, rather than a note claiming a breach that lost
                // its race.
                let applied = cases
                    .breach_deadline(deadline.case, &deadline.name, now)
                    .await
                    .map_err(RuntimeError::from_store)?;
                if applied {
                    self.account_for_breach(&deadline, report, ledger).await?;
                }
            } else if deadline.needs_warning(now) {
                ledger
                    .note(
                        self.store(),
                        Some(deadline.case),
                        deadline.case.to_string(),
                        SweptAction::DeadlineWarned,
                        Some(format!(
                            "'{}' comes due {}",
                            deadline.name, deadline.resolved_at
                        )),
                    )
                    .await?;
                // A run that met the obligation since `due` read it has
                // settled it; the warning is moot, not a failure of the tick.
                // A warning is noted before it is set, so the note above is
                // corrected on the record: the last word about the subject is
                // what happened to it.
                match cases
                    .set_deadline_state(deadline.case, &deadline.name, DeadlineState::Warned)
                    .await
                {
                    Ok(()) => warned += 1,
                    Err(StoreError::DeadlineFinal { .. }) => {
                        ledger
                            .note(
                                self.store(),
                                Some(deadline.case),
                                deadline.case.to_string(),
                                SweptAction::NotApplied,
                                Some(format!(
                                    "warning for '{}' not applied: the obligation was settled first",
                                    deadline.name
                                )),
                            )
                            .await?;
                    }
                    Err(e) => return Err(RuntimeError::from_store(e)),
                }
            }
        }
        Ok(warned)
    }

    /// Write the account of an applied breach — the breach and the escalation
    /// it caused — then retire the obligation's owed mark.
    ///
    /// The notes land before the mark is cleared, so a crash between them
    /// repeats the account next tick rather than losing it: a duplicate note
    /// is honest about a breach that happened, a missing one is not.
    async fn account_for_breach(
        &self,
        deadline: &crate::core::Deadline,
        report: &mut SweepReport,
        ledger: &mut SweepLedger,
    ) -> Result<(), RuntimeError> {
        let Some(cases) = self.cases() else {
            return Ok(());
        };
        ledger
            .note(
                self.store(),
                Some(deadline.case),
                deadline.case.to_string(),
                SweptAction::DeadlineBreached,
                Some(format!(
                    "'{}' was due {} and was not met",
                    deadline.name, deadline.resolved_at
                )),
            )
            .await?;
        ledger
            .note(
                self.store(),
                Some(deadline.case),
                deadline.case.to_string(),
                SweptAction::CaseEscalated,
                Some(format!("obligation '{}' was breached", deadline.name)),
            )
            .await?;
        cases
            .mark_breach_noted(deadline.case, &deadline.name)
            .await
            .map_err(RuntimeError::from_store)?;
        tracing::error!(
            target: super::telemetry::DEADLINE_BREACHED,
            case = %deadline.case,
            obligation = %deadline.name,
            due = %deadline.resolved_at,
        );
        self.meter().count(metrics::DEADLINE_BREACHES, "");
        report.breached += 1;
        Ok(())
    }

    /// Apply the declared expiry policy to tasks nobody answered.
    async fn sweep_tasks(
        &self,
        now: Timestamp,
        report: &mut SweepReport,
        ledger: &mut SweepLedger,
    ) -> Result<(), RuntimeError> {
        let Some(tasks) = self.tasks() else {
            return Ok(());
        };

        let overdue = tasks
            .overdue(now, TASK_BATCH)
            .await
            .map_err(RuntimeError::from_store)?;
        if overdue.len() >= TASK_BATCH {
            report.saturated.tasks = true;
        }
        for task in overdue {
            match task.on_expiry {
                // Widen the audience, clear the stale reservation, and keep
                // waiting. `escalate` is the write that removes the task from
                // `overdue`, so it goes last (the note is already durable) —
                // a crash between the two re-selects the task next tick and
                // repeats the note, which is the affordable direction.
                OnExpiry::Escalate => {
                    ledger
                        .note(
                            self.store(),
                            task.case,
                            task.id.to_hex(),
                            SweptAction::TaskEscalated,
                            Some("nobody answered inside the window".to_owned()),
                        )
                        .await?;
                    tasks
                        .escalate(task.id)
                        .await
                        .map_err(RuntimeError::from_store)?;
                    report.tasks_escalated += 1;
                }

                // Deny or proceed: both are decisions that were made in advance,
                // and both resume the run with a recorded answer rather than
                // leaving it hanging.
                policy => {
                    let decision = Decision::expired(policy);
                    ledger
                        .note(
                            self.store(),
                            task.case,
                            task.id.to_hex(),
                            SweptAction::TaskExpired,
                            // The *declared* policy, so the record says what was
                            // decided in advance rather than only what happened.
                            Some(format!("window closed; applied {policy:?}")),
                        )
                        .await?;
                    // A duplicate means an answer is already on record: a
                    // reviewer's whose settlement was lost, or this policy's
                    // own from a tick that died before settling. Whichever it
                    // was, the task is settled to it here — skipped, it would
                    // stay overdue and every tick write a fresh expiry note — and a
                    // reviewer's answer puts the loss of this one on the
                    // record after the note that claimed it.
                    let delivery = self.answer_task(task.id, &decision).await?;
                    if delivery == crate::core::Delivery::Duplicate {
                        match self.settle_answered(task.id).await? {
                            Some(crate::case::Minter::Operator(by)) => {
                                ledger
                                    .note(
                                        self.store(),
                                        task.case,
                                        task.id.to_hex(),
                                        SweptAction::NotApplied,
                                        Some(format!(
                                            "expiry not applied: '{}' answered first",
                                            by.actor()
                                        )),
                                    )
                                    .await?;
                            }
                            Some(crate::case::Minter::Nobody) => report.tasks_expired += 1,
                            None => {}
                        }
                        continue;
                    }
                    if tasks
                        .set_state(task.id, TaskState::Expired)
                        .await
                        .map_err(RuntimeError::from_store)?
                    {
                        report.tasks_expired += 1;
                    }
                }
            }
        }
        Ok(())
    }

    /// Settle a task whose answer is already on record to that answer's
    /// state, returning who minted it — or `None` when no answer is buffered.
    ///
    /// The delivery that met a duplicate learns here whose answer is already
    /// on record: a person's settles the task `Completed`, the expiry
    /// policy's `Expired`. Idempotent — a task already settled stays as it is.
    async fn settle_answered(
        &self,
        id: crate::core::TaskId,
    ) -> Result<Option<crate::case::Minter>, RuntimeError> {
        let (Some(tasks), Some(events)) = (self.tasks(), self.events()) else {
            return Ok(None);
        };
        let minter = events
            .minter(SOURCE_WORKLIST, &decision_event_id(id))
            .await
            .map_err(RuntimeError::from_store)?;
        let state = match &minter {
            Some(crate::case::Minter::Operator(_)) => TaskState::Completed,
            Some(crate::case::Minter::Nobody) => TaskState::Expired,
            None => return Ok(None),
        };
        if tasks
            .set_state(id, state)
            .await
            .map_err(RuntimeError::from_store)?
        {
            Ok(minter)
        } else {
            // Already settled: nothing changed, so nothing is reported.
            Ok(None)
        }
    }

    /// Record a human decision and resume the run waiting on it.
    ///
    /// The decision is delivered as an ordinary inbound event, so it travels the
    /// same buffered, deduplicated, single-consumer path as any other message —
    /// including the case where the run has not yet reached its wait. It is
    /// the one door into the reserved kind a waiting task listens for, which
    /// is why it is not public: an answer that skipped [`decide_task`]'s claim
    /// and eligibility checks would be an approval nobody was checked for.
    ///
    /// [`decide_task`]: Self::decide_task
    pub(crate) async fn answer_task(
        &self,
        id: crate::core::TaskId,
        decision: &Decision,
    ) -> Result<crate::core::Delivery, RuntimeError> {
        let event = InboundEvent::new(
            // The plane itself, not an outside party: a decision comes from the
            // worklist this runtime owns, and naming it as such keeps a
            // counterparty from minting an event that looks like one.
            SOURCE_WORKLIST,
            // One decision per task: the event id makes a double submit a
            // duplicate rather than a second answer.
            decision_event_id(id),
            super::ctx::TASK_DECIDED,
            serde_json::to_value(decision)?,
        )
        .correlate(crate::core::CorrelationKey::new("task", id.to_hex()));

        // The decider, lifted out of the payload and onto the event.
        //
        // The whole decision travels as this effect's output, and that output
        // is a **sealed** journal field: erase the run's payloads and the
        // reason goes, as it must, taking the approver's name with it. Every
        // operator act that *stops* a run already names its actor in the
        // clear; this is the same rule for the one act that lets a run carry
        // on. An expiry names nobody, and nobody is what it carries.
        let event = match decision.decided.operator() {
            Some(by) => event.minted_by(by.clone()),
            None => event,
        };

        self.deliver_minted(&event).await
    }

    /// Complete a task on behalf of a person, enforcing eligibility first.
    ///
    /// The claim is checked before the decision is recorded: an approval from
    /// somebody who was not permitted to give it is worse than no approval,
    /// because it looks like one.
    ///
    /// # Only a person's answer
    ///
    /// [`Decided::OnExpiry`](crate::core::Decided::OnExpiry) is refused here.
    /// It is the sweeper's to apply, from the policy the task declared, and it
    /// is the one decision with nobody to claim against: four-eyes has no
    /// actor to exclude and eligibility has no name to check, so every control
    /// this function exists to run would pass by having nothing to test. A
    /// caller reaching this door with one is either applying an expiry from
    /// outside the sweeper or dressing a person's answer as the policy's.
    pub async fn decide_task(
        &self,
        id: crate::core::TaskId,
        decision: &Decision,
        roles: &[String],
    ) -> Result<crate::core::Delivery, RuntimeError> {
        let tasks = self
            .tasks()
            .ok_or_else(|| RuntimeError::PlanContract("this runtime has no task store".into()))?;

        let Some(by) = decision.decided.operator() else {
            return Err(RuntimeError::PlanContract(
                "an expired window is the sweeper's to apply: this door records a person's \
                 answer, and an expiry has no actor to check a claim or an exclusion against"
                    .to_owned(),
            ));
        };

        // The claim protocol's refusals surface as
        // [`RuntimeError::TaskClaim`], exactly as the claim verb reports them.
        // Wrapped as a policy denial they would claim a rule fired when none
        // did, and collapse three different answers into one class.
        // An approval is only of what its decider can be shown. A plane whose
        // key ring cannot open this proposal — a terminal holding no key, a
        // key an erasure destroyed, an envelope that did not decode — reads
        // the row back withheld, and an approval of that is an approval of
        // arguments nobody saw. Refused before the claim, so the task stays
        // open for a surface that can show it. A rejection refuses unseen
        // arguments, which is safe.
        //
        // The reason is the one the store carried out of band, never the
        // proposal's shape: clear arguments spelled like an envelope are
        // arguments, and reading the shape let untrusted input make a task
        // nobody could approve.
        if decision.approved
            && let Some(task) = tasks.task(id).await.map_err(RuntimeError::from_store)?
            && let Some(reason) = task.withheld
        {
            return Err(RuntimeError::ProposalWithheld {
                task: id.to_string(),
                reason,
            });
        }

        let claimed = tasks.claim(id, by.actor(), roles).await?;

        // Bound to what the store holds as this person decides. The run
        // compares it with the task it proposed, so a row edited in between
        // — its arguments swapped for ones the reviewer would pass — yields an
        // approval of something nobody proposed, and the run refuses it.
        let mut decision = decision.clone();
        decision.reviewed = Some(claimed.justification.digest());
        let delivery = self.answer_task(id, &decision).await?;
        // An answer to this task is already on record. This decider's own —
        // a resubmission after the settlement below was lost — settles the
        // task and succeeds. Anybody else's, or the expiry policy's applied
        // since the claim, is the answer the run acted on: the task is settled
        // to it and this decision is refused rather than recorded over it.
        if matches!(delivery, crate::core::Delivery::Duplicate) {
            let events = self.events().ok_or_else(|| {
                RuntimeError::PlanContract("this runtime has no event store".into())
            })?;
            let minter = events
                .minter(SOURCE_WORKLIST, &decision_event_id(id))
                .await
                .map_err(RuntimeError::from_store)?;
            let own = matches!(
                &minter,
                Some(crate::case::Minter::Operator(first)) if first.actor() == by.actor()
            );
            self.settle_answered(id).await?;
            if own {
                return Ok(delivery);
            }
            return Err(RuntimeError::TaskClaim(
                crate::core::ClaimError::AlreadyAnswered { task: id },
            ));
        }
        tasks
            .set_state(id, TaskState::Completed)
            .await
            .map_err(RuntimeError::from_store)?;
        Ok(delivery)
    }
}

/// The id of the one event that answers a task: derived from the task, so a
/// second answer is a duplicate rather than a second decision.
fn decision_event_id(id: crate::core::TaskId) -> String {
    format!("task-decision:{}", id.to_hex())
}