agentplane 0.45.0

//! The error taxonomy.
//!
//! Every variant here is a *loud* failure. Principle P7 — no silent anything —
//! is enforced by the absence of fallbacks: there is no "log a warning and
//! continue" path for divergence, truncation, or a failed compensation, because
//! the dominant production failure mode is the one nothing reported.

use serde::{Deserialize, Serialize};

use crate::core::Spend;
use crate::core::{Digest, EffectKey, Sensitivity, Seq};

/// Render an error's `Debug` as its `Display`, for the types user code holds.
///
/// # Why this is not gratuitous
///
/// `fn main() -> Result<(), E>` reports a failure with **`Debug`**, not
/// `Display` — that is what `std`'s `Termination` impl does, and it is the shape
/// every getting-started program in every Rust project uses. So a derived
/// `Debug` throws away the entire error taxonomy at exactly the moment a
/// newcomer meets it. Before this, the first failure anyone hit read:
///
/// ```text
/// Error: NoProvider("demo.greet")
/// ```
///
/// while the message written for that variant — *no skill provides capability
/// 'demo.greet'* — was never shown to anybody. Every carefully-worded refusal in
/// this crate was invisible on the one path that matters most, and the
/// build-time diagnostics being excellent made the contrast worse rather than
/// better: the same person got a paragraph of guidance from `build()` and a
/// tuple from `run()`.
///
/// The same applies to `unwrap()`/`expect()` on a `Result`, which also print
/// `Debug`, and to `assert!(matches!(..), "{err:?}")` in a user's own tests.
///
/// # Why `Display` alone is complete here
///
/// Every variant of every type below either interpolates its inner error into
/// its own message (`"policy denied: {0}"`) or is `#[error(transparent)]`. There
/// is therefore no information in the source chain that `Display` does not
/// already print, and walking it as well would print the inner error twice for
/// every transparent variant.
///
/// # What is deliberately unaffected
///
/// Programmatic inspection: these are `#[non_exhaustive]` enums and `matches!`
/// on a variant is unchanged, which is how code should branch on a failure. This
/// governs only how a failure *reads*. Applied to the types a user's own code
/// holds — not to every error in the crate — because an inner error reached
/// through `{0}` or `transparent` is already rendered by its holder.
macro_rules! debug_is_display {
    ($($t:ty),+ $(,)?) => {$(
        impl ::core::fmt::Debug for $t {
            fn fmt(&self, f: &mut ::core::fmt::Formatter<'_>) -> ::core::fmt::Result {
                ::core::fmt::Display::fmt(self, f)
            }
        }
    )+};
}
pub(crate) use debug_is_display;

/// The rendered rate-limit failure, including the window the peer named.
///
/// The window belongs on the *message* because the message is what
/// `EffectFailed` stores: an operator reading the journal has to be able to tell
/// a thirty-second throttle from a two-hour one, and the two decide different
/// things about a run that failed for it.
///
/// Deliberately only on the message. A replayed failure is reconstructed from
/// the recorded string, so a skill branching on a *typed* window would take one
/// path live and another on replay — which is why the advice informs the
/// runtime's own schedule and never a caller's control flow.
fn rate_limited_message(detail: &str, retry_after: Option<std::time::Duration>) -> String {
    match retry_after {
        Some(window) => {
            format!("effect rate limited: {detail} (the peer asked for {window:?})")
        }
        None => format!("effect rate limited: {detail}"),
    }
}

/// How many capabilities a refusal lists before it summarises the rest.
///
/// A bounded list shaped exactly like a complete one is the silent-truncation
/// shape, so the message says how many it did not print rather than quietly
/// stopping — the same rule `SweepReport::saturated` follows for a capped sweep.
const LISTED_CAPABILITIES: usize = 10;

fn no_provider_message(target: &str, available: &[String]) -> String {
    if available.is_empty() {
        return format!(
            "no skill provides capability '{target}', and this plane has none at all — \
             register one with `RuntimeBuilder::skill(..)`, or an agent's with \
             `.agent(Agent::new(&manifest))`"
        );
    }
    let shown = available
        .iter()
        .take(LISTED_CAPABILITIES)
        .cloned()
        .collect::<Vec<_>>()
        .join(", ");
    let rest = available.len().saturating_sub(LISTED_CAPABILITIES);
    let and_more = if rest == 0 {
        String::new()
    } else {
        format!(", and {rest} more")
    };
    format!(
        "no skill provides capability '{target}' — this plane provides: {shown}{and_more}. \
         `run` takes a capability, not a skill name; a skill declares its own with \
         `SkillDescriptor::new(..).provides(..)`"
    )
}

