agentplane/runtime/

effects.rs

//! Built-in effects.
//!
//! These are the runtime's own drivers — the small set of non-deterministic
//! things it needs before any tool or model exists. Each is journaled like any
//! other effect, which is why a replayed run sees the instant the original run
//! saw rather than the instant now.

use async_trait::async_trait;
use serde_json::json;

use crate::core::{Effect, EffectDescriptor, EffectError, Recovery, Sensitivity, Timestamp};

/// Reads the wall clock.
///
/// Non-mutating and freely retryable: re-reading a clock after a crash is
/// harmless, which is why this is one of the few effects that does not need an
/// operator when its outcome is unknown.
#[derive(Debug, Clone, Copy)]
pub struct Clock;

#[async_trait]
impl Effect for Clock {
    /// Trusted: does not cross a trust boundary.
    ///
    /// The journaled clock. The instant comes from the runtime and is written
    /// to the journal, so it is the runtime's own data rather than the world's —
    /// and a timestamp that arrived untrusted could not be used for the deadline
    /// arithmetic it exists for.
    fn trust(&self) -> crate::core::Trust {
        crate::core::Trust::Trusted
    }

    type Output = Timestamp;

    fn descriptor(&self) -> EffectDescriptor {
        EffectDescriptor::nullary("clock.now")
    }

    fn mutates(&self) -> bool {
        false
    }

    fn recovery(&self) -> Recovery {
        Recovery::Retry
    }

    fn max_sensitivity(&self) -> Sensitivity {
        Sensitivity::Public
    }

    // The single legitimate ambient clock read in the crate: its result is
    // written to the journal as `EffectDone`, so every later replay reads that
    // record instead of calling this again.
    #[allow(clippy::disallowed_methods)]
    async fn perform(&self) -> Result<Timestamp, EffectError> {
        Ok(Timestamp::now_utc())
    }
}

/// A test fixture standing in for an externally visible action.
///
/// Counts its own invocations so tests can assert the property the whole design
/// exists for: **replay does not perform it again**.
///
/// `#[doc(hidden)]` because it is public for the test suite's benefit and is not
/// part of the API. It would be a poor model for a real effect: it declares that
/// it mutates and still asks to be **retried** after an unknown outcome, which
/// is the combination every real mutating effect must not have — the trait's own
/// default gives a mutating effect `RequiresOperator` for exactly that reason.
/// The override is here so the engine's retry paths have something retryable to
/// exercise, and nowhere else.
///
/// It lives here rather than in `testkit` because `just test-default` runs the
/// durability and recovery suites under the default feature set, and moving it
/// behind a feature would quietly stop those from running there.
#[doc(hidden)]
#[derive(Debug, Clone)]
pub struct Recorded {
    pub name: String,
    pub payload: serde_json::Value,
    pub calls: std::sync::Arc<std::sync::atomic::AtomicUsize>,
    pub mutates: bool,
    pub max_sensitivity: Sensitivity,
}

impl Recorded {
    /// The payload starts as `null`, and it is also what
    /// [`sink_arguments`](Effect::sink_arguments) binds.
    ///
    /// So a `Recorded` dispatched through `StepCtx::sink` with anything but
    /// `Tainted::…(json!(null))` is refused by the argument binding, which is
    /// working as intended — the gate's rule is that what policy checked and
    /// what the sink sends are the same value — but it is a refusal about the
    /// fixture rather than about the run. Set both sides with
    /// [`payload`](Self::payload), or dispatch through `cx.effect`.
    pub fn new(name: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            payload: json!(null),
            calls: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)),
            mutates: true,
            max_sensitivity: Sensitivity::Secret,
        }
    }

    #[must_use]
    pub fn counter(mut self, c: std::sync::Arc<std::sync::atomic::AtomicUsize>) -> Self {
        self.calls = c;
        self
    }

    #[must_use]
    pub fn payload(mut self, v: serde_json::Value) -> Self {
        self.payload = v;
        self
    }

    #[must_use]
    pub fn ceiling(mut self, s: Sensitivity) -> Self {
        self.max_sensitivity = s;
        self
    }

    #[must_use]
    pub fn read_only(mut self) -> Self {
        self.mutates = false;
        self
    }
}

#[async_trait]
impl Effect for Recorded {
    /// Trusted: does not cross a trust boundary.
    ///
    /// A value the runtime recorded for itself. Its trust is whatever the
    /// original source's was, carried separately; this wrapper adds none.
    fn trust(&self) -> crate::core::Trust {
        crate::core::Trust::Trusted
    }

    type Output = serde_json::Value;

    fn descriptor(&self) -> EffectDescriptor {
        EffectDescriptor::new(format!("test.{}", self.name), self.payload.clone())
    }

    fn mutates(&self) -> bool {
        self.mutates
    }

    fn max_sensitivity(&self) -> Sensitivity {
        self.max_sensitivity
    }

    fn sink_arguments(&self) -> Option<&serde_json::Value> {
        Some(&self.payload)
    }

    fn recovery(&self) -> Recovery {
        Recovery::Retry
    }

