khive_runtime/

error.rs

//! Runtime error types.

use std::fmt;
use std::time::Duration;

use thiserror::Error;
use uuid::Uuid;

/// Convenience alias for `Result<T, RuntimeError>`.
pub type RuntimeResult<T> = Result<T, RuntimeError>;

/// What this dispatch boundary can establish about its own domain operation.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DomainDisposition {
    /// The handler returned its canonical success value before a later failure.
    Committed,
    /// The operation was refused before its handler was invoked.
    NotCommitted,
    /// The handler failed, so its domain effects are not established here.
    Unknown,
}

impl DomainDisposition {
    /// Stable wire spelling, shared by request and daemon envelopes.
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Committed => "committed",
            Self::NotCommitted => "not_committed",
            Self::Unknown => "unknown",
        }
    }
}

/// An unchanged runtime error with provenance from one dispatch boundary.
#[derive(Debug, Error)]
#[error("{source}")]
pub struct DispatchError {
    #[source]
    source: RuntimeError,
    disposition: DomainDisposition,
}

impl DispatchError {
    pub(crate) fn new(source: RuntimeError, disposition: DomainDisposition) -> Self {
        Self {
            source,
            disposition,
        }
    }

    pub(crate) fn before_dispatch(source: RuntimeError) -> Self {
        Self::new(source, DomainDisposition::NotCommitted)
    }

    pub(crate) fn after_handler(source: RuntimeError, domain_succeeded: bool) -> Self {
        Self::new(
            source,
            if domain_succeeded {
                DomainDisposition::Committed
            } else {
                DomainDisposition::Unknown
            },
        )
    }

    /// Borrow the original error without changing its classification.
    pub fn source(&self) -> &RuntimeError {
        &self.source
    }

    /// Read provenance belonging to this dispatch, not to a nested operation.
    pub fn disposition(&self) -> DomainDisposition {
        self.disposition
    }

    /// Separate the original error from its dispatch provenance.
    pub fn into_parts(self) -> (RuntimeError, DomainDisposition) {
        (self.source, self.disposition)
    }

    /// Compatibility path for callers that consume runtime errors directly.
    pub fn into_source(self) -> RuntimeError {
        self.source
    }
}

/// Typed cause of a post-dispatch audit or durable-receipt failure.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AuditObligationReason {
    /// The audit batch reported one of its closed terminal outcomes.
    Terminal(crate::audit_batch::AuditTerminalReason),
    /// Receipt setup failed before submission produced a terminal outcome.
    GitDigestReceiptFailure,
}

/// An obligation failure preserves its typed cause and any storage source.
#[derive(Debug, Error)]
#[error("{message}")]
pub struct AuditObligationFailure {
    /// Machine-readable terminal or receipt-setup cause.
    pub reason: AuditObligationReason,
    /// Domain verb whose post-dispatch obligation failed.
    pub verb: String,
    /// Stable caller-visible explanation; excludes the canonical domain value.
    pub message: String,
    #[source]
    source: Option<khive_storage::StorageError>,
}

impl AuditObligationFailure {
    /// Preserve a terminal audit outcome without parsing its display text.
    pub fn new(verb: impl Into<String>, reason: crate::audit_batch::AuditTerminalReason) -> Self {
        let verb = verb.into();
        Self {
            message: format!("audit obligation commit failed for verb {verb:?}: {reason:?}"),
            verb,
            reason: AuditObligationReason::Terminal(reason),
            source: None,
        }
    }

    pub(crate) fn from_store(verb: &str, source: khive_storage::StorageError) -> Self {
        let mut failure = Self::new(verb, crate::audit_batch::AuditTerminalReason::StoreFailure);
        failure.message = format!("audit obligation commit failed for verb {verb:?}: {source}");
        failure.source = Some(source);
        failure
    }

    pub(crate) fn git_digest_receipt(branch: &'static str) -> Self {
        Self {
            reason: AuditObligationReason::GitDigestReceiptFailure,
            verb: "git.digest".into(),
            message: format!(
                "git_digest_receipt_persist_failed: {branch}; git.digest writes may have committed, but no durable success receipt was confirmed; inspect ingest state before retrying"
            ),
            source: None,
        }
    }

    /// Closed wire-code mapping: a new batch terminal reason requires an arm.
    pub const fn wire_code(&self) -> &'static str {
        use crate::audit_batch::AuditTerminalReason;
        match self.reason {
            AuditObligationReason::Terminal(reason) => match reason {
                AuditTerminalReason::PreflightRejected => "preflight_rejected",
                AuditTerminalReason::AdmissionClosed => "admission_closed",
                AuditTerminalReason::QueueAdmissionExhausted => "queue_admission_exhausted",
                AuditTerminalReason::AdmissionDeadlineExpired => "admission_deadline_expired",
                AuditTerminalReason::ResolutionDeadlineExpired => "resolution_deadline_expired",
                AuditTerminalReason::IdentityConflict => "identity_conflict",
                AuditTerminalReason::StoreFailure => "store_failure",
                AuditTerminalReason::RetryExhausted => "retry_exhausted",
                AuditTerminalReason::IdempotencyUnsupported => "idempotency_unsupported",
                AuditTerminalReason::DriverPanicked => "driver_panicked",
                AuditTerminalReason::DriverCancelled => "driver_cancelled",
                AuditTerminalReason::DriverJoinLost => "driver_join_lost",
                AuditTerminalReason::DriverExitedInconsistent => "driver_exited_inconsistent",
                AuditTerminalReason::DriverAppendAbandoned => "driver_append_abandoned",
                AuditTerminalReason::StoreWedged => "store_wedged",
            },
            AuditObligationReason::GitDigestReceiptFailure => "git_digest_receipt_failure",
        }
    }
}

/// Stable ADR-135 F6 stage and wire code for a finite-wait pooled writer
/// checkout that expires before SQLite executes.
pub const WRITER_POOL_CHECKOUT_TIMEOUT_STAGE: &str = "writer_pool_checkout_timeout";

/// Stable wire code/stage for a bounded write-queue enqueue that never
/// accepted the request within its configured deadline (#1382, #1643).
pub const WRITER_QUEUE_SATURATED_STAGE: &str = "writer_queue_saturated";

/// Stable wire code/stage for SQLite write-lock contention after the writer
/// queue accepted a request but before its operation closure ran.
pub const WRITER_TASK_BEGIN_BUSY_STAGE: &str = "writer_task_begin_busy";

/// Stable wire code/stage for an ordinary writer request whose operation or
/// COMMIT failed after the task proved its transaction was rolled back.
pub const WRITER_TASK_REQUEST_FAILED_STAGE: &str = "writer_task_request_failed";

/// Stable wire code/stage for a request reported by a permanently retired
/// writer seam.
pub const WRITER_TASK_TERMINATED_STAGE: &str = "writer_task_terminated";

/// Stable wire code/stage for a bounded storage-admission wait (a
/// reader/writer handle slot or a pooled reader checkout) that elapsed
/// before anything was acquired. The operation never started, so the
/// failure is safe to retry.
pub const STORAGE_ADMISSION_TIMEOUT_STAGE: &str = "storage_admission_timeout";

/// Stable wire code/stage for a cached read-only transaction proactively
/// rolled back after pinning a WAL snapshot past the configured
/// `read_tx_max_age` bound (#1846). Always safe to retry: the eviction only
/// ever fires on a read-only handle, so no side effect can exist for a
/// retry to duplicate. Shared by both
/// [`khive_storage::StorageError::ReadTransactionAgeEvicted`] (clean
/// rollback) and
/// [`khive_storage::StorageError::ReadTransactionAgeEvictionCleanupFailed`]
/// (rollback denied or failed outright) — the rendered `message` field
/// distinguishes the two; retry behavior is identical for both.
pub const READ_TX_AGE_EVICTED_STAGE: &str = "read_tx_age_evicted";