/// Failures reaching the operator.
#[derive(thiserror::Error)]
#[non_exhaustive]
pub enum RuntimeError {
    #[error("policy denied: {0}")]
    PolicyDenied(#[from] PolicyError),

    /// The delegation chain presented for this run cannot act here, now.
    ///
    /// Its own variant rather than a [`PolicyDenied`](Self::PolicyDenied):
    /// no rule fired, and the two call for different responses — a denial is
    /// an answer to retry nowhere, an expired chain is an answer to retry with
    /// a fresh credential. Transparent, so the refusal keeps the chain's own
    /// words: which link, until when, for which plane.
    #[error(transparent)]
    Delegation(#[from] crate::core::DelegationError),

    /// The worklist's own protocol refused a claim or a decision.
    ///
    /// Its own variant rather than a [`PolicyDenied`](Self::PolicyDenied),
    /// because no policy fired: the refusal is the claim protocol's — four-eyes
    /// exclusion, a missing role, a task somebody else holds, an id that names
    /// nothing. Transparent, so the refusal keeps the store's own words; typed,
    /// so a surface can answer honestly — "does not exist", "not yours to
    /// decide" and "held by Bob" call for three different responses, and a
    /// class that flattens them teaches a caller to retry the permanent and
    /// abandon the transient.
    #[error(transparent)]
    TaskClaim(#[from] crate::core::ClaimError),

    /// An approval was offered for a task whose proposal this plane cannot
    /// show.
    ///
    /// Its own class, apart from a claim refusal and from
    /// [`PlanContract`](Self::PlanContract), because a caller does something
    /// different with it: decide from a plane that holds the key ring, or
    /// reject — a rejection of the unseen is safe and still records. Nothing
    /// was claimed or recorded, and the task stays open.
    #[error(
        "task {task} holds a proposal this plane cannot show — {reason} — so an approval \
         would be of arguments nobody was shown; decide it where the key ring that sealed \
         it is wired, or reject it"
    )]
    ProposalWithheld {
        task: String,
        reason: crate::core::Withheld,
    },

    #[error("plan contract violation: {0}")]
    PlanContract(String),

    /// This process serves no plane for the tenant named.
    ///
    /// Refused rather than defaulted, which is the whole point: a fallback
    /// plane would answer an unregistered tenant with somebody else's data, and
    /// it would look exactly like working software.
    #[error(
        "this process serves no plane for tenant '{0}' — refused rather than \
         defaulted, because a fallback would serve another tenant's data"
    )]
    UnknownTenant(String),

    /// An event from outside named a kind this plane mints for itself.
    ///
    /// A human task's answer travels as an event in the `agentplane.`
    /// namespace, so accepting one from outside would let whoever may post an
    /// event decide a task. Refused at every intake; the worklist is the one
    /// door into the namespace.
    #[error(
        "event kind '{kind}' is in the `agentplane.` namespace, which only this plane mints — \
         a task is decided on the worklist, never by posting its answer as an event"
    )]
    ReservedEventKind { kind: String },

    /// An open run would continue under policy semantics other than the bundle
    /// recorded at admission.
    ///
    /// Journaled as the run's quarantine reason rather than raised, so the
    /// message is the one a person reads off the run, and it names the verbs
    /// that answer it.
    #[error(
        "the policy bundle changed under an open run: admitted under {}, and this plane holds {} \
         — the run is quarantined; reopen it with `quarantine` and `replay` it on a plane \
         holding the recorded bundle, or abandon it with `quarantine`",
        bundle_named(.recorded.as_ref()),
        bundle_named(.configured.as_ref())
    )]
    PolicyBundleChanged {
        recorded: Option<crate::core::Digest>,
        configured: Option<crate::core::Digest>,
    },

    /// An open run would continue under a different **declaration** than the
    /// one it was admitted under.
    ///
    /// The bundle above covers who may authorize; this covers what the agent
    /// *is*. A declarative agent's behaviour is its manifest — the prompt, the
    /// tool grants, the model, the ceilings — so editing it and resuming runs
    /// one program over another's journal. Refused before anything replays,
    /// which is the difference between a named remedy and discovering the same
    /// fact as a key mismatch several effects in.
    ///
    /// Reported only where both sides name a declaration. A coded skill's
    /// behaviour is the embedder's binary, which this crate cannot identify and
    /// does not claim to; there, divergence is the answer, later and less
    /// precisely.
    ///
    /// Journaled as the run's quarantine reason rather than raised, as the
    /// bundle refusal is.
    #[error(
        "the declaration for `{agent}` changed under an open run: admitted under {recorded}, \
         and this plane holds {configured} — the run is quarantined; reopen it with \
         `quarantine` and `replay` it under the revision that wrote the journal, or abandon \
         it with `quarantine`"
    )]
    DeclarationChanged {
        agent: String,
        recorded: crate::core::Digest,
        configured: crate::core::Digest,
    },

    /// The history was written under a different canonicalization rule.
    ///
    /// Not a divergence, and reporting it as one is the defect this exists to
    /// remove: every effect key comes out of the canonicalizer, so a rule change
    /// moves all of them at once and a healthy run replays as *non-determinism*.
    /// The run is **unverifiable by this build**, which is a different claim and
    /// the one the evidence supports.
    ///
    /// The journal chain is unaffected — it hashes the bytes it stored rather
    /// than re-canonicalizing them — so the history is intact and readable; it
    /// simply cannot be re-derived here. Before format freeze the answer is to
    /// recreate; after it, a build that means to read old history implements the
    /// old rule and selects on this number.
    #[error(
        "this run's derived digests were produced by canonicalization rule \
         {recorded} and this build implements {implemented}, so its effect keys \
         cannot be recomputed here. The journal is intact — the chain hashes \
         stored bytes, not re-canonicalized ones — and this is not a divergence"
    )]
    CanonicalizationChanged { recorded: u16, implemented: u16 },

    /// The run's own history is sealed to a key that was destroyed, so this
    /// build cannot read the plan it must replay.
    ///
    /// **A completed erasure, not a fault**, and the two call for opposite
    /// responses — which is why this is its own variant rather than the
    /// deserialization error the payload's shape produces. A sealed payload
    /// arrives at the parser as `{"$sealed": "…"}`, and the parser says what a
    /// parser says: a field is missing. An operator reading that goes looking
    /// for a corrupt journal, for a version skew, for a bug. The journal is
    /// intact, the chain still verifies, and nothing is wrong with the build:
    /// the data is gone because somebody asked for it to be.
    ///
    /// **What is still available is the part that matters.** A run whose data
    /// is erased can never execute again — its recorded effects cannot be read
    /// back, so there is nothing to resume onto — but it can still be
    /// *concluded*. A cancellation and an abandonment are recorded in the
    /// clear and need no plan, so an operator is never left holding a run with
    /// no verb that clears it.
    #[error(
        "run {run}'s recorded plan is sealed to a destroyed key: its payloads were \
         erased, so it cannot be replayed or resumed. The journal is intact and \
         still verifies, and nothing is wrong with this build"
    )]
    PayloadsErased { run: String },

    /// The run's history is sealed, and this plane holds no key ring to open
    /// it.
    ///
    /// Not an erasure, and kept apart from [`PayloadsErased`](Self::PayloadsErased)
    /// because the two send a reader opposite ways: *erased* says the data is
    /// gone for good, this says it is intact and this plane cannot read it — an
    /// operator's terminal, a restored export, a verifier handed no key. A ring
    /// that is wired answers for itself: a destroyed key is an erasure, an
    /// unreachable one fails the read.
    #[error(
        "run {run}'s recorded plan is sealed and this plane holds no key ring to open \
         it: nothing is known to be erased — replay it where the key ring it was sealed \
         under is wired"
    )]
    PayloadsSealed { run: String },

    /// Nothing on this plane answers to the name `run` was given.
    ///
    /// Carries what the plane *does* provide, because the question a reader has
    /// next is always "then what should I have asked for?" — and the plane is
    /// the only party that can answer it. A refusal that names the missing thing
    /// and not the available ones sends somebody back to their own source to
    /// reconstruct a list this error was already holding.
    #[error("{}", no_provider_message(target, available))]
    NoProvider {
        /// The capability (or skill name) that was asked for.
        target: String,
        /// Every capability this plane provides, sorted. Empty means no skills.
        available: Vec<String>,
    },

    /// The tenant is at a ceiling, so nothing was admitted.
    ///
    /// Distinct from a policy denial, because they call for opposite responses.
    /// A denial says *you may not*, and retrying is pointless. A quota refusal
    /// says *not right now*, and the caller should come back — a concurrency
    /// ceiling clears when a run finishes. Collapsing them would teach callers
    /// to retry denials or to give up on back-pressure.
    /// Transparent, because the variant carries a halt as well as a ceiling
    /// and the two must not share a prefix: an operator reading `quota: … is
    /// halted` has been told a stop is back-pressure, which is the confusion
    /// `QuotaError::Halted` exists to prevent.
    #[error(transparent)]
    QuotaExceeded(#[from] crate::quota::QuotaError),

    /// This instance is shutting down, so nothing was admitted.
    ///
    /// Back-pressure, not a verdict: the work is fine and another instance will
    /// take it. Kept apart from a quota refusal because the two clear on
    /// different terms — a ceiling clears when a run finishes here, this one
    /// clears when the caller reaches a different process — and apart from a
    /// halt, which means *stop asking anybody*.
    ///
    /// Nothing was written: no lease, no quota slot, no journal. A caller may
    /// retry immediately, elsewhere.
    #[error(
        "this instance is draining and did not admit the run — retry; \
         nothing was written and another instance can take it"
    )]
    Draining,

    /// A live pass finished, but its durable quota receipt did not commit.
    ///
    /// The run deliberately keeps its lease. Once it expires, the abandonment
    /// sweep derives the same settlement from the journal and retries it under
    /// the idempotent `(run, epoch)` key; releasing here would remove that retry
    /// handle and turn a store outage into permanent under-accounting.
    #[error(
        "quota settlement for run {run} at epoch {epoch} is pending: {detail} — \
         the run remains leased so recovery can retry without charging twice"
    )]
    QuotaSettlementPending {
        run: String,
        epoch: u64,
        detail: String,
    },

    /// The journal's hash chain does not verify. Either a record was altered
    /// after the fact, or a writer produced bytes it did not hash.
    #[error("journal integrity broken at seq {seq}: {detail}")]
    ChainBroken { seq: Seq, detail: String },

    /// A write was rejected because another instance owns this run at a higher
    /// epoch. Not an error to retry blindly: this instance has been fenced and
    /// must drop the run.
    #[error("fenced at run {run}: held epoch {held}, store is at {current}")]
    Fenced {
        run: String,
        held: u64,
        current: u64,
    },

    /// Another instance holds a live lease on this run. Retryable *after* the
    /// lease expires — unlike [`Fenced`](Self::Fenced), which never is.
    #[error("run {run} is leased by '{owner}' for another {remaining_secs}s")]
    LeaseHeld {
        run: String,
        owner: String,
        remaining_secs: u64,
    },

    /// An operator's answer was offered to a run that is not asking a question.
    ///
    /// Reopening and abandoning are answers to a **quarantine** specifically,
    /// and a run in any other state has either not stopped, stopped for a
    /// reason a resume already addresses, or ended. Refused by name rather than
    /// recorded and ignored: an intervention that is acknowledged and then not
    /// acted on is worse than one that is declined, because the operator stops
    /// looking.
    #[error(
        "run {run} is '{status}', not quarantined — only a run the runtime could not decide can \
         be reopened or abandoned"
    )]
    NotQuarantined { run: String, status: String },

    /// An assertion was offered about an effect whose outcome is already known.
    ///
    /// A person may supply the fact the journal lacks; they may not replace one
    /// it holds. Overwriting a recorded landing with "it did not happen" would
    /// let an operator talk a run out of compensating work that stands in the
    /// world — with the record showing an orderly reconciliation.
    #[error(
        "effect {effect} in run {run} is not in doubt — its outcome is on the record, and an \
         assertion may supply a missing fact but never replace a recorded one"
    )]
    NotUndecided { run: String, effect: String },

    /// Something that unwinds was asked for on a run that must not unwind.
    ///
    /// Cancelling promises to reverse what the run did and put the world back,
    /// which is exactly what a run holding an unknown outcome may not do. Its
    /// own variant rather than a generic refusal, because the operator's next
    /// move is named in it and a caller matching on the class should be able to
    /// route them there.
    #[error(
        "run {run} is quarantined, and cancelling unwinds — which is exactly what a run holding \
         an unknown outcome must not do. Answer the doubt and reopen it, or abandon it, which \
         closes it without unwinding"
    )]
    CannotUnwind { run: String },

    /// A cancellation was asked of a run that has already concluded.
    ///
    /// Refused rather than recorded: a sealed run is not reopened by anybody
    /// changing their mind, and a stored request against it would answer the
    /// operator "recorded" for a stop that can never happen.
    #[error("run {run} already concluded as '{outcome}'; there is nothing left to stop")]
    AlreadyConcluded { run: String, outcome: String },

    #[error(transparent)]
    Store(#[from] StoreError),

    #[error(transparent)]
    Encoding(#[from] serde_json::Error),
}

/// What a failure says about whether the call reached the outside world.
///
/// This is the distinction retry safety rests on, and it is not the same
/// question as "was the error transient". A refused connection and a timed-out
/// request are both transient; only one of them is safe to repeat against a
/// ledger.
///
/// The vocabulary is borrowed from distributed transactions, where a
/// participant whose outcome is unknown after a failure has been called
/// **in-doubt** since the XA specification. The situation is identical: the
/// journal cannot distinguish "never applied" from "applied, and the
/// acknowledgement was lost", and no amount of retrying makes it decidable.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Disposition {
    /// The call provably never took effect — refused before dispatch, or
    /// rejected by the peer with the request intact. Safe to repeat, even for
    /// something that mutates.
    DidNotHappen,

    /// The outcome is unknown. The request may or may not have been applied,
    /// and nothing observable distinguishes the two.
    ///
    /// Identical in kind to the orphan a crash leaves behind, so it is resolved
    /// the same way: by the effect's declared [`Recovery`](crate::core::Recovery),
    /// never by guessing.
    InDoubt,

    /// It definitely took effect, and something went wrong afterwards — most
    /// often a response that would not decode.
    ///
    /// Never retried. A repeat would be a second real performance, and the
    /// second one would fail to decode exactly like the first.
    Landed,
}

