khive_runtime/

error_projection.rs

//! Canonical structured projection of typed runtime failures.
//!
//! Transport nesting limits and removal of a nested operation's domain result
//! belong to the transport boundary, not to this lossless shared projection.

use khive_storage::{StorageCapability, StorageError};
use serde_json::{json, Value};

use crate::{DomainDisposition, RuntimeError};

/// Project the original typed error, preserving every structured source field.
///
/// `disposition` describes the enclosing operation. Named runtime refusals keep
/// their existing wire override. This function moves an obligation's result
/// without cloning or recursively serializing it; callers apply their own depth
/// limits before any recursive operation on the returned value.
pub fn runtime_error_value(error: RuntimeError, disposition: DomainDisposition) -> Value {
    // These named outcomes carry their own domain proof. Do not infer general
    // write disposition from a conflict or unavailable variant.
    let receipt_projection = crate::visibility_receipts::receipt_error_projection(&error);
    let named_disposition = match &error {
        _ if receipt_projection.is_some_and(|(_, replay)| replay) => Some("not_committed"),
        // An immutable-stream policy refusal is a definite no-write. It is matched by the
        // shared predicate rather than by a reason string so the two consumers of that
        // predicate and this projection cannot drift into disagreeing about it.
        error if error.is_stream_policy_refusal() => Some("not_committed"),
        // A refusal that wrote a receipt is a definite no-write of the thing it
        // refused: the receipt exists precisely to record that nothing ran.
        RuntimeError::RefusedWithReceipt(_) => Some("not_committed"),
        RuntimeError::Khive(k) => match (k.kind(), k.details().and_then(|d| d.get("reason"))) {
            (
                khive_types::ErrorKind::Internal,
                Some("post_commit_degraded" | "embedding_input_truncated"),
            ) => Some("committed"),
            // These keyed conflicts certify that this request wrote no domain record.
            (
                khive_types::ErrorKind::Conflict,
                Some("key_conflict" | "fence_conflict" | "idempotency_key_conflict"),
            ) => Some("not_committed"),
            (khive_types::ErrorKind::Unavailable, Some("key_holder_unresolved")) => Some("unknown"),
            // ADR-174 A1.1: a stream member refusal carries
            // `domain_disposition: not_committed` wherever it surfaces. In
            // per-member mode it is the member's own value and the runtime
            // writes the field itself; in atomic mode the refusal is raised
            // as the call's error, where without these rows the boundary's
            // `unknown` would stand and the caller could not tell a batch
            // that wrote nothing from one whose outcome is unestablished.
            (khive_types::ErrorKind::Conflict, Some("seq_conflict")) => Some("not_committed"),
            (khive_types::ErrorKind::Conflict, Some("unknown_op")) => Some("not_committed"),
            (khive_types::ErrorKind::Conflict, Some("version_conflict" | "identity_conflict")) => {
                Some("not_committed")
            }
            (khive_types::ErrorKind::Conflict, Some("expired" | "live_until_unreadable")) => {
                Some("not_committed")
            }
            (khive_types::ErrorKind::NotFound, Some("stream_write_not_found")) => {
                Some("not_committed")
            }
            // An exact keyed replay whose holder was hard-deleted before its
            // provenance read: the attempt's unit rolled back on the key claim.
            (khive_types::ErrorKind::NotFound, Some("keyed_replay_holder_missing")) => {
                Some("not_committed")
            }
            (khive_types::ErrorKind::InvalidInput, Some("member_unavailable")) => {
                Some("not_committed")
            }
            _ => None,
        },
        _ => None,
    };
    // The refusal text is the same Display string every consumer already
    // matches on; the receipt fields ride beside it.
    let denial_message =
        matches!(error, RuntimeError::PermissionDenied { .. }).then(|| error.to_string());
    let payload = match error {
        RuntimeError::WithResolution { context } => {
            let crate::error::ResolutionFailureContext { source, resolution } = context;
            let mut value = runtime_error_value(*source, disposition);
            let details = value
                .as_object_mut()
                .expect("runtime error projection is an object")
                .entry("details")
                .or_insert_with(|| json!({}));
            if !details.is_object() {
                *details = json!({});
            }
            details["resolution"] = json!(resolution);
            return value;
        }
        RuntimeError::RefusedWithEvents { context } => {
            let crate::error::RefusalEventContext { source, recordings } = context;
            // Project the source first so named dispositions, code/details,
            // detector location and refusal text retain their original meaning.
            let mut value = runtime_error_value(*source, disposition);
            if !recordings.is_empty() {
                value["refusal_recorded"] = json!(recordings.iter().all(|recording| {
                    matches!(recording, crate::RefusalEventRecording::Recorded { .. })
                }));
                value["refusal_events"] = json!(recordings);
            }
            return value;
        }
        RuntimeError::PermissionDenied {
            verb,
            reason,
            receipt,
        } => json!({
            "kind": "runtime_error",
            "code": "permission_denied",
            "message": denial_message.unwrap_or_default(),
            "verb": verb,
            "reason": reason,
            "audit_event_id": receipt.audit_event_id.map(|id| id.to_string()),
            "audit_outcome": receipt.audit_outcome.wire_code(),
        }),
        RuntimeError::SecretDetected(matched) => {
            let message = RuntimeError::SecretDetected(matched.clone()).to_string();
            json!({
                "kind": "runtime_error",
                "code": "secret_detected",
                "detector": matched.detector,
                "location": matched.location,
                "message": message,
            })
        }
        RuntimeError::RefusedWithReceipt(refusal) => {
            let crate::error::ReceiptRefusal {
                code,
                message,
                receipt_id,
                reason,
                detail,
            } = *refusal;
            // Surface evidence first, contract fields last: a detail member that
            // happens to share a name with a contract field cannot shadow it.
            let mut error = match detail {
                Value::Object(members) => members,
                _ => serde_json::Map::new(),
            };
            error.insert("kind".into(), json!("runtime_error"));
            error.insert("code".into(), json!(code));
            error.insert("message".into(), json!(message));
            error.insert("receipt_id".into(), json!(receipt_id));
            error.insert("reason".into(), json!(reason));
            Value::Object(error)
        }
        RuntimeError::AuditObligation {
            failure,
            domain_result,
        } => {
            let mut error = serde_json::Map::from_iter([
                ("kind".into(), json!("obligation")),
                ("code".into(), json!(failure.wire_code())),
                ("message".into(), json!(failure.to_string())),
            ]);
            error.insert("domain_result".into(), domain_result);
            Value::Object(error)
        }
        RuntimeError::Khive(k) => {
            let mut value = serde_json::to_value(&k)
                .unwrap_or_else(|_| json!({"kind": "internal", "message": k.to_string()}));
            if k.kind() == khive_types::ErrorKind::InvalidInput
                && k.details().and_then(|d| d.get("reason")) == Some("external_id_unverifiable")
            {
                value["code"] = json!("external_id_unverifiable");
            }
            value
        }
        RuntimeError::RemoteFetchError { remote, message } => json!({
            "kind": "remote_fetch_error",
            "remote": remote,
            "message": message,
        }),
        missing @ RuntimeError::NotFound(_) => json!({
            "kind": "not_found",
            "code": null,
            "details": null,
            "message": missing.to_string(),
        }),
        other @ (RuntimeError::Storage(_)
        | RuntimeError::Sqlite(_)
        | RuntimeError::Query(_)
        | RuntimeError::InvalidInput(_)
        | RuntimeError::UnknownVerb(_)
        | RuntimeError::Unconfigured(_)
        | RuntimeError::UnknownModel(_)
        | RuntimeError::Embedding(_)
        | RuntimeError::Ambiguous(_)
        | RuntimeError::Fusion(_)
        | RuntimeError::UnknownFusionStrategy(_)
        | RuntimeError::Internal(_)
        | RuntimeError::IncompatibleEventStore(_)
        | RuntimeError::GuardedWriteFailed(_)
        | RuntimeError::MissingPackDependency(_)
        | RuntimeError::MissingPackDependencies(_)
        | RuntimeError::CircularPackDependency(_)
        | RuntimeError::PackRedeclared { .. }
        | RuntimeError::VerbCollision { .. }
        | RuntimeError::ReservedEnvelopeParam { .. }
        | RuntimeError::GateUnavailable { .. }
        | RuntimeError::NamespaceMismatch { .. }
        | RuntimeError::AmbiguousPrefix { .. }
        | RuntimeError::CrossBackendMergeUnsupported { .. }
        | RuntimeError::UnknownRemote { .. }
        | RuntimeError::RemoteCacheMissing { .. }
        | RuntimeError::AmbiguousId { .. }
        | RuntimeError::CrossNamespaceWrite { .. }
        | RuntimeError::WriteBudgetExceeded { .. }
        | RuntimeError::DeadlineExceeded { .. }) => {
            if let Some(capacity) = sqlite_capacity_failure(&other) {
                let mut value = capacity.into_value(other.to_string());
                if let Some(context) = other.writer_task_failure_context() {
                    value["request_state"] = json!(context.request_state.to_string());
                    value["task_terminated"] = json!(context.task_terminated);
                }
                value
            } else if let Some(context) = other.writer_task_failure_context() {
                json!({"kind":"storage", "code":context.stage, "stage":context.stage,
                    "message":other.to_string(), "retryable":context.retryable,
                    "request_state":context.request_state.to_string(), "task_terminated":context.task_terminated})
            } else if let Some(context) = other.retryable_failure_context() {
                let timeout_ms = u64::try_from(context.timeout.as_millis()).unwrap_or(u64::MAX);
                let mut value = json!({"kind":"unavailable", "code":context.stage, "stage":context.stage,
                    "message":other.to_string(), "retryable":true, "timeout_ms":timeout_ms,
                    "capability":context.capability.map(storage_capability_wire_name),
                    "operation":context.operation, "scope":context.scope, "retry_after_ms":context.retry_after_ms});
                if let Some(pool_identity) = context.pool_identity {
                    value["pool_identity"] = json!(pool_identity);
                }
                value
            } else {
                json!({"kind":"runtime_error", "message":other.to_string()})
            }
        }
    };
    let mut value = payload;
    if let Some((retryable, _)) = receipt_projection {
        value["retryable"] = json!(retryable);
    }
    value["domain_disposition"] = json!(disposition.as_str());
    if let Some(named) = named_disposition {
        value["domain_disposition"] = json!(named);
    }
    value
}