/// Stable ADR-131:251 `scope` discriminator carried on a
/// [`WRITER_QUEUE_SATURATED_STAGE`] failure — distinguishes write-queue
/// admission saturation from other `unavailable` failure kinds that share
/// the same `retryable: true` shape but are not bounded by the admission
/// deadline (e.g. [`WRITER_POOL_CHECKOUT_TIMEOUT_STAGE`], which has no
/// ADR-131-defined scope and carries `None`).
pub const WRITER_ADMISSION_SCOPE: &str = "writer_admission";

/// Structured context for a pre-execution admission failure: a finite-wait
/// pooled writer checkout timeout, a bounded write-queue enqueue timeout, or
/// a bounded storage-admission wait (reader/writer handle slot or pooled
/// reader checkout). All happen before SQLite executes the request, so all
/// are safe to classify as retryable — the request was never accepted, let
/// alone started.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AdmissionFailureContext {
    /// Stable wire stage/code: one of [`WRITER_POOL_CHECKOUT_TIMEOUT_STAGE`],
    /// [`WRITER_QUEUE_SATURATED_STAGE`], or
    /// [`STORAGE_ADMISSION_TIMEOUT_STAGE`].
    pub stage: &'static str,
    /// The configured deadline that elapsed.
    pub timeout: Duration,
    /// Storage capability the request was scoped to, when known. The write
    /// queue is process-local and unscoped by capability, so this is `None`
    /// for [`WRITER_QUEUE_SATURATED_STAGE`].
    pub capability: Option<khive_storage::StorageCapability>,
    /// Storage operation name, when known.
    pub operation: Option<String>,
    /// Database identity supplied by the pool that refused admission, when known.
    pub pool_identity: Option<String>,
    /// ADR-131:251 `scope` discriminator. `Some(`[`WRITER_ADMISSION_SCOPE`]`)`
    /// for [`WRITER_QUEUE_SATURATED_STAGE`]; `None` for
    /// [`WRITER_POOL_CHECKOUT_TIMEOUT_STAGE`], which ADR-131 does not define a
    /// scope for.
    pub scope: Option<&'static str>,
    /// ADR-131:251 `retry_after_ms` hint — set equal to the admission
    /// deadline actually applied to the rejected operation, so a retrying
    /// caller waits at least one full admission window before retrying.
    /// `Some(timeout_ms)` for [`WRITER_QUEUE_SATURATED_STAGE`]; `None`
    /// otherwise.
    pub retry_after_ms: Option<u64>,
}

/// Structured context for a failure that is proven safe to retry.
///
/// This includes the two pre-admission failures above and writer-task BEGIN
/// contention after queue acceptance but before operation execution. Callers
/// must inspect `scope`: only `writer_admission` means the queue never
/// accepted the operation.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RetryableFailureContext {
    /// Stable wire stage/code.
    pub stage: &'static str,
    /// Configured wait that ended before execution.
    pub timeout: Duration,
    /// Storage capability when the failure is capability-scoped.
    pub capability: Option<khive_storage::StorageCapability>,
    /// Storage operation when known.
    pub operation: Option<String>,
    /// Database identity supplied by the pool that refused admission, when known.
    pub pool_identity: Option<String>,
    /// Admission scope only when the queue never accepted the operation.
    pub scope: Option<&'static str>,
    /// Server backoff hint when a governing contract defines one.
    pub retry_after_ms: Option<u64>,
}

/// Structured writer-request finality recovered from a storage error.
///
/// `request_state` answers whether effects are known absent or ambiguous;
/// `task_terminated` separately answers whether the writer seam remains
/// usable. `retryable` retains the source error's transient policy and must
/// not be inferred from rollback finality alone.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct WriterTaskFailureContext {
    /// Stable MCP code/stage for the error family.
    pub stage: &'static str,
    /// What the writer can prove about this request's effects.
    pub request_state: khive_storage::WriterTaskRequestState,
    /// Whether the writer seam was permanently retired.
    pub task_terminated: bool,
    /// Whether the source failure is transient independently of finality.
    pub retryable: bool,
}

impl From<AdmissionFailureContext> for RetryableFailureContext {
    fn from(context: AdmissionFailureContext) -> Self {
        Self {
            stage: context.stage,
            timeout: context.timeout,
            capability: context.capability,
            operation: context.operation,
            pool_identity: context.pool_identity,
            scope: context.scope,
            retry_after_ms: context.retry_after_ms,
        }
    }
}

/// Typed disposition for a channel message whose `comm.ingest` write failed.
///
/// Every bucket carries the stable [`RuntimeError`] variant name that selected
/// it. Consumers can therefore make retry/quarantine decisions and report a
/// reason without inspecting the error's rendered text.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ChannelIngestFailureClass {
    /// A pre-execution admission failure that is safe to retry indefinitely.
    Retryable { reason: &'static str },
    /// A deterministic policy refusal that should quarantine immediately.
    Permanent { reason: &'static str },
    /// An error with no explicit retry or permanent policy classification.
    Unknown { reason: &'static str },
}

impl ChannelIngestFailureClass {
    /// Stable lowercase bucket name persisted on quarantine notifications.
    pub const fn name(self) -> &'static str {
        match self {
            Self::Retryable { .. } => "retryable",
            Self::Permanent { .. } => "permanent",
            Self::Unknown { .. } => "unknown",
        }
    }

    /// Stable typed error-variant name; never derived from `Display` output.
    pub const fn reason(self) -> &'static str {
        match self {
            Self::Retryable { reason } | Self::Permanent { reason } | Self::Unknown { reason } => {
                reason
            }
        }
    }
}

/// Structured context recovered from either a direct SQLite runtime error or
/// a typed SQLite source preserved inside a storage-driver wrapper.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WriterPoolCheckoutTimeoutContext {
    /// Finite checkout deadline that elapsed before the pooled mutex was
    /// acquired.
    pub timeout: Duration,
    /// Storage capability that wrapped the SQLite source, if one did.
    pub capability: Option<khive_storage::StorageCapability>,
    /// Storage operation that wrapped the SQLite source, if one did.
    pub operation: Option<String>,
}

/// A guarded edge write (`link`/`link_many`) was refused because one or both
/// endpoints no longer existed at write time. Names the exact endpoint(s)
/// missing instead of a generic "source or target" message, and, for a batch
/// write, which entry in the batch failed first.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct GuardedWriteFailure {
    /// Index of the failing entry within a `link_many` batch. `None` for the
    /// singleton `link` path, which has no batch to index into.
    pub entry_index: Option<usize>,
    /// The source endpoint id, present only when it is the (or one of the)
    /// missing endpoint(s).
    pub missing_source: Option<Uuid>,
    /// The target endpoint id, present only when it is the (or one of the)
    /// missing endpoint(s).
    pub missing_target: Option<Uuid>,
}

impl fmt::Display for GuardedWriteFailure {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let mut missing = Vec::new();
        if let Some(source) = self.missing_source {
            missing.push(format!("source {source}"));
        }
        if let Some(target) = self.missing_target {
            missing.push(format!("target {target}"));
        }
        let missing = if missing.is_empty() {
            "endpoint(s)".to_string()
        } else {
            missing.join(" and ")
        };
        match self.entry_index {
            Some(index) => write!(
                f,
                "batch entry {index}: {missing} no longer exist at write time"
            ),
            None => write!(f, "{missing} no longer exist at write time"),
        }
    }
}