    async fn perform(&self) -> Result<serde_json::Value, EffectError> {
        let n = self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
        Ok(json!({ "call": n, "payload": self.payload }))
    }
}

/// Resolves a deadline description to an instant via the configured calendar.
///
/// This is an effect, not a plain function call, for one reason: the answer
/// depends on a calendar whose rules change over time. Journaling the resolved
/// instant means a corrected holiday table or a new regulatory notice cannot
/// retroactively move an obligation that was already registered — the instant
/// is a recorded fact, and the calendar digest beside it says which ruleset
/// produced it.
#[derive(Debug, Clone)]
pub struct ResolveDeadline {
    pub(crate) calendar: std::sync::Arc<dyn crate::core::Calendar>,
    pub(crate) name: String,
    pub(crate) from: Timestamp,
    pub(crate) spec: crate::core::DeadlineSpec,
}

/// What a calendar produced, and which ruleset produced it.
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct ResolvedDeadline {
    #[serde(with = "time::serde::rfc3339")]
    pub at: Timestamp,
    pub calendar_digest: crate::core::Digest,
}

#[async_trait]
impl Effect for ResolveDeadline {
    /// Trusted: does not cross a trust boundary.
    ///
    /// A `Calendar` is configured by the operator, not supplied by the
    /// world. Its answer is an instant the deployment chose the rules for.
    fn trust(&self) -> crate::core::Trust {
        crate::core::Trust::Trusted
    }

    type Output = ResolvedDeadline;

    fn descriptor(&self) -> EffectDescriptor {
        EffectDescriptor::new(
            "deadline.resolve",
            json!({
                "name": self.name,
                "spec": self.spec,
                // `from` is part of the identity: the same rule anchored at a
                // different instant is a different obligation.
                "from": self.from.unix_timestamp(),
            }),
        )
    }

    fn mutates(&self) -> bool {
        false
    }

    fn recovery(&self) -> Recovery {
        Recovery::Retry
    }

    async fn perform(&self) -> Result<ResolvedDeadline, EffectError> {
        let at = self
            .calendar
            .resolve(self.from, &self.spec)
            .map_err(|e| EffectError::Rejected(e.to_string()))?;

        // Truncate to whole seconds, deliberately and once.
        //
        // A deadline is a wall-clock instant; sub-second precision in a
        // regulatory window is noise. More importantly, the instant is recorded
        // in two places — the journal and the case store — and if they disagree
        // by a fraction of a second then "which one is the obligation?" has two
        // answers. Truncating here means they cannot drift.
        let at = at
            .replace_nanosecond(0)
            .map_err(|e| EffectError::Rejected(format!("deadline instant out of range: {e}")))?;

        Ok(ResolvedDeadline {
            at,
            calendar_digest: self.calendar.digest(),
        })
    }
}

/// Reading a case's state.
///
/// **A journaled effect, and it has to be.** Case state is mutable storage
/// shared by every run correlated to the case; reading it is exactly as
/// non-deterministic as reading a clock. Read directly, a strict replay would
/// see the *current* state and the run's own logic reach a different answer
/// from the same journal — the divergence the whole design exists to make
/// impossible.
#[derive(Debug, Clone)]
pub struct ReadCaseState {
    pub(crate) cases: std::sync::Arc<dyn crate::case::CaseStore>,
    pub(crate) case: crate::core::CaseId,
}

/// A case's state, and the revision it was read at.
///
/// The version travels with the value because a write has to name it. Handing
/// back the value alone is what makes a lost update easy to write.
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct CaseSnapshot {
    pub state: serde_json::Value,
    pub version: crate::core::CaseVersion,
}

#[async_trait]
impl Effect for ReadCaseState {
    type Output = CaseSnapshot;

    /// Trusted: the plane's own storage, not the world's.
    ///
    /// The *contents* may well have come from an untrusted source, and the
    /// caller receives them labeled accordingly — but the store itself is not a
    /// trust boundary the way a tool or a peer is.
    fn trust(&self) -> crate::core::Trust {
        crate::core::Trust::Trusted
    }

    fn descriptor(&self) -> EffectDescriptor {
        EffectDescriptor::new("case.read_state", json!({ "case": self.case.to_string() }))
    }

    fn mutates(&self) -> bool {
        false
    }

    fn recovery(&self) -> Recovery {
        Recovery::Retry
    }

    async fn perform(&self) -> Result<CaseSnapshot, EffectError> {
        let case = self
            .cases
            .case(self.case)
            .await
            .map_err(|e| EffectError::Other(e.to_string()))?
            .ok_or_else(|| EffectError::Rejected(format!("no case {}", self.case)))?;
        Ok(CaseSnapshot {
            state: case.state,
            version: case.version,
        })
    }
}