impl Disposition {
    /// The variant name, for a metric label.
    ///
    /// Deliberately not `Display`: a metric dimension must be bounded, and a
    /// rendered message carries values. One label per distinct limit or detail
    /// string is a cardinality explosion that takes a metrics backend down —
    /// which is why every dimension in `runtime::metrics` comes from an accessor
    /// like this one rather than from a formatted error.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::DidNotHappen => "did_not_happen",
            Self::InDoubt => "in_doubt",
            Self::Landed => "landed",
        }
    }

    /// Whether repeating the call is safe on its own terms, before the
    /// effect's [`Recovery`](crate::core::Recovery) gets a say.
    #[must_use]
    pub fn is_definitely_safe_to_repeat(self) -> bool {
        matches!(self, Self::DidNotHappen)
    }
}

/// Failure of a single external interaction.
///
/// Every variant declares a [`Disposition`], because the runtime cannot infer
/// one from a message and must not guess. Anything that does not say is treated
/// as [`InDoubt`](Disposition::InDoubt) — the same conservative default that
/// makes [`Recovery::RequiresOperator`](crate::core::Recovery::RequiresOperator)
/// the fallback for an effect that does not declare itself.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum EffectError {
    /// The driver could not dispatch at all — no connection, no credentials,
    /// no route. Nothing reached the peer.
    #[error("driver '{driver}' unavailable: {detail}")]
    Unavailable { driver: String, detail: String },

    /// The peer received the call and refused it. The request is intact and
    /// nothing was applied.
    ///
    /// Retried under the effect's policy, because this covers transient
    /// refusals — an overloaded gateway, a rate limit, a 5xx. A refusal that
    /// no retry can change is [`Refused`](Self::Refused), and conflating the
    /// two is how a caller retries a decision that will never change.
    #[error("effect rejected: {0}")]
    Rejected(String),

    /// The peer declined *for now* and said when to come back.
    ///
    /// A rate limit is a wait, not a fault, and the peer is the only party who
    /// knows how long. Separate from [`Rejected`](Self::Rejected) — which is
    /// also retried — because a computed backoff is a guess, and here there is
    /// nothing to guess about: the runtime waits the named time, bounded by
    /// [`RetryPolicy::max_advice`](crate::core::RetryPolicy::max_advice).
    ///
    /// `retry_after` is `None` where the peer named no window, which is
    /// ordinary — the effect's own schedule applies, and the variant still
    /// carries what that schedule cannot infer: repeating sooner is pointless
    /// rather than merely unlucky.
    ///
    /// Nothing was applied and nothing metered, so a mutating effect is as safe
    /// to repeat here as a read.
    #[error("{}", rate_limited_message(detail, *retry_after))]
    RateLimited {
        detail: String,
        retry_after: Option<std::time::Duration>,
    },

    /// The peer understood the request and said **no** — an answer, not a
    /// fault. Nothing was applied, and nothing was metered.
    ///
    /// Distinct from [`Rejected`](Self::Rejected) in the one way the retry
    /// loop can act on: repeating this call asks the same rule the same
    /// question, so no attempt is spent on it — the first refusal is the
    /// final one. A model provider's 400 (unknown model, malformed request,
    /// input filtered) is the canonical case: three retries with backoff
    /// against a request that is *wrong* burn wall-clock and teach the
    /// operator that retries are noise.
    ///
    /// The bit is recorded on the failure (`EffectFailed.permanent`), because
    /// the retry decision is replayed from history and a replay that could
    /// not see it would expect a retry the live run never made.
    #[error("effect refused: {0}")]
    Refused(String),

    /// The peer accepted the request and never answered in time.
    ///
    /// The canonical in-doubt case, and the one that separates this runtime
    /// from a retry loop: a timed-out payment may well have been taken.
    #[error("driver '{driver}' did not answer within {waited_ms}ms")]
    Timeout { driver: String, waited_ms: u64 },

    /// The connection died mid-flight, after the request went out.
    #[error("driver '{driver}' interrupted: {detail}")]
    Interrupted { driver: String, detail: String },

    /// The call consumed metered resources and then failed.
    ///
    /// A model stream that dies after five hundred tokens has *spent* them: the
    /// provider bills for what it generated whether or not the answer arrived.
    /// Every other failure variant reports nothing consumed, which is right for
    /// a refused connection and wrong for this — and wrong in the direction that
    /// matters, because the token and cost ceilings exist to bound exactly the
    /// runaway a flaky provider produces.
    ///
    /// Carries its own disposition because only the driver knows: a stream that
    /// died mid-response definitely reached the provider, while a request
    /// refused before generation did not.
    #[error("effect consumed resources and failed: {detail}")]
    Metered {
        detail: String,
        spend: Spend,
        disposition: Disposition,
    },

    /// The peer performed the operation and reported that it failed.
    ///
    /// Distinct from [`Rejected`](EffectError::Rejected), which means the peer
    /// declined *before* doing anything. Here the work was attempted, so a
    /// repeat is a second attempt — and whether the first one changed something
    /// before failing is not knowable from the answer.
    ///
    /// Treated as `Landed` rather than `InDoubt` deliberately. `InDoubt` invites
    /// the effect's `Recovery` to resolve it, and for an outcome the peer has
    /// already reported there is nothing to resolve: asking again returns the
    /// same error, and repeating the call is the only other option.
    #[error("effect performed and failed: {0}")]
    Performed(String),

    /// It landed and answered, and the answer did not match the declared type.
    #[error("effect output did not match its declared type: {0}")]
    OutputShape(#[from] serde_json::Error),

    /// Every permitted attempt failed, and this is the last one's verdict.
    ///
    /// Carries the **disposition** rather than flattening it. A driver that
    /// said [`Rejected`](Self::Rejected) — refused before anything happened —
    /// must not be reported upward as undecidable merely because the runtime
    /// stopped retrying. Anything deciding whether it is safe to unwind would
    /// then refuse to, for a call that provably did nothing; and the failure
    /// that most needs an operator would be indistinguishable from the one that
    /// needs nobody.
    #[error("{detail}")]
    Final {
        detail: String,
        disposition: Disposition,
    },

    #[error("{0}")]
    Other(String),
}

