khive_runtime/
error_projection.rs

1//! Canonical structured projection of typed runtime failures.
2//!
3//! Transport nesting limits and removal of a nested operation's domain result
4//! belong to the transport boundary, not to this lossless shared projection.
5
6use khive_storage::{StorageCapability, StorageError};
7use serde_json::{json, Value};
8
9use crate::{DomainDisposition, RuntimeError};
10
11/// Project the original typed error, preserving every structured source field.
12///
13/// `disposition` describes the enclosing operation. Named runtime refusals keep
14/// their existing wire override. This function moves an obligation's result
15/// without cloning or recursively serializing it; callers apply their own depth
16/// limits before any recursive operation on the returned value.
17pub fn runtime_error_value(error: RuntimeError, disposition: DomainDisposition) -> Value {
18    // These named outcomes carry their own domain proof. Do not infer general
19    // write disposition from a conflict or unavailable variant.
20    let sqlite_write_failure = match error.refusal_source() {
21        RuntimeError::Storage(error) => error.sqlite_write_failure(),
22        _ => None,
23    };
24    let outcome_has_precedence = error.writer_task_failure_context().is_some_and(|context| {
25        context.task_terminated
26            || context.request_state == khive_storage::WriterTaskRequestState::SideEffectsUnknown
27    });
28    let receipt_projection = crate::visibility_receipts::receipt_error_projection(&error);
29    let named_disposition = match &error {
30        _ if receipt_projection.is_some_and(|(_, replay)| replay) => Some("not_committed"),
31        // An immutable-stream policy refusal is a definite no-write. It is matched by the
32        // shared predicate rather than by a reason string so the two consumers of that
33        // predicate and this projection cannot drift into disagreeing about it.
34        error if error.is_stream_policy_refusal() => Some("not_committed"),
35        // A refusal that wrote a receipt is a definite no-write of the thing it
36        // refused: the receipt exists precisely to record that nothing ran.
37        RuntimeError::RefusedWithReceipt(_) => Some("not_committed"),
38        RuntimeError::Khive(k) => match (k.kind(), k.details().and_then(|d| d.get("reason"))) {
39            (
40                khive_types::ErrorKind::Internal,
41                Some("post_commit_degraded" | "embedding_input_truncated"),
42            ) => Some("committed"),
43            // These keyed conflicts certify that this request wrote no domain record.
44            (
45                khive_types::ErrorKind::Conflict,
46                Some("key_conflict" | "fence_conflict" | "idempotency_key_conflict"),
47            ) => Some("not_committed"),
48            (khive_types::ErrorKind::Unavailable, Some("key_holder_unresolved")) => Some("unknown"),
49            // ADR-174 A1.1: a stream member refusal carries
50            // `domain_disposition: not_committed` wherever it surfaces. In
51            // per-member mode it is the member's own value and the runtime
52            // writes the field itself; in atomic mode the refusal is raised
53            // as the call's error, where without these rows the boundary's
54            // `unknown` would stand and the caller could not tell a batch
55            // that wrote nothing from one whose outcome is unestablished.
56            (khive_types::ErrorKind::Conflict, Some("seq_conflict")) => Some("not_committed"),
57            (khive_types::ErrorKind::Conflict, Some("unknown_op")) => Some("not_committed"),
58            (khive_types::ErrorKind::Conflict, Some("version_conflict" | "identity_conflict")) => {
59                Some("not_committed")
60            }
61            (khive_types::ErrorKind::Conflict, Some("expired" | "live_until_unreadable")) => {
62                Some("not_committed")
63            }
64            (khive_types::ErrorKind::NotFound, Some("stream_write_not_found")) => {
65                Some("not_committed")
66            }
67            // An exact keyed replay whose holder was hard-deleted before its
68            // provenance read: the attempt's unit rolled back on the key claim.
69            (khive_types::ErrorKind::NotFound, Some("keyed_replay_holder_missing")) => {
70                Some("not_committed")
71            }
72            (khive_types::ErrorKind::InvalidInput, Some("member_unavailable")) => {
73                Some("not_committed")
74            }
75            _ => None,
76        },
77        _ => None,
78    };
79    // The refusal text is the same Display string every consumer already
80    // matches on; the receipt fields ride beside it.
81    let denial_message =
82        matches!(error, RuntimeError::PermissionDenied { .. }).then(|| error.to_string());
83    let payload = match error {
84        RuntimeError::CanonicalAdmission { op_index, source } => {
85            // Preparation precedes all domain DML. Keep the original typed
86            // refusal (including secret detector/permission receipt) and add
87            // only the constituent's index; never stringify captured input.
88            let mut value = runtime_error_value(*source, DomainDisposition::NotCommitted);
89            value["op_index"] = json!(op_index);
90            value["domain_disposition"] = json!("not_committed");
91            value["retryable"] = json!(false);
92            return value;
93        }
94        RuntimeError::LintFix { source, detail } => {
95            let mut value = runtime_error_value(*source, DomainDisposition::Unknown);
96            let object = value
97                .as_object_mut()
98                .expect("runtime error projection is an object");
99            let details = object.entry("details").or_insert_with(|| json!({}));
100            if !details.is_object() {
101                *details = json!({});
102            }
103            details["lint_fix"] = json!(detail);
104            // Earlier actions may already have committed. A named refusal of
105            // the last action cannot certify the whole call as not committed.
106            value["domain_disposition"] = json!("unknown");
107            value["retryable"] = json!(false);
108            return value;
109        }
110
111        RuntimeError::WithResolution { context } => {
112            let crate::error::ResolutionFailureContext { source, resolution } = context;
113            let mut value = runtime_error_value(*source, disposition);
114            let details = value
115                .as_object_mut()
116                .expect("runtime error projection is an object")
117                .entry("details")
118                .or_insert_with(|| json!({}));
119            if !details.is_object() {
120                *details = json!({});
121            }
122            details["resolution"] = json!(resolution);
123            return value;
124        }
125        RuntimeError::RefusedWithEvents { context } => {
126            let crate::error::RefusalEventContext { source, recordings } = context;
127            // Project the source first so named dispositions, code/details,
128            // detector location and refusal text retain their original meaning.
129            let mut value = runtime_error_value(*source, disposition);
130            if !recordings.is_empty() {
131                value["refusal_recorded"] = json!(recordings.iter().all(|recording| {
132                    matches!(recording, crate::RefusalEventRecording::Recorded { .. })
133                }));
134                value["refusal_events"] = json!(recordings);
135            }
136            return value;
137        }
138        RuntimeError::PermissionDenied {
139            verb,
140            reason,
141            receipt,
142        } => json!({
143            "kind": "runtime_error",
144            "code": "permission_denied",
145            "message": denial_message.unwrap_or_default(),
146            "verb": verb,
147            "reason": reason,
148            "audit_event_id": receipt.audit_event_id.map(|id| id.to_string()),
149            "audit_outcome": receipt.audit_outcome.wire_code(),
150        }),
151        RuntimeError::SecretDetected(matched) => {
152            let message = RuntimeError::SecretDetected(matched.clone()).to_string();
153            json!({
154                "kind": "runtime_error",
155                "code": "secret_detected",
156                "detector": matched.detector,
157                "location": matched.location,
158                "message": message,
159            })
160        }
161        RuntimeError::RefusedWithReceipt(refusal) => {
162            let crate::error::ReceiptRefusal {
163                code,
164                message,
165                receipt_id,
166                reason,
167                detail,
168            } = *refusal;
169            // Surface evidence first, contract fields last: a detail member that
170            // happens to share a name with a contract field cannot shadow it.
171            let mut error = match detail {
172                Value::Object(members) => members,
173                _ => serde_json::Map::new(),
174            };
175            error.insert("kind".into(), json!("runtime_error"));
176            error.insert("code".into(), json!(code));
177            error.insert("message".into(), json!(message));
178            error.insert("receipt_id".into(), json!(receipt_id));
179            error.insert("reason".into(), json!(reason));
180            Value::Object(error)
181        }
182        RuntimeError::AuditObligation {
183            failure,
184            domain_result,
185        } => {
186            let mut error = serde_json::Map::from_iter([
187                ("kind".into(), json!("obligation")),
188                ("code".into(), json!(failure.wire_code())),
189                ("message".into(), json!(failure.to_string())),
190            ]);
191            error.insert("domain_result".into(), domain_result);
192            Value::Object(error)
193        }
194        RuntimeError::Khive(k) => {
195            let mut value = serde_json::to_value(&k)
196                .unwrap_or_else(|_| json!({"kind": "internal", "message": k.to_string()}));
197            if k.kind() == khive_types::ErrorKind::InvalidInput
198                && k.details().and_then(|d| d.get("reason")) == Some("external_id_unverifiable")
199            {
200                value["code"] = json!("external_id_unverifiable");
201            }
202            value
203        }
204        RuntimeError::RemoteFetchError { remote, message } => json!({
205            "kind": "remote_fetch_error",
206            "remote": remote,
207            "message": message,
208        }),
209        missing @ RuntimeError::NotFound(_) => json!({
210            "kind": "not_found",
211            "code": null,
212            "details": null,
213            "message": missing.to_string(),
214        }),
215        other @ (RuntimeError::Storage(_)
216        | RuntimeError::Query(_)
217        | RuntimeError::InvalidInput(_)
218        | RuntimeError::UnknownVerb(_)
219        | RuntimeError::Unconfigured(_)
220        | RuntimeError::UnknownModel(_)
221        | RuntimeError::Embedding(_)
222        | RuntimeError::Ambiguous(_)
223        | RuntimeError::Fusion(_)
224        | RuntimeError::UnknownFusionStrategy(_)
225        | RuntimeError::Internal(_)
226        | RuntimeError::IncompatibleEventStore(_)
227        | RuntimeError::GuardedWriteFailed(_)
228        | RuntimeError::MissingPackDependency(_)
229        | RuntimeError::MissingPackDependencies(_)
230        | RuntimeError::CircularPackDependency(_)
231        | RuntimeError::PackRedeclared { .. }
232        | RuntimeError::VerbCollision { .. }
233        | RuntimeError::ReservedEnvelopeParam { .. }
234        | RuntimeError::GateUnavailable { .. }
235        | RuntimeError::NamespaceMismatch { .. }
236        | RuntimeError::AmbiguousPrefix { .. }
237        | RuntimeError::CrossBackendMergeUnsupported { .. }
238        | RuntimeError::UnknownRemote { .. }
239        | RuntimeError::RemoteCacheMissing { .. }
240        | RuntimeError::AmbiguousId { .. }
241        | RuntimeError::CrossNamespaceWrite { .. }
242        | RuntimeError::WriteBudgetExceeded { .. }
243        | RuntimeError::DeadlineExceeded { .. }) => {
244            if let Some(capacity) = sqlite_capacity_failure(&other) {
245                let mut value = capacity.into_value(other.to_string());
246                if let Some(context) = other.writer_task_failure_context() {
247                    value["request_state"] = json!(context.request_state.to_string());
248                    value["task_terminated"] = json!(context.task_terminated);
249                }
250                value
251            } else if let Some(context) = other.writer_task_failure_context() {
252                json!({"kind":"storage", "code":context.stage, "stage":context.stage,
253                    "message":other.to_string(), "retryable":context.retryable,
254                    "request_state":context.request_state.to_string(), "task_terminated":context.task_terminated})
255            } else if let Some(context) = other.retryable_failure_context() {
256                let timeout_ms = u64::try_from(context.timeout.as_millis()).unwrap_or(u64::MAX);
257                let mut value = json!({"kind":"unavailable", "code":context.stage, "stage":context.stage,
258                    "message":other.to_string(), "retryable":true, "timeout_ms":timeout_ms,
259                    "capability":context.capability.map(storage_capability_wire_name),
260                    "operation":context.operation, "scope":context.scope, "retry_after_ms":context.retry_after_ms});
261                if let Some(pool_identity) = context.pool_identity {
262                    value["pool_identity"] = json!(pool_identity);
263                }
264                value
265            } else {
266                json!({"kind":"runtime_error", "message":other.to_string()})
267            }
268        }
269    };
270    let mut value = payload;
271    if let Some(failure) = sqlite_write_failure {
272        // Capacity refusal retains its existing caller-visible precedence.
273        if !outcome_has_precedence
274            && !failure.settlement_unknown
275            && failure.primary_code != rusqlite::ffi::SQLITE_FULL
276        {
277            value["code"] = json!(failure.stage.as_str());
278            value["stage"] = json!(failure.stage.as_str());
279            if value["kind"] == "runtime_error" {
280                value["kind"] = json!("storage");
281            }
282        }
283        value["sqlite_write_stage"] = json!(failure.stage.as_str());
284        value["sqlite_primary_code"] = json!(failure.primary_code);
285        value["sqlite_extended_code"] = json!(failure.extended_code);
286    }
287    if let Some((retryable, _)) = receipt_projection {
288        value["retryable"] = json!(retryable);
289    }
290    value["domain_disposition"] = json!(disposition.as_str());
291    if let Some(named) = named_disposition {
292        value["domain_disposition"] = json!(named);
293    }
294    value
295}
296
297enum SqliteCapacityFailure {
298    Refused {
299        volume: String,
300        available_bytes: u64,
301        reserve_bytes: u64,
302        required_headroom_bytes: u64,
303    },
304    Unavailable {
305        phase: khive_storage::CapacityUnavailablePhase,
306    },
307    NativeFull {
308        primary_code: i32,
309        extended_code: i32,
310    },
311}
312
313impl SqliteCapacityFailure {
314    fn into_value(self, message: String) -> Value {
315        match self {
316            Self::Refused {
317                volume,
318                available_bytes,
319                reserve_bytes,
320                required_headroom_bytes,
321            } => json!({
322                "kind": "storage",
323                "code": "sqlite_capacity_refused",
324                "stage": "sqlite_capacity_refused",
325                "message": message,
326                "retryable": false,
327                "capability": "sql",
328                "volume": volume,
329                "available_bytes": available_bytes,
330                "reserve_bytes": reserve_bytes,
331                "required_headroom_bytes": required_headroom_bytes,
332            }),
333            Self::Unavailable { phase } => json!({
334                "kind": "storage",
335                "code": "sqlite_capacity_unavailable",
336                "stage": "sqlite_capacity_unavailable",
337                "message": message,
338                "retryable": phase == khive_storage::CapacityUnavailablePhase::LockContended,
339                "capability": "sql",
340                "phase": phase.as_str(),
341            }),
342            Self::NativeFull {
343                primary_code,
344                extended_code,
345            } => json!({
346                "kind": "storage",
347                "code": "sqlite_disk_full",
348                "stage": "sqlite_disk_full",
349                "message": message,
350                "retryable": false,
351                "capability": "sql",
352                "sqlite_primary_code": primary_code,
353                "sqlite_extended_code": extended_code,
354            }),
355        }
356    }
357}
358
359fn sqlite_capacity_failure(error: &RuntimeError) -> Option<SqliteCapacityFailure> {
360    match error {
361        RuntimeError::Storage(storage) => sqlite_capacity_failure_from_storage(storage),
362        _ => None,
363    }
364}
365
366fn sqlite_capacity_failure_from_storage(error: &StorageError) -> Option<SqliteCapacityFailure> {
367    match error {
368        StorageError::CapacityFloor {
369            capability: StorageCapability::Sql,
370            volume,
371            available_bytes,
372            floor_bytes,
373            required_headroom_bytes,
374        } => Some(SqliteCapacityFailure::Refused {
375            volume: volume.clone(),
376            available_bytes: *available_bytes,
377            reserve_bytes: *floor_bytes,
378            required_headroom_bytes: *required_headroom_bytes,
379        }),
380        StorageError::CapacityUnavailable {
381            capability: StorageCapability::Sql,
382            phase,
383            ..
384        } => Some(SqliteCapacityFailure::Unavailable { phase: *phase }),
385        StorageError::WriterTaskTerminated {
386            sqlite_full_codes: Some((primary_code, extended_code)),
387            ..
388        } if *primary_code == rusqlite::ffi::SQLITE_FULL
389            && *extended_code >= 0
390            && (*extended_code & 0xff) == *primary_code =>
391        {
392            Some(SqliteCapacityFailure::NativeFull {
393                primary_code: *primary_code,
394                extended_code: *extended_code,
395            })
396        }
397        StorageError::SqliteWrite { failure, .. }
398            if failure.primary_code == rusqlite::ffi::SQLITE_FULL =>
399        {
400            Some(SqliteCapacityFailure::NativeFull {
401                primary_code: failure.primary_code,
402                extended_code: failure.extended_code,
403            })
404        }
405        StorageError::WriterTaskRequestFailed { source, .. }
406        | StorageError::SqliteWrite { source, .. } => sqlite_capacity_failure_from_storage(source),
407        StorageError::Driver { source, .. } => {
408            if let Some(sqlite) = source.downcast_ref::<khive_db::SqliteError>() {
409                sqlite_capacity_failure_from_sqlite(sqlite)
410            } else {
411                native_sqlite_full(source.as_ref())
412            }
413        }
414        _ => None,
415    }
416}
417
418fn sqlite_capacity_failure_from_sqlite(
419    error: &khive_db::SqliteError,
420) -> Option<SqliteCapacityFailure> {
421    match error {
422        khive_db::SqliteError::CapacityFloor {
423            volume,
424            available_bytes,
425            floor_bytes,
426            required_headroom_bytes,
427        } => Some(SqliteCapacityFailure::Refused {
428            volume: volume.clone(),
429            available_bytes: *available_bytes,
430            reserve_bytes: *floor_bytes,
431            required_headroom_bytes: *required_headroom_bytes,
432        }),
433        khive_db::SqliteError::CapacityUnavailable { phase, .. } => {
434            Some(SqliteCapacityFailure::Unavailable { phase: *phase })
435        }
436        khive_db::SqliteError::Rusqlite(sqlite)
437        | khive_db::SqliteError::Write { source: sqlite, .. } => native_sqlite_full(sqlite),
438        _ => None,
439    }
440}
441
442fn native_sqlite_full(error: &(dyn std::error::Error + 'static)) -> Option<SqliteCapacityFailure> {
443    let mut source = Some(error);
444    while let Some(current) = source {
445        if let Some(rusqlite::Error::SqliteFailure(code, _)) =
446            current.downcast_ref::<rusqlite::Error>()
447        {
448            if code.code == rusqlite::ErrorCode::DiskFull {
449                return Some(SqliteCapacityFailure::NativeFull {
450                    primary_code: code.extended_code & 0xff,
451                    extended_code: code.extended_code,
452                });
453            }
454        }
455        source = current.source();
456    }
457    None
458}
459
460fn storage_capability_wire_name(capability: StorageCapability) -> &'static str {
461    match capability {
462        StorageCapability::Sql => "sql",
463        StorageCapability::Notes => "notes",
464        StorageCapability::Entities => "entities",
465        StorageCapability::Graph => "graph",
466        StorageCapability::Events => "events",
467        StorageCapability::Vectors => "vectors",
468        StorageCapability::Sparse => "sparse",
469        StorageCapability::Text => "text",
470        StorageCapability::Blob => "blob",
471        StorageCapability::Attachments => "attachments",
472    }
473}
474
475#[cfg(test)]
476#[path = "error_projection_tests.rs"]
477mod tests;
khive_runtime/error_projection.rs

khive_runtime/
error_projection.rs