/// Writing a case's state, if it has not moved.
///
/// Journaled for two reasons that are usually the same reason: replay must not
/// perform it again, and the *outcome* — the new version — is something the run
/// goes on to use. A replay that re-derived the version by writing again would
/// be writing again.
#[derive(Debug, Clone)]
pub struct WriteCaseState {
    pub(crate) cases: std::sync::Arc<dyn crate::case::CaseStore>,
    pub(crate) case: crate::core::CaseId,
    pub(crate) expected: crate::core::CaseVersion,
    pub(crate) state: serde_json::Value,
}

#[async_trait]
impl Effect for WriteCaseState {
    type Output = crate::core::CaseVersion;

    fn trust(&self) -> crate::core::Trust {
        crate::core::Trust::Trusted
    }

    fn descriptor(&self) -> EffectDescriptor {
        EffectDescriptor::new(
            "case.write_state",
            json!({
                "case": self.case.to_string(),
                // Both in the key: writing different bytes is a different
                // effect, and so is writing the same bytes against a different
                // revision — the second one is a different claim about what the
                // case said when the decision was made.
                "expected": self.expected.0,
                "state": self.state,
            }),
        )
    }

    /// It changes state other runs can observe. That is what mutating means.
    fn mutates(&self) -> bool {
        true
    }

    /// Not retryable, and the version is why.
    ///
    /// A retry re-sends the same `expected`. If the first attempt landed, the
    /// case has already moved past it and the retry fails as a conflict — a
    /// confusing report of a write that in fact succeeded. Whether it landed is
    /// exactly what the effect protocol is for.
    fn recovery(&self) -> Recovery {
        Recovery::Reconcile
    }

    async fn perform(&self) -> Result<crate::core::CaseVersion, EffectError> {
        self.cases
            .put_state(self.case, self.expected, self.state.clone())
            .await
            .map_err(|e| match e {
                // A conflict is a decision the caller has to make again, not a
                // transport failure. `Rejected` so it does not read as in-doubt.
                crate::core::StoreError::CaseConflict { .. } => {
                    EffectError::Rejected(e.to_string())
                }
                other => EffectError::Other(other.to_string()),
            })
    }
}

/// Recalling memories, as an effect.
///
/// Journaled because memory is mutable state outside the chain: a lookup done
/// inside the deterministic zone would make a replayed run retrieve whatever the
/// store holds *now*, reach different conclusions, and produce a history that
/// disagrees with itself.
///
/// The output is the **selection** — ids, versions and content digests — and not
/// the content. Two reasons, and both matter. Personal data must not enter a
/// hash chain that cannot be redacted; and pinning versions means a replay
/// re-materialises exactly what was read rather than re-running a ranking that
/// drifts as the corpus grows.
#[derive(Debug, Clone)]
pub struct RecallMemory {
    pub(crate) memories: std::sync::Arc<dyn crate::memory::MemoryStore>,
    pub(crate) query: crate::memory::Recall,
}

/// Turning text into a vector, as an observation rather than a computation.
///
/// An embedding service does not promise the same floats for the same text —
/// batching, hardware and an unannounced model revision each move the last bits.
/// The vector then enters [`SemanticRecall`]'s key, so a run that computed one
/// outside the protocol would derive a different key on replay and quarantine
/// itself, with nothing on the record saying why.
///
/// Journaling it makes the vector history rather than a recomputation, puts the
/// revision that produced it beside it, and meters the call like any other.
#[derive(Debug, Clone)]
pub struct Embed {
    pub(crate) embedder: std::sync::Arc<dyn crate::memory::Embedder>,
    pub(crate) text: String,
    pub(crate) arguments: serde_json::Value,
    /// The highest sensitivity this embedding service may be shown.
    ///
    /// Declared per call rather than fixed, because embedding is an egress:
    /// the text goes to a provider, and how confidential a text may be before
    /// it stops being allowed to leave is a deployment's decision, not this
    /// type's. A fixed [`Public`](Sensitivity::Public) would be that decision
    /// made once, in the strictest possible place, and would refuse every
    /// query worth embedding — a model's paraphrase, a recalled memory, a
    /// customer's own question — since anything that crossed a trust boundary
    /// is already [`Internal`](Sensitivity::Internal).
    pub(crate) max_sensitivity: Sensitivity,
}

#[async_trait]
impl Effect for Embed {
    type Output = crate::memory::Embedding;

    fn descriptor(&self) -> EffectDescriptor {
        EffectDescriptor::new(
            "memory.embed",
            json!({
                "revision": self.embedder.revision(),
                "text": self.text,
            }),
        )
    }

    fn mutates(&self) -> bool {
        false
    }

    fn recovery(&self) -> Recovery {
        Recovery::Retry
    }

    fn max_sensitivity(&self) -> Sensitivity {
        self.max_sensitivity
    }

    fn sink_arguments(&self) -> Option<&serde_json::Value> {
        Some(&self.arguments)
    }

    async fn perform(&self) -> Result<Self::Output, EffectError> {
        // The revision is read from the driver rather than taken from a
        // caller, so the space a vector lives in is a fact about the wiring
        // and never a claim. `IndexIdentity` is where the cost of a claim is
        // written down.
        let crate::memory::Embedded { vector, mut usage } =
            self.embedder
                .embed(&self.text)
                .await
                .map_err(|error| EffectError::Other(error.to_string()))?;
        usage.minor_units = self
            .embedder
            .pricing()
            .map_or(0, |pricing| pricing.price(&usage));
        Ok(crate::memory::Embedding {
            vector,
            revision: self.embedder.revision(),
            usage,
        })
    }

