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::LintFix { source, detail } => {
85            let mut value = runtime_error_value(*source, DomainDisposition::Unknown);
86            let object = value
87                .as_object_mut()
88                .expect("runtime error projection is an object");
89            let details = object.entry("details").or_insert_with(|| json!({}));
90            if !details.is_object() {
91                *details = json!({});
92            }
93            details["lint_fix"] = json!(detail);
94            // Earlier actions may already have committed. A named refusal of
95            // the last action cannot certify the whole call as not committed.
96            value["domain_disposition"] = json!("unknown");
97            value["retryable"] = json!(false);
98            return value;
99        }
100
101        RuntimeError::WithResolution { context } => {
102            let crate::error::ResolutionFailureContext { source, resolution } = context;
103            let mut value = runtime_error_value(*source, disposition);
104            let details = value
105                .as_object_mut()
106                .expect("runtime error projection is an object")
107                .entry("details")
108                .or_insert_with(|| json!({}));
109            if !details.is_object() {
110                *details = json!({});
111            }
112            details["resolution"] = json!(resolution);
113            return value;
114        }
115        RuntimeError::RefusedWithEvents { context } => {
116            let crate::error::RefusalEventContext { source, recordings } = context;
117            // Project the source first so named dispositions, code/details,
118            // detector location and refusal text retain their original meaning.
119            let mut value = runtime_error_value(*source, disposition);
120            if !recordings.is_empty() {
121                value["refusal_recorded"] = json!(recordings.iter().all(|recording| {
122                    matches!(recording, crate::RefusalEventRecording::Recorded { .. })
123                }));
124                value["refusal_events"] = json!(recordings);
125            }
126            return value;
127        }
128        RuntimeError::PermissionDenied {
129            verb,
130            reason,
131            receipt,
132        } => json!({
133            "kind": "runtime_error",
134            "code": "permission_denied",
135            "message": denial_message.unwrap_or_default(),
136            "verb": verb,
137            "reason": reason,
138            "audit_event_id": receipt.audit_event_id.map(|id| id.to_string()),
139            "audit_outcome": receipt.audit_outcome.wire_code(),
140        }),
141        RuntimeError::SecretDetected(matched) => {
142            let message = RuntimeError::SecretDetected(matched.clone()).to_string();
143            json!({
144                "kind": "runtime_error",
145                "code": "secret_detected",
146                "detector": matched.detector,
147                "location": matched.location,
148                "message": message,
149            })
150        }
151        RuntimeError::RefusedWithReceipt(refusal) => {
152            let crate::error::ReceiptRefusal {
153                code,
154                message,
155                receipt_id,
156                reason,
157                detail,
158            } = *refusal;
159            // Surface evidence first, contract fields last: a detail member that
160            // happens to share a name with a contract field cannot shadow it.
161            let mut error = match detail {
162                Value::Object(members) => members,
163                _ => serde_json::Map::new(),
164            };
165            error.insert("kind".into(), json!("runtime_error"));
166            error.insert("code".into(), json!(code));
167            error.insert("message".into(), json!(message));
168            error.insert("receipt_id".into(), json!(receipt_id));
169            error.insert("reason".into(), json!(reason));
170            Value::Object(error)
171        }
172        RuntimeError::AuditObligation {
173            failure,
174            domain_result,
175        } => {
176            let mut error = serde_json::Map::from_iter([
177                ("kind".into(), json!("obligation")),
178                ("code".into(), json!(failure.wire_code())),
179                ("message".into(), json!(failure.to_string())),
180            ]);
181            error.insert("domain_result".into(), domain_result);
182            Value::Object(error)
183        }
184        RuntimeError::Khive(k) => {
185            let mut value = serde_json::to_value(&k)
186                .unwrap_or_else(|_| json!({"kind": "internal", "message": k.to_string()}));
187            if k.kind() == khive_types::ErrorKind::InvalidInput
188                && k.details().and_then(|d| d.get("reason")) == Some("external_id_unverifiable")
189            {
190                value["code"] = json!("external_id_unverifiable");
191            }
192            value
193        }
194        RuntimeError::RemoteFetchError { remote, message } => json!({
195            "kind": "remote_fetch_error",
196            "remote": remote,
197            "message": message,
198        }),
199        missing @ RuntimeError::NotFound(_) => json!({
200            "kind": "not_found",
201            "code": null,
202            "details": null,
203            "message": missing.to_string(),
204        }),
205        other @ (RuntimeError::Storage(_)
206        | RuntimeError::Query(_)
207        | RuntimeError::InvalidInput(_)
208        | RuntimeError::UnknownVerb(_)
209        | RuntimeError::Unconfigured(_)
210        | RuntimeError::UnknownModel(_)
211        | RuntimeError::Embedding(_)
212        | RuntimeError::Ambiguous(_)
213        | RuntimeError::Fusion(_)
214        | RuntimeError::UnknownFusionStrategy(_)
215        | RuntimeError::Internal(_)
216        | RuntimeError::IncompatibleEventStore(_)
217        | RuntimeError::GuardedWriteFailed(_)
218        | RuntimeError::MissingPackDependency(_)
219        | RuntimeError::MissingPackDependencies(_)
220        | RuntimeError::CircularPackDependency(_)
221        | RuntimeError::PackRedeclared { .. }
222        | RuntimeError::VerbCollision { .. }
223        | RuntimeError::ReservedEnvelopeParam { .. }
224        | RuntimeError::GateUnavailable { .. }
225        | RuntimeError::NamespaceMismatch { .. }
226        | RuntimeError::AmbiguousPrefix { .. }
227        | RuntimeError::CrossBackendMergeUnsupported { .. }
228        | RuntimeError::UnknownRemote { .. }
229        | RuntimeError::RemoteCacheMissing { .. }
230        | RuntimeError::AmbiguousId { .. }
231        | RuntimeError::CrossNamespaceWrite { .. }
232        | RuntimeError::WriteBudgetExceeded { .. }
233        | RuntimeError::DeadlineExceeded { .. }) => {
234            if let Some(capacity) = sqlite_capacity_failure(&other) {
235                let mut value = capacity.into_value(other.to_string());
236                if let Some(context) = other.writer_task_failure_context() {
237                    value["request_state"] = json!(context.request_state.to_string());
238                    value["task_terminated"] = json!(context.task_terminated);
239                }
240                value
241            } else if let Some(context) = other.writer_task_failure_context() {
242                json!({"kind":"storage", "code":context.stage, "stage":context.stage,
243                    "message":other.to_string(), "retryable":context.retryable,
244                    "request_state":context.request_state.to_string(), "task_terminated":context.task_terminated})
245            } else if let Some(context) = other.retryable_failure_context() {
246                let timeout_ms = u64::try_from(context.timeout.as_millis()).unwrap_or(u64::MAX);
247                let mut value = json!({"kind":"unavailable", "code":context.stage, "stage":context.stage,
248                    "message":other.to_string(), "retryable":true, "timeout_ms":timeout_ms,
249                    "capability":context.capability.map(storage_capability_wire_name),
250                    "operation":context.operation, "scope":context.scope, "retry_after_ms":context.retry_after_ms});
251                if let Some(pool_identity) = context.pool_identity {
252                    value["pool_identity"] = json!(pool_identity);
253                }
254                value
255            } else {
256                json!({"kind":"runtime_error", "message":other.to_string()})
257            }
258        }
259    };
260    let mut value = payload;
261    if let Some(failure) = sqlite_write_failure {
262        // Capacity refusal retains its existing caller-visible precedence.
263        if !outcome_has_precedence
264            && !failure.settlement_unknown
265            && failure.primary_code != rusqlite::ffi::SQLITE_FULL
266        {
267            value["code"] = json!(failure.stage.as_str());
268            value["stage"] = json!(failure.stage.as_str());
269            if value["kind"] == "runtime_error" {
270                value["kind"] = json!("storage");
271            }
272        }
273        value["sqlite_write_stage"] = json!(failure.stage.as_str());
274        value["sqlite_primary_code"] = json!(failure.primary_code);
275        value["sqlite_extended_code"] = json!(failure.extended_code);
276    }
277    if let Some((retryable, _)) = receipt_projection {
278        value["retryable"] = json!(retryable);
279    }
280    value["domain_disposition"] = json!(disposition.as_str());
281    if let Some(named) = named_disposition {
282        value["domain_disposition"] = json!(named);
283    }
284    value
285}
286
287enum SqliteCapacityFailure {
288    Refused {
289        volume: String,
290        available_bytes: u64,
291        reserve_bytes: u64,
292        required_headroom_bytes: u64,
293    },
294    Unavailable {
295        phase: khive_storage::CapacityUnavailablePhase,
296    },
297    NativeFull {
298        primary_code: i32,
299        extended_code: i32,
300    },
301}
302
303impl SqliteCapacityFailure {
304    fn into_value(self, message: String) -> Value {
305        match self {
306            Self::Refused {
307                volume,
308                available_bytes,
309                reserve_bytes,
310                required_headroom_bytes,
311            } => json!({
312                "kind": "storage",
313                "code": "sqlite_capacity_refused",
314                "stage": "sqlite_capacity_refused",
315                "message": message,
316                "retryable": false,
317                "capability": "sql",
318                "volume": volume,
319                "available_bytes": available_bytes,
320                "reserve_bytes": reserve_bytes,
321                "required_headroom_bytes": required_headroom_bytes,
322            }),
323            Self::Unavailable { phase } => json!({
324                "kind": "storage",
325                "code": "sqlite_capacity_unavailable",
326                "stage": "sqlite_capacity_unavailable",
327                "message": message,
328                "retryable": phase == khive_storage::CapacityUnavailablePhase::LockContended,
329                "capability": "sql",
330                "phase": phase.as_str(),
331            }),
332            Self::NativeFull {
333                primary_code,
334                extended_code,
335            } => json!({
336                "kind": "storage",
337                "code": "sqlite_disk_full",
338                "stage": "sqlite_disk_full",
339                "message": message,
340                "retryable": false,
341                "capability": "sql",
342                "sqlite_primary_code": primary_code,
343                "sqlite_extended_code": extended_code,
344            }),
345        }
346    }
347}
348
349fn sqlite_capacity_failure(error: &RuntimeError) -> Option<SqliteCapacityFailure> {
350    match error {
351        RuntimeError::Storage(storage) => sqlite_capacity_failure_from_storage(storage),
352        _ => None,
353    }
354}
355
356fn sqlite_capacity_failure_from_storage(error: &StorageError) -> Option<SqliteCapacityFailure> {
357    match error {
358        StorageError::CapacityFloor {
359            capability: StorageCapability::Sql,
360            volume,
361            available_bytes,
362            floor_bytes,
363            required_headroom_bytes,
364        } => Some(SqliteCapacityFailure::Refused {
365            volume: volume.clone(),
366            available_bytes: *available_bytes,
367            reserve_bytes: *floor_bytes,
368            required_headroom_bytes: *required_headroom_bytes,
369        }),
370        StorageError::CapacityUnavailable {
371            capability: StorageCapability::Sql,
372            phase,
373            ..
374        } => Some(SqliteCapacityFailure::Unavailable { phase: *phase }),
375        StorageError::WriterTaskTerminated {
376            sqlite_full_codes: Some((primary_code, extended_code)),
377            ..
378        } if *primary_code == rusqlite::ffi::SQLITE_FULL
379            && *extended_code >= 0
380            && (*extended_code & 0xff) == *primary_code =>
381        {
382            Some(SqliteCapacityFailure::NativeFull {
383                primary_code: *primary_code,
384                extended_code: *extended_code,
385            })
386        }
387        StorageError::SqliteWrite { failure, .. }
388            if failure.primary_code == rusqlite::ffi::SQLITE_FULL =>
389        {
390            Some(SqliteCapacityFailure::NativeFull {
391                primary_code: failure.primary_code,
392                extended_code: failure.extended_code,
393            })
394        }
395        StorageError::WriterTaskRequestFailed { source, .. }
396        | StorageError::SqliteWrite { source, .. } => sqlite_capacity_failure_from_storage(source),
397        StorageError::Driver { source, .. } => {
398            if let Some(sqlite) = source.downcast_ref::<khive_db::SqliteError>() {
399                sqlite_capacity_failure_from_sqlite(sqlite)
400            } else {
401                native_sqlite_full(source.as_ref())
402            }
403        }
404        _ => None,
405    }
406}
407
408fn sqlite_capacity_failure_from_sqlite(
409    error: &khive_db::SqliteError,
410) -> Option<SqliteCapacityFailure> {
411    match error {
412        khive_db::SqliteError::CapacityFloor {
413            volume,
414            available_bytes,
415            floor_bytes,
416            required_headroom_bytes,
417        } => Some(SqliteCapacityFailure::Refused {
418            volume: volume.clone(),
419            available_bytes: *available_bytes,
420            reserve_bytes: *floor_bytes,
421            required_headroom_bytes: *required_headroom_bytes,
422        }),
423        khive_db::SqliteError::CapacityUnavailable { phase, .. } => {
424            Some(SqliteCapacityFailure::Unavailable { phase: *phase })
425        }
426        khive_db::SqliteError::Rusqlite(sqlite)
427        | khive_db::SqliteError::Write { source: sqlite, .. } => native_sqlite_full(sqlite),
428        _ => None,
429    }
430}
431
432fn native_sqlite_full(error: &(dyn std::error::Error + 'static)) -> Option<SqliteCapacityFailure> {
433    let mut source = Some(error);
434    while let Some(current) = source {
435        if let Some(rusqlite::Error::SqliteFailure(code, _)) =
436            current.downcast_ref::<rusqlite::Error>()
437        {
438            if code.code == rusqlite::ErrorCode::DiskFull {
439                return Some(SqliteCapacityFailure::NativeFull {
440                    primary_code: code.extended_code & 0xff,
441                    extended_code: code.extended_code,
442                });
443            }
444        }
445        source = current.source();
446    }
447    None
448}
449
450fn storage_capability_wire_name(capability: StorageCapability) -> &'static str {
451    match capability {
452        StorageCapability::Sql => "sql",
453        StorageCapability::Notes => "notes",
454        StorageCapability::Entities => "entities",
455        StorageCapability::Graph => "graph",
456        StorageCapability::Events => "events",
457        StorageCapability::Vectors => "vectors",
458        StorageCapability::Sparse => "sparse",
459        StorageCapability::Text => "text",
460        StorageCapability::Blob => "blob",
461        StorageCapability::Attachments => "attachments",
462    }
463}
464
465#[cfg(test)]
466#[path = "error_projection_tests.rs"]
467mod tests;
khive_runtime/error_projection.rs

khive_runtime/
error_projection.rs
