1use std::fmt;
4use std::time::Duration;
5
6use thiserror::Error;
7use uuid::Uuid;
8
9pub type RuntimeResult<T> = Result<T, RuntimeError>;
11
12#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
14#[serde(rename_all = "snake_case")]
15pub enum DomainDisposition {
16 Committed,
18 NotCommitted,
20 Unknown,
22}
23
24impl DomainDisposition {
25 pub const fn as_str(self) -> &'static str {
27 match self {
28 Self::Committed => "committed",
29 Self::NotCommitted => "not_committed",
30 Self::Unknown => "unknown",
31 }
32 }
33}
34
35#[derive(Debug, Error)]
37#[error("{source}")]
38pub struct DispatchError {
39 #[source]
40 source: RuntimeError,
41 disposition: DomainDisposition,
42}
43
44impl DispatchError {
45 pub(crate) fn new(source: RuntimeError, disposition: DomainDisposition) -> Self {
46 Self {
47 source,
48 disposition,
49 }
50 }
51
52 pub(crate) fn before_dispatch(source: RuntimeError) -> Self {
53 Self::new(source, DomainDisposition::NotCommitted)
54 }
55
56 pub(crate) fn after_handler(source: RuntimeError, domain_succeeded: bool) -> Self {
57 Self::new(
58 source,
59 if domain_succeeded {
60 DomainDisposition::Committed
61 } else {
62 DomainDisposition::Unknown
63 },
64 )
65 }
66
67 pub fn source(&self) -> &RuntimeError {
69 &self.source
70 }
71
72 pub fn disposition(&self) -> DomainDisposition {
74 self.disposition
75 }
76
77 pub fn into_parts(self) -> (RuntimeError, DomainDisposition) {
79 (self.source, self.disposition)
80 }
81
82 pub fn into_source(self) -> RuntimeError {
84 self.source
85 }
86}
87
88#[derive(Debug, Clone, Copy, PartialEq, Eq)]
90pub enum AuditObligationReason {
91 Terminal(crate::audit_batch::AuditTerminalReason),
93 GitDigestReceiptFailure,
95}
96
97#[derive(Debug, Error)]
99#[error("{message}")]
100pub struct AuditObligationFailure {
101 pub reason: AuditObligationReason,
103 pub verb: String,
105 pub message: String,
107 #[source]
108 source: Option<khive_storage::StorageError>,
109}
110
111impl AuditObligationFailure {
112 pub fn new(verb: impl Into<String>, reason: crate::audit_batch::AuditTerminalReason) -> Self {
114 let verb = verb.into();
115 Self {
116 message: format!("audit obligation commit failed for verb {verb:?}: {reason:?}"),
117 verb,
118 reason: AuditObligationReason::Terminal(reason),
119 source: None,
120 }
121 }
122
123 pub(crate) fn from_store(verb: &str, source: khive_storage::StorageError) -> Self {
124 let mut failure = Self::new(verb, crate::audit_batch::AuditTerminalReason::StoreFailure);
125 failure.message = format!("audit obligation commit failed for verb {verb:?}: {source}");
126 failure.source = Some(source);
127 failure
128 }
129
130 pub(crate) fn git_digest_receipt(branch: &'static str) -> Self {
131 Self {
132 reason: AuditObligationReason::GitDigestReceiptFailure,
133 verb: "git.digest".into(),
134 message: format!(
135 "git_digest_receipt_persist_failed: {branch}; git.digest writes may have committed, but no durable success receipt was confirmed; inspect ingest state before retrying"
136 ),
137 source: None,
138 }
139 }
140
141 pub const fn wire_code(&self) -> &'static str {
143 use crate::audit_batch::AuditTerminalReason;
144 match self.reason {
145 AuditObligationReason::Terminal(reason) => match reason {
146 AuditTerminalReason::PreflightRejected => "preflight_rejected",
147 AuditTerminalReason::AdmissionClosed => "admission_closed",
148 AuditTerminalReason::QueueAdmissionExhausted => "queue_admission_exhausted",
149 AuditTerminalReason::AdmissionDeadlineExpired => "admission_deadline_expired",
150 AuditTerminalReason::ResolutionDeadlineExpired => "resolution_deadline_expired",
151 AuditTerminalReason::IdentityConflict => "identity_conflict",
152 AuditTerminalReason::StoreFailure => "store_failure",
153 AuditTerminalReason::RetryExhausted => "retry_exhausted",
154 AuditTerminalReason::IdempotencyUnsupported => "idempotency_unsupported",
155 AuditTerminalReason::DriverPanicked => "driver_panicked",
156 AuditTerminalReason::DriverCancelled => "driver_cancelled",
157 AuditTerminalReason::DriverJoinLost => "driver_join_lost",
158 AuditTerminalReason::DriverExitedInconsistent => "driver_exited_inconsistent",
159 AuditTerminalReason::DriverAppendAbandoned => "driver_append_abandoned",
160 AuditTerminalReason::StoreWedged => "store_wedged",
161 },
162 AuditObligationReason::GitDigestReceiptFailure => "git_digest_receipt_failure",
163 }
164 }
165}
166
167pub const WRITER_POOL_CHECKOUT_TIMEOUT_STAGE: &str = "writer_pool_checkout_timeout";
170
171pub use khive_db::{SQLITE_WAL_CAPACITY_REFUSED_STAGE, SQLITE_WAL_CAPACITY_UNAVAILABLE_STAGE};
173
174pub const WRITER_QUEUE_SATURATED_STAGE: &str = "writer_queue_saturated";
177
178pub const WRITER_TASK_BEGIN_BUSY_STAGE: &str = "writer_task_begin_busy";
181
182pub const WRITER_TASK_REQUEST_FAILED_STAGE: &str = "writer_task_request_failed";
185
186pub const WRITER_TASK_TERMINATED_STAGE: &str = "writer_task_terminated";
189
190pub const STORAGE_ADMISSION_TIMEOUT_STAGE: &str = "storage_admission_timeout";
195
196pub const READ_TX_AGE_EVICTED_STAGE: &str = "read_tx_age_evicted";
207
208pub const WRITER_ADMISSION_SCOPE: &str = "writer_admission";
215
216#[derive(Debug, Clone, PartialEq, Eq)]
223pub struct AdmissionFailureContext {
224 pub stage: &'static str,
228 pub timeout: Duration,
230 pub capability: Option<khive_storage::StorageCapability>,
234 pub operation: Option<String>,
236 pub pool_identity: Option<String>,
238 pub scope: Option<&'static str>,
243 pub retry_after_ms: Option<u64>,
249}
250
251#[derive(Debug, Clone, PartialEq, Eq)]
258pub struct RetryableFailureContext {
259 pub stage: &'static str,
261 pub timeout: Duration,
263 pub capability: Option<khive_storage::StorageCapability>,
265 pub operation: Option<String>,
267 pub pool_identity: Option<String>,
269 pub scope: Option<&'static str>,
271 pub retry_after_ms: Option<u64>,
273}
274
275#[derive(Debug, Clone, Copy, PartialEq, Eq)]
282pub struct WriterTaskFailureContext {
283 pub stage: &'static str,
285 pub request_state: khive_storage::WriterTaskRequestState,
287 pub task_terminated: bool,
289 pub retryable: bool,
291}
292
293impl From<AdmissionFailureContext> for RetryableFailureContext {
294 fn from(context: AdmissionFailureContext) -> Self {
295 Self {
296 stage: context.stage,
297 timeout: context.timeout,
298 capability: context.capability,
299 operation: context.operation,
300 pool_identity: context.pool_identity,
301 scope: context.scope,
302 retry_after_ms: context.retry_after_ms,
303 }
304 }
305}
306
307#[derive(Debug, Clone, Copy, PartialEq, Eq)]
313pub enum ChannelIngestFailureClass {
314 Retryable { reason: &'static str },
316 Permanent { reason: &'static str },
318 Unknown { reason: &'static str },
320}
321
322impl ChannelIngestFailureClass {
323 pub const fn name(self) -> &'static str {
325 match self {
326 Self::Retryable { .. } => "retryable",
327 Self::Permanent { .. } => "permanent",
328 Self::Unknown { .. } => "unknown",
329 }
330 }
331
332 pub const fn reason(self) -> &'static str {
334 match self {
335 Self::Retryable { reason } | Self::Permanent { reason } | Self::Unknown { reason } => {
336 reason
337 }
338 }
339 }
340}
341
342#[derive(Debug, Clone, PartialEq, Eq)]
345pub struct WriterPoolCheckoutTimeoutContext {
346 pub timeout: Duration,
349 pub capability: Option<khive_storage::StorageCapability>,
351 pub operation: Option<String>,
353}
354
355#[derive(Debug, Clone, PartialEq, Eq)]
360pub struct GuardedWriteFailure {
361 pub entry_index: Option<usize>,
364 pub missing_source: Option<Uuid>,
367 pub missing_target: Option<Uuid>,
370}
371
372impl fmt::Display for GuardedWriteFailure {
373 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
374 let mut missing = Vec::new();
375 if let Some(source) = self.missing_source {
376 missing.push(format!("source {source}"));
377 }
378 if let Some(target) = self.missing_target {
379 missing.push(format!("target {target}"));
380 }
381 let missing = if missing.is_empty() {
382 "endpoint(s)".to_string()
383 } else {
384 missing.join(" and ")
385 };
386 match self.entry_index {
387 Some(index) => write!(
388 f,
389 "batch entry {index}: {missing} no longer exist at write time"
390 ),
391 None => write!(f, "{missing} no longer exist at write time"),
392 }
393 }
394}
395
396impl std::error::Error for GuardedWriteFailure {}
397
398#[derive(Debug, Clone, PartialEq, Eq)]
400pub struct MissingPackDependency {
401 pub from: String,
402 pub requires: String,
403}
404
405impl fmt::Display for MissingPackDependency {
406 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
407 write!(
408 f,
409 "pack '{}' requires '{}', but '{}' is not in the loaded pack set",
410 self.from, self.requires, self.requires
411 )
412 }
413}
414
415impl std::error::Error for MissingPackDependency {}
416
417#[derive(Debug, Clone, PartialEq, Eq)]
419pub struct MissingPackDependencies {
420 pub missing: Vec<MissingPackDependency>,
421}
422
423impl fmt::Display for MissingPackDependencies {
424 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
425 let parts: Vec<String> = self.missing.iter().map(ToString::to_string).collect();
426 write!(f, "{}", parts.join("; "))
427 }
428}
429
430impl std::error::Error for MissingPackDependencies {}
431
432#[derive(Debug, Clone, PartialEq, Eq)]
434pub struct CircularPackDependency {
435 pub cycle: Vec<String>,
436}
437
438impl fmt::Display for CircularPackDependency {
439 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
440 write!(
441 f,
442 "circular dependency detected among packs: {}",
443 self.cycle.join(" -> ")
444 )
445 }
446}
447
448impl std::error::Error for CircularPackDependency {}
449
450#[derive(Debug, Clone, Copy, PartialEq, Eq)]
457pub enum DenialAuditOutcome {
458 Committed,
461 NotCommitted(&'static str),
464 NoStore,
466 NotAudited,
469}
470
471impl DenialAuditOutcome {
472 pub fn wire_code(&self) -> String {
475 match self {
476 Self::Committed => "committed".to_string(),
477 Self::NotCommitted(code) => format!("not_committed:{code}"),
478 Self::NoStore => "no_store".to_string(),
479 Self::NotAudited => "not_audited".to_string(),
480 }
481 }
482}
483
484#[derive(Debug, Clone, PartialEq, Eq)]
488pub struct DenialReceipt {
489 pub audit_event_id: Option<uuid::Uuid>,
491 pub audit_outcome: DenialAuditOutcome,
493}
494
495impl DenialReceipt {
496 pub fn not_audited() -> Self {
498 Self {
499 audit_event_id: None,
500 audit_outcome: DenialAuditOutcome::NotAudited,
501 }
502 }
503
504 pub fn no_store() -> Self {
506 Self {
507 audit_event_id: None,
508 audit_outcome: DenialAuditOutcome::NoStore,
509 }
510 }
511}
512
513#[derive(Debug, Clone)]
515pub struct ReceiptRefusal {
516 pub code: &'static str,
518 pub message: String,
520 pub receipt_id: String,
522 pub reason: String,
524 pub detail: serde_json::Value,
526}
527
528#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
530#[serde(rename_all = "snake_case")]
531pub enum RefusalRecordingErrorClass {
532 EventStoreUnavailable,
533 EventAppendFailed,
534}
535
536impl RefusalRecordingErrorClass {
537 pub const fn as_str(self) -> &'static str {
538 match self {
539 Self::EventStoreUnavailable => "event_store_unavailable",
540 Self::EventAppendFailed => "event_append_failed",
541 }
542 }
543}
544
545#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
548#[serde(untagged)]
549pub enum RefusalEventRecording {
550 Recorded {
551 item_index: usize,
552 subject: Uuid,
553 event_id: Uuid,
554 },
555 Failed {
556 item_index: usize,
557 subject: Uuid,
558 error_class: RefusalRecordingErrorClass,
559 },
560}
561
562#[derive(Debug)]
568pub struct RefusalEventContext {
569 pub source: Box<RuntimeError>,
570 pub recordings: Vec<RefusalEventRecording>,
571}
572
573impl std::fmt::Display for RefusalEventContext {
574 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
575 std::fmt::Display::fmt(self.source.as_ref(), formatter)
576 }
577}
578
579impl std::error::Error for RefusalEventContext {
580 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
581 Some(self.source.as_ref())
582 }
583}
584
585#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
587pub struct ResolutionFacts {
588 pub project_id: Uuid,
589 pub duplicate_anchor_ids: Vec<Uuid>,
590 pub slug_backfilled: bool,
591 pub project_created: bool,
592 pub orphaned_project_id: Option<Uuid>,
593 pub orphaned_note_count: u64,
594}
595
596#[derive(Debug)]
598pub struct ResolutionFailureContext {
599 pub source: Box<RuntimeError>,
600 pub resolution: ResolutionFacts,
601}
602
603impl std::fmt::Display for ResolutionFailureContext {
604 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
605 std::fmt::Display::fmt(self.source.as_ref(), formatter)
606 }
607}
608
609impl std::error::Error for ResolutionFailureContext {
610 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
611 Some(self.source.as_ref())
612 }
613}
614
615#[derive(Debug, Error)]
619pub enum RuntimeError {
620 #[error("canonical operation {op_index} refused")]
623 CanonicalAdmission {
624 op_index: usize,
625 #[source]
626 source: Box<RuntimeError>,
627 },
628 #[error("{source}")]
630 LintFix {
631 #[source]
632 source: Box<RuntimeError>,
633 detail: Box<crate::graph_lint::LintFixFailureDetail>,
634 },
635
636 #[error("{failure}")]
638 AuditObligation {
639 #[source]
641 failure: Box<AuditObligationFailure>,
642 domain_result: serde_json::Value,
644 },
645
646 #[error("storage: {0}")]
647 Storage(#[from] khive_storage::StorageError),
648
649 #[error("query: {0}")]
650 Query(#[from] khive_query::QueryError),
651
652 #[error("not found: {0}")]
653 NotFound(String),
654
655 #[error("invalid input: {0}")]
656 InvalidInput(String),
657
658 #[error("invalid input: {0}")]
659 UnknownVerb(String),
660
661 #[error("unconfigured: {0} is not set")]
662 Unconfigured(String),
663
664 #[error("unknown embedding model: {0}")]
665 UnknownModel(String),
666
667 #[error("embedding: {0}")]
668 Embedding(#[from] lattice_embed::EmbedError),
669
670 #[error("ambiguous: {0}")]
671 Ambiguous(String),
672
673 #[error("fusion: {0}")]
674 Fusion(#[from] khive_fusion::FuseError),
675
676 #[error("unknown fusion strategy: {0}")]
680 UnknownFusionStrategy(String),
681
682 #[error("internal: {0}")]
683 Internal(String),
684
685 #[error("audit batch incompatible event store: {0}")]
691 IncompatibleEventStore(String),
692
693 #[error("guarded edge write refused: {0}")]
694 GuardedWriteFailed(GuardedWriteFailure),
695
696 #[error("missing pack dependency: {0}")]
697 MissingPackDependency(MissingPackDependency),
698
699 #[error("missing pack dependencies: {0}")]
700 MissingPackDependencies(MissingPackDependencies),
701
702 #[error("{0}")]
703 CircularPackDependency(CircularPackDependency),
704
705 #[error("pack '{name}' registered twice (indices {first_idx} and {second_idx})")]
706 PackRedeclared {
707 name: String,
708 first_idx: usize,
709 second_idx: usize,
710 },
711
712 #[error(
716 "verb collision: verb {verb:?} declared by both pack {first_pack:?} and pack \
717 {second_pack:?}; rename one handler or use Visibility::Subhandler for internal verbs"
718 )]
719 VerbCollision {
720 verb: String,
721 first_pack: String,
722 second_pack: String,
723 },
724
725 #[error(
727 "pack {pack:?} handler {verb:?} declares request-envelope parameter {param:?}; rename the verb argument"
728 )]
729 ReservedEnvelopeParam {
730 pack: String,
731 verb: String,
732 param: String,
733 },
734
735 #[error("{}", .0.message)]
753 RefusedWithReceipt(Box<ReceiptRefusal>),
754
755 #[error(transparent)]
759 RefusedWithEvents { context: RefusalEventContext },
760
761 #[error(transparent)]
764 WithResolution { context: ResolutionFailureContext },
765
766 #[error("permission denied for verb {verb:?}: {reason}")]
776 PermissionDenied {
777 verb: String,
778 reason: String,
779 receipt: Box<DenialReceipt>,
780 },
781
782 #[error("gate unavailable for verb {verb:?}: {reason}")]
788 GateUnavailable { verb: String, reason: String },
789
790 #[error("{0}")]
794 Khive(khive_types::KhiveError),
795
796 #[error("not found in this namespace")]
801 NamespaceMismatch { id: uuid::Uuid },
802
803 #[error("ambiguous prefix {prefix:?}: matches {}", format_uuid_list(matches))]
809 AmbiguousPrefix {
810 prefix: String,
811 matches: Vec<uuid::Uuid>,
812 },
813
814 #[error(
819 "cross-backend merge is not supported: \
820 into_id {into_id} is on backend '{into_backend}', \
821 from_id {from_id} is on backend '{from_backend}'. \
822 Both entities must be on the same backend to merge."
823 )]
824 CrossBackendMergeUnsupported {
825 into_id: uuid::Uuid,
826 from_id: uuid::Uuid,
827 into_backend: String,
828 from_backend: String,
829 },
830
831 #[error("unknown remote: {name:?}")]
834 UnknownRemote { name: String },
835
836 #[error("remote cache missing for remote={remote:?} namespace={namespace:?}")]
838 RemoteCacheMissing { remote: String, namespace: String },
839
840 #[error("ambiguous id {id:?}: matched {count} records")]
842 AmbiguousId { id: String, count: usize },
843
844 #[error("cross-namespace write denied: cannot write to remote namespace {namespace:?}")]
846 CrossNamespaceWrite { namespace: String },
847
848 #[error("remote fetch error for remote={remote:?}: {message}")]
851 RemoteFetchError { remote: String, message: String },
852
853 #[error(
859 "write budget exceeded: max_new_entries={max_new_entries}, \
860 attempted_new_entries={attempted_new_entries}"
861 )]
862 WriteBudgetExceeded {
863 max_new_entries: u64,
864 attempted_new_entries: u64,
865 },
866
867 #[error("write blocked: {0}")]
873 SecretDetected(crate::secret_gate::SecretMatch),
874
875 #[error("{operation} exceeded its {budget_ms}ms deadline (elapsed {elapsed_ms}ms)")]
889 DeadlineExceeded {
890 operation: String,
891 budget_ms: u64,
892 elapsed_ms: u64,
893 },
894}
895
896impl From<khive_db::SqliteError> for RuntimeError {
900 fn from(error: khive_db::SqliteError) -> Self {
901 Self::Storage(match error {
902 khive_db::SqliteError::RequestReadStopped(error) => error,
903 error => error.into_storage_error(khive_storage::StorageCapability::Sql, "runtime"),
904 })
905 }
906}
907
908impl RuntimeError {
909 pub fn with_resolution(self, resolution: ResolutionFacts) -> Self {
911 Self::WithResolution {
912 context: ResolutionFailureContext {
913 source: Box::new(self),
914 resolution,
915 },
916 }
917 }
918
919 pub fn with_refusal_events(self, mut recordings: Vec<RefusalEventRecording>) -> Self {
922 if recordings.is_empty() {
923 return self;
924 }
925 match self {
926 Self::RefusedWithEvents { mut context } => {
927 context.recordings.append(&mut recordings);
928 Self::RefusedWithEvents { context }
929 }
930 source => Self::RefusedWithEvents {
931 context: RefusalEventContext {
932 source: Box::new(source),
933 recordings,
934 },
935 },
936 }
937 }
938
939 pub fn refusal_source(&self) -> &Self {
942 let mut error = self;
943 loop {
944 error = match error {
945 Self::RefusedWithEvents { context } => &context.source,
946 Self::WithResolution { context } => context.source.as_ref(),
947 Self::LintFix { source, .. } => source.as_ref(),
948 Self::CanonicalAdmission { source, .. } => source.as_ref(),
949 _ => return error,
950 };
951 }
952 }
953
954 pub fn is_stream_policy_refusal(&self) -> bool {
961 matches!(self.refusal_source(), Self::Khive(error)
962 if error.kind() == khive_types::ErrorKind::Conflict
963 && error.details().and_then(|details| details.get("reason"))
964 == Some("stream_member"))
965 }
966
967 pub fn permission_denied(verb: impl Into<String>, reason: impl Into<String>) -> Self {
969 Self::PermissionDenied {
970 verb: verb.into(),
971 reason: reason.into(),
972 receipt: Box::new(DenialReceipt::not_audited()),
973 }
974 }
975
976 pub fn internal_with_context(context: impl fmt::Display, source: impl fmt::Display) -> Self {
981 Self::Internal(format!("{context}: {source}"))
982 }
983
984 pub fn channel_ingest_failure_class(&self) -> ChannelIngestFailureClass {
992 let source = self.refusal_source();
993 let reason = source.variant_name();
994 if source.retryable_failure_context().is_some() {
995 ChannelIngestFailureClass::Retryable { reason }
996 } else if matches!(source, Self::SecretDetected(_)) {
997 ChannelIngestFailureClass::Permanent { reason }
998 } else {
999 ChannelIngestFailureClass::Unknown { reason }
1000 }
1001 }
1002
1003 fn variant_name(&self) -> &'static str {
1005 match self {
1006 Self::AuditObligation { .. } => "AuditObligation",
1007 Self::Storage(_) => "Storage",
1008 Self::Query(_) => "Query",
1009 Self::NotFound(_) => "NotFound",
1010 Self::InvalidInput(_) => "InvalidInput",
1011 Self::UnknownVerb(_) => "UnknownVerb",
1012 Self::Unconfigured(_) => "Unconfigured",
1013 Self::UnknownModel(_) => "UnknownModel",
1014 Self::Embedding(_) => "Embedding",
1015 Self::Ambiguous(_) => "Ambiguous",
1016 Self::Fusion(_) => "Fusion",
1017 Self::UnknownFusionStrategy(_) => "UnknownFusionStrategy",
1018 Self::Internal(_) => "Internal",
1019 Self::GuardedWriteFailed(_) => "GuardedWriteFailed",
1020 Self::MissingPackDependency(_) => "MissingPackDependency",
1021 Self::MissingPackDependencies(_) => "MissingPackDependencies",
1022 Self::CircularPackDependency(_) => "CircularPackDependency",
1023 Self::PackRedeclared { .. } => "PackRedeclared",
1024 Self::VerbCollision { .. } => "VerbCollision",
1025 Self::ReservedEnvelopeParam { .. } => "ReservedEnvelopeParam",
1026 Self::PermissionDenied { .. } => "PermissionDenied",
1027 Self::GateUnavailable { .. } => "GateUnavailable",
1028 Self::Khive(_) => "Khive",
1029 Self::NamespaceMismatch { .. } => "NamespaceMismatch",
1030 Self::AmbiguousPrefix { .. } => "AmbiguousPrefix",
1031 Self::CrossBackendMergeUnsupported { .. } => "CrossBackendMergeUnsupported",
1032 Self::UnknownRemote { .. } => "UnknownRemote",
1033 Self::RemoteCacheMissing { .. } => "RemoteCacheMissing",
1034 Self::AmbiguousId { .. } => "AmbiguousId",
1035 Self::CrossNamespaceWrite { .. } => "CrossNamespaceWrite",
1036 Self::RemoteFetchError { .. } => "RemoteFetchError",
1037 Self::WriteBudgetExceeded { .. } => "WriteBudgetExceeded",
1038 Self::SecretDetected(_) => "SecretDetected",
1039 Self::DeadlineExceeded { .. } => "DeadlineExceeded",
1040 Self::IncompatibleEventStore(_) => "IncompatibleEventStore",
1041 Self::RefusedWithReceipt { .. } => "RefusedWithReceipt",
1042 Self::RefusedWithEvents { context } => context.source.variant_name(),
1043 Self::WithResolution { context } => context.source.variant_name(),
1044 Self::LintFix { source, .. } => source.variant_name(),
1045 Self::CanonicalAdmission { source, .. } => source.variant_name(),
1046 }
1047 }
1048
1049 pub fn writer_pool_checkout_timeout_context(&self) -> Option<WriterPoolCheckoutTimeoutContext> {
1056 let (sqlite_error, capability, operation) = match self.refusal_source() {
1057 Self::Storage(khive_storage::StorageError::Driver {
1058 capability,
1059 operation,
1060 source,
1061 }) => (
1062 source.downcast_ref::<khive_db::SqliteError>()?,
1063 Some(*capability),
1064 Some(operation.to_string()),
1065 ),
1066 _ => return None,
1067 };
1068
1069 let khive_db::SqliteError::WriterPoolCheckoutTimeout { timeout } = sqlite_error else {
1070 return None;
1071 };
1072 Some(WriterPoolCheckoutTimeoutContext {
1073 timeout: *timeout,
1074 capability,
1075 operation,
1076 })
1077 }
1078
1079 pub fn admission_failure_context(&self) -> Option<AdmissionFailureContext> {
1084 let source = self.refusal_source();
1085 if let Some(context) = self.writer_pool_checkout_timeout_context() {
1086 return Some(AdmissionFailureContext {
1087 stage: WRITER_POOL_CHECKOUT_TIMEOUT_STAGE,
1088 timeout: context.timeout,
1089 capability: context.capability,
1090 operation: context.operation,
1091 pool_identity: None,
1092 scope: None,
1093 retry_after_ms: None,
1094 });
1095 }
1096 if let Self::Storage(khive_storage::StorageError::WriteQueueFull { timeout_ms }) = source {
1097 return Some(AdmissionFailureContext {
1098 stage: WRITER_QUEUE_SATURATED_STAGE,
1099 timeout: Duration::from_millis(*timeout_ms),
1100 capability: None,
1101 operation: None,
1102 pool_identity: None,
1103 scope: Some(WRITER_ADMISSION_SCOPE),
1104 retry_after_ms: Some(*timeout_ms),
1105 });
1106 }
1107 if let Self::Storage(khive_storage::StorageError::AdmissionTimeout {
1108 operation,
1109 timeout_ms,
1110 pool_identity,
1111 }) = source
1112 {
1113 return Some(AdmissionFailureContext {
1114 stage: STORAGE_ADMISSION_TIMEOUT_STAGE,
1115 timeout: Duration::from_millis(*timeout_ms),
1116 capability: None,
1117 operation: Some(operation.to_string()),
1118 pool_identity: pool_identity.clone(),
1119 scope: None,
1120 retry_after_ms: None,
1121 });
1122 }
1123 None
1124 }
1125
1126 pub fn writer_task_failure_context(&self) -> Option<WriterTaskFailureContext> {
1130 let Self::Storage(error) = self.refusal_source() else {
1131 return None;
1132 };
1133 match error.without_sqlite_write_stage() {
1134 khive_storage::StorageError::WriterTaskRequestFailed {
1135 request_state,
1136 source,
1137 } => Some(WriterTaskFailureContext {
1138 stage: WRITER_TASK_REQUEST_FAILED_STAGE,
1139 request_state: *request_state,
1140 task_terminated: false,
1141 retryable: source.is_retryable(),
1142 }),
1143 khive_storage::StorageError::WriterTaskTerminated { request_state, .. } => {
1144 Some(WriterTaskFailureContext {
1145 stage: WRITER_TASK_TERMINATED_STAGE,
1146 request_state: *request_state,
1147 task_terminated: true,
1148 retryable: false,
1149 })
1150 }
1151 _ => None,
1152 }
1153 }
1154
1155 pub fn retryable_failure_context(&self) -> Option<RetryableFailureContext> {
1158 let mut enclosing = self;
1159 loop {
1160 match enclosing {
1161 Self::LintFix { .. } => return None,
1162 Self::RefusedWithEvents { context } => enclosing = &context.source,
1163 Self::WithResolution { context } => enclosing = &context.source,
1164 _ => break,
1165 }
1166 }
1167
1168 let source = self.refusal_source();
1169 if let Some(context) = self.admission_failure_context() {
1170 return Some(context.into());
1171 }
1172 if let Self::Storage(error) = source {
1173 if let khive_storage::StorageError::WriterTaskBusy { timeout_ms } =
1174 error.without_sqlite_write_stage()
1175 {
1176 return Some(RetryableFailureContext {
1177 stage: WRITER_TASK_BEGIN_BUSY_STAGE,
1178 timeout: Duration::from_millis(*timeout_ms),
1179 capability: None,
1180 operation: Some("writer_task_begin".to_string()),
1181 pool_identity: None,
1182 scope: None,
1183 retry_after_ms: None,
1184 });
1185 }
1186 }
1187 if let Self::Storage(khive_storage::StorageError::ReadTransactionAgeEvicted {
1188 operation,
1189 max_age_secs,
1190 }) = source
1191 {
1192 return Some(RetryableFailureContext {
1193 stage: READ_TX_AGE_EVICTED_STAGE,
1194 timeout: Duration::from_secs(*max_age_secs),
1195 capability: Some(khive_storage::StorageCapability::Sql),
1196 operation: Some(operation.to_string()),
1197 pool_identity: None,
1198 scope: None,
1199 retry_after_ms: None,
1200 });
1201 }
1202 if let Self::Storage(
1203 khive_storage::StorageError::ReadTransactionAgeEvictionCleanupFailed {
1204 operation,
1205 max_age_secs,
1206 ..
1207 },
1208 ) = source
1209 {
1210 return Some(RetryableFailureContext {
1211 stage: READ_TX_AGE_EVICTED_STAGE,
1212 timeout: Duration::from_secs(*max_age_secs),
1213 capability: Some(khive_storage::StorageCapability::Sql),
1214 operation: Some(operation.to_string()),
1215 pool_identity: None,
1216 scope: None,
1217 retry_after_ms: None,
1218 });
1219 }
1220 None
1221 }
1222}
1223
1224pub fn fts_text_leg_or_err<T>(
1231 result: Result<Vec<T>, RuntimeError>,
1232 context: &'static str,
1233 query: &str,
1234) -> RuntimeResult<Vec<T>> {
1235 match result {
1236 Ok(hits) => Ok(hits),
1237 Err(RuntimeError::Storage(se)) if se.is_fts5_syntax_error() => {
1238 tracing::warn!(
1239 error = %se,
1240 query = %query,
1241 context,
1242 "FTS text leg failed on a parser syntax error; failing loud (#569)"
1243 );
1244 Err(RuntimeError::InvalidInput(format!(
1245 "{context}: FTS query could not be parsed: {se}"
1246 )))
1247 }
1248 Err(e) => Err(e),
1249 }
1250}
1251
1252fn format_uuid_list(uuids: &[uuid::Uuid]) -> String {
1253 uuids
1254 .iter()
1255 .map(uuid::Uuid::to_string)
1256 .collect::<Vec<_>>()
1257 .join(", ")
1258}
1259
1260impl From<khive_types::EntityTypeError> for RuntimeError {
1264 fn from(e: khive_types::EntityTypeError) -> Self {
1265 Self::InvalidInput(e.to_string())
1266 }
1267}
1268
1269impl From<khive_types::KhiveError> for RuntimeError {
1270 fn from(e: khive_types::KhiveError) -> Self {
1271 Self::Khive(e)
1272 }
1273}
1274
1275#[cfg(test)]
1276#[path = "error_refusal_event_context_tests.rs"]
1277mod refusal_event_context_tests;
1278
1279#[cfg(test)]
1280#[path = "error_stream_policy_refusal_tests.rs"]
1281mod stream_policy_refusal_tests;
1282
1283#[cfg(test)]
1284#[path = "error_channel_ingest_failure_class_tests.rs"]
1285mod channel_ingest_failure_class_tests;