    fn spend(&self, output: &Self::Output) -> crate::core::Spend {
        output.usage.spend()
    }
}

/// Semantic ranking is an effect because an index and embedding implementation
/// can change independently of durable memory truth.
///
/// # The journaled selection is the screened one
///
/// The index is derived, so it keeps naming versions after they stop being
/// current truth: a superseded version after a correction, an expired one
/// after its window, an erased one after a retention sweep. Each is routine —
/// a derived index is stale by construction between rebuilds — and none may
/// decide what a model is shown or whether the query answers at all. So live
/// dispatch screens every hit against the authoritative store at the query's
/// journaled cutoff and records only the survivors; a hit whose version is no
/// longer the one a fresh recall would see simply leaves the selection.
///
/// What screening deliberately does **not** absorb: a *present* version whose
/// content or security metadata moved under its digest, and a hit outside the
/// query's scope. Those are integrity findings about the store and the
/// retriever respectively, they stay refusals, and they are checked where
/// replay also runs — in [`StepCtx::semantic_recall`].
///
/// [`StepCtx::semantic_recall`]: crate::runtime::StepCtx::semantic_recall
#[derive(Debug, Clone)]
pub struct SemanticRecall {
    pub(crate) retriever: std::sync::Arc<dyn crate::memory::SemanticRetriever>,
    pub(crate) memories: std::sync::Arc<dyn crate::memory::MemoryStore>,
    pub(crate) query: crate::memory::SemanticQuery,
    pub(crate) arguments: serde_json::Value,
}

#[async_trait]
impl Effect for SemanticRecall {
    type Output = Vec<crate::memory::SemanticHit>;

    fn descriptor(&self) -> EffectDescriptor {
        EffectDescriptor::new(
            "memory.semantic-recall",
            json!({
                "retriever": self.retriever.profile(),
                "query": self.query,
            }),
        )
    }

    fn mutates(&self) -> bool {
        false
    }

    fn recovery(&self) -> Recovery {
        Recovery::Retry
    }

    fn max_sensitivity(&self) -> crate::core::Sensitivity {
        self.query.max_sensitivity
    }

    fn sink_arguments(&self) -> Option<&serde_json::Value> {
        Some(&self.arguments)
    }

    async fn perform(&self) -> Result<Self::Output, EffectError> {
        let hits = self
            .retriever
            .search(&self.query)
            .await
            .map_err(|error| EffectError::Other(error.to_string()))?;
        // Refused before any store read: every hit past here costs one, and a
        // seam answering more than it was asked is misbehaving whatever the
        // survivors would have been.
        if hits.len() > self.query.limit {
            return Err(EffectError::Other(format!(
                "semantic retriever returned {} hits for a limit of {}",
                hits.len(),
                self.query.limit
            )));
        }
        let mut survivors = Vec::with_capacity(hits.len());
        for hit in hits {
            // Refused before the selection is recorded, because it could not
            // be recorded: a non-finite f32 journals as `null` and the record
            // would fail to read back on replay.
            if !hit.score.is_finite() {
                return Err(EffectError::Other(
                    "semantic retriever returned a non-finite score".to_owned(),
                ));
            }
            let current = self
                .memories
                .current(&hit.selected.id, Some(self.query.as_of))
                .await
                .map_err(|error| EffectError::Other(error.to_string()))?;
            // Kept only while the named version is the one a fresh recall at
            // the cutoff would see. A different current version, an expired
            // window, and an erased id all mean the index is behind durable
            // truth, which is its normal state between rebuilds — not a
            // finding, and not this query's failure.
            if current.is_some_and(|item| item.version == hit.selected.version) {
                survivors.push(hit);
            }
        }
        Ok(survivors)
    }
}

#[async_trait]
impl Effect for RecallMemory {
    type Output = Vec<crate::memory::Selected>;

    /// Trusted: this plane's own store, not the world's.
    ///
    /// The *contents* are another matter entirely, and the caller receives them
    /// labelled from each item's declared provenance — see
    /// [`StepCtx::recall`](crate::runtime::StepCtx::recall).
    fn trust(&self) -> crate::core::Trust {
        crate::core::Trust::Trusted
    }

    fn descriptor(&self) -> EffectDescriptor {
        EffectDescriptor::new(
            "memory.recall",
            serde_json::to_value(&self.query).unwrap_or(json!(null)),
        )
    }

    fn mutates(&self) -> bool {
        false
    }

    fn recovery(&self) -> Recovery {
        Recovery::Retry
    }

    async fn perform(&self) -> Result<Self::Output, EffectError> {
        let found = self
            .memories
            .recall(&self.query)
            .await
            .map_err(|e| EffectError::Other(e.to_string()))?;
        Ok(found
            .iter()
            .map(|item| crate::memory::Selected {
                id: item.id.clone(),
                version: item.version,
                digest: item.selection_digest(),
            })
            .collect())
    }
}