enum SqliteCapacityFailure {
    Refused {
        volume: String,
        available_bytes: u64,
        reserve_bytes: u64,
        required_headroom_bytes: u64,
    },
    Unavailable {
        phase: khive_storage::CapacityUnavailablePhase,
    },
    NativeFull {
        primary_code: i32,
        extended_code: i32,
    },
}

impl SqliteCapacityFailure {
    fn into_value(self, message: String) -> Value {
        match self {
            Self::Refused {
                volume,
                available_bytes,
                reserve_bytes,
                required_headroom_bytes,
            } => json!({
                "kind": "storage",
                "code": "sqlite_capacity_refused",
                "stage": "sqlite_capacity_refused",
                "message": message,
                "retryable": false,
                "capability": "sql",
                "volume": volume,
                "available_bytes": available_bytes,
                "reserve_bytes": reserve_bytes,
                "required_headroom_bytes": required_headroom_bytes,
            }),
            Self::Unavailable { phase } => json!({
                "kind": "storage",
                "code": "sqlite_capacity_unavailable",
                "stage": "sqlite_capacity_unavailable",
                "message": message,
                "retryable": false,
                "capability": "sql",
                "phase": phase.as_str(),
            }),
            Self::NativeFull {
                primary_code,
                extended_code,
            } => json!({
                "kind": "storage",
                "code": "sqlite_disk_full",
                "stage": "sqlite_disk_full",
                "message": message,
                "retryable": false,
                "capability": "sql",
                "sqlite_primary_code": primary_code,
                "sqlite_extended_code": extended_code,
            }),
        }
    }
}