impl EffectError {
    /// What this failure cost, if anything.
    ///
    /// Zero for everything that never reached a meter. The runtime bills this on
    /// the failure path, so a call that burned tokens and then died is counted
    /// against the run's ceiling rather than being free.
    ///
    /// Written out variant by variant rather than defaulting the rest, because
    /// the default would be **free** and the ceilings this feeds — tokens,
    /// cost, `max_effects` — exist to bound exactly the runaway a flaky
    /// provider produces. A variant added later that carries a meter would
    /// compile, pass every test here, and silently spend nothing; enumerated,
    /// it does not build until somebody has answered what it cost.
    #[must_use]
    pub fn spend(&self) -> Spend {
        match self {
            Self::Metered { spend, .. } => *spend,
            // Nothing reached a meter: refused before dispatch, refused by the
            // peer, or answered and rejected. `Final` carries the last
            // attempt's verdict and no meter of its own — every attempt was
            // billed as it failed, and adding them again here would double
            // every retried run's spend.
            Self::Unavailable { .. }
            | Self::Rejected(_)
            | Self::RateLimited { .. }
            | Self::Refused(_)
            | Self::Timeout { .. }
            | Self::Interrupted { .. }
            | Self::Performed(_)
            | Self::OutputShape(_)
            | Self::Final { .. }
            | Self::Other(_) => Spend::default(),
        }
    }

    /// When the peer asked to be called again, if it named a time.
    ///
    /// Unbounded on purpose: this reports what was *said*. What this deployment
    /// will act on is
    /// [`RetryPolicy::max_advice`](crate::core::RetryPolicy::max_advice) — one
    /// bound, at the place that owns the schedule.
    #[must_use]
    pub const fn retry_after(&self) -> Option<std::time::Duration> {
        match self {
            Self::RateLimited { retry_after, .. } => *retry_after,
            _ => None,
        }
    }

    /// The class of fault, as one low-cardinality word.
    ///
    /// What `error.type` reports on an effect span. Rendered messages carry the
    /// detail and are useless to group by; the disposition says what the failure
    /// means for the *world* and deliberately collapses faults that differ for an
    /// operator — a refused credential and a timed-out socket are both
    /// `DidNotHappen`. This is the third question, *what went wrong*, and it is
    /// the one "which driver fails how" is asked in.
    ///
    /// Exhaustive rather than defaulted, for the reason
    /// [`spend`](Self::spend) is: a variant added later would otherwise arrive
    /// under whatever word the catch-all happened to say.
    #[must_use]
    pub const fn class(&self) -> &'static str {
        match self {
            Self::Unavailable { .. } => "unavailable",
            Self::Rejected(_) => "rejected",
            Self::RateLimited { .. } => "rate_limited",
            Self::Refused(_) => "refused",
            Self::Timeout { .. } => "timeout",
            Self::Interrupted { .. } => "interrupted",
            Self::Metered { .. } => "metered",
            Self::Performed(_) => "performed",
            Self::OutputShape(_) => "output_shape",
            Self::Final { .. } => "final",
            Self::Other(_) => "other",
        }
    }

    /// What this failure says about whether the call reached the outside world.
    #[must_use]
    pub fn disposition(&self) -> Disposition {
        match self {
            Self::Metered { disposition, .. } | Self::Final { disposition, .. } => *disposition,
            Self::Unavailable { .. }
            | Self::Rejected(_)
            | Self::RateLimited { .. }
            | Self::Refused(_) => Disposition::DidNotHappen,
            Self::OutputShape(_) | Self::Performed(_) => Disposition::Landed,
            // `Other` shares the in-doubt arm deliberately: an error that does
            // not say what it did is treated as dangerous. A driver that wants
            // its failures retried has to state that they did not happen.
            Self::Timeout { .. } | Self::Interrupted { .. } | Self::Other(_) => {
                Disposition::InDoubt
            }
        }
    }
}

/// Failure inside a skill.
#[derive(thiserror::Error)]
#[non_exhaustive]
pub enum SkillError {
    #[error("input did not match the declared schema: {0}")]
    Input(String),