#[derive(Debug, Clone)]
pub struct TouchMemory {
    pub(crate) memories: std::sync::Arc<dyn crate::memory::MemoryStore>,
    pub(crate) ids: Vec<String>,
    pub(crate) at: crate::core::Timestamp,
}

#[async_trait]
impl Effect for TouchMemory {
    type Output = ();

    fn descriptor(&self) -> EffectDescriptor {
        EffectDescriptor::new(
            "memory.touch",
            json!({ "ids": self.ids, "at": crate::core::format_timestamp(self.at) }),
        )
    }

    fn recovery(&self) -> Recovery {
        // Stores take max(existing, at + window), so repeating is idempotent.
        Recovery::Retry
    }

    async fn perform(&self) -> Result<Self::Output, EffectError> {
        self.memories
            .touch(&self.ids, self.at)
            .await
            .map_err(|error| EffectError::Other(error.to_string()))
    }
}

/// Writing a memory, as an effect.
///
/// A write is external mutable state, so a replay that performed it again would
/// append a second version of a memory the run wrote once — and the version
/// number the run went on to use would be wrong.
#[derive(Debug, Clone)]
pub struct RememberMemory {
    pub(crate) memories: std::sync::Arc<dyn crate::memory::MemoryStore>,
    pub(crate) item: crate::memory::MemoryItem,
}

#[async_trait]
impl Effect for RememberMemory {
    type Output = u64;

    fn descriptor(&self) -> EffectDescriptor {
        EffectDescriptor::new(
            "memory.remember",
            json!({
                "id": self.item.id,
                "subject": self.item.subject,
                "purpose": self.item.purpose,
                "provenance": self.item.provenance,
                "sensitivity": self.item.sensitivity,
                "trust": self.item.trust,
                "created_at": crate::core::format_timestamp(self.item.created_at),
                "expires_at": self.item.expires_at.map(crate::core::format_timestamp),
                // In the key beside `expires_at`, because both are lifecycle
                // the write *creates*: `selection_digest` already commits to
                // this field, and an effect identity that omitted it let a
                // build change silently re-derive a different retention for
                // the same journaled write instead of diverging.
                "access_retention_seconds": self.item.access_retention_seconds,
                "derived_from": self.item.derived_from,
                // The content's digest, not the content: an effect key is
                // recorded verbatim, and a memory's content belongs in a store
                // that can be erased.
                "content": self.item.digest().to_hex(),
            }),
        )
    }

    async fn perform(&self) -> Result<u64, EffectError> {
        self.memories
            .remember(&self.item)
            .await
            .map_err(|e| EffectError::Other(e.to_string()))
    }
}

/// Erase expired memories at one journaled cutoff.
#[derive(Debug, Clone)]
pub struct SweepExpiredMemory {
    pub(crate) memories: std::sync::Arc<dyn crate::memory::MemoryStore>,
    pub(crate) at: crate::core::Timestamp,
}

#[async_trait]
impl Effect for SweepExpiredMemory {
    type Output = usize;

    fn descriptor(&self) -> EffectDescriptor {
        EffectDescriptor::new(
            "memory.sweep-expired",
            json!({ "at": crate::core::format_timestamp(self.at) }),
        )
    }

    async fn perform(&self) -> Result<Self::Output, EffectError> {
        self.memories
            .sweep_expired(self.at)
            .await
            .map(|swept| swept.len())
            .map_err(|error| EffectError::Other(error.to_string()))
    }
}

/// Changing a case's status, as an effect.
///
/// # Why this is not a plain store call
///
/// A case's status is shared mutable state that outlives the run: several runs
/// and an operator all write it over months. A write performed outside the
/// journal has two holes, and the second one is worse.
///
/// It is performed **again on every replay**. Replaying last quarter's history
/// to answer a question would close a case that has since been reopened — a
/// replay is supposed to read history, not rewrite the world it happened in.
///
/// And it leaves **no record**. *Who closed this case, and when* is exactly the
/// question the journal exists to answer, and a status that changed without one
/// is a change nobody can attribute.
#[derive(Debug, Clone)]
pub struct SetCaseStatus {
    pub(crate) cases: std::sync::Arc<dyn crate::case::CaseStore>,
    pub(crate) case: crate::core::CaseId,
    pub(crate) status: crate::core::CaseStatus,
}

#[async_trait]
impl Effect for SetCaseStatus {
    type Output = ();

    /// The runtime's own write against a store it owns. Nothing crosses a trust
    /// boundary, and the value is `()`.
    fn trust(&self) -> crate::core::Trust {
        crate::core::Trust::Trusted
    }

    fn descriptor(&self) -> EffectDescriptor {
        EffectDescriptor::new(
            "case.set_status",
            json!({ "case": self.case.to_string(), "status": self.status }),
        )
    }

    /// It changes state other runs observe.
    fn mutates(&self) -> bool {
        true
    }