impl std::error::Error for GuardedWriteFailure {}

/// A single missing pack dependency.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MissingPackDependency {
    pub from: String,
    pub requires: String,
}

impl fmt::Display for MissingPackDependency {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            f,
            "pack '{}' requires '{}', but '{}' is not in the loaded pack set",
            self.from, self.requires, self.requires
        )
    }
}

impl std::error::Error for MissingPackDependency {}

/// Multiple missing pack dependencies collected into one error.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MissingPackDependencies {
    pub missing: Vec<MissingPackDependency>,
}

impl fmt::Display for MissingPackDependencies {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let parts: Vec<String> = self.missing.iter().map(ToString::to_string).collect();
        write!(f, "{}", parts.join("; "))
    }
}

impl std::error::Error for MissingPackDependencies {}

/// Circular pack dependency detected during topological sort.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CircularPackDependency {
    pub cycle: Vec<String>,
}

impl fmt::Display for CircularPackDependency {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            f,
            "circular dependency detected among packs: {}",
            self.cycle.join(" -> ")
        )
    }
}

impl std::error::Error for CircularPackDependency {}

/// All errors produced by the khive-runtime layer.
///
/// Where the `GateDenied` audit row of a refused dispatch ended up.
///
/// A refusal reaches the caller whatever happens to its audit row; this value
/// says whether the row the caller could cite exists.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DenialAuditOutcome {
    /// The row committed, or an identical row was already present, and
    /// `audit_event_id` on the refusal names it.
    Committed,
    /// The row was built and submitted but did not commit; the audit
    /// obligation wire code says why.
    NotCommitted(&'static str),
    /// No event store is configured, so no row was written.
    NoStore,
    /// The refusal comes from a path that writes no audit row (namespace
    /// authorization, channel policy).
    NotAudited,
}

impl DenialAuditOutcome {
    /// Closed wire spelling: `committed`, `not_committed:<code>`, `no_store`,
    /// `not_audited`.
    pub fn wire_code(&self) -> String {
        match self {
            Self::Committed => "committed".to_string(),
            Self::NotCommitted(code) => format!("not_committed:{code}"),
            Self::NoStore => "no_store".to_string(),
            Self::NotAudited => "not_audited".to_string(),
        }
    }
}

/// What a refused dispatch can cite: the `GateDenied` audit row's id when
/// one committed, and where the row ended up otherwise. Boxed on the error
/// so a refusal does not widen every `Result<_, RuntimeError>` in the tree.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DenialReceipt {
    /// The audit row's event id when `audit_outcome` is `Committed`.
    pub audit_event_id: Option<uuid::Uuid>,
    /// Whether the row the caller could cite exists.
    pub audit_outcome: DenialAuditOutcome,
}

impl DenialReceipt {
    /// The refusal's path writes no audit row.
    pub fn not_audited() -> Self {
        Self {
            audit_event_id: None,
            audit_outcome: DenialAuditOutcome::NotAudited,
        }
    }

    /// No event store is configured, so no row was written.
    pub fn no_store() -> Self {
        Self {
            audit_event_id: None,
            audit_outcome: DenialAuditOutcome::NoStore,
        }
    }
}

/// Payload of [`RuntimeError::RefusedWithReceipt`].
#[derive(Debug, Clone)]
pub struct ReceiptRefusal {
    /// Wire `code`, naming which refusal surface produced this.
    pub code: &'static str,
    /// The refusal's own sentence, unchanged.
    pub message: String,
    /// Id of the durable row the refusal wrote.
    pub receipt_id: String,
    /// Why the call was refused, as the receipt records it.
    pub reason: String,
    /// Surface-specific structured evidence, an object or `Value::Null`.
    pub detail: serde_json::Value,
}

/// Bounded failure evidence; never carries rejected input or a storage error string.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
#[serde(rename_all = "snake_case")]
pub enum RefusalRecordingErrorClass {
    EventStoreUnavailable,
    EventAppendFailed,
}

impl RefusalRecordingErrorClass {
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::EventStoreUnavailable => "event_store_unavailable",
            Self::EventAppendFailed => "event_append_failed",
        }
    }
}

/// Event evidence for one eligible existing target in a refused request.
/// `Failed` means no confirmed event, not proof that an ambiguous append wrote nothing.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
#[serde(untagged)]
pub enum RefusalEventRecording {
    Recorded {
        item_index: usize,
        subject: Uuid,
        event_id: Uuid,
    },
    Failed {
        item_index: usize,
        subject: Uuid,
        error_class: RefusalRecordingErrorClass,
    },
}

/// Original refusal and the independent evidence of recording it.
///
/// The runtime variant transparently forwards this context's error source, which
/// deliberately exposes the contained `RuntimeError`, not its `Box` or its own
/// underlying source. This keeps typed consumers at the original refusal first.
#[derive(Debug)]
pub struct RefusalEventContext {
    pub source: Box<RuntimeError>,
    pub recordings: Vec<RefusalEventRecording>,
}

impl std::fmt::Display for RefusalEventContext {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        std::fmt::Display::fmt(self.source.as_ref(), formatter)
    }
}

impl std::error::Error for RefusalEventContext {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        Some(self.source.as_ref())
    }
}

/// Variants cover storage, query, validation, namespace isolation, and permission failures.
/// Callers should match on `InvalidInput` for bad arguments, `NotFound` for missing records,
/// and `NamespaceMismatch` (reported as not-found) for cross-namespace access attempts.
#[derive(Debug, Error)]
pub enum RuntimeError {
    /// The domain handler succeeded, but its post-dispatch obligation did not.
    #[error("{failure}")]
    AuditObligation {
        /// Typed reason and source of the obligation failure.
        #[source]
        failure: Box<AuditObligationFailure>,
        /// Exact canonical value held after the domain handler succeeded.
        domain_result: serde_json::Value,
    },

    #[error("storage: {0}")]
    Storage(#[from] khive_storage::StorageError),

    #[error("sqlite: {0}")]
    Sqlite(khive_db::SqliteError),

    #[error("query: {0}")]
    Query(#[from] khive_query::QueryError),

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

    #[error("invalid input: {0}")]
    InvalidInput(String),

    #[error("invalid input: {0}")]
    UnknownVerb(String),

    #[error("unconfigured: {0} is not set")]
    Unconfigured(String),

    #[error("unknown embedding model: {0}")]
    UnknownModel(String),

    #[error("embedding: {0}")]
    Embedding(#[from] lattice_embed::EmbedError),

    #[error("ambiguous: {0}")]
    Ambiguous(String),

    #[error("fusion: {0}")]
    Fusion(#[from] khive_fusion::FuseError),

    /// `FusionStrategy::Custom { name, .. }` named a strategy no pack has
    /// registered via `KhiveRuntime::register_fusion_strategy` (ADR-012).
    /// Fails closed — never falls back to RRF or any other default.
    #[error("unknown fusion strategy: {0}")]
    UnknownFusionStrategy(String),

    #[error("internal: {0}")]
    Internal(String),

    /// An `EventStore` was configured via `with_event_store` but does not
    /// implement ADR-133's `preflight_event`/`append_events_idempotent`
    /// pair (`EventStore::supports_idempotent_audit_batch` reports
    /// `false`). Raised at `build()` time rather than left to fail every
    /// audited dispatch silently.
    #[error("audit batch incompatible event store: {0}")]
    IncompatibleEventStore(String),

    #[error("guarded edge write refused: {0}")]
    GuardedWriteFailed(GuardedWriteFailure),

    #[error("missing pack dependency: {0}")]
    MissingPackDependency(MissingPackDependency),

    #[error("missing pack dependencies: {0}")]
    MissingPackDependencies(MissingPackDependencies),

    #[error("{0}")]
    CircularPackDependency(CircularPackDependency),

    #[error("pack '{name}' registered twice (indices {first_idx} and {second_idx})")]
    PackRedeclared {
        name: String,
        first_idx: usize,
        second_idx: usize,
    },

    /// Two packs declared the same `Visibility::Verb` handler name.
    /// `Visibility::Subhandler` entries are pack-prefixed and do not
    /// participate in cross-pack collision checks.
    #[error(
        "verb collision: verb {verb:?} declared by both pack {first_pack:?} and pack \
         {second_pack:?}; rename one handler or use Visibility::Subhandler for internal verbs"
    )]
    VerbCollision {
        verb: String,
        first_pack: String,
        second_pack: String,
    },

    /// A handler advertised a parameter that request parsing owns at the envelope level.
    #[error(
        "pack {pack:?} handler {verb:?} declares request-envelope parameter {param:?}; rename the verb argument"
    )]
    ReservedEnvelopeParam {
        pack: String,
        verb: String,
        param: String,
    },