    #[error(transparent)]
    Step(#[from] StepError),

    /// A tool call could not be prepared: the tool is not in the operator's
    /// catalogue, or the arguments do not match what it declared.
    ///
    /// Here so that `?` works on `ToolCall::prepare`, which is the second thing
    /// the getting-started page teaches and the first thing every skill that
    /// touches the world does. Without it the published snippet did not compile
    /// — *the trait `From<ToolError>` is not implemented for `SkillError`* — and
    /// every real caller wrote the same
    /// `.map_err(|e| SkillError::Other(e.to_string()))` incantation, which
    /// throws the typed error away and leaves three copies of one decision.
    /// A skill-facing operation deserves a skill-facing conversion.
    #[error(transparent)]
    Tool(#[from] crate::tools::ToolError),

    #[error("{0}")]
    Other(String),
}

/// Failure surfaced to a skill through [`StepCtx`](crate::runtime::StepCtx).
#[derive(thiserror::Error)]
#[non_exhaustive]
pub enum StepError {
    #[error(transparent)]
    Effect(#[from] EffectError),

    #[error(transparent)]
    Policy(#[from] PolicyError),

    #[error(transparent)]
    Store(#[from] StoreError),

    #[error("{0}")]
    Encoding(#[from] serde_json::Error),

    /// A tool call could not be prepared: the tool is not in the operator's
    /// catalogue, or the arguments do not match what it declared.
    ///
    /// The same conversion [`SkillError::Tool`] provides, one level down,
    /// and it is load-bearing for the same reason. `sink_with` is how a
    /// governed tool call is written — the closure hands back whatever
    /// building the effect produced — and [`BuildsEffect`] accepts a
    /// `Result` only when its error converts to *this* type. Without this
    /// variant the trait's stated purpose ("`ToolCall::prepare`, which
    /// refuses a tool the catalogue does not hold, returns the `Result` it
    /// already produces") was a capability nothing could use, and the
    /// published snippet that exercises it did not compile.
    ///
    /// A refusal here dispatched nothing, so it fails the step rather than
    /// leaving doubt: the catalogue was consulted before any call left.
    ///
    /// [`BuildsEffect`]: crate::runtime::BuildsEffect
    #[error(transparent)]
    Tool(#[from] crate::tools::ToolError),

    /// The outcome of an effect cannot be determined, and its declared
    /// [`Recovery`](crate::core::Recovery) forbids guessing.
    ///
    /// Reached two ways, which are the same situation from different
    /// directions: a crash landed between "sent" and "recorded", or the call
    /// itself failed [`InDoubt`](Disposition::InDoubt). Either way the journal
    /// cannot distinguish "never applied" from "applied, acknowledgement lost",
    /// and for anything that mutates, the runtime escalates rather than guess.
    ///
    /// A distinct variant rather than a message, because the executor
    /// quarantines on it — and a run's disposition must not hinge on the
    /// wording of a string.
    #[error(
        "effect {key} is undecidable ({detail}); recovery mode {recovery:?} forbids \
         guessing — run quarantined"
    )]
    Undecidable {
        key: EffectKey,
        recovery: crate::core::Recovery,
        detail: String,
    },

    /// The effect's outcome is known to this process and could not be recorded.
    ///
    /// The call already returned when the journal refused the terminal record,
    /// so the journal holds an announcement with no outcome — an orphan a
    /// resume resolves by the declared recovery, exactly as it would after a
    /// crash. What this variant preserves is the knowledge the crash case does
    /// not have: *this* pass saw the answer, and `disposition` says whether
    /// the call reached the world.
    ///
    /// A distinct variant rather than a [`Store`](Self::Store) error because a
    /// consumer deciding what the failure permits branches on exactly that
    /// knowledge: an effect group asked to abort may claim *taken back whole*
    /// only over members that provably did not externalise, and a store error
    /// raised after the call landed is not that — flattening the two would
    /// make the cheap abort available over a send that already went out.
    #[error(
        "effect {key} completed ({disposition:?}) but its outcome could not be recorded: {detail}"
    )]
    Unrecorded {
        key: EffectKey,
        /// What the call did to the world, as this process observed it.
        disposition: crate::core::Disposition,
        detail: String,
    },

    /// A read this run pinned no longer returns what it was pinned to.
    ///
    /// I1 exempts a read whose answer cannot change from the effect protocol —
    /// bytes by their own digest, a memory by id *and* version — on the
    /// condition that the immutability claim is checked rather than assumed.
    /// This is that check failing: the identifier still resolves, and not to
    /// what the journal recorded choosing.
    ///
    /// Its own variant because the executor quarantines on it, for the reason
    /// [`Undecidable`](Self::Undecidable) is a variant and not a message: a
    /// run's disposition may not hinge on the wording of a string. A store that
    /// cannot be reached fails a run that can be run again; one that answers
    /// successfully with different content has made every run that read it
    /// unreproducible, including ones that already finished.
    ///
    /// Deliberately **not** raised when the pinned thing is *gone*. Erasure is a
    /// recorded decision, and routing it here would fill the integrity backlog
    /// with lawful deletions.
    #[error(
        "{what} is no longer what this run pinned it to ({detail}) — the history \
         cannot be reproduced, so the run is quarantined rather than failed"
    )]
    Unreproducible { what: String, detail: String },

    /// Surfaced when replay finds the recorded run took a different path.
    ///
    /// **`detail` is the field that makes this actionable**, and the two keys
    /// are not. A digest pair says two calls differ and nothing about how; the
    /// party who has to act is whoever changed the code, and what they need is
    /// which call moved. It is composed from the fields that are *clear* on the
    /// record — an effect's kind and its attempt — so the conclusion that
    /// carries it says which call moved without opening a sealed payload.
    #[error("non-determinism at seq {seq}: {detail} (history {expected}, this build {actual})")]
    NonDeterminism {
        seq: Seq,
        expected: EffectKey,
        actual: EffectKey,
        detail: String,
    },

    /// A limit stopped the run before it spent more.
    ///
    /// Not a fault: the run did what it was told, and what it was told included
    /// a ceiling. Distinct from an ordinary failure so an operator can tell
    /// "this needs a bigger budget" from "this is broken".
    #[error(transparent)]
    Budget(#[from] crate::core::BudgetExceeded),

    /// **Not a failure.** The run is waiting for something that has not
    /// happened, and its frame has been persisted.
    ///
    /// Propagate it with `?`. A skill that catches this turns a durable wait
    /// into a silent hang: the subscription stays registered, the event
    /// eventually arrives, and it resumes a run that has already decided it
    /// finished. It is modelled as an error only because that is how control
    /// leaves a skill — the run is healthy.
    #[error("suspended: {0}")]
    Suspended(crate::core::SuspendReason),

    /// Policy refused the effect.
    ///
    /// Separate from `Budget` because the two are answered differently: a limit
    /// is raised, a rule is argued with. Collapsing them would put "ask for more
    /// quota" and "you are not allowed to do this" behind one message.
    #[error("policy denied '{action}' on '{resource}': {reason}")]
    Denied {
        action: String,
        resource: String,
        reason: String,
    },

    /// A member did not fit the group it was added to.
    ///
    /// A footprint violation, a mutating effect declared as a read, a nested
    /// group, or an empty footprint. Every one of these is caught **before**
    /// the effect runs, which is the only time catching it is free.
    #[error("effect group '{group}': {detail}")]
    GroupFootprint { group: String, detail: String },

    /// A group was taken back whole, and nothing it did is standing.
    ///
    /// Not a quarantine and not a silent failure: every reversible member was
    /// reversed, no deferred member ran, and `what` says which condition
    /// stopped it. A caller may handle this and carry on, which is the point of
    /// grouping in the first place.
    #[error("effect group aborted and fully reversed: {what}")]
    GroupAborted { what: String },

    /// A group could be neither committed nor taken back.
    ///
    /// A reversal failed, or a member is in doubt. The run is quarantined,
    /// because a partially unwound group is a state nobody declared and no
    /// later code can reason about. This is the honest report of the situation
    /// that other systems surface as a success with a warning.
    #[error("effect group '{group}' could not be settled: {detail} — run quarantined")]
    GroupUnsettled { group: String, detail: String },

    /// Strict replay reached the end of history and the code asked for another
    /// effect. The recorded run did less than this code does — divergence that
    /// ordered key comparison alone cannot see, because there is nothing left to
    /// compare against.
    #[error(
        "replay overrun: journal is exhausted but the run requested `{kind}` ({actual}) — \
         this build performs more effects than the recorded one"
    )]
    ReplayOverrun { actual: EffectKey, kind: String },
}

impl SkillError {
    /// The class of fault, as one low-cardinality word — what a loud event
    /// reports as `error_type` in place of the message.
    ///
    /// Exhaustive for the reason [`EffectError::class`] is.
    #[must_use]
    pub const fn class(&self) -> &'static str {
        match self {
            Self::Input(_) => "input",
            Self::Step(step) => step.class(),
            Self::Tool(_) => "tool",
            Self::Other(_) => "other",
        }
    }
}

impl StepError {
    /// The class of fault, as one low-cardinality word.
    ///
    /// An effect's failure reports the effect's own class, so the word is the
    /// one its span already carried as `error.type`.
    #[must_use]
    pub const fn class(&self) -> &'static str {
        match self {
            Self::Effect(e) => e.class(),
            Self::Policy(_) => "policy",
            Self::Store(_) => "store",
            Self::Encoding(_) => "encoding",
            Self::Tool(_) => "tool",
            Self::Undecidable { .. } => "undecidable",
            Self::Unrecorded { .. } => "unrecorded",
            Self::Unreproducible { .. } => "unreproducible",
            Self::NonDeterminism { .. } => "nondeterminism",
            Self::Budget(_) => "budget",
            Self::Suspended(_) => "suspended",
            Self::Denied { .. } => "denied",
            Self::GroupFootprint { .. } => "group_footprint",
            Self::GroupAborted { .. } => "group_aborted",
            Self::GroupUnsettled { .. } => "group_unsettled",
            Self::ReplayOverrun { .. } => "replay_overrun",
        }
    }
}