    /// Setting a status is idempotent in the value: the second write of
    /// `Closed` leaves the case exactly as the first did. So a repeat is safe,
    /// unlike a versioned state write where the retry would report a conflict
    /// for a write that in fact succeeded.
    fn recovery(&self) -> Recovery {
        Recovery::Retry
    }

    async fn perform(&self) -> Result<(), EffectError> {
        self.cases
            .set_status(self.case, self.status)
            .await
            .map_err(|e| EffectError::Other(e.to_string()))
    }
}

/// Moving a deadline out of `Pending`, as an effect.
///
/// Same reasoning as [`SetCaseStatus`], plus one of its own: writing the
/// transition to the store and *then* journaling it leaves a crash in between
/// holding an obligation marked met with nothing saying who met it. Announce
/// before act.
#[derive(Debug, Clone)]
pub struct TransitionDeadline {
    pub(crate) cases: std::sync::Arc<dyn crate::case::CaseStore>,
    pub(crate) case: crate::core::CaseId,
    pub(crate) name: String,
    pub(crate) to: crate::core::DeadlineState,
}

#[async_trait]
impl Effect for TransitionDeadline {
    /// The state it was in before, so the record can say what changed rather
    /// than only what it became.
    type Output = crate::core::DeadlineState;

    fn trust(&self) -> crate::core::Trust {
        crate::core::Trust::Trusted
    }

    fn descriptor(&self) -> EffectDescriptor {
        EffectDescriptor::new(
            "case.transition_deadline",
            json!({
                "case": self.case.to_string(),
                "name": self.name,
                "to": self.to,
            }),
        )
    }

    fn mutates(&self) -> bool {
        true
    }

    /// Idempotent in the value, as above.
    fn recovery(&self) -> Recovery {
        Recovery::Retry
    }

    async fn perform(&self) -> Result<crate::core::DeadlineState, EffectError> {
        let before = self
            .cases
            .deadlines(self.case)
            .await
            .map_err(|e| EffectError::Other(e.to_string()))?
            .into_iter()
            .find(|d| d.name == self.name)
            .map_or(crate::core::DeadlineState::Pending, |d| d.state);

        self.cases
            .set_deadline_state(self.case, &self.name, self.to)
            .await
            .map_err(|e| EffectError::Other(e.to_string()))?;
        Ok(before)
    }
}

/// Opening a human task **without waiting for it**.
///
/// # Why this is not `StepCtx::task` with the wait removed
///
/// [`StepCtx::task`] is a *decision*: it opens the row, subscribes to the
/// answer, and suspends until somebody gives one. Its whole shape is the wait.
/// This is a *notification*: the run has already reached its answer, and a
/// person is being told rather than asked. Nothing resumes on the decision,
/// because nothing is blocked on it.
///
/// Both derive the task id from the effect key rather than minting one, and for
/// the same reason: a resumed run must address the row it already opened instead
/// of opening a second one for the same finding. That is also what makes this
/// safe to [`Retry`](Recovery::Retry) — `TaskStore::open` is idempotent on the
/// id, so an interrupted attempt cannot leave two rows in a worklist.
///
/// [`StepCtx::task`]: crate::runtime::StepCtx::task
#[derive(Debug, Clone)]
pub struct OpenTask {
    pub(crate) tasks: std::sync::Arc<dyn crate::case::TaskStore>,
    pub(crate) run: crate::core::RunId,
    pub(crate) case: crate::core::CaseId,
    pub(crate) spec: crate::core::TaskSpec,
    /// When the row was created, from the run's journaled clock.
    pub(crate) at: Timestamp,
    /// When the obligation bounding it falls due.
    pub(crate) due_at: Timestamp,
    /// Filled by [`Effect::attach`]; the runtime owns the key, not the caller.
    pub(crate) key: Option<crate::core::EffectKey>,
}

#[async_trait]
impl Effect for OpenTask {
    type Output = crate::core::TaskId;

    /// Trusted: the id is derived by the runtime from its own effect key.
    ///
    /// The *justification* carried into the row may well be untrusted, and it
    /// arrives at this effect through the sink gate like any other outbound
    /// value — see [`StepCtx::open_task`](crate::runtime::StepCtx::open_task).
    fn trust(&self) -> crate::core::Trust {
        crate::core::Trust::Trusted
    }

    fn descriptor(&self) -> EffectDescriptor {
        EffectDescriptor::new(
            "task.open",
            json!({
                "case": self.case.to_string(),
                "kind": self.spec.kind,
                "deadline": self.spec.deadline,
                "roles": self.spec.candidate_roles,
                "escalate_to": self.spec.on_expiry.escalate_roles(),
                "priority": self.spec.priority.as_str(),
                // In the key: two findings that differ only in what they say
                // are two rows, and collapsing them would silently drop one.
                "justification": self.spec.justification,
            }),
        )
    }

    /// It creates durable state a person acts on. That is what mutating means,
    /// and saying otherwise would exempt the justification from the taint gate.
    fn mutates(&self) -> bool {
        true
    }