    /// A handler refused the call and wrote a durable receipt naming the refusal.
    ///
    /// The durable id rides as its own field. Without it a consumer has to find
    /// the id inside the refusal sentence with a regular expression, which makes
    /// the wording of a message part of the contract and breaks silently the
    /// first time someone rewords it. `message` keeps whatever text the refusal
    /// already produced, so a reader that does parse it today keeps working.
    ///
    /// `reason` is the refusal's own explanation, the same value the receipt
    /// stores, so a consumer keeping its own command record does not parse the
    /// sentence for it either. `detail` carries the surface's other structured
    /// evidence (for exec, the resolved output cap); it must be a JSON object or
    /// null, and its members are projected beside the fields above without
    /// overriding them.
    ///
    /// Boxed so the refusal's evidence does not widen every `Result` in the
    /// runtime: the error enum is copied on each `?`, the refusal is rare.
    #[error("{}", .0.message)]
    RefusedWithReceipt(Box<ReceiptRefusal>),

    /// Original typed refusal, with recording evidence independent of the refused write.
    /// Construct through [`Self::with_refusal_events`] so no eligible targets means
    /// an unchanged, unadorned refusal.
    #[error(transparent)]
    RefusedWithEvents { context: RefusalEventContext },

    /// Gate denied this verb invocation.
    ///
    /// Returned by `VerbRegistry::dispatch` when the configured `Gate` returns
    /// `GateDecision::Deny`. The pack is never invoked. The `reason` field
    /// carries the deny message produced by the gate implementation.
    ///
    /// `receipt` carries the id of the `GateDenied` audit row when one
    /// committed, so the caller can cite the refusal, and says whether such
    /// a row exists.
    #[error("permission denied for verb {verb:?}: {reason}")]
    PermissionDenied {
        verb: String,
        reason: String,
        receipt: Box<DenialReceipt>,
    },

    /// The configured gate could not produce an authorization decision.
    ///
    /// This is distinct from [`Self::PermissionDenied`]: the pack or
    /// intercepted operation is never invoked, but the gate did not answer
    /// with an explicit denial.
    #[error("gate unavailable for verb {verb:?}: {reason}")]
    GateUnavailable { verb: String, reason: String },

    /// A structured [`khive_types::KhiveError`] converted into the runtime
    /// layer. The full structured error is preserved so callers can inspect
    /// `kind`, `code`, `details`, and `retry_hint` without information loss.
    #[error("{0}")]
    Khive(khive_types::KhiveError),

    /// Record exists but belongs to a different namespace than the provided token.
    ///
    /// Externally reported as "not found in this namespace" to avoid leaking
    /// cross-namespace existence information (timing-oracle mitigation).
    #[error("not found in this namespace")]
    NamespaceMismatch { id: uuid::Uuid },

    /// A short-prefix lookup matched more than one record.
    ///
    /// `prefix` is the 8+ hex-char prefix supplied by the caller.
    /// `matches` holds the full UUIDs of all matching records (at most 2 are
    /// reported to bound the scan — callers must supply the full UUID to disambiguate).
    #[error("ambiguous prefix {prefix:?}: matches {}", format_uuid_list(matches))]
    AmbiguousPrefix {
        prefix: String,
        matches: Vec<uuid::Uuid>,
    },

    /// Cross-backend `merge_entity` is unsupported in v1.
    ///
    /// Both entities must reside on the same backend. To merge entities on different
    /// backends, manually export `from_id`, delete it, and re-import on `into_id`'s backend.
    #[error(
        "cross-backend merge is not supported: \
         into_id {into_id} is on backend '{into_backend}', \
         from_id {from_id} is on backend '{from_backend}'. \
         Both entities must be on the same backend to merge."
    )]
    CrossBackendMergeUnsupported {
        into_id: uuid::Uuid,
        from_id: uuid::Uuid,
        into_backend: String,
        from_backend: String,
    },

    // ── Remote Resolution and Content-Hash Verification ──────────────────────
    /// A `kg://` ref names a remote not declared in `schema.yaml`.
    #[error("unknown remote: {name:?}")]
    UnknownRemote { name: String },

    /// A remote cache entry is absent and `--fetch` was not requested.
    #[error("remote cache missing for remote={remote:?} namespace={namespace:?}")]
    RemoteCacheMissing { remote: String, namespace: String },

    /// A short ID matches multiple entities in the same namespace or remote cache.
    #[error("ambiguous id {id:?}: matched {count} records")]
    AmbiguousId { id: String, count: usize },

    /// A write operation targeted a remote namespace, which is read-only.
    #[error("cross-namespace write denied: cannot write to remote namespace {namespace:?}")]
    CrossNamespaceWrite { namespace: String },

    /// Remote cache setup or repair failed. Producers redact the source and
    /// sanitize diagnostics before constructing this wire-visible error.
    #[error("remote fetch error for remote={remote:?}: {message}")]
    RemoteFetchError { remote: String, message: String },

    /// A caller-supplied write budget was exceeded during a Compound apply.
    ///
    /// `max_new_entries` is the limit passed by the caller. `attempted_new_entries`
    /// is `consumed + 1`, i.e. the create that would have exceeded the cap.
    /// `None` budget never produces this error (unlimited path).
    #[error(
        "write budget exceeded: max_new_entries={max_new_entries}, \
         attempted_new_entries={attempted_new_entries}"
    )]
    WriteBudgetExceeded {
        max_new_entries: u64,
        attempted_new_entries: u64,
    },

    /// Write blocked: content matches a secret pattern.
    ///
    /// The `SecretMatch` carries the detector name and a masked excerpt
    /// (`first6...Nchars`). The full candidate is never stored in the error.
    /// Store a pointer (env-var name, keychain item) rather than the raw value.
    #[error("write blocked: {0}")]
    SecretDetected(crate::secret_gate::SecretMatch),

    /// A bounded per-operation deadline elapsed before the operation
    /// completed (#889). The operation may still be running in the
    /// background (this is a client-observable timeout, not a cancellation
    /// signal to the underlying work) — callers should treat this as "no
    /// answer within budget", not "the operation failed or was rolled back".
    ///
    /// Distinct from `#836`'s narrower `ann_ready_timeout_ms`, which bounds
    /// only a single cold-miss ANN-build wait inside the recall vector leg
    /// and degrades to an in-band FTS-only result. This variant bounds the
    /// *entire* operation end-to-end and is surfaced as a typed error so a
    /// caller under sustained contention gets a fast, clear answer instead
    /// of hanging until an upstream client-side ceiling (observed at 300s in
    /// production, #889) fires instead.
    #[error("{operation} exceeded its {budget_ms}ms deadline (elapsed {elapsed_ms}ms)")]
    DeadlineExceeded {
        operation: String,
        budget_ms: u64,
        elapsed_ms: u64,
    },
}