/// Authorization failure.
///
/// Evaluation is total and side-effect free, so this never means "the policy
/// engine was unreachable" — that state cannot arise.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum PolicyError {
    #[error("principal '{principal}' may not '{action}' on '{resource}'")]
    Denied {
        principal: String,
        action: String,
        resource: String,
    },

    /// A sink-gate refusal read back from the journal.
    ///
    /// Carries the recorded wording rather than re-deriving a variant from it,
    /// for the reason [`BudgetExceeded::Recorded`](crate::core::BudgetExceeded)
    /// does: the run stopped for the reason it recorded, and a replay that
    /// re-worded it makes an auditor compare a run's status against its own
    /// replay and find a difference that means nothing.
    ///
    /// It is a [`PolicyError`] rather than a
    /// [`StepError::Denied`](crate::core::StepError::Denied) because *which
    /// gate refused* survives the round trip and matters: a sink gate refuses
    /// one call, and a tool-calling loop may tell the model and try another
    /// route, where an authorization denial ends the run. Collapsing the two on
    /// the way back would end a run the original finished.
    #[error("{reason}")]
    Recorded { reason: String },

    /// An argument derived from untrusted data reached a mutating sink without
    /// an explicit, policy-authorized release.
    #[error("untrusted data may not reach mutating sink '{sink}' without an authorized release")]
    TaintGate { sink: String },

    /// A sink did not expose the value it will send, so the runtime cannot bind
    /// the information-flow decision to the outbound call.
    #[error("sink '{sink}' does not bind the arguments it sends to the value checked by policy")]
    UnboundSinkArguments { sink: String },

    /// A caller tried to dispatch an outbound-value effect through the generic
    /// effect API, bypassing information-flow enforcement.
    #[error("sink '{sink}' must be dispatched with StepCtx::sink so its outbound value is checked")]
    SinkGateRequired { sink: String },

    /// The labeled value presented to the gate differs from the value the sink
    /// will send.
    ///
    /// The rule is exact equality, so *that* they differ is the whole verdict;
    /// the reader also needs *where*, or is left with two documents and a diff
    /// by eye — and the commonest case, a bound payload still at its `null`
    /// default beside a labelled object, is the least visible. So the refusal
    /// names the first differing RFC 6901 pointer (`""` at the root) and both
    /// canonical digests. Neither value is printed: the labelled one is the
    /// data these gates exist to keep out of a log, and a digest identifies it
    /// to whoever already holds it while disclosing nothing to anyone else.
    #[error(
        "sink '{sink}' attempted to send arguments other than the labeled value policy \
         checked — they first differ at '{at}' (bound {bound}, sent {sent})"
    )]
    SinkArgumentsMismatch {
        sink: String,
        /// RFC 6901 pointer to the first difference; `""` is the whole value.
        at: String,
        /// Digest of the canonical bytes the effect will actually send.
        bound: Digest,
        /// Digest of the canonical bytes presented to the gate.
        sent: Digest,
    },

    /// A field the sink declares security-sensitive is absent from the value.
    #[error("sink '{sink}' requires protected field '{path}', but the argument is absent")]
    ProtectedFieldMissing { sink: String, path: String },

    /// Untrusted data attempted to choose a protected sink argument.
    #[error("untrusted data may not select protected field '{path}' of sink '{sink}'")]
    ProtectedFieldTaint { sink: String, path: String },

    /// A destination-scoped release covers this value — but for a different
    /// sink. Named explicitly, because "untrusted" alone would send the
    /// operator hunting for a missing release that in fact exists and was
    /// simply granted somewhere else. The message names only the two
    /// destinations — never the release's basis or evidence, which would hand
    /// a probing model the reviewer's reasoning.
    #[error(
        "untrusted data may not reach mutating sink '{sink}': the value's release \
         names destination '{granted}', and this sink is '{actual}'"
    )]
    ReleaseDestination {
        sink: String,
        granted: String,
        actual: String,
    },

    /// The field-scoped twin of [`ReleaseDestination`](Self::ReleaseDestination).
    #[error(
        "untrusted data may not select protected field '{path}' of sink '{sink}': \
         the field's release names destination '{granted}', and this sink is '{actual}'"
    )]
    ProtectedFieldReleaseDestination {
        sink: String,
        path: String,
        granted: String,
        actual: String,
    },

    /// A protected field derives from a source outside its operator declaration.
    #[error(
        "protected field '{path}' of sink '{sink}' derives from undeclared source '{actual_source}'"
    )]
    ProtectedFieldSource {
        sink: String,
        path: String,
        actual_source: String,
    },

    /// A protected field carries a value outside its declared set.
    ///
    /// The message names the constraint and deliberately not the value: the
    /// value is untrusted-influenced by construction, and echoing it hands an
    /// injected payload a path into logs and refusal channels. The manifest
    /// is where a reader sees what is permitted.
    #[error(
        "protected field '{path}' of sink '{sink}' carries a value outside the \
         declared set — the manifest enumerates what may stand in this field, \
         and this value is not one of them"
    )]
    ProtectedFieldValue { sink: String, path: String },

    /// A protected field exceeds its own sensitivity ceiling.
    #[error(
        "protected field '{path}' sensitivity {actual} exceeds sink '{sink}' field ceiling {ceiling}"
    )]
    ProtectedFieldSensitivity {
        sink: String,
        path: String,
        actual: Sensitivity,
        ceiling: Sensitivity,
    },

    /// A field-specific release was requested for a value whose field lineage
    /// was never tracked.
    #[error(
        "release scope contains a missing or untracked field; use Tainted::object/array before releasing selected fields"
    )]
    UntrackedReleaseField,

    /// A serialized release bypassed the safe constructors and violated the
    /// typed-release invariants.
    #[error("invalid release: {detail}")]
    InvalidRelease { detail: String },

    /// A value's sensitivity exceeds what this agent may write into the
    /// journal.
    ///
    /// Distinct from [`EgressCeiling`](Self::EgressCeiling), and the
    /// distinction is the whole point: egress asks *may this leave*, this asks
    /// *may this be written down forever*. The journal is append-only, so an
    /// argument recorded there — a prompt, a tool call's arguments — is never
    /// removed. A deployment with an erasure obligation has two answers and
    /// this ceiling is the first: **refuse** the data at dispatch, rather than
    /// meet an impossibility at the erasure request. The second is to **seal**
    /// it — `RuntimeBuilder::keyring` puts payloads under a per-case key that
    /// `erase_case` destroys — and the two compose: a deployment may seal
    /// everything and still refuse the classes it would rather never hold.
    ///
    /// The message names both, because a reader who has configured a key ring
    /// and then meets this refusal would otherwise conclude the seal is not
    /// working.
    #[error(
        "sensitivity {actual} exceeds the journal ceiling {ceiling} for sink \
         '{sink}' — the journal is append-only, so this argument could not be \
         removed afterwards. Put the bytes in a blob and pass the digest, or \
         configure a key ring so payloads are sealed under a key erasure destroys"
    )]
    JournalCeiling {
        sink: String,
        actual: crate::core::Sensitivity,
        ceiling: crate::core::Sensitivity,
    },

    /// A value's sensitivity exceeds what the sink is allowed to receive. This
    /// is the exfiltration path that matters: not the network, but a
    /// legitimate-looking tool call carrying a secret read three steps ago.
    #[error("sensitivity {actual} exceeds sink '{sink}' ceiling {ceiling}")]
    EgressCeiling {
        sink: String,
        actual: Sensitivity,
        ceiling: Sensitivity,
    },

    /// A handoff would make the authority chain deeper than this agent's
    /// reviewed declaration permits.
    #[error("delegation depth {actual} exceeds sink '{sink}' ceiling {ceiling}")]
    DelegationDepth {
        sink: String,
        actual: usize,
        ceiling: usize,
    },
}

/// What a model may be told about a refusal.
///
/// # A denial reason is an oracle
///
/// Every message in [`PolicyError`] is written for an operator reading a
/// journal, and each one is precise on purpose: which principal, which sink,
/// what sensitivity, which ceiling. That precision is exactly what makes it
/// unsafe to hand back to a model.
///
/// An agent loop that feeds the refusal into its next prompt turns the policy
/// into a queryable service. Injected content steering the agent can probe it:
/// vary the request, watch which variants come back refused, and read the
/// boundary off the answers. `EgressCeiling` is the sharpest case — it reports
/// the *sensitivity of the data* and the sink's ceiling, so a few probes
/// classify data the run was never allowed to reveal, without any of it ever
/// crossing the boundary.
///
/// So the split is deliberate: **the journal keeps everything, the model is told
/// one uniform sentence.** An auditor needs to know why; the thing that might be
/// attacking the policy must not learn anything it can differentiate.
///
/// This does not remove the denied/allowed bit itself. Nothing can, short of
/// fabricating success. What bounds *that* channel is
/// [`Budget::max_denials`](crate::core::Budget::max_denials), which counts the
/// refusals a model is shown — these — as well as the engine denials it is not:
/// a run that keeps being refused is probing, and it is stopped.
pub const REFUSED: &str = "this action was not permitted";

impl PolicyError {
    /// The one sentence a model may be shown.
    ///
    /// Uniform across every variant, deliberately — see [`REFUSED`]. Use
    /// [`Display`](std::fmt::Display) for the journal and for operators, and
    /// this for anything that reaches a prompt.
    #[must_use]
    pub const fn for_model(&self) -> &'static str {
        REFUSED
    }
}