    /// Idempotent on the derived id, so a lost acknowledgement costs a repeated
    /// write rather than a second row.
    fn recovery(&self) -> Recovery {
        Recovery::Retry
    }

    /// # This is not a sink, and that is the one decision here worth arguing
    ///
    /// It binds no outbound arguments, so the whole-value taint gate and the
    /// egress ceiling do not run — the same arrangement the blocking
    /// [`StepCtx::task`](crate::runtime::StepCtx::task) has always had, and for
    /// the same reason.
    ///
    /// A worklist row's *entire purpose* is to put untrusted content in front of
    /// a person. The gate refuses an untrusted value at a mutating sink with no
    /// protected fields, which is right for a payment and exactly wrong here: it
    /// would mean a task could only ever carry content nobody needs to review,
    /// and every real finding — a model's answer, by construction untrusted —
    /// would be refused. The control a reviewer is for *is* the review.
    ///
    /// A ceiling on what a worklist may hold is a coherent thing to want, and if
    /// it is ever added it belongs on both paths at once. One of the two
    /// enforcing it would make *notifying* a compliance desk stricter than
    /// *asking* one, which is a distinction nobody could defend from the
    /// outside.
    fn max_sensitivity(&self) -> Sensitivity {
        Sensitivity::Secret
    }

    fn attach(&mut self, provenance: &crate::core::Provenance) {
        // `dispatch` when the runtime supplied one, so two attempts at one
        // notification address one row. `effect` otherwise, which is the
        // single-attempt case where the two are the same call.
        self.key = Some(provenance.dispatch.unwrap_or(provenance.effect));
    }

    async fn perform(&self) -> Result<Self::Output, EffectError> {
        let key = self.key.ok_or_else(|| {
            // Unreachable through `StepCtx`, which always attaches. Loud rather
            // than minting an id: a task whose id is not derived from the run
            // reopens on every resume, and a worklist that grows a row per
            // resume is worse than a run that fails.
            EffectError::Other(
                "a task was opened without its effect key — an id that is not derived \
                 from the run would open a second row on every resume"
                    .to_owned(),
            )
        })?;
        let id = crate::core::TaskId::derive(self.run, key);
        self.tasks
            .open(&crate::core::Task {
                id,
                run: self.run,
                case: Some(self.case),
                kind: self.spec.kind.clone(),
                justification: self.spec.justification.clone(),
                candidate_roles: self.spec.candidate_roles.clone(),
                escalate_to: self.spec.on_expiry.escalate_roles().to_vec(),
                excluded_actors: self.spec.excluded_actors.clone(),
                assignee: None,
                priority: self.spec.priority,
                state: crate::core::TaskState::Open,
                on_expiry: self.spec.on_expiry.policy(),
                created_at: self.at,
                due_at: Some(self.due_at),
                withheld: None,
            })
            .await
            .map_err(|e| EffectError::Other(e.to_string()))?;
        Ok(id)
    }
}

/// The standing approvals a holder holds that still stand at `at`.
///
/// A read of mutable state outside the journal, so an effect: a replay matches
/// against what the run saw, not against what the store holds now. Revoked and
/// expired ones are left out here, at the run's journaled instant; the draw
/// that follows a match checks both again, atomically, at the store.
#[derive(Debug, Clone)]
pub struct ListStandingApprovals {
    pub(crate) authorities: std::sync::Arc<dyn crate::authority::AuthorityStore>,
    pub(crate) holder: crate::authority::Holder,
    pub(crate) at: Timestamp,
}

#[async_trait]
impl Effect for ListStandingApprovals {
    type Output = Vec<crate::authority::StandingAuthority>;

    fn descriptor(&self) -> EffectDescriptor {
        EffectDescriptor::new(
            "authority.approvals",
            json!({
                "holder": self.holder,
                "at": crate::core::format_timestamp(self.at),
            }),
        )
    }

    fn recovery(&self) -> Recovery {
        Recovery::Retry
    }

    async fn perform(&self) -> Result<Self::Output, EffectError> {
        let now = self.at.unix_timestamp();
        let held =
            self.authorities
                .held_by(&self.holder)
                .await
                .map_err(|e| EffectError::Unavailable {
                    driver: "authority-store".to_owned(),
                    detail: e.to_string(),
                })?;
        Ok(held
            .into_iter()
            .filter(|state| {
                !state.authority.approves.is_empty()
                    && state.revoked.is_none()
                    && state
                        .authority
                        .expires_at
                        .is_some_and(|expires| now < expires.unix_timestamp())
            })
            .map(|state| state.authority)
            .collect())
    }
}

/// Drawing on a standing authority, as an effect.
///
/// # Why this cannot be a plain call
///
/// The remaining balance is mutable state outside the journal. A skill that read
/// it inside the deterministic zone would make replay depend on what the store
/// happens to hold *now*: a run replayed after a later draw would see a
/// different balance, take a different branch, and produce a history that
/// disagrees with itself. Journaling the draw makes the receipt part of the
/// record, so replay reads what the run actually got.
///
/// # The key it deduplicates on is `dispatch`, not `effect`
///
/// [`Provenance::effect`] hashes the attempt number, so a retry carries a
/// *different* key — which is right for the journal and exactly wrong here. Two
/// attempts at one draw must consume the authority once, so the store is keyed
/// on [`Provenance::dispatch`], the identifier the crate defines as stable
/// across retries of the same call. Getting this backwards spends a customer's
/// authorization twice for one purchase, and only under retry, which is the
/// hardest condition to notice in testing.
///
/// [`Provenance::effect`]: crate::core::Provenance::effect
/// [`Provenance::dispatch`]: crate::core::Provenance::dispatch
#[derive(Debug, Clone)]
pub struct DrawOnAuthority {
    pub(crate) authorities: std::sync::Arc<dyn crate::authority::AuthorityStore>,
    pub(crate) id: crate::authority::AuthorityId,
    /// Whom the drawing run acts for — see [`crate::authority::holder_of`].
    /// `None` for a run that holds nothing.
    pub(crate) holder: Option<crate::authority::Holder>,
    pub(crate) amount: crate::core::Spend,
    pub(crate) at: Timestamp,
    /// Filled by [`Effect::attach`]; the runtime owns the key, not the caller.
    pub(crate) key: Option<crate::core::EffectKey>,
}

#[async_trait]
impl Effect for DrawOnAuthority {
    type Output = crate::authority::Drawn;

    fn descriptor(&self) -> EffectDescriptor {
        EffectDescriptor::new(
            "authority.draw",
            json!({
                "authority": self.id,
                "holder": self.holder,
                "amount": self.amount,
                "at": crate::core::format_timestamp(self.at),
            }),
        )
    }

    fn attach(&mut self, provenance: &crate::core::Provenance) {
        // `dispatch` when the runtime supplied one, `effect` otherwise. The
        // fallback is the single-attempt case, where the two are the same call.
        self.key = Some(provenance.dispatch.unwrap_or(provenance.effect));
    }

    /// It changes durable state that authorizes spending, and says so to every
    /// rule reading `context.mutates`. The id it draws on is checked before
    /// this effect exists: [`StepCtx::draw`](crate::runtime::StepCtx::draw)
    /// refuses one untrusted data chose.
    fn mutates(&self) -> bool {
        true
    }

    /// Safe to repeat, because the store deduplicates on the dispatch key and
    /// returns the original receipt. Without that idempotence this would have to
    /// be `Reconcile`, and every interrupted draw would wake an operator.
    fn recovery(&self) -> Recovery {
        Recovery::Retry
    }

    async fn perform(&self) -> Result<Self::Output, EffectError> {
        let key = self.key.ok_or_else(|| {
            // Unreachable through `StepCtx`, which always attaches. Loud rather
            // than defaulting to some other key: a draw deduplicated on the
            // wrong identifier is a double-spend, and guessing here would hide
            // it behind a plausible success.
            EffectError::Other(
                "a draw reached the store without its dispatch key — deduplicating on \
                 anything else would let a retry spend the authority twice"
                    .to_owned(),
            )
        })?;
        // The holder is part of the terms, which are written once and never
        // updated — so reading it before the draw leaves no window a
        // concurrent writer could change it through.
        let held = self
            .authorities
            .state(&self.id)
            .await
            .map_err(|e| crate::authority::AuthorityError::Unavailable(e.to_string()))
            .and_then(|state| {
                let state = state
                    .ok_or_else(|| crate::authority::AuthorityError::Unknown(self.id.clone()))?;
                if self.holder.as_ref() == Some(&state.authority.holder) {
                    Ok(())
                } else {
                    Err(crate::authority::AuthorityError::NotHolder {
                        authority: self.id.clone(),
                        holder: state.authority.holder,
                        drawer: self
                            .holder
                            .as_ref()
                            .map_or_else(|| "nobody".to_owned(), ToString::to_string),
                    })
                }
            });
        let drawn = match held {
            Ok(()) => {
                self.authorities
                    .draw(&self.id, key, self.amount, self.at)
                    .await
            }
            Err(refused) => Err(refused),
        };
        drawn.map_err(|error| match &error {
            // The store could not answer. Retrying is safe *because of*
            // the receipt dedup — even a draw whose commit acknowledgement
            // was lost returns its original receipt on the retry — which
            // is the same reason `recovery()` is `Retry`.
            crate::authority::AuthorityError::Unavailable(_) => EffectError::Unavailable {
                driver: "authority-store".to_owned(),
                detail: error.to_string(),
            },
            // Answers, not faults: nothing was consumed, and no retry of
            // the same draw changes a revocation, an expiry, a spent
            // ceiling or an unknown id. `Other` would read as **in-doubt**,
            // so a refusal the module docs call "not retryable, ever" would
            // be retried under the full policy, reported upward as "may well
            // have been applied", and would quarantine any group it was
            // deferred in where the cheap abort is the truthful settlement.
            // The refusal's own message says which one it was.
            _ => EffectError::refused(error.to_string()),
        })
    }
}