impl From<khive_db::SqliteError> for RuntimeError {
    fn from(error: khive_db::SqliteError) -> Self {
        match error {
            khive_db::SqliteError::RequestReadStopped(error) => Self::Storage(error),
            error => Self::Sqlite(error),
        }
    }
}

impl RuntimeError {
    /// Attach only evidence for established, eligible targets. Missing/new targets
    /// contribute no entry, and an empty batch leaves the error variant unchanged.
    pub fn with_refusal_events(self, mut recordings: Vec<RefusalEventRecording>) -> Self {
        if recordings.is_empty() {
            return self;
        }
        match self {
            Self::RefusedWithEvents { mut context } => {
                context.recordings.append(&mut recordings);
                Self::RefusedWithEvents { context }
            }
            source => Self::RefusedWithEvents {
                context: RefusalEventContext {
                    source: Box::new(source),
                    recordings,
                },
            },
        }
    }

    /// Inspect the original typed error without treating event-recording failure
    /// as the failed domain operation or losing its policy/retry classification.
    pub fn refusal_source(&self) -> &Self {
        let mut error = self;
        while let Self::RefusedWithEvents { context } = error {
            error = &context.source;
        }
        error
    }

    /// Whether the immutable stream record policy refused this write.
    ///
    /// Only the structured runtime marker establishes this class. Ordinary
    /// conflicts, caller sequence preconditions, and raw storage failures do
    /// not acquire a refusal classification from their rendered messages.
    /// These membership guards refuse before applying the requested domain write.
    pub fn is_stream_policy_refusal(&self) -> bool {
        matches!(self.refusal_source(), Self::Khive(error)
            if error.kind() == khive_types::ErrorKind::Conflict
                && error.details().and_then(|details| details.get("reason"))
                    == Some("stream_member"))
    }

    /// A gate refusal from a path that writes no audit row.
    pub fn permission_denied(verb: impl Into<String>, reason: impl Into<String>) -> Self {
        Self::PermissionDenied {
            verb: verb.into(),
            reason: reason.into(),
            receipt: Box::new(DenialReceipt::not_audited()),
        }
    }

    /// Classify a failed inbound channel write without inspecting rendered
    /// error text.
    ///
    /// Existing typed safe-retry failures remain retryable. Secret detection is
    /// the first deterministic policy refusal and is permanent. `None` from
    /// [`Self::retryable_failure_context`] is deliberately not interpreted as
    /// permanent: every other variant starts in the bounded `Unknown` bucket.
    pub fn channel_ingest_failure_class(&self) -> ChannelIngestFailureClass {
        let source = self.refusal_source();
        let reason = source.variant_name();
        if source.retryable_failure_context().is_some() {
            ChannelIngestFailureClass::Retryable { reason }
        } else if matches!(source, Self::SecretDetected(_)) {
            ChannelIngestFailureClass::Permanent { reason }
        } else {
            ChannelIngestFailureClass::Unknown { reason }
        }
    }

    /// Stable top-level variant name used by typed policy classifiers.
    fn variant_name(&self) -> &'static str {
        match self {
            Self::AuditObligation { .. } => "AuditObligation",
            Self::Storage(_) => "Storage",
            Self::Sqlite(_) => "Sqlite",
            Self::Query(_) => "Query",
            Self::NotFound(_) => "NotFound",
            Self::InvalidInput(_) => "InvalidInput",
            Self::UnknownVerb(_) => "UnknownVerb",
            Self::Unconfigured(_) => "Unconfigured",
            Self::UnknownModel(_) => "UnknownModel",
            Self::Embedding(_) => "Embedding",
            Self::Ambiguous(_) => "Ambiguous",
            Self::Fusion(_) => "Fusion",
            Self::UnknownFusionStrategy(_) => "UnknownFusionStrategy",
            Self::Internal(_) => "Internal",
            Self::GuardedWriteFailed(_) => "GuardedWriteFailed",
            Self::MissingPackDependency(_) => "MissingPackDependency",
            Self::MissingPackDependencies(_) => "MissingPackDependencies",
            Self::CircularPackDependency(_) => "CircularPackDependency",
            Self::PackRedeclared { .. } => "PackRedeclared",
            Self::VerbCollision { .. } => "VerbCollision",
            Self::ReservedEnvelopeParam { .. } => "ReservedEnvelopeParam",
            Self::PermissionDenied { .. } => "PermissionDenied",
            Self::GateUnavailable { .. } => "GateUnavailable",
            Self::Khive(_) => "Khive",
            Self::NamespaceMismatch { .. } => "NamespaceMismatch",
            Self::AmbiguousPrefix { .. } => "AmbiguousPrefix",
            Self::CrossBackendMergeUnsupported { .. } => "CrossBackendMergeUnsupported",
            Self::UnknownRemote { .. } => "UnknownRemote",
            Self::RemoteCacheMissing { .. } => "RemoteCacheMissing",
            Self::AmbiguousId { .. } => "AmbiguousId",
            Self::CrossNamespaceWrite { .. } => "CrossNamespaceWrite",
            Self::RemoteFetchError { .. } => "RemoteFetchError",
            Self::WriteBudgetExceeded { .. } => "WriteBudgetExceeded",
            Self::SecretDetected(_) => "SecretDetected",
            Self::DeadlineExceeded { .. } => "DeadlineExceeded",
            Self::IncompatibleEventStore(_) => "IncompatibleEventStore",
            Self::RefusedWithReceipt { .. } => "RefusedWithReceipt",
            Self::RefusedWithEvents { context } => context.source.variant_name(),
        }
    }

    /// Recover a finite-wait pool-checkout timeout without inspecting rendered
    /// error text.
    ///
    /// Store implementations retain [`khive_db::SqliteError`] as the typed
    /// source of `StorageError::Driver`; this method carries that structure
    /// through the runtime wrapper for the MCP wire serializer. A direct
    /// `RuntimeError::Sqlite` follows the same classification path.
    pub fn writer_pool_checkout_timeout_context(&self) -> Option<WriterPoolCheckoutTimeoutContext> {
        let (sqlite_error, capability, operation) = match self.refusal_source() {
            Self::Sqlite(error) => (error, None, None),
            Self::Storage(khive_storage::StorageError::Driver {
                capability,
                operation,
                source,
            }) => (
                source.downcast_ref::<khive_db::SqliteError>()?,
                Some(*capability),
                Some(operation.to_string()),
            ),
            _ => return None,
        };

        let khive_db::SqliteError::WriterPoolCheckoutTimeout { timeout } = sqlite_error else {
            return None;
        };
        Some(WriterPoolCheckoutTimeoutContext {
            timeout: *timeout,
            capability,
            operation,
        })
    }

    /// Recover either pre-execution write-admission failure this process can
    /// produce, by typed variant rather than rendered message text (#1643).
    /// Both are safe to classify as retryable: the request was never
    /// accepted, so no partial side effect can exist to roll back.
    pub fn admission_failure_context(&self) -> Option<AdmissionFailureContext> {
        let source = self.refusal_source();
        if let Some(context) = self.writer_pool_checkout_timeout_context() {
            return Some(AdmissionFailureContext {
                stage: WRITER_POOL_CHECKOUT_TIMEOUT_STAGE,
                timeout: context.timeout,
                capability: context.capability,
                operation: context.operation,
                pool_identity: None,
                scope: None,
                retry_after_ms: None,
            });
        }
        if let Self::Storage(khive_storage::StorageError::WriteQueueFull { timeout_ms }) = source {
            return Some(AdmissionFailureContext {
                stage: WRITER_QUEUE_SATURATED_STAGE,
                timeout: Duration::from_millis(*timeout_ms),
                capability: None,
                operation: None,
                pool_identity: None,
                scope: Some(WRITER_ADMISSION_SCOPE),
                retry_after_ms: Some(*timeout_ms),
            });
        }
        if let Self::Storage(khive_storage::StorageError::AdmissionTimeout {
            operation,
            timeout_ms,
            pool_identity,
        }) = source
        {
            return Some(AdmissionFailureContext {
                stage: STORAGE_ADMISSION_TIMEOUT_STAGE,
                timeout: Duration::from_millis(*timeout_ms),
                capability: None,
                operation: Some(operation.to_string()),
                pool_identity: pool_identity.clone(),
                scope: None,
                retry_after_ms: None,
            });
        }
        None
    }

    /// Recover writer-request finality without parsing the rendered storage
    /// error. Ordinary proven rollbacks and terminal writer failures are
    /// deliberately distinct even when they carry the same request state.
    pub fn writer_task_failure_context(&self) -> Option<WriterTaskFailureContext> {
        let Self::Storage(error) = self.refusal_source() else {
            return None;
        };
        match error {
            khive_storage::StorageError::WriterTaskRequestFailed {
                request_state,
                source,
            } => Some(WriterTaskFailureContext {
                stage: WRITER_TASK_REQUEST_FAILED_STAGE,
                request_state: *request_state,
                task_terminated: false,
                retryable: source.is_retryable(),
            }),
            khive_storage::StorageError::WriterTaskTerminated { request_state } => {
                Some(WriterTaskFailureContext {
                    stage: WRITER_TASK_TERMINATED_STAGE,
                    request_state: *request_state,
                    task_terminated: true,
                    retryable: false,
                })
            }
            _ => None,
        }
    }

    /// Recover every typed failure for which this process can prove that
    /// retrying the one failed operation cannot duplicate a side effect.
    pub fn retryable_failure_context(&self) -> Option<RetryableFailureContext> {
        let source = self.refusal_source();
        if let Some(context) = self.admission_failure_context() {
            return Some(context.into());
        }
        if let Self::Storage(khive_storage::StorageError::WriterTaskBusy { timeout_ms }) = source {
            return Some(RetryableFailureContext {
                stage: WRITER_TASK_BEGIN_BUSY_STAGE,
                timeout: Duration::from_millis(*timeout_ms),
                capability: None,
                operation: Some("writer_task_begin".to_string()),
                pool_identity: None,
                scope: None,
                retry_after_ms: None,
            });
        }
        if let Self::Storage(khive_storage::StorageError::ReadTransactionAgeEvicted {
            operation,
            max_age_secs,
        }) = source
        {
            return Some(RetryableFailureContext {
                stage: READ_TX_AGE_EVICTED_STAGE,
                timeout: Duration::from_secs(*max_age_secs),
                capability: Some(khive_storage::StorageCapability::Sql),
                operation: Some(operation.to_string()),
                pool_identity: None,
                scope: None,
                retry_after_ms: None,
            });
        }
        if let Self::Storage(
            khive_storage::StorageError::ReadTransactionAgeEvictionCleanupFailed {
                operation,
                max_age_secs,
                ..
            },
        ) = source
        {
            return Some(RetryableFailureContext {
                stage: READ_TX_AGE_EVICTED_STAGE,
                timeout: Duration::from_secs(*max_age_secs),
                capability: Some(khive_storage::StorageCapability::Sql),
                operation: Some(operation.to_string()),
                pool_identity: None,
                scope: None,
                retry_after_ms: None,
            });
        }
        None
    }
}

/// Resolve an FTS text-leg search result, failing loud on parser syntax
/// errors instead of silently degrading to vector-only fusion.
///
/// A genuine backend outage (pool exhaustion, connection failure, etc.) is
/// NOT a bad query and is returned as-is via the fallthrough `Err(e)` arm;
/// `is_fts5_syntax_error` is the narrow gate that tells the two apart.
pub fn fts_text_leg_or_err<T>(
    result: Result<Vec<T>, RuntimeError>,
    context: &'static str,
    query: &str,
) -> RuntimeResult<Vec<T>> {
    match result {
        Ok(hits) => Ok(hits),
        Err(RuntimeError::Storage(se)) if se.is_fts5_syntax_error() => {
            tracing::warn!(
                error = %se,
                query = %query,
                context,
                "FTS text leg failed on a parser syntax error; failing loud (#569)"
            );
            Err(RuntimeError::InvalidInput(format!(
                "{context}: FTS query could not be parsed: {se}"
            )))
        }
        Err(e) => Err(e),
    }
}

fn format_uuid_list(uuids: &[uuid::Uuid]) -> String {
    uuids
        .iter()
        .map(uuid::Uuid::to_string)
        .collect::<Vec<_>>()
        .join(", ")
}

/// Maps the dependency-light `khive-types` entity-type resolution error onto
/// `RuntimeError::InvalidInput` at the pack boundary: `khive-types` cannot
/// depend on `khive-runtime`, so it cannot produce `RuntimeError` directly.
impl From<khive_types::EntityTypeError> for RuntimeError {
    fn from(e: khive_types::EntityTypeError) -> Self {
        Self::InvalidInput(e.to_string())
    }
}

impl From<khive_types::KhiveError> for RuntimeError {
    fn from(e: khive_types::KhiveError) -> Self {
        Self::Khive(e)
    }
}

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

    fn failed() -> Vec<RefusalEventRecording> {
        vec![RefusalEventRecording::Failed {
            item_index: 2,
            subject: Uuid::from_u128(11),
            error_class: RefusalRecordingErrorClass::EventAppendFailed,
        }]
    }

    #[test]
    fn refusal_events_preserve_the_source_type_display_and_policy_class() {
        let original = RuntimeError::SecretDetected(crate::secret_gate::SecretMatch {
            detector: "fixture",
            trigger: None,
            masked: "never-in-display".into(),
            location: Some("atoms[2].content".into()),
        });
        let message = original.to_string();
        let wrapped = original.with_refusal_events(failed());
        assert_eq!(wrapped.to_string(), message);
        assert!(matches!(
            wrapped.refusal_source(),
            RuntimeError::SecretDetected(_)
        ));
        assert!(std::error::Error::source(&wrapped)
            .unwrap()
            .downcast_ref::<RuntimeError>()
            .is_some_and(|error| matches!(error, RuntimeError::SecretDetected(_))));
        assert_eq!(
            wrapped.channel_ingest_failure_class(),
            ChannelIngestFailureClass::Permanent {
                reason: "SecretDetected"
            }
        );
        assert!(wrapped.retryable_failure_context().is_none());

        let lookalike = RuntimeError::InvalidInput(message).with_refusal_events(failed());
        assert_eq!(
            lookalike.channel_ingest_failure_class(),
            ChannelIngestFailureClass::Unknown {
                reason: "InvalidInput"
            }
        );
        let policy = RuntimeError::Khive(
            khive_types::KhiveError::conflict("immutable")
                .with_details(khive_types::Details::new([("reason", "stream_member")])),
        )
        .with_refusal_events(failed());
        assert!(policy.is_stream_policy_refusal());
    }

    #[test]
    fn refusal_events_preserve_retry_and_writer_finality_contexts() {
        let checkout = RuntimeError::Sqlite(khive_db::SqliteError::WriterPoolCheckoutTimeout {
            timeout: Duration::from_millis(17),
        })
        .with_refusal_events(failed());
        assert_eq!(
            checkout
                .writer_pool_checkout_timeout_context()
                .unwrap()
                .timeout,
            Duration::from_millis(17)
        );
        assert_eq!(
            checkout.admission_failure_context().unwrap().stage,
            WRITER_POOL_CHECKOUT_TIMEOUT_STAGE
        );
        assert_eq!(
            checkout.retryable_failure_context().unwrap().stage,
            WRITER_POOL_CHECKOUT_TIMEOUT_STAGE
        );

        let queued =
            RuntimeError::Storage(khive_storage::StorageError::WriteQueueFull { timeout_ms: 23 })
                .with_refusal_events(failed());
        assert_eq!(
            queued.retryable_failure_context().unwrap().stage,
            WRITER_QUEUE_SATURATED_STAGE
        );
        assert_eq!(
            queued.channel_ingest_failure_class(),
            ChannelIngestFailureClass::Retryable { reason: "Storage" }
        );

        let busy =
            RuntimeError::Storage(khive_storage::StorageError::WriterTaskBusy { timeout_ms: 31 })
                .with_refusal_events(failed());
        assert_eq!(
            busy.retryable_failure_context().unwrap().stage,
            WRITER_TASK_BEGIN_BUSY_STAGE
        );

        let stopped = RuntimeError::Storage(khive_storage::StorageError::WriterTaskTerminated {
            request_state: khive_storage::WriterTaskRequestState::SideEffectsUnknown,
        })
        .with_refusal_events(failed());
        let context = stopped.writer_task_failure_context().unwrap();
        assert_eq!(
            context.request_state,
            khive_storage::WriterTaskRequestState::SideEffectsUnknown
        );
        assert!(context.task_terminated);
        assert!(!context.retryable);
        assert!(stopped.retryable_failure_context().is_none());
    }

    #[test]
    fn refusal_events_preserve_the_original_error_and_its_underlying_source_chain() {
        let original = RuntimeError::Storage(khive_storage::StorageError::driver(
            khive_storage::StorageCapability::Notes,
            "refusal-context-fixture",
            std::io::Error::new(std::io::ErrorKind::PermissionDenied, "fixture refusal"),
        ));
        let message = original.to_string();
        let wrapped = original.with_refusal_events(failed());
        assert_eq!(wrapped.to_string(), message);

        let runtime = std::error::Error::source(&wrapped)
            .unwrap()
            .downcast_ref::<RuntimeError>()
            .expect("the first source must retain the runtime classification");
        assert!(matches!(runtime, RuntimeError::Storage(_)));
        let storage = std::error::Error::source(runtime)
            .unwrap()
            .downcast_ref::<khive_storage::StorageError>()
            .expect("the original runtime error must retain its storage source");
        let driver = std::error::Error::source(storage)
            .unwrap()
            .downcast_ref::<std::io::Error>()
            .expect("the storage source chain must remain traversable");
        assert_eq!(driver.kind(), std::io::ErrorKind::PermissionDenied);
    }

    #[test]
    fn refusal_events_with_no_eligible_targets_do_not_wrap_or_grow_metadata() {
        let error = RuntimeError::InvalidInput("unchanged".into()).with_refusal_events(vec![]);
        assert!(matches!(error, RuntimeError::InvalidInput(ref message) if message == "unchanged"));
        let error = error
            .with_refusal_events(failed())
            .with_refusal_events(vec![RefusalEventRecording::Recorded {
                item_index: 3,
                subject: Uuid::from_u128(12),
                event_id: Uuid::from_u128(13),
            }]);
        let source = std::error::Error::source(&error)
            .unwrap()
            .downcast_ref::<RuntimeError>()
            .expect("appending recordings must not add another error wrapper");
        assert!(matches!(source, RuntimeError::InvalidInput(_)));
        assert!(std::error::Error::source(source).is_none());
        let RuntimeError::RefusedWithEvents { context } = error else {
            panic!("recordings must accompany the original typed source");
        };
        let RefusalEventContext { source, recordings } = context;
        assert!(matches!(*source, RuntimeError::InvalidInput(_)));
        assert_eq!(recordings.len(), 2);
        assert_eq!(recordings[0], failed()[0]);
    }

    #[test]
    fn refusal_recording_classes_serialize_only_the_closed_safe_spellings() {
        for (class, name) in [
            (
                RefusalRecordingErrorClass::EventStoreUnavailable,
                "event_store_unavailable",
            ),
            (
                RefusalRecordingErrorClass::EventAppendFailed,
                "event_append_failed",
            ),
        ] {
            assert_eq!(class.as_str(), name);
            assert_eq!(
                serde_json::to_value(class).unwrap(),
                serde_json::json!(name)
            );
        }
    }
}

#[cfg(test)]
mod stream_policy_refusal_tests {
    use super::RuntimeError;
    use khive_types::{Details, KhiveError};

    #[test]
    fn stream_policy_refusal_uses_structured_kind_and_marker() {
        let policy = RuntimeError::Khive(
            KhiveError::conflict("changed diagnostic wording")
                .with_details(Details::new([("reason", "stream_member")])),
        );
        assert!(policy.is_stream_policy_refusal());

        let wrong_kind = RuntimeError::Khive(
            KhiveError::invalid_input("stream entries are immutable")
                .with_details(Details::new([("reason", "stream_member")])),
        );
        assert!(!wrong_kind.is_stream_policy_refusal());
    }

    #[test]
    fn stream_policy_refusal_excludes_other_conflicts_and_message_lookalikes() {
        for marker in [
            None,
            Some("seq_conflict"),
            Some("key_conflict"),
            Some("version_conflict"),
            Some("fence_conflict"),
            Some("identity_conflict"),
            Some("expired"),
            Some("live_until_unreadable"),
            Some("key_ambiguous"),
            Some("unknown_op"),
            Some("stream_member_extra"),
        ] {
            let mut error = KhiveError::conflict("stream entries are immutable");
            if let Some(marker) = marker {
                error = error.with_details(Details::new([("reason", marker)]));
            }
            assert!(
                !RuntimeError::Khive(error).is_stream_policy_refusal(),
                "unrelated conflict marker {marker:?}"
            );
        }

        let internal = RuntimeError::Internal("conflict: stream entries are immutable".into());
        assert!(!internal.is_stream_policy_refusal());
        let driver = RuntimeError::Storage(khive_storage::StorageError::driver(
            khive_storage::StorageCapability::Sql,
            "inline.execute",
            std::io::Error::other("stream_member"),
        ));
        assert!(!driver.is_stream_policy_refusal());
    }
}

#[cfg(test)]
mod channel_ingest_failure_class_tests {
    use super::{ChannelIngestFailureClass, RuntimeError};
    use crate::secret_gate::SecretMatch;
    use std::time::Duration;

    #[test]
    fn secret_detected_is_permanent_by_typed_variant_not_display_text() {
        let first = RuntimeError::SecretDetected(SecretMatch {
            detector: "fixture",
            trigger: None,
            masked: "first-rendering".to_string(),
            location: None,
        });
        let second = RuntimeError::SecretDetected(SecretMatch {
            detector: "fixture",
            trigger: Some("token"),
            masked: "completely-different-rendering".to_string(),
            location: None,
        });

        assert_ne!(first.to_string(), second.to_string());
        assert_eq!(
            first.channel_ingest_failure_class(),
            ChannelIngestFailureClass::Permanent {
                reason: "SecretDetected"
            }
        );
        assert_eq!(
            second.channel_ingest_failure_class(),
            ChannelIngestFailureClass::Permanent {
                reason: "SecretDetected"
            },
            "rendered error details must not participate in ingest classification"
        );
    }

    #[test]
    fn admission_failures_are_retryable_and_unclassified_errors_are_unknown() {
        let retryable =
            RuntimeError::Storage(khive_storage::StorageError::WriteQueueFull { timeout_ms: 25 });
        assert_eq!(
            retryable.channel_ingest_failure_class(),
            ChannelIngestFailureClass::Retryable { reason: "Storage" }
        );

        let begin_busy =
            RuntimeError::Storage(khive_storage::StorageError::WriterTaskBusy { timeout_ms: 175 });
        let context = begin_busy
            .retryable_failure_context()
            .expect("contended BEGIN must remain typed and retryable");
        assert_eq!(context.stage, "writer_task_begin_busy");
        assert_eq!(context.timeout, Duration::from_millis(175));
        assert_eq!(context.operation.as_deref(), Some("writer_task_begin"));
        assert_eq!(context.scope, None);
        assert_eq!(context.retry_after_ms, None);
        assert_eq!(
            begin_busy.channel_ingest_failure_class(),
            ChannelIngestFailureClass::Retryable { reason: "Storage" }
        );

        let unknown = RuntimeError::InvalidInput(
            "write blocked: SecretDetected text must not affect classification".to_string(),
        );
        assert_eq!(
            unknown.channel_ingest_failure_class(),
            ChannelIngestFailureClass::Unknown {
                reason: "InvalidInput"
            },
            "a rendered message resembling SecretDetected must remain Unknown unless its typed variant is SecretDetected"
        );
    }

    #[test]
    fn writer_task_failure_context_separates_request_finality_from_task_liveness() {
        use khive_storage::{StorageError, WriterTaskRequestState};

        let rolled_back = RuntimeError::Storage(StorageError::WriterTaskRequestFailed {
            request_state: WriterTaskRequestState::TransactionRolledBack,
            source: Box::new(StorageError::Pool {
                operation: "writer_task_commit".into(),
                message: "commit refused".into(),
            }),
        });
        let rolled_back_context = rolled_back
            .writer_task_failure_context()
            .expect("proven rollback must remain typed through RuntimeError");
        assert_eq!(
            rolled_back_context.request_state,
            WriterTaskRequestState::TransactionRolledBack
        );
        assert_eq!(
            rolled_back_context.stage,
            super::WRITER_TASK_REQUEST_FAILED_STAGE
        );
        assert!(!rolled_back_context.task_terminated);
        assert!(rolled_back_context.retryable);

        let unknown = RuntimeError::Storage(StorageError::WriterTaskTerminated {
            request_state: WriterTaskRequestState::SideEffectsUnknown,
        });
        let unknown_context = unknown
            .writer_task_failure_context()
            .expect("ambiguous finality must remain typed through RuntimeError");
        assert_eq!(
            unknown_context.request_state,
            WriterTaskRequestState::SideEffectsUnknown
        );
        assert_eq!(unknown_context.stage, super::WRITER_TASK_TERMINATED_STAGE);
        assert!(unknown_context.task_terminated);
        assert!(!unknown_context.retryable);
    }

    #[test]
    fn storage_admission_timeout_is_a_retryable_admission_failure() {
        let admission = RuntimeError::Storage(khive_storage::StorageError::AdmissionTimeout {
            operation: "sql_bridge.writer_handle".into(),
            timeout_ms: 30_000,
            pool_identity: None,
        });
        let context = admission
            .admission_failure_context()
            .expect("a storage admission timeout happens before the operation starts");
        assert_eq!(context.stage, super::STORAGE_ADMISSION_TIMEOUT_STAGE);
        assert_eq!(context.timeout, Duration::from_millis(30_000));
        assert_eq!(
            context.operation.as_deref(),
            Some("sql_bridge.writer_handle")
        );
        assert_eq!(context.scope, None);
        assert_eq!(context.retry_after_ms, None);
        assert_eq!(
            admission.channel_ingest_failure_class(),
            ChannelIngestFailureClass::Retryable { reason: "Storage" }
        );

        // The general mid-flight timeout stays unclassified: it proves
        // nothing about whether work was in flight when the deadline expired.
        let mid_flight = RuntimeError::Storage(khive_storage::StorageError::Timeout {
            operation: "sql_bridge.reader_open".into(),
        });
        assert!(mid_flight.admission_failure_context().is_none());
        assert!(mid_flight.retryable_failure_context().is_none());
    }

    /// Binds both age-eviction outcomes a cached-reader rollback can produce
    /// (clean rollback vs. failed/denied cleanup) to the same retryable wire
    /// stage: a regression that reverted either branch back to generic
    /// `StorageError::Transaction` would drop this match arm entirely and
    /// fail here, since `Transaction` does have a `retryable_failure_context`
    /// mapping distinct from `super::READ_TX_AGE_EVICTED_STAGE`.
    #[test]
    fn both_read_tx_age_eviction_outcomes_map_to_the_same_retryable_stage() {
        let clean = RuntimeError::Storage(khive_storage::StorageError::ReadTransactionAgeEvicted {
            operation: "query_all".into(),
            max_age_secs: 120,
        });
        let clean_context = clean
            .retryable_failure_context()
            .expect("a clean age eviction must be typed-retryable");
        assert_eq!(clean_context.stage, super::READ_TX_AGE_EVICTED_STAGE);
        assert_eq!(clean_context.timeout, Duration::from_secs(120));
        assert_eq!(clean_context.operation.as_deref(), Some("query_all"));
        assert_eq!(
            clean_context.capability,
            Some(khive_storage::StorageCapability::Sql)
        );

        let cleanup_failed = RuntimeError::Storage(
            khive_storage::StorageError::ReadTransactionAgeEvictionCleanupFailed {
                operation: "query_all".into(),
                max_age_secs: 120,
                message: "rollback failed: disk I/O error".into(),
            },
        );
        let cleanup_failed_context = cleanup_failed
            .retryable_failure_context()
            .expect("a failed cleanup rollback must remain typed-retryable");
        assert_eq!(
            cleanup_failed_context.stage,
            super::READ_TX_AGE_EVICTED_STAGE
        );
        assert_eq!(cleanup_failed_context.timeout, Duration::from_secs(120));
        assert_eq!(
            cleanup_failed_context.operation.as_deref(),
            Some("query_all")
        );
        assert_eq!(
            cleanup_failed_context.capability,
            Some(khive_storage::StorageCapability::Sql)
        );

        assert_ne!(
            clean.to_string(),
            cleanup_failed.to_string(),
            "the rendered message must distinguish a clean eviction from a failed cleanup even \
             though both map to the same wire stage"
        );
        assert!(cleanup_failed.to_string().contains("rollback failed"));
    }
}