/// Persistence failure.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum StoreError {
    #[error("backend: {0}")]
    Backend(String),

    #[error("not found: {0}")]
    NotFound(String),

    /// A single record exceeded the size a journal will hold.
    ///
    /// Refused rather than written, and the distinction is the whole point. The
    /// journal is append-only and hash-chained: an oversized record cannot be
    /// pruned later, cannot be rewritten, and is replayed on every read of that
    /// run. Every durable-execution engine in the field caps this — Temporal at
    /// 2 MB with a claim check above ~256 KiB, Restate at 32 MiB after
    /// oversized entries drove it into an unrecoverable state — and the failure
    /// they all avoid is the one where the write succeeds and the problem
    /// surfaces months later as a store nobody can read quickly.
    ///
    /// The fix is at the call site, not here: put the bytes somewhere addressed
    /// by a digest and journal the digest.
    #[error(
        "record of {bytes} bytes exceeds the {limit}-byte journal limit — \
         journal a digest and keep the bytes outside the chain"
    )]
    RecordTooLarge { bytes: usize, limit: usize },

    /// The `(run_id, effect_key)` unique index rejected a second start for one
    /// effect. Exactly-once is a database invariant here, not a code path.
    #[error("effect {0} already started in this run")]
    DuplicateEffect(EffectKey),

    /// The `(tenant, idempotency_key)` unique index rejected a second
    /// admission. At-most-once admission is a database invariant, not a code
    /// path — for the reason exactly-once above is.
    ///
    /// Carries the run that **holds** the key, which is the answer the loser
    /// needs. The runtime turns this into
    /// [`Admission::Replayed`](crate::runtime::Admission::Replayed) or
    /// [`Admission::InFlight`](crate::runtime::Admission::InFlight) before a
    /// caller sees it.
    #[error("admission key '{key}' is already held by run {run}")]
    DuplicateAdmission { key: String, run: String },

    /// A batch was reopened under a different plan.
    ///
    /// One batch runs one frozen plan: its audit identity is "this act, under
    /// this plan", and item 60,001 settling under an edited plan would be a
    /// second act wearing the first one's name. Both digests are named so the
    /// operator can see which two artifacts disagree; the remedy is a new
    /// batch, never a widened old one.
    #[error(
        "batch {batch} is bound to plan {stored}, and this resume offers {offered} — \
         one batch runs one frozen plan; start a new batch for a new plan"
    )]
    BatchPlanChanged {
        batch: String,
        stored: String,
        offered: String,
    },

    /// A case-state write named a version the case has moved past.
    ///
    /// Somebody else wrote to this case between the read and the write. The
    /// caller must re-read and decide again — *not* retry the same write, which
    /// would be the lost update this error exists to prevent.
    #[error("case {case} has moved to {current}; the write was made against {expected}")]
    CaseConflict {
        case: String,
        expected: u64,
        current: u64,
    },

    /// Closure was refused because the case still owes obligations.
    ///
    /// A business refusal, not a store fault, and the type is what keeps the
    /// two apart: closure is the moment people stop looking, so an unmet
    /// deadline must survive it — while a store that is merely unreachable
    /// must not read as the rule firing, or an outage becomes indistinguishable
    /// from enforcement. The remedy is the caller's: meet or cancel the
    /// obligations, then close.
    #[error(
        "case {case} still has {outstanding} open obligation(s); closure is \
         when people stop looking, so meet or cancel them before closing"
    )]
    ObligationsOutstanding { case: String, outstanding: usize },

    /// A write was refused because the matter is closed.
    ///
    /// *A closed case owes nothing* is a property of the store, not of the order
    /// two callers ran in: closure refuses an outstanding obligation, and this
    /// refuses the write that would add one afterwards. Without both halves the
    /// sweep breaches the late obligation and escalates, so a matter audited as
    /// settled acquires a duty and misses it with no run and no operator
    /// involved.
    ///
    /// Reopening is the caller's to do, and deliberately explicit: a store that
    /// reopened the case for them would decide on somebody's behalf that an
    /// audited closure did not stand.
    #[error(
        "case {case} is closed; reopen it before registering an obligation, because \
         closure is when people stop looking"
    )]
    CaseClosed { case: String },

    /// A write was refused because the address it names has been erased.
    ///
    /// Content addressing makes the address the content, so a run producing the
    /// same bytes a second time lands on the erased object. Taking the write
    /// would put the data back — silently, under a tombstone that still records
    /// when and why it went — which is an erasure reported as discharged and
    /// then reversed by ordinary work.
    ///
    /// Typed rather than a [`Backend`](Self::Backend) string for the reason
    /// every refusal here is: retrying cannot help, the store is healthy, and a
    /// business rule wearing a storage fault's type is read as an outage by
    /// everything that classifies one.
    #[error(
        "blob {digest} was erased at {at} ({reason}); storing these bytes again would \
         put back what somebody asked to have removed"
    )]
    BlobErased {
        digest: String,
        at: i64,
        reason: String,
    },

    /// An obligation was moved out of a state it may not leave.
    ///
    /// Met, breached and withdrawn are the three ways an obligation ends, and
    /// the second has to be terminal: the state column is the only record that
    /// a window closed unmet, and moving it to `met` would take the miss off
    /// the obligation listing and out of the row at once. Answering late is a
    /// fact to record beside the breach — an acknowledgement carries the note —
    /// not a way to unsay it.
    #[error(
        "obligation '{obligation}' on case {case} is {from} and may not become \
         {to}; how an obligation ended is not editable, so record a late answer \
         as an account of the breach rather than as a state"
    )]
    DeadlineFinal {
        case: String,
        obligation: String,
        from: String,
        to: String,
    },

    /// An account was offered for an obligation that was not breached.
    ///
    /// Accounting for a breach is only meaningful once there is one. Accepting
    /// it earlier would let a pending obligation be taken off the listing
    /// before it was ever due — which is the listing's whole purpose, answered
    /// in advance.
    #[error("obligation '{obligation}' on case {case} is {state}, not breached")]
    NotBreached {
        case: String,
        obligation: String,
        state: String,
    },

    /// The writer did not present the current lease epoch. A stale writer has
    /// been taken over; a future epoch was never acquired. Neither owns the run,
    /// and neither may retry blindly.
    #[error("fenced: run {run} is owned at epoch {current}, writer held {held}")]
    Fenced {
        run: String,
        held: u64,
        current: u64,
    },

    /// A transaction's `COMMIT` was sent and no acknowledgement arrived.
    ///
    /// The one in-doubt window a native transaction keeps: the server either
    /// committed or did not, but the *client's knowledge* of which was lost —
    /// a connection dropped between sending `COMMIT` and receiving its answer.
    /// A commit the server **refused** (a serialization or constraint failure,
    /// returned as a database error) is not this; that is a clean rollback and
    /// stays an ordinary [`Backend`](Self::Backend) error. The two must stay
    /// distinguishable, because they call for opposite handling: a refusal is
    /// a cheap abort, and an unknown outcome must be treated as a standing
    /// write until somebody reconciles it — settling `Aborted` over it would
    /// be the journal claiming *taken back whole* about a write that may
    /// stand.
    #[error(
        "the transaction's outcome is unknown — COMMIT may or may not have \
         been applied: {detail}"
    )]
    CommitUnknown { detail: String },

    /// The run is sealed; its journal is frozen.
    ///
    /// A seal freezes the chain head the Merkle log's leaf commits to. An
    /// append past it — even by the caller that legitimately holds the current
    /// epoch — advances the true head past the leaf every checkpoint attests,
    /// so the store refuses it inside the same transaction that would have
    /// written it. The executor's own refusal to resume a closed run is
    /// application logic a future caller can bypass; this is the constraint
    /// that cannot be.
    #[error("run {run} is sealed as '{outcome}'; a sealed journal accepts no appends")]
    RunSealed { run: String, outcome: String },

    /// A memory item is under legal hold, so the erasure touched nothing.
    ///
    /// Typed rather than a [`Backend`](Self::Backend) string for the reason
    /// every refusal here is: the store is healthy and retrying cannot help —
    /// the hold has to be released by whoever placed it.
    #[error("memory '{id}' is under legal hold, so nothing was erased")]
    UnderLegalHold { id: String },

    /// Another instance holds a *live* lease. Distinct from being fenced: this
    /// writer is not stale, it is simply not the owner yet. The correct response
    /// is to wait for expiry (or for an operator to force a takeover), which is
    /// why it is a separate variant rather than a `Fenced` with placeholder
    /// numbers in it.
    #[error("run {run} is leased by '{owner}' at epoch {epoch} for another {remaining_secs}s")]
    LeaseHeld {
        run: String,
        owner: String,
        epoch: u64,
        remaining_secs: u64,
    },

    /// A renewal presented an `(owner, epoch)` the lease is not currently held
    /// under.
    ///
    /// The one honest reading is *this caller no longer owns the run*: the
    /// lease was released by a clean exit, lapsed and was reclaimed, or was
    /// taken over. Distinct from [`LeaseHeld`](Self::LeaseHeld) — that refuses
    /// a newcomer, this refuses a former owner — and the responses are
    /// opposite: a newcomer waits, a former owner stops. What a renewal must
    /// never do with this state is *claim*: a heartbeat that re-acquires a
    /// released lease holds a run its owner already handed back, and the sweep
    /// then "recovers" a run that concluded cleanly.
    #[error(
        "run {run}: the lease is not held at epoch {epoch} by this owner — it was \
         released, lapsed, or taken over, and a renewal never claims"
    )]
    LeaseNotHeld { run: String, epoch: u64 },

    #[error("corrupt record at seq {seq}: {detail}")]
    Corrupt { seq: Seq, detail: String },

    /// A record was written at a schema version this build does not read.
    ///
    /// **Not corruption, and the difference is what an operator does next.**
    /// The bytes are intact and they hash to what the chain says; what is
    /// missing is a reader that knows the shape. Classified apart from
    /// [`Corrupt`](Self::Corrupt) for the reason
    /// [`KeyError::UnknownFormat`](crate::keyring::KeyError::UnknownFormat) is:
    /// a rolling deploy that put a writer ahead of its readers would otherwise
    /// reach a human as *the history has been altered*, which is the one alarm
    /// that must stay believable.
    ///
    /// It fails closed all the same. A record whose version this build cannot
    /// account for is refused rather than read with the fields it does not know
    /// taking their serde defaults — a false answer to an audit question is
    /// worse than a refusal to answer.
    #[error(
        "record {kind} is v{version} and this build reads v{reads} — the bytes are intact and \
         hash as written; what is missing is a reader that knows the shape. If v{version} is the \
         newer one, deploy readers before writers"
    )]
    UnknownRecordVersion {
        kind: String,
        version: u16,
        reads: u16,
    },

    /// A record at the version this build writes whose shape this build does
    /// not parse.
    ///
    /// **Also a build skew, and before the format freeze it is the only one
    /// that can occur.** A shape change is a hard cut here and a hard cut does
    /// not bump `v` ([`RecordKind::version`](crate::journal::RecordKind::version)
    /// answers 1 for every kind), so the version a reader compares is the same
    /// on both sides of the change and
    /// [`UnknownRecordVersion`](Self::UnknownRecordVersion) never fires. What
    /// reaches a reader instead is a field it has never heard of, or one whose
    /// type moved.
    ///
    /// It is not damage, and the proof is the order the read happens in: the
    /// stored hash is verified before the body is parsed, so a parse that fails
    /// afterwards is a statement about the *reader*. Reporting it as an edited
    /// record would spend the alarm [`Corrupt`](Self::Corrupt) exists to keep.
    #[error(
        "record {kind} is v{version}, the version this build writes, and its shape does not \
         parse here: {detail}. The bytes are intact and hash as written, so another build \
         wrote this journal — run the build that wrote it, or read this history from its \
         export"
    )]
    UnreadableRecordShape {
        kind: String,
        version: u16,
        detail: String,
    },

    #[error(transparent)]
    Encoding(#[from] serde_json::Error),
}