fn sqlite_capacity_failure(error: &RuntimeError) -> Option<SqliteCapacityFailure> {
    match error {
        RuntimeError::Storage(storage) => sqlite_capacity_failure_from_storage(storage),
        RuntimeError::Sqlite(sqlite) => sqlite_capacity_failure_from_sqlite(sqlite),
        _ => None,
    }
}

fn sqlite_capacity_failure_from_storage(error: &StorageError) -> Option<SqliteCapacityFailure> {
    match error {
        StorageError::CapacityFloor {
            capability: StorageCapability::Sql,
            volume,
            available_bytes,
            floor_bytes,
            required_headroom_bytes,
        } => Some(SqliteCapacityFailure::Refused {
            volume: volume.clone(),
            available_bytes: *available_bytes,
            reserve_bytes: *floor_bytes,
            required_headroom_bytes: *required_headroom_bytes,
        }),
        StorageError::CapacityUnavailable {
            capability: StorageCapability::Sql,
            phase,
            ..
        } => Some(SqliteCapacityFailure::Unavailable { phase: *phase }),
        StorageError::WriterTaskTerminated {
            sqlite_full_codes: Some((primary_code, extended_code)),
            ..
        } if *primary_code == rusqlite::ffi::SQLITE_FULL
            && *extended_code >= 0
            && (*extended_code & 0xff) == *primary_code =>
        {
            Some(SqliteCapacityFailure::NativeFull {
                primary_code: *primary_code,
                extended_code: *extended_code,
            })
        }
        StorageError::WriterTaskRequestFailed { source, .. } => {
            sqlite_capacity_failure_from_storage(source)
        }
        StorageError::Driver { source, .. } => {
            if let Some(sqlite) = source.downcast_ref::<khive_db::SqliteError>() {
                sqlite_capacity_failure_from_sqlite(sqlite)
            } else {
                native_sqlite_full(source.as_ref())
            }
        }
        _ => None,
    }
}

fn sqlite_capacity_failure_from_sqlite(
    error: &khive_db::SqliteError,
) -> Option<SqliteCapacityFailure> {
    match error {
        khive_db::SqliteError::CapacityFloor {
            volume,
            available_bytes,
            floor_bytes,
            required_headroom_bytes,
        } => Some(SqliteCapacityFailure::Refused {
            volume: volume.clone(),
            available_bytes: *available_bytes,
            reserve_bytes: *floor_bytes,
            required_headroom_bytes: *required_headroom_bytes,
        }),
        khive_db::SqliteError::CapacityUnavailable { phase, .. } => {
            Some(SqliteCapacityFailure::Unavailable { phase: *phase })
        }
        khive_db::SqliteError::Rusqlite(sqlite) => native_sqlite_full(sqlite),
        _ => None,
    }
}

fn native_sqlite_full(error: &(dyn std::error::Error + 'static)) -> Option<SqliteCapacityFailure> {
    let mut source = Some(error);
    while let Some(current) = source {
        if let Some(rusqlite::Error::SqliteFailure(code, _)) =
            current.downcast_ref::<rusqlite::Error>()
        {
            if code.code == rusqlite::ErrorCode::DiskFull {
                return Some(SqliteCapacityFailure::NativeFull {
                    primary_code: code.extended_code & 0xff,
                    extended_code: code.extended_code,
                });
            }
        }
        source = current.source();
    }
    None
}

fn storage_capability_wire_name(capability: StorageCapability) -> &'static str {
    match capability {
        StorageCapability::Sql => "sql",
        StorageCapability::Notes => "notes",
        StorageCapability::Entities => "entities",
        StorageCapability::Graph => "graph",
        StorageCapability::Events => "events",
        StorageCapability::Vectors => "vectors",
        StorageCapability::Sparse => "sparse",
        StorageCapability::Text => "text",
        StorageCapability::Blob => "blob",
        StorageCapability::Attachments => "attachments",
    }
}

#[cfg(test)]
mod tests {
    use super::runtime_error_value;
    use crate::{
        AuditObligationFailure, DenialAuditOutcome, DenialReceipt, DomainDisposition, RuntimeError,
    };
    use khive_storage::{CapacityUnavailablePhase, StorageCapability, StorageError};
    use khive_types::{Details, ErrorCode, ErrorDomain, KhiveError};
    use serde_json::json;

    #[test]
    fn capacity_phase_projection_preserves_writer_request_settlement() {
        for phase in [
            CapacityUnavailablePhase::Identity,
            CapacityUnavailablePhase::Lock,
            CapacityUnavailablePhase::Probe,
        ] {
            let value = runtime_error_value(
                RuntimeError::Storage(StorageError::WriterTaskRequestFailed {
                    request_state: khive_storage::WriterTaskRequestState::TransactionRolledBack,
                    source: Box::new(StorageError::CapacityUnavailable {
                        capability: StorageCapability::Sql,
                        phase,
                        message: "unavailable".into(),
                    }),
                }),
                DomainDisposition::Unknown,
            );
            assert_eq!(value["code"], "sqlite_capacity_unavailable");
            assert_eq!(value["phase"], phase.as_str());
            assert_eq!(value["request_state"], "transaction_rolled_back");
            assert_eq!(value["task_terminated"], false);
            assert_eq!(value["retryable"], false);
        }
    }

    #[test]
    fn sqlite_capacity_stages_preserve_typed_evidence_and_do_not_retry() {
        let refused = runtime_error_value(
            RuntimeError::Storage(StorageError::CapacityFloor {
                capability: StorageCapability::Sql,
                volume: "/volume".to_string(),
                available_bytes: 99,
                floor_bytes: 100,
                required_headroom_bytes: 12,
            }),
            DomainDisposition::Unknown,
        );
        assert_eq!(refused["stage"], "sqlite_capacity_refused");
        assert_eq!(refused["available_bytes"], 99);
        assert_eq!(refused["reserve_bytes"], 100);
        assert_eq!(refused["required_headroom_bytes"], 12);
        assert_eq!(refused["retryable"], false);

        let unavailable = runtime_error_value(
            RuntimeError::Storage(StorageError::CapacityUnavailable {
                capability: StorageCapability::Sql,
                phase: CapacityUnavailablePhase::Lock,
                message: "bounded lease timed out".to_string(),
            }),
            DomainDisposition::Unknown,
        );
        assert_eq!(unavailable["stage"], "sqlite_capacity_unavailable");
        assert_eq!(unavailable["phase"], "lock");
        assert_eq!(unavailable["retryable"], false);

        let full = runtime_error_value(
            RuntimeError::Storage(StorageError::driver(
                StorageCapability::Sql,
                "write",
                rusqlite::Error::SqliteFailure(
                    rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_FULL),
                    None,
                ),
            )),
            DomainDisposition::Unknown,
        );
        assert_eq!(full["stage"], "sqlite_disk_full");
        assert_eq!(full["sqlite_primary_code"], rusqlite::ffi::SQLITE_FULL);
        assert_eq!(full["retryable"], false);
    }

    #[test]
    fn unsettled_sqlite_writer_converts_to_side_effects_unknown() {
        for error in [
            khive_db::SqliteError::InheritedWriterTransaction,
            khive_db::SqliteError::WriterSettlementUnknown,
        ] {
            let value = runtime_error_value(RuntimeError::from(error), DomainDisposition::Unknown);
            assert_eq!(value["request_state"], "side_effects_unknown");
            assert_eq!(value["task_terminated"], true);
            assert_eq!(value["retryable"], false);
        }
    }

    #[test]
    fn poisoned_writer_refusal_converts_to_not_started() {
        let value = runtime_error_value(
            RuntimeError::from(khive_db::SqliteError::WriterPoisoned),
            DomainDisposition::Unknown,
        );
        assert_eq!(value["request_state"], "not_started");
        assert_eq!(value["task_terminated"], true);
        assert_eq!(value["retryable"], false);
    }

    #[test]
    fn missing_subject_projection_is_typed_without_matching_error_text() {
        for disposition in [DomainDisposition::Unknown, DomainDisposition::NotCommitted] {
            let error = RuntimeError::NotFound("subject missing".into());
            let message = error.to_string();
            let value = runtime_error_value(error, disposition);
            assert_eq!(value["kind"], "not_found");
            assert_eq!(value.get("code"), Some(&serde_json::Value::Null));
            assert_eq!(value.get("details"), Some(&serde_json::Value::Null));
            assert_eq!(value["message"], message);
            assert_eq!(value["domain_disposition"], disposition.as_str());
        }
        for error in [
            RuntimeError::Internal("not found: damaged store index".into()),
            RuntimeError::InvalidInput("not found: malformed reference".into()),
        ] {
            let value = runtime_error_value(error, DomainDisposition::Unknown);
            assert_eq!(value["kind"], "runtime_error");
        }
    }

    #[test]
    fn projection_preserves_structured_source_fields_and_serialized_bytes() {
        let source = KhiveError::unavailable("append reply missing")
            .with_code(ErrorCode::new(ErrorDomain::Db, 71))
            .with_details(Details::new([
                ("operation", "append"),
                ("driver_phase", "awaiting_reply"),
                ("retry_hint", "do_not_repeat"),
                ("extra", "preserve me\nincluding escapes"),
            ]));
        let mut expected = serde_json::to_value(&source).unwrap();
        expected["domain_disposition"] = json!("unknown");
        let expected_bytes = serde_json::to_vec(&expected).unwrap();
        let actual = runtime_error_value(source.into(), DomainDisposition::Unknown);
        assert_eq!(actual, expected);
        assert_eq!(serde_json::to_vec(&actual).unwrap(), expected_bytes);
    }

    #[test]
    fn embedding_truncation_after_write_projects_committed_disposition() {
        let error = KhiveError::internal("embedding input truncated").with_details(Details::new([
            ("reason", "embedding_input_truncated"),
            ("record_id", "00000000-0000-0000-0000-000000000001"),
            ("committed", "true"),
            ("retryable", "false"),
        ]));
        let value = runtime_error_value(error.into(), DomainDisposition::Unknown);
        assert_eq!(value["domain_disposition"], "committed");
        assert_eq!(
            value["details"]["record_id"],
            "00000000-0000-0000-0000-000000000001"
        );
    }

    #[test]
    fn resolution_wrapper_adds_details_without_changing_source_classification() {
        let source = RuntimeError::Khive(KhiveError::conflict("source conflict").with_details(
            Details::new([("reason", "seq_conflict"), ("extra", "retained")]),
        ));
        let selected = uuid::Uuid::from_u128(1);
        let duplicate = uuid::Uuid::from_u128(2);
        let error = source.with_resolution(crate::error::ResolutionFacts {
            project_id: selected,
            duplicate_anchor_ids: vec![duplicate],
            slug_backfilled: true,
            project_created: false,
            orphaned_project_id: None,
            orphaned_note_count: 0,
        });
        assert!(matches!(error.refusal_source(), RuntimeError::Khive(_)));
        let value = runtime_error_value(error, DomainDisposition::Unknown);
        assert_eq!(value["domain_disposition"], "not_committed");
        assert_eq!(value["details"]["reason"], "seq_conflict");
        assert_eq!(value["details"]["extra"], "retained");
        assert_eq!(
            value["details"]["resolution"]["project_id"],
            selected.to_string()
        );
        assert_eq!(
            value["details"]["resolution"]["duplicate_anchor_ids"],
            json!([duplicate])
        );
    }

    #[test]
    fn resolution_wrapper_projects_remote_fetch_type_and_anchor_facts() {
        let selected = uuid::Uuid::from_u128(3);
        let error = RuntimeError::RemoteFetchError {
            remote: "https://example.com/repo".into(),
            message: "cache repair failed".into(),
        }
        .with_resolution(crate::error::ResolutionFacts {
            project_id: selected,
            duplicate_anchor_ids: vec![],
            slug_backfilled: false,
            project_created: false,
            orphaned_project_id: None,
            orphaned_note_count: 0,
        });
        let value = runtime_error_value(error, DomainDisposition::Unknown);
        assert_eq!(value["kind"], "remote_fetch_error");
        assert_eq!(value["remote"], "https://example.com/repo");
        assert_eq!(value["message"], "cache repair failed");
        assert_eq!(value["domain_disposition"], "unknown");
        assert_eq!(
            value["details"]["resolution"]["project_id"],
            selected.to_string()
        );
        assert_eq!(value["details"]["resolution"]["project_created"], false);
    }

    #[test]
    fn named_refusals_keep_their_override_without_classifying_arbitrary_conflicts() {
        for (reason, expected) in [
            ("seq_conflict", "not_committed"),
            ("fence_conflict", "not_committed"),
            ("arbitrary_conflict", "unknown"),
        ] {
            let source = KhiveError::conflict("same rendered message")
                .with_details(Details::new([("reason", reason), ("extra", "retained")]));
            let value = runtime_error_value(source.into(), DomainDisposition::Unknown);
            assert_eq!(value["domain_disposition"], expected);
            assert_eq!(value["details"]["extra"], "retained");
        }
    }

    #[test]
    fn idempotency_key_conflict_projects_not_committed() {
        let source = KhiveError::conflict("different content under an existing key").with_details(
            Details::new([
                ("reason", "idempotency_key_conflict"),
                ("key", "operation-1"),
                ("existing_id", "holder-1"),
            ]),
        );
        let value = runtime_error_value(source.into(), DomainDisposition::Unknown);
        assert_eq!(value["kind"], "conflict");
        assert_eq!(value["details"]["reason"], "idempotency_key_conflict");
        assert_eq!(
            value["domain_disposition"], "not_committed",
            "keyed refusal must project not_committed"
        );
    }

    #[test]
    fn shared_projection_retains_obligation_result_and_denial_receipt() {
        let domain_result = json!({"rows": [{"id": "recorded", "extra": [null, true, 17]}]});
        let failure = Box::new(AuditObligationFailure::new(
            "stream.append",
            crate::audit_batch::AuditTerminalReason::StoreFailure,
        ));
        let expected = json!({
            "kind": "obligation", "code": failure.wire_code(),
            "message": failure.to_string(), "domain_result": domain_result,
            "domain_disposition": "unknown",
        });
        let projected = runtime_error_value(
            RuntimeError::AuditObligation {
                failure,
                domain_result,
            },
            DomainDisposition::Unknown,
        );
        assert_eq!(projected, expected);
        assert_eq!(
            serde_json::to_vec(&projected).unwrap(),
            serde_json::to_vec(&expected).unwrap()
        );

        let event_id = uuid::Uuid::from_u128(17);
        let denied = RuntimeError::PermissionDenied {
            verb: "stream.append".into(),
            reason: "policy".into(),
            receipt: Box::new(DenialReceipt {
                audit_event_id: Some(event_id),
                audit_outcome: DenialAuditOutcome::Committed,
            }),
        };
        let expected = json!({
            "kind": "runtime_error", "code": "permission_denied", "message": denied.to_string(),
            "verb": "stream.append", "reason": "policy", "audit_event_id": event_id.to_string(),
            "audit_outcome": "committed", "domain_disposition": "not_committed",
        });
        assert_eq!(
            runtime_error_value(denied, DomainDisposition::NotCommitted),
            expected
        );
    }

    /// A refusal that wrote a durable receipt must name it as its own field.
    /// The consumer contract is `error.receipt_id`, not a substring of the
    /// sentence: the sentence is free to be reworded and a regular expression
    /// over it breaks without failing anything.
    #[test]
    fn a_refusal_receipt_id_is_a_field_and_not_only_a_substring_of_the_message() {
        let error = RuntimeError::RefusedWithReceipt(Box::new(crate::error::ReceiptRefusal {
            code: "exec_refused",
            message: "exec.run refused: tool not registered (receipt_id=r-1)".into(),
            receipt_id: "r-1".into(),
            reason: "tool not registered".into(),
            detail: json!({ "effective_max_output_bytes": 65536, "receipt_id": "shadow" }),
        }));

        // Disposition is deliberately the wrong one on the way in: a receipt-bearing
        // refusal establishes its own no-write, so the boundary's guess is overridden.
        let value = runtime_error_value(error, DomainDisposition::Unknown);

        assert_eq!(
            value["receipt_id"], "r-1",
            "a detail member cannot shadow a contract field"
        );
        assert_eq!(value["code"], "exec_refused");
        assert_eq!(value["kind"], "runtime_error");
        assert_eq!(value["domain_disposition"], "not_committed");
        assert_eq!(value["reason"], "tool not registered");
        assert_eq!(value["effective_max_output_bytes"], 65536);
        assert_eq!(
            value["message"], "exec.run refused: tool not registered (receipt_id=r-1)",
            "the existing wording is kept so a reader that parses it today keeps working"
        );
    }

    /// The control for the arm above: a refusal carrying no receipt is a
    /// different error entirely and must not grow a `receipt_id`. Without this,
    /// an implementation that stamped the field unconditionally would pass.
    #[test]
    fn an_ordinary_invalid_input_has_no_receipt_id() {
        let value = runtime_error_value(
            RuntimeError::InvalidInput("exec.run refused: tool not registered".into()),
            DomainDisposition::Unknown,
        );

        assert!(
            value.get("receipt_id").is_none(),
            "an error with no durable receipt must not name one: {value}"
        );
        assert_eq!(value["domain_disposition"], "unknown");
    }

    #[test]
    fn terminal_full_projection_requires_consistent_native_codes() {
        for codes in [
            None,
            Some((5, 5)),
            Some((13, 5)),
            Some((5, 13)),
            Some((13, -243)),
        ] {
            let error = StorageError::WriterTaskTerminated {
                request_state: khive_storage::WriterTaskRequestState::SideEffectsUnknown,
                sqlite_full_codes: codes,
            };
            let value = runtime_error_value(
                crate::RuntimeError::Storage(error),
                crate::DomainDisposition::Unknown,
            );
            assert_eq!(value["stage"], "writer_task_terminated");
            assert!(value.get("sqlite_primary_code").is_none());
            assert_eq!(value["retryable"], false);
        }
        let extended = rusqlite::ffi::SQLITE_FULL | (3 << 8);
        let error = StorageError::WriterTaskTerminated {
            request_state: khive_storage::WriterTaskRequestState::SideEffectsUnknown,
            sqlite_full_codes: Some((rusqlite::ffi::SQLITE_FULL, extended)),
        };
        let value = runtime_error_value(
            crate::RuntimeError::Storage(error),
            crate::DomainDisposition::Unknown,
        );
        assert_eq!(value["stage"], "sqlite_disk_full");
        assert_eq!(value["code"], "sqlite_disk_full");
        assert_eq!(value["sqlite_primary_code"], rusqlite::ffi::SQLITE_FULL);
        assert_eq!(value["sqlite_extended_code"], extended);
        assert_eq!(value["request_state"], "side_effects_unknown");
        assert_eq!(value["task_terminated"], true);
        assert_eq!(value["retryable"], false);
        assert_eq!(value["domain_disposition"], "unknown");
    }

    #[test]
    fn projection_keeps_typed_writer_state_and_capability_spelling() {
        let error = RuntimeError::Storage(khive_storage::StorageError::writer_task_terminated(
            khive_storage::WriterTaskRequestState::SideEffectsUnknown,
        ));
        let value = runtime_error_value(error, DomainDisposition::Unknown);
        assert_eq!(value["request_state"], "side_effects_unknown");
        assert_eq!(value["task_terminated"], true);
        assert_eq!(value["retryable"], false);
        assert_eq!(value["domain_disposition"], "unknown");

        let error = RuntimeError::Storage(khive_storage::StorageError::driver(
            StorageCapability::Sql,
            "append checkout",
            khive_db::SqliteError::WriterPoolCheckoutTimeout {
                timeout: std::time::Duration::from_millis(17),
            },
        ));
        let value = runtime_error_value(error, DomainDisposition::Unknown);
        assert_eq!(value["capability"], "sql");
        assert_eq!(value["timeout_ms"], 17);
        assert_eq!(value["operation"], "append checkout");
        assert_eq!(value["domain_disposition"], "unknown");
    }

    #[test]
    fn refusal_events_project_mixed_recording_without_changing_the_original_secret_error() {
        use crate::{RefusalEventRecording, RefusalRecordingErrorClass};
        let source = || {
            RuntimeError::SecretDetected(crate::secret_gate::SecretMatch {
                detector: "fixture",
                trigger: None,
                masked: "PRIVATE-MASKED-EXCERPT".into(),
                location: Some("atoms[1].properties[0].value".into()),
            })
        };
        let subject_a = uuid::Uuid::from_u128(11);
        let subject_b = uuid::Uuid::from_u128(12);
        let event = uuid::Uuid::from_u128(13);
        let expected = runtime_error_value(source(), DomainDisposition::Unknown);
        let mut actual = runtime_error_value(
            source().with_refusal_events(vec![
                RefusalEventRecording::Recorded {
                    item_index: 1,
                    subject: subject_a,
                    event_id: event,
                },
                RefusalEventRecording::Failed {
                    item_index: 4,
                    subject: subject_b,
                    error_class: RefusalRecordingErrorClass::EventAppendFailed,
                },
            ]),
            DomainDisposition::Unknown,
        );
        assert_eq!(actual["refusal_recorded"], false);
        assert_eq!(
            actual["refusal_events"],
            json!([
                {"item_index":1, "subject":subject_a, "event_id":event},
                {"item_index":4, "subject":subject_b, "error_class":"event_append_failed"},
            ])
        );
        assert!(!actual.to_string().contains("PRIVATE-MASKED-EXCERPT"));
        assert!(actual.get("receipt_id").is_none());
        let object = actual.as_object_mut().unwrap();
        object.remove("refusal_recorded");
        object.remove("refusal_events");
        assert_eq!(actual, expected);
    }

    #[test]
    fn refusal_events_keep_typed_details_and_named_disposition_with_all_recorded() {
        use crate::RefusalEventRecording;
        let source = || {
            RuntimeError::Khive(
                KhiveError::conflict("fixed reason")
                    .with_code(ErrorCode::new(ErrorDomain::Db, 71))
                    .with_details(Details::new([
                        ("reason", "seq_conflict"),
                        ("extra", "unchanged"),
                    ])),
            )
        };
        let mut expected = runtime_error_value(source(), DomainDisposition::Unknown);
        expected["refusal_recorded"] = json!(true);
        expected["refusal_events"] = json!([
            {"item_index":0, "subject":uuid::Uuid::from_u128(1), "event_id":uuid::Uuid::from_u128(2)},
            {"item_index":3, "subject":uuid::Uuid::from_u128(3), "event_id":uuid::Uuid::from_u128(4)},
        ]);
        let actual = runtime_error_value(
            source().with_refusal_events(vec![
                RefusalEventRecording::Recorded {
                    item_index: 0,
                    subject: uuid::Uuid::from_u128(1),
                    event_id: uuid::Uuid::from_u128(2),
                },
                RefusalEventRecording::Recorded {
                    item_index: 3,
                    subject: uuid::Uuid::from_u128(3),
                    event_id: uuid::Uuid::from_u128(4),
                },
            ]),
            DomainDisposition::Unknown,
        );
        assert_eq!(actual, expected);
        assert_eq!(actual["domain_disposition"], "not_committed");
    }

    #[test]
    fn refusal_events_do_not_invent_receipts_for_failed_or_absent_recordings() {
        use crate::{RefusalEventRecording, RefusalRecordingErrorClass};
        for class in [
            RefusalRecordingErrorClass::EventStoreUnavailable,
            RefusalRecordingErrorClass::EventAppendFailed,
        ] {
            let error = RuntimeError::InvalidInput("validation refused".into())
                .with_refusal_events(vec![RefusalEventRecording::Failed {
                    item_index: 0,
                    subject: uuid::Uuid::from_u128(1),
                    error_class: class,
                }]);
            let actual = runtime_error_value(error, DomainDisposition::Unknown);
            assert_eq!(actual["refusal_recorded"], false);
            assert_eq!(actual["refusal_events"][0]["error_class"], class.as_str());
            assert!(actual["refusal_events"][0].get("event_id").is_none());
            assert!(actual.get("receipt_id").is_none());
            assert_eq!(actual["message"], "invalid input: validation refused");
        }
        let actual = runtime_error_value(
            RuntimeError::InvalidInput("validation refused".into()).with_refusal_events(vec![]),
            DomainDisposition::Unknown,
        );
        assert!(actual.get("refusal_recorded").is_none());
        assert!(actual.get("refusal_events").is_none());
    }
}