deepstrike_core/runtime/kernel/wire/
fault.rs

1//! Fault taxonomy and the closed prepare result (spec §7.13).
2//!
3//! Two shapes, one property. [`KernelPreparation`] is a **closed** union with no `faults` field on
4//! any success arm, so "a step that carries both actions and faults" is not constructible — the
5//! a historical step-level fault list made a partially-applied transition representable,
6//! and one host then ignored the vector entirely while another turned it into a panic.
7//!
8//! Every rejection is therefore zero-mutation by construction: [`KernelPreparation::Rejected`]
9//! carries a fault and nothing else — no record to append, no token to commit, no step to publish.
10
11use std::fmt;
12
13use serde::de::{self, Deserializer, Visitor};
14use serde::{Deserialize, Serialize, Serializer};
15
16use super::effect::{Digest, wire_opaque_ref};
17use super::scalar::{SCALAR_ERROR_MARKER, WireScalarError, WireU64};
18
19// ---------------------------------------------------------------------------------------------
20// §7.13 · fault codes
21// ---------------------------------------------------------------------------------------------
22
23/// Why the kernel refused an input.
24#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
25#[serde(rename_all = "snake_case")]
26pub enum KernelFaultCode {
27    MalformedEnvelope,
28    OperationMismatch,
29    ClockRegression,
30    InvalidLifecycle,
31    InvalidConfig,
32    InvalidAuthority,
33    ResourceLimitExceeded,
34    DuplicateInputConflict,
35    /// A host resolved a pending effect with the success payload of another effect kind, or
36    /// resolved an effect the kernel is not waiting on.
37    UnexpectedEffectOutcome,
38    TransactionConflict,
39    CheckpointIncompatible,
40    CheckpointCorrupted,
41    /// A durable **journal record** no longer matches the digest it carries, or the chain it sits
42    /// in no longer links up.
43    ///
44    /// Deliberately not folded into [`Self::CheckpointCorrupted`]: the two have different recovery
45    /// ladders. A corrupted checkpoint can be answered by falling back to an older checkpoint and
46    /// replaying more tail; a corrupted record is a journal-integrity failure with nothing to fall
47    /// back to, because the record chain *is* the operation's history.
48    RecordCorrupted,
49    /// The **only** retryable code (GAP-2).
50    ///
51    /// Returned as a `Rejected` preparation with zero mutation — the input was never accepted. The
52    /// host takes a checkpoint candidate, installs it and acks it (§12.3), then retries with the
53    /// *same* `input_id`. That retry is a brand-new prepare and must not fall into
54    /// [`Self::DuplicateInputConflict`].
55    ///
56    /// This code exists to replace the snapshot-overflow latch, whose double consequence — snapshots
57    /// permanently disabled snapshots and later preparations — was a hard failure on the
58    /// only durable host and a silent degradation everywhere else.
59    CheckpointRequired,
60    /// The kernel was about to emit an effect whose kind the operation's `host_effect_support`
61    /// declaration does not cover (DEC-8, GAP-6).
62    ///
63    /// Fail-closed **before emission**: the fault is committed and no effect is published, so the
64    /// host is never handed an effect it cannot execute. This is the declaration-time half of the
65    /// pair whose runtime half is
66    /// [`HostEffectFailureKind::ProtocolError`](super::effect::HostEffectFailureKind::ProtocolError)
67    /// (DEC-7).
68    UnsupportedEffect,
69}
70
71impl KernelFaultCode {
72    pub const ALL: [Self; 15] = [
73        Self::MalformedEnvelope,
74        Self::OperationMismatch,
75        Self::ClockRegression,
76        Self::InvalidLifecycle,
77        Self::InvalidConfig,
78        Self::InvalidAuthority,
79        Self::ResourceLimitExceeded,
80        Self::DuplicateInputConflict,
81        Self::UnexpectedEffectOutcome,
82        Self::TransactionConflict,
83        Self::CheckpointIncompatible,
84        Self::CheckpointCorrupted,
85        Self::RecordCorrupted,
86        Self::CheckpointRequired,
87        Self::UnsupportedEffect,
88    ];
89
90    pub fn as_str(self) -> &'static str {
91        match self {
92            Self::MalformedEnvelope => "malformed_envelope",
93            Self::OperationMismatch => "operation_mismatch",
94            Self::ClockRegression => "clock_regression",
95            Self::InvalidLifecycle => "invalid_lifecycle",
96            Self::InvalidConfig => "invalid_config",
97            Self::InvalidAuthority => "invalid_authority",
98            Self::ResourceLimitExceeded => "resource_limit_exceeded",
99            Self::DuplicateInputConflict => "duplicate_input_conflict",
100            Self::UnexpectedEffectOutcome => "unexpected_effect_outcome",
101            Self::TransactionConflict => "transaction_conflict",
102            Self::CheckpointIncompatible => "checkpoint_incompatible",
103            Self::CheckpointCorrupted => "checkpoint_corrupted",
104            Self::RecordCorrupted => "record_corrupted",
105            Self::CheckpointRequired => "checkpoint_required",
106            Self::UnsupportedEffect => "unsupported_effect",
107        }
108    }
109
110    /// Whether re-submitting the same `input_id` unchanged can succeed. Exactly one code says yes.
111    pub fn is_retryable(self) -> bool {
112        matches!(self, Self::CheckpointRequired)
113    }
114}
115
116impl fmt::Display for KernelFaultCode {
117    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
118        f.write_str(self.as_str())
119    }
120}
121
122/// A structured rejection. Malformed JSON, unknown fields/variants and revision mismatches all
123/// arrive here too — the same shape in all four languages, rather than one language's exception.
124#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
125#[serde(deny_unknown_fields)]
126pub struct KernelFault {
127    pub code: KernelFaultCode,
128    #[serde(default, skip_serializing_if = "String::is_empty")]
129    pub message: String,
130}
131
132impl KernelFault {
133    pub fn new(code: KernelFaultCode, message: impl Into<String>) -> Self {
134        Self {
135            code,
136            message: message.into(),
137        }
138    }
139
140    pub fn is_retryable(&self) -> bool {
141        self.code.is_retryable()
142    }
143}
144
145impl fmt::Display for KernelFault {
146    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
147        if self.message.is_empty() {
148            f.write_str(self.code.as_str())
149        } else {
150            write!(f, "{}: {}", self.code.as_str(), self.message)
151        }
152    }
153}
154
155impl std::error::Error for KernelFault {}
156
157wire_opaque_ref!(
158    /// Handle for a prepared-but-uncommitted transition. Handed out only by
159    /// [`KernelPreparation::Prepared`]: a replay has nothing to commit and a rejection has nothing
160    /// to abort.
161    PrepareToken,
162    "prepare token"
163);
164
165// ---------------------------------------------------------------------------------------------
166// §7.13 · the closed prepare result
167// ---------------------------------------------------------------------------------------------
168
169/// The result of preparing one input.
170///
171/// Generic over the durable record and the planned step: Task 6 owns those two contracts, and this
172/// task fixes only the **shape** of the result — which arms exist, and what each may carry.
173#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
174#[serde(tag = "status", rename_all = "snake_case")]
175pub enum KernelPreparation<Record, Step> {
176    /// A new record was built and is waiting for the host to append it and then commit the token.
177    Prepared(PreparedTransition<Record, Step>),
178    /// This input maps onto a record that **already exists**. No new record is produced and
179    /// `step_seq` points at the existing one.
180    ///
181    /// Two triggers, one shape:
182    ///
183    /// 1. **input-level replay** — the same `input_id` with the same canonical payload;
184    /// 2. **effect-level dedup** (DEC-1) — a *new* `input_id` resolving an already-completed
185    ///    effect with the same payload; the cancellation dedup branch behaves identically.
186    ///
187    /// The second trigger is the one that used to be reported as `Prepared` while returning the
188    /// old `step_seq`. A host then built a transaction whose `step_seq` did not increase, its CAS
189    /// successor check rejected it as an integrity error, and the run died — a live dead end on
190    /// the only durable host.
191    Replayed(ReplayedTransition<Record, Step>),
192    /// Nothing was accepted, nothing was staged, no state moved.
193    Rejected(RejectedTransition),
194}
195
196impl<Record, Step> KernelPreparation<Record, Step> {
197    pub fn record(&self) -> Option<&Record> {
198        match self {
199            Self::Prepared(prepared) => Some(&prepared.record),
200            Self::Replayed(replayed) => replayed.record.as_ref(),
201            Self::Rejected(_) => None,
202        }
203    }
204
205    /// Only a `Prepared` transition has something to commit.
206    pub fn token(&self) -> Option<&PrepareToken> {
207        match self {
208            Self::Prepared(prepared) => Some(&prepared.token),
209            Self::Replayed(_) | Self::Rejected(_) => None,
210        }
211    }
212
213    pub fn step(&self) -> Option<&Step> {
214        match self {
215            Self::Prepared(prepared) => Some(&prepared.planned_step),
216            Self::Replayed(replayed) => replayed.committed_step.as_ref(),
217            Self::Rejected(_) => None,
218        }
219    }
220
221    /// The sequence of the record this preparation refers to. `Prepared` does not have one yet —
222    /// its record is not in the journal until the host appends it.
223    pub fn step_seq(&self) -> Option<WireU64> {
224        match self {
225            Self::Replayed(replayed) => Some(replayed.step_seq),
226            Self::Prepared(_) | Self::Rejected(_) => None,
227        }
228    }
229
230    pub fn fault(&self) -> Option<&KernelFault> {
231        match self {
232            Self::Rejected(rejected) => Some(&rejected.fault),
233            Self::Prepared(_) | Self::Replayed(_) => None,
234        }
235    }
236
237    /// Whether this preparation left the operation byte-identical. True for exactly the rejected
238    /// arm — the type-level statement of the zero-mutation rule.
239    pub fn is_zero_mutation(&self) -> bool {
240        matches!(self, Self::Rejected(_))
241    }
242
243    /// Whether the host may re-submit the same `input_id` unchanged.
244    pub fn is_retryable(&self) -> bool {
245        self.fault().is_some_and(KernelFault::is_retryable)
246    }
247}
248
249#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
250#[serde(deny_unknown_fields)]
251pub struct PreparedTransition<Record, Step> {
252    pub token: PrepareToken,
253    pub record: Record,
254    /// Ephemeral. The planned step is returned to the caller but never enters the durable record —
255    /// a rebuild re-derives it from the canonical input and checks its digest, which is what keeps
256    /// rendered provider contexts out of the journal.
257    pub planned_step: Step,
258}
259
260/// The answer to an input this operation already accepted.
261///
262/// Two strengths of answer, and §12.3 rule 10 is what decides which one a caller gets:
263///
264/// * **reproduction** — above a restored checkpoint's `base_step_seq` the runtime still holds the
265///   record and the step it committed, so a redelivery is answered with both;
266/// * **acknowledgement** — below it, the step was never durable (§22.12) and the record may already
267///   have been reclaimed under an acked checkpoint. What survives is the ledger entry, and that is
268///   what answers: this input is step N, record D. A caller retrying a lost response learns exactly
269///   what it needed to; nothing is fabricated to fill the other two fields.
270///
271/// `record_digest` is therefore the one field that is always present — it is the identity of the
272/// transition, where `record` and `committed_step` are the (possibly reclaimed) *contents* of it.
273#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
274#[serde(deny_unknown_fields)]
275pub struct ReplayedTransition<Record, Step> {
276    #[serde(default = "Option::default")]
277    pub record: Option<Record>,
278    pub record_digest: Digest,
279    #[serde(default = "Option::default")]
280    pub committed_step: Option<Step>,
281    pub step_seq: WireU64,
282}
283
284#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
285#[serde(deny_unknown_fields)]
286pub struct RejectedTransition {
287    pub fault: KernelFault,
288}
289
290#[cfg(test)]
291mod tests {
292    use std::collections::BTreeSet;
293
294    use serde::{Deserialize, Serialize};
295    use serde_json::json;
296
297    use super::super::*;
298
299    /// Stand-ins for the Task 6 record/step types. [`KernelPreparation`] is generic over them
300    /// precisely so this task can fix the *shape* of the prepare result without squatting on the
301    /// durable-record contract.
302    #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
303    #[serde(deny_unknown_fields)]
304    struct StubRecord {
305        step_seq: WireU64,
306    }
307
308    #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
309    #[serde(deny_unknown_fields)]
310    struct StubStep {
311        effects: u32,
312    }
313
314    type Preparation = KernelPreparation<StubRecord, StubStep>;
315
316    fn prepared() -> Preparation {
317        KernelPreparation::Prepared(PreparedTransition {
318            token: PrepareToken::new("prepare-1").unwrap(),
319            record: StubRecord {
320                step_seq: WireU64::new(4),
321            },
322            planned_step: StubStep { effects: 1 },
323        })
324    }
325
326    fn replayed() -> Preparation {
327        KernelPreparation::Replayed(ReplayedTransition {
328            record: Some(StubRecord {
329                step_seq: WireU64::new(2),
330            }),
331            record_digest: Digest::new("sha256:replayed").unwrap(),
332            committed_step: Some(StubStep { effects: 1 }),
333            step_seq: WireU64::new(2),
334        })
335    }
336
337    fn rejected(code: KernelFaultCode) -> Preparation {
338        KernelPreparation::Rejected(RejectedTransition {
339            fault: KernelFault::new(code, "rejected"),
340        })
341    }
342
343    // -----------------------------------------------------------------------------------------
344    // fault codes (§7.13, GAP-2 / GAP-6)
345    // -----------------------------------------------------------------------------------------
346
347    #[test]
348    fn the_fault_taxonomy_is_the_fifteen_declared_codes() {
349        let labels: BTreeSet<&str> = KernelFaultCode::ALL.iter().map(|c| c.as_str()).collect();
350        assert_eq!(
351            labels,
352            BTreeSet::from([
353                "malformed_envelope",
354                "operation_mismatch",
355                "clock_regression",
356                "invalid_lifecycle",
357                "invalid_config",
358                "invalid_authority",
359                "resource_limit_exceeded",
360                "duplicate_input_conflict",
361                "unexpected_effect_outcome",
362                "transaction_conflict",
363                "checkpoint_incompatible",
364                "checkpoint_corrupted",
365                "record_corrupted",
366                "checkpoint_required",
367                "unsupported_effect",
368            ])
369        );
370        assert_eq!(KernelFaultCode::ALL.len(), 15);
371
372        for code in KernelFaultCode::ALL {
373            let text = serde_json::to_string(&code).unwrap();
374            assert_eq!(text, format!("\"{}\"", code.as_str()));
375            let back: KernelFaultCode = serde_json::from_str(&text).unwrap();
376            assert_eq!(back, code);
377        }
378    }
379
380    #[test]
381    fn checkpoint_required_is_the_only_retryable_fault_code() {
382        for code in KernelFaultCode::ALL {
383            assert_eq!(
384                code.is_retryable(),
385                code == KernelFaultCode::CheckpointRequired,
386                "{} must{} be retryable",
387                code.as_str(),
388                if code == KernelFaultCode::CheckpointRequired {
389                    ""
390                } else {
391                    " not"
392                }
393            );
394        }
395        assert!(KernelFault::new(KernelFaultCode::CheckpointRequired, "").is_retryable());
396        assert!(!KernelFault::new(KernelFaultCode::DuplicateInputConflict, "").is_retryable());
397    }
398
399    #[test]
400    fn unknown_fault_codes_are_rejected() {
401        for raw in ["\"snapshot_overflow\"", "\"ok\"", "3", "null"] {
402            assert!(
403                serde_json::from_str::<KernelFaultCode>(raw).is_err(),
404                "{raw} must not decode as a fault code"
405            );
406        }
407    }
408
409    // -----------------------------------------------------------------------------------------
410    // zero mutation (§7.13)
411    // -----------------------------------------------------------------------------------------
412
413    #[test]
414    fn a_rejected_preparation_carries_no_record_no_token_and_no_step() {
415        for code in KernelFaultCode::ALL {
416            let preparation = rejected(code);
417            assert!(preparation.record().is_none(), "{}", code.as_str());
418            assert!(preparation.token().is_none(), "{}", code.as_str());
419            assert!(preparation.step().is_none(), "{}", code.as_str());
420            assert!(preparation.step_seq().is_none(), "{}", code.as_str());
421            assert_eq!(preparation.fault().map(|f| f.code), Some(code));
422            assert!(preparation.is_zero_mutation());
423        }
424    }
425
426    #[test]
427    fn a_successful_preparation_can_never_carry_a_fault() {
428        for preparation in [prepared(), replayed()] {
429            assert!(preparation.fault().is_none());
430            assert!(!preparation.is_zero_mutation());
431
432            let mut all = BTreeSet::new();
433            let value = serde_json::to_value(&preparation).unwrap();
434            if let serde_json::Value::Object(map) = &value {
435                for key in map.keys() {
436                    all.insert(key.clone());
437                }
438            }
439            assert!(
440                !all.contains("fault") && !all.contains("faults"),
441                "a fault-bearing success step must not be constructible: {value}"
442            );
443        }
444    }
445
446    // -----------------------------------------------------------------------------------------
447    // the closed prepare result (§7.13, DEC-1)
448    // -----------------------------------------------------------------------------------------
449
450    #[test]
451    fn preparation_has_exactly_three_shapes() {
452        let statuses: BTreeSet<String> = [
453            prepared(),
454            replayed(),
455            rejected(KernelFaultCode::InvalidLifecycle),
456        ]
457        .iter()
458        .map(|preparation| {
459            serde_json::to_value(preparation).unwrap()["status"]
460                .as_str()
461                .unwrap()
462                .to_string()
463        })
464        .collect();
465        assert_eq!(
466            statuses,
467            BTreeSet::from([
468                "prepared".to_string(),
469                "replayed".to_string(),
470                "rejected".to_string(),
471            ])
472        );
473
474        for shape in ["accepted", "deferred", "prepared_with_faults"] {
475            let raw = json!({ "status": shape });
476            assert!(
477                serde_json::from_value::<Preparation>(raw).is_err(),
478                "{shape} is not a preparation shape"
479            );
480        }
481    }
482
483    #[test]
484    fn replayed_points_at_the_existing_record_step_seq() {
485        let preparation = replayed();
486        assert_eq!(preparation.step_seq(), Some(WireU64::new(2)));
487        assert_eq!(
488            preparation.record().map(|record| record.step_seq),
489            Some(WireU64::new(2)),
490            "a replay must point at the record that already exists, not mint a new one"
491        );
492        assert!(
493            preparation.token().is_none(),
494            "a replay has nothing to commit, so it hands out no prepare token"
495        );
496    }
497
498    #[test]
499    fn preparation_round_trips_and_rejects_unknown_fields() {
500        for preparation in [
501            prepared(),
502            replayed(),
503            rejected(KernelFaultCode::CheckpointRequired),
504        ] {
505            let value = serde_json::to_value(&preparation).unwrap();
506            let back: Preparation = serde_json::from_value(value).unwrap();
507            assert_eq!(back, preparation);
508        }
509
510        let extra = json!({
511            "status": "rejected",
512            "fault": { "code": "invalid_lifecycle", "message": "terminal already committed" },
513            "retry_after_ms": 500,
514        });
515        assert!(serde_json::from_value::<Preparation>(extra).is_err());
516    }
517}
deepstrike_core/runtime/kernel/wire/fault.rs

deepstrike_core/runtime/kernel/wire/
fault.rs