/// A policy bundle as a refusal names it: its digest, or the absence of an
/// engine, which is a bundle of its own and not a blank.
fn bundle_named(digest: Option<&crate::core::Digest>) -> String {
    digest.map_or_else(|| "no policy engine".to_owned(), ToString::to_string)
}

impl RuntimeError {
    /// Lift a store error into the operator-facing taxonomy.
    ///
    /// Two promotions matter, because both change what a human should do:
    ///
    /// * **Fenced** — "I lost ownership of this run" (drop it; another instance
    ///   has it), as opposed to "the database is unhappy" (retry).
    /// * **Corrupt → [`ChainBroken`](Self::ChainBroken)** — the journal does not
    ///   verify. That is never a retryable storage hiccup; it means the history
    ///   has been altered and nothing downstream of it can be trusted. Leaving
    ///   it as a generic store error would bury the one failure that must never
    ///   be shrugged off.
    #[must_use]
    pub fn from_store(e: StoreError) -> Self {
        match e {
            StoreError::Fenced { run, held, current } => Self::Fenced { run, held, current },
            StoreError::LeaseHeld {
                run,
                owner,
                remaining_secs,
                ..
            } => Self::LeaseHeld {
                run,
                owner,
                remaining_secs,
            },
            StoreError::Corrupt { seq, detail } => Self::ChainBroken { seq, detail },
            other => Self::Store(other),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::{Disposition, LISTED_CAPABILITIES, RuntimeError, SkillError, StepError};

    /// An embedder-facing predicate the crate itself never calls, so a wrong
    /// `matches!` arm would be invisible without this. It does not claim to
    /// *be* a control, which is what separates it from
    /// `PolicyError::for_model`; it is still a decision an embedder makes
    /// recovery choices on.
    #[test]
    fn only_a_call_that_never_left_is_safe_to_repeat_on_its_own_terms() {
        assert!(Disposition::DidNotHappen.is_definitely_safe_to_repeat());
        // The two that matter. `InDoubt` is the whole reason this is not a
        // negation of `Landed`: a timed-out payment may well have been taken.
        assert!(!Disposition::InDoubt.is_definitely_safe_to_repeat());
        assert!(!Disposition::Landed.is_definitely_safe_to_repeat());
    }

    // ── How a failure reads ─────────────────────────────────────────────────

    /// `Debug` must render the message, because that is what `main` prints.
    ///
    /// `fn main() -> Result<(), E>` reports through `Debug`, not `Display`, so a
    /// derived `Debug` on these types makes every message in this file
    /// unreachable on the path a newcomer takes first. This is exactly the
    /// deletable-in-silence shape: re-adding `#[derive(Debug)]` compiles, passes
    /// every other test, and quietly returns the crate to printing
    /// `NoProvider("demo.greet")` at the one moment guidance matters most.
    #[test]
    fn a_failure_debugs_as_the_message_it_carries() {
        let e = RuntimeError::NoProvider {
            target: "demo.greet".to_owned(),
            available: vec!["demo.other".to_owned()],
        };
        assert_eq!(format!("{e:?}"), e.to_string());
        assert!(
            !format!("{e:?}").starts_with("NoProvider"),
            "the derived Debug is back: {e:?}"
        );

        // Every type user code holds, not only the one that prompted this.
        let skill = SkillError::Other("boom".to_owned());
        assert_eq!(format!("{skill:?}"), skill.to_string());
        let step = StepError::Encoding(serde_json::from_str::<i32>("x").unwrap_err());
        assert_eq!(format!("{step:?}"), step.to_string());
    }

    /// A refusal names what the plane *does* provide.
    #[test]
    fn an_unknown_capability_is_told_what_exists() {
        let e = RuntimeError::NoProvider {
            target: "demo.greeet".to_owned(),
            available: vec!["demo.greet".to_owned(), "demo.sum".to_owned()],
        };
        let msg = e.to_string();
        assert!(msg.contains("demo.greeet"), "{msg}");
        assert!(msg.contains("demo.greet, demo.sum"), "{msg}");
    }

    /// An empty plane says so, rather than listing nothing and looking complete.
    #[test]
    fn an_empty_plane_says_it_has_no_skills() {
        let e = RuntimeError::NoProvider {
            target: "demo.greet".to_owned(),
            available: Vec::new(),
        };
        let msg = e.to_string();
        assert!(msg.contains("has none at all"), "{msg}");
        assert!(msg.contains("RuntimeBuilder::skill"), "{msg}");
    }

    /// A capped list says how many it did not print.
    ///
    /// A bounded result shaped exactly like a complete one is shape 12, and it
    /// applies to a diagnostic as much as to a worklist: a reader who scans ten
    /// capabilities and does not find theirs must be able to tell "it is not
    /// here" from "the message stopped".
    #[test]
    fn a_capped_capability_list_admits_the_cap() {
        let available: Vec<String> = (0..LISTED_CAPABILITIES + 3)
            .map(|i| format!("cap.{i}"))
            .collect();
        let msg = RuntimeError::NoProvider {
            target: "nope".to_owned(),
            available,
        }
        .to_string();
        assert!(msg.contains("and 3 more"), "{msg}");
    }
}

debug_is_display!(RuntimeError, SkillError, StepError);

/// Log a fault on this plane's side, and return the one sentence a served
/// surface tells its caller about it.
///
/// The detail goes to the operator's log and nowhere else. Rendered to the
/// caller, a store's error names its DSN, a table, a host — facts about this
/// deployment a counterparty or a calling model has no use for and a prober
/// has every use for. One helper for every surface that serves strangers, so
/// no surface words its own and one of them leaks.
#[cfg(any(feature = "a2a-server", feature = "mcp-server"))]
pub(crate) fn withheld_fault(
    surface: &'static str,
    doing: &str,
    error: &dyn std::fmt::Display,
) -> &'static str {
    tracing::error!(
        target: "agentplane::served",
        surface,
        doing,
        %error,
        "a served request failed internally"
    );
    "internal error"
}