Skip to main content

deepstrike_core/runtime/kernel/wire/
checkpoint.rs

1//! The logical checkpoint and its bounded tail (spec §12).
2//!
3//! A checkpoint is *not* a snapshot of the kernel's internals. It is a stable, versioned DTO whose
4//! shape is a contract in its own right, and every design rule below corrects the retired
5//! full-journal recovery format:
6//!
7//! 1. **Nothing here is derived from a private layout.** [`LogicalKernelState`] is built by an
8//!    explicit projection — [`LogicalStateProjection`] from the semantic driver, the transition
9//!    partition from the transaction — so a field added to `LoopStateMachine` cannot silently
10//!    change the checkpoint format, and a field this DTO needs cannot silently disappear. The old
11//!    snapshot serialised the whole last planned step, rendered context and all, which made
12//!    the blob a function of the *rendered prompt* rather than of the state.
13//! 2. **Every piece of correctness state has exactly one home.** The four partitions of §12.1 —
14//!    transition / syscall / scheduler / context_vm — partition the state, they do not overlap it:
15//!    pending effects, the input replay ledger and the terminal live in `transition`, task attempts
16//!    in `scheduler`, P3 handles in `context_vm`, and the checkpoint header repeats none of them.
17//!    `single_ownership_is_structural` proves it by scanning the serialised document.
18//! 3. **The bounded tail is exact.** `tail_inputs` covers `(base_step_seq, through_step_seq]` with
19//!    no hole, no duplicate and nothing outside the range — checked at construction, so a
20//!    checkpoint that would replay a different history than the journal did is not constructible.
21//! 4. **Three digests, three questions.** `state_digest` answers "is this the logical state that
22//!    was captured", `tail_digest` answers "is this the tail that was captured", and
23//!    `checkpoint_digest` answers "is this the whole checkpoint, header included". They all use the
24//!    record layer's canonical bytes, so a host validator that already implements §7.1.1 for
25//!    records needs no second serialiser.
26//!
27//! What this module deliberately does **not** do: install, restore, rebase or ack. §12.3's second
28//! half is Task 16. What exists here is *generation* — [`KernelTransaction::checkpoint_candidate`]
29//! and the shapes it produces — plus the verification a restore will call into.
30
31use std::fmt;
32
33use serde::de::{self, Deserializer, Visitor};
34use serde::{Deserialize, Serialize, Serializer};
35
36use super::KERNEL_CHECKPOINT_VERSION;
37use super::config::ResolvedOperationConfig;
38use super::effect::{Digest, KernelEffect, LaunchToken, wire_opaque_ref};
39use super::envelope::{AbiRevision, OperationLifecycle};
40use super::fault::{KernelFault, KernelFaultCode};
41use super::record::{NormalizedInput, RecordError, canonical_bytes, canonical_digest};
42use super::root::{ExecutionFocus, LogicalAgentSpec, LogicalTask, RootKind};
43use super::scalar::{
44    AttemptId, BoundedJson, CanonicalBytes, EffectId, InputId, MemoryBindingId, NodeId,
45    OperationId, SCALAR_ERROR_MARKER, SignalId, TaskId, WireScalarError, WireU64, WorkflowId,
46};
47use super::syscall::MemoryKind;
48use super::terminal::KernelTerminal;
49
50// ---------------------------------------------------------------------------------------------
51// errors
52// ---------------------------------------------------------------------------------------------
53
54/// Prefix of every checkpoint-layer rejection, so all four hosts classify on one marker.
55pub const CHECKPOINT_ERROR_MARKER: &str = "kernel checkpoint rejected";
56
57/// Why a checkpoint could not be assembled, decoded or verified.
58///
59/// The split is the recovery ladder, not the field that failed. `Incompatible` means "this blob is
60/// not for this kernel or not for this operation" — a host answers it by looking for a different
61/// checkpoint. `Corrupted` means "this blob claims to be ours and does not hold together" — the
62/// only answer is an older checkpoint plus more tail.
63#[derive(Debug, Clone, PartialEq, Eq)]
64pub enum CheckpointError {
65    /// Wrong checkpoint revision, wrong ABI revision, wrong operation, or a genesis/head this
66    /// operation never had.
67    Incompatible(String),
68    /// A digest disagrees with the bytes it summarises, or the bounded tail does not cover
69    /// `(base_step_seq, through_step_seq]` exactly.
70    Corrupted(String),
71    /// The value has no canonical byte representation at all.
72    NotCanonical(String),
73}
74
75impl CheckpointError {
76    pub fn message(&self) -> &str {
77        match self {
78            Self::Incompatible(message)
79            | Self::Corrupted(message)
80            | Self::NotCanonical(message) => message,
81        }
82    }
83
84    pub fn code(&self) -> KernelFaultCode {
85        match self {
86            Self::Incompatible(_) => KernelFaultCode::CheckpointIncompatible,
87            Self::Corrupted(_) => KernelFaultCode::CheckpointCorrupted,
88            Self::NotCanonical(_) => KernelFaultCode::MalformedEnvelope,
89        }
90    }
91
92    /// Host-facing projection (§7.13).
93    pub fn fault(&self) -> KernelFault {
94        KernelFault::new(self.code(), self.to_string())
95    }
96}
97
98impl fmt::Display for CheckpointError {
99    /// The rendered form names its own code.
100    ///
101    /// Not cosmetic: a checkpoint rejected inside `serde` reaches the caller as a *string*, and
102    /// [`from_checkpoint_bytes`](KernelCheckpoint::from_checkpoint_bytes) has to recover the class
103    /// from it. Printing the code is what keeps that recovery from being a guess based on which
104    /// words happen to be in the message.
105    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
106        write!(
107            f,
108            "{CHECKPOINT_ERROR_MARKER} ({}): {}",
109            self.code().as_str(),
110            self.message()
111        )
112    }
113}
114
115impl std::error::Error for CheckpointError {}
116
117impl From<RecordError> for CheckpointError {
118    fn from(error: RecordError) -> Self {
119        Self::NotCanonical(error.message().to_string())
120    }
121}
122
123// ---------------------------------------------------------------------------------------------
124// §12.1 · the checkpoint format revision
125// ---------------------------------------------------------------------------------------------
126
127/// The checkpoint format revision, fail-closed on the wire.
128///
129/// A separate axis from [`AbiRevision`] on purpose: the wire contract and the checkpoint layout
130/// can move independently, and DEC-6 renamed the field to `checkpoint_version` so this value could
131/// start at 1 without colliding with retired recovery-format revisions.
132/// Deserialising anything else fails at the boundary, so "an unrecognised checkpoint version" can
133/// never reach a restore.
134#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
135pub struct CheckpointRevision(u32);
136
137impl CheckpointRevision {
138    pub const CURRENT: Self = Self(KERNEL_CHECKPOINT_VERSION);
139
140    pub const fn get(self) -> u32 {
141        self.0
142    }
143}
144
145impl Default for CheckpointRevision {
146    fn default() -> Self {
147        Self::CURRENT
148    }
149}
150
151impl fmt::Display for CheckpointRevision {
152    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
153        write!(f, "{}", self.0)
154    }
155}
156
157impl Serialize for CheckpointRevision {
158    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
159        serializer.serialize_u32(self.0)
160    }
161}
162
163impl<'de> Deserialize<'de> for CheckpointRevision {
164    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
165        struct RevisionVisitor;
166
167        impl Visitor<'_> for RevisionVisitor {
168            type Value = CheckpointRevision;
169
170            fn expecting(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
171                write!(
172                    f,
173                    "the kernel checkpoint revision {KERNEL_CHECKPOINT_VERSION}"
174                )
175            }
176
177            fn visit_u64<E: de::Error>(self, value: u64) -> Result<Self::Value, E> {
178                if value == u64::from(KERNEL_CHECKPOINT_VERSION) {
179                    Ok(CheckpointRevision::CURRENT)
180                } else {
181                    Err(E::custom(
182                        CheckpointError::Incompatible(format!(
183                            "unsupported checkpoint version {value}; this kernel reads only \
184                             version {KERNEL_CHECKPOINT_VERSION}"
185                        ))
186                        .to_string(),
187                    ))
188                }
189            }
190
191            fn visit_i64<E: de::Error>(self, value: i64) -> Result<Self::Value, E> {
192                if value < 0 {
193                    return Err(E::custom(
194                        CheckpointError::Incompatible(format!(
195                            "unsupported checkpoint version {value}"
196                        ))
197                        .to_string(),
198                    ));
199                }
200                self.visit_u64(value as u64)
201            }
202        }
203
204        deserializer.deserialize_u32(RevisionVisitor)
205    }
206}
207
208wire_opaque_ref!(
209    /// Handle the host returns to `ack_checkpoint` once the blob is durably installed (§12.3).
210    ///
211    /// It is **not** a [`KernelInput`](super::envelope::KernelInput) (§12.3 rule 4): acking is
212    /// runtime maintenance, it writes no record, and a crash between install and ack is recovered
213    /// from the installed checkpoint anyway. Deriving it from the checkpoint's own digest is what
214    /// makes "ack a checkpoint that was never handed out" unrepresentable.
215    CheckpointAckToken,
216    "checkpoint ack token"
217);
218
219// ---------------------------------------------------------------------------------------------
220// §12.1 · the bounded tail
221// ---------------------------------------------------------------------------------------------
222
223/// One accepted input inside a checkpoint's bounded tail.
224///
225/// It carries the normalised input (so the tail can be **replayed**) and the digest of the record
226/// that input produced (so the replay can be **verified** against the journal it came from).
227/// Neither alone is enough: digests without inputs make a checkpoint auditable but useless, and
228/// inputs without digests make a restore that silently disagrees with the journal possible.
229#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
230#[serde(deny_unknown_fields)]
231pub struct CanonicalInput {
232    pub step_seq: WireU64,
233    pub record_digest: Digest,
234    pub input: NormalizedInput,
235}
236
237impl CanonicalInput {
238    /// Project one durable record into its tail entry.
239    pub fn from_record(record: &super::record::KernelRecord) -> Result<Self, CheckpointError> {
240        Ok(Self {
241            step_seq: record.step_seq(),
242            record_digest: record.record_digest().clone(),
243            input: record.normalized_input()?,
244        })
245    }
246}
247
248// ---------------------------------------------------------------------------------------------
249// §12.1 · the four logical partitions
250// ---------------------------------------------------------------------------------------------
251
252/// The versioned logical state of one operation, partitioned as §12.1 requires.
253///
254/// The four fields are a *partition*: each piece of correctness state appears in exactly one of
255/// them, and the checkpoint header above repeats none of it. That is the property the historical
256/// snapshot lacked — it stored pending effects at the top level, again inside `last_step`, and a
257/// third time in the resumed-outcome vectors, so "which copy is authoritative" was decided by
258/// whichever restore path happened to run.
259#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
260#[serde(deny_unknown_fields)]
261pub struct LogicalKernelState {
262    pub transition: TransitionStateV1,
263    pub syscall: SyscallStateV1,
264    pub scheduler: SchedulerStateV1,
265    pub context_vm: ContextVmStateV1,
266}
267
268/// §12.1 · operation lifecycle, execution focus, the effect ledger, input replay, cancellation and
269/// the terminal.
270///
271/// The step sequence is deliberately **not** here: the checkpoint header already states
272/// `base_step_seq` and `through_step_seq`, and §12.1's "the header must not duplicate sub-state"
273/// cuts both ways.
274#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
275#[serde(deny_unknown_fields)]
276pub struct TransitionStateV1 {
277    pub lifecycle: OperationLifecycle,
278    /// The configuration the genesis record froze (§8.1, §15.2 item 8).
279    ///
280    /// Here rather than "read it off the genesis record", because §12.3 rule 6 lets an acked
281    /// checkpoint reclaim the journal prefix *including genesis*: after that, the checkpoint is the
282    /// only place the resolved configuration still exists, and every later step is planned against
283    /// it. `genesis_digest` in the header keeps binding the identity; this carries the content.
284    pub resolved_config: ResolvedOperationConfig,
285    /// Immutable after the root start commits (§6.1.5).
286    #[serde(default)]
287    pub root_kind: Option<RootKind>,
288    /// Where control is (§7.4). Moves only on a committed transition, so a checkpoint states it
289    /// rather than deriving it.
290    #[serde(default)]
291    pub focus: Option<ExecutionFocus>,
292    /// The operation's only clock fact (§11.2): the newest accepted `observed_at_ms`. A restore
293    /// must not accept an input that precedes it.
294    pub last_observed_at_ms: WireU64,
295    /// Effects published by committed records and not yet resolved. **The** home of pending
296    /// effects — no other partition, and not the header.
297    #[serde(default)]
298    pub pending_effects: Vec<KernelEffect>,
299    /// Effects already answered, with the digest of the outcome that answered them. This is what
300    /// makes a redelivered `ResolveEffect` a `Replayed` instead of a second record (DEC-1).
301    #[serde(default)]
302    pub resolved_effects: Vec<ResolvedEffectState>,
303    /// Launch tokens the kernel minted with a `SpawnTasks` effect. Effect-resolution bookkeeping,
304    /// not task state — the task table lives in [`SchedulerStateV1`].
305    #[serde(default)]
306    pub launch_tokens: Vec<LaunchTokenState>,
307    /// §12.3 rule 7 · the input replay/dedupe ledger. An ack must never empty it: it is what turns
308    /// a redelivery into an idempotent answer instead of a second durable record.
309    #[serde(default)]
310    pub accepted_inputs: Vec<AcceptedInputState>,
311    /// The cancellation this operation already accepted, so a retry of it is answered rather than
312    /// refused by the terminal it created.
313    #[serde(default)]
314    pub accepted_cancellation: Option<AcceptedCancellationState>,
315    /// The committed terminal. **The** home of the terminal.
316    #[serde(default)]
317    pub terminal: Option<KernelTerminal>,
318}
319
320#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
321#[serde(deny_unknown_fields)]
322pub struct ResolvedEffectState {
323    pub effect_id: EffectId,
324    pub outcome_digest: Digest,
325    pub input_id: InputId,
326    pub step_seq: WireU64,
327}
328
329#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
330#[serde(deny_unknown_fields)]
331pub struct LaunchTokenState {
332    pub launch_token: LaunchToken,
333    pub step_seq: WireU64,
334}
335
336/// One entry of the replay ledger (§12.3 rules 7 and 10).
337///
338/// The digest is what makes the ledger answerable on its own. Below `base_step_seq` a restored
339/// runtime holds no step and, once retention has reclaimed the prefix, no record either — so the
340/// guarantee a redelivery gets down there is **idempotent acknowledgement, not step reproduction**:
341/// "input X is step N, record D". That is exactly what a caller retrying a lost response needs, and
342/// it is all a checkpoint has to remember.
343#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
344#[serde(deny_unknown_fields)]
345pub struct AcceptedInputState {
346    pub input_id: InputId,
347    pub step_seq: WireU64,
348    pub record_digest: Digest,
349}
350
351#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
352#[serde(deny_unknown_fields)]
353pub struct AcceptedCancellationState {
354    /// Digest of the canonical cancel command, so "the same cancellation" is decided by bytes.
355    pub command_digest: Digest,
356    pub input_id: InputId,
357    pub step_seq: WireU64,
358}
359
360/// §12.1 · governance revision, the live policy, the rate-limit window, and the provider-tool
361/// causation the P1 gate derives a caller from.
362#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
363#[serde(deny_unknown_fields)]
364pub struct SyscallStateV1 {
365    /// §13.2 · the revision two concurrent policy writers race on. `None` before the genesis
366    /// record installs the policy.
367    #[serde(default)]
368    pub policy_revision: Option<WireU64>,
369    /// The live-mutable configuration as patched so far. Distinct from the genesis record's
370    /// resolved configuration, which is frozen — this is the value the gate reads today.
371    #[serde(default)]
372    pub live_config: Option<ResolvedOperationConfig>,
373    /// §7.6 · the provider calls this operation is waiting on and the tool surface each one
374    /// advertised. A tool call naming anything else has no causation to derive from.
375    #[serde(default)]
376    pub provider_calls: Vec<PendingProviderCallState>,
377    /// Tool call ids that already produced a syscall. A causation is spent once.
378    #[serde(default)]
379    pub consumed_call_ids: Vec<String>,
380    /// §22.13 · what the kernel authored for each pending memory write. The resolution reports
381    /// these, never what the host echoes back.
382    #[serde(default)]
383    pub authored_memory_writes: Vec<AuthoredMemoryWriteState>,
384    #[serde(default)]
385    pub authored_memory_queries: Vec<AuthoredMemoryQueryState>,
386    /// Rolling window of accepted memory-write timestamps, in the operation's own clock. The
387    /// window is a gate input, so dropping it at a checkpoint would hand the run a fresh quota.
388    #[serde(default)]
389    pub memory_write_window_ms: Vec<WireU64>,
390}
391
392#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
393#[serde(deny_unknown_fields)]
394pub struct PendingProviderCallState {
395    pub effect_id: EffectId,
396    /// The task whose turn issued the call — the caller a syscall inherits.
397    pub task_id: TaskId,
398    pub exposed_tools: Vec<String>,
399}
400
401#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
402#[serde(deny_unknown_fields)]
403pub struct AuthoredMemoryWriteState {
404    pub effect_id: EffectId,
405    pub binding_id: MemoryBindingId,
406    pub name: String,
407    pub kind: MemoryKind,
408    pub size_bytes: u32,
409}
410
411#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
412#[serde(deny_unknown_fields)]
413pub struct AuthoredMemoryQueryState {
414    pub effect_id: EffectId,
415    pub binding_id: MemoryBindingId,
416    pub text: String,
417    pub requested_k: u32,
418}
419
420/// §12.1 · the P2 plane: task control blocks and their attempts, budgets and waits, the workflow
421/// graph, queued signals plus dedupe memory, and the milestone cascade.
422#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
423#[serde(deny_unknown_fields)]
424pub struct SchedulerStateV1 {
425    pub turn: u32,
426    pub total_tokens: WireU64,
427    pub rounds_completed: u32,
428    pub subagents_spawned: u32,
429    /// First observed clock of the run, the anchor the wall-budget axis measures against.
430    #[serde(default)]
431    pub started_at_ms: Option<WireU64>,
432    /// §13.2 · the wall-clock budget an `UpdateDeadline` command last projected onto the axis.
433    /// Not derivable from the configuration — the command sets a duration measured from
434    /// [`Self::started_at_ms`] — so a restore that dropped it would un-bound the run.
435    #[serde(default)]
436    pub wall_budget_ms: Option<WireU64>,
437    /// The task table. **The** home of task lifecycle and, with [`Self::attempts`], of task
438    /// identity.
439    #[serde(default)]
440    pub tasks: Vec<TaskControlState>,
441    /// §10.4 · the attempt the kernel minted for each live task. A completion naming an attempt
442    /// that is not here has no authority, which is why the mapping is checkpointed rather than
443    /// re-derived from task ids.
444    #[serde(default)]
445    pub attempts: Vec<TaskAttemptState>,
446    #[serde(default)]
447    pub workflow: Option<WorkflowGraphState>,
448    #[serde(default)]
449    pub queued_signals: Vec<QueuedSignalState>,
450    /// Router dedupe keys in eviction order, including keys for already-dispatched signals.
451    #[serde(default)]
452    pub signal_dedupe_keys: Vec<String>,
453    #[serde(default)]
454    pub milestone: Option<MilestoneState>,
455    /// Session-disorder measurement and alert-gate state. These values intentionally survive a
456    /// canonical crash restore even though they do not participate in a same-process turn
457    /// rollback: otherwise the first post-restore sample forgets prior failures and rollbacks.
458    #[serde(default)]
459    pub entropy: EntropyState,
460}
461
462#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
463#[serde(deny_unknown_fields)]
464pub struct EntropyState {
465    /// Oldest to newest, bounded by the kernel entropy window.
466    #[serde(default)]
467    pub window: Vec<EntropyTurnState>,
468    /// Rollbacks observed after the newest completed turn.
469    pub rollbacks_pending: u32,
470    /// Threshold watch hysteresis state.
471    pub disarmed: bool,
472    /// Most recent alert turn, retained after re-arming for cooldown enforcement.
473    #[serde(default)]
474    pub last_alert_turn: Option<u32>,
475}
476
477#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
478#[serde(deny_unknown_fields)]
479pub struct EntropyTurnState {
480    pub errored_results: u32,
481    pub total_results: u32,
482    pub rollbacks: u32,
483}
484
485/// One task control block, projected. The lifecycle travels as its label rather than as the
486/// internal enum: `TaskLifecycle::Done(TerminationReason)` is a semantic-kernel shape, and a
487/// checkpoint that mirrored it would be a checkpoint of a private layout.
488///
489/// The label alone is not *invertible*, though, and Task 16 needs it to be: a restore that rebuilt
490/// a finished task without the reason it finished for would hand the next transition a different
491/// task table than the uninterrupted run had. So the two data-carrying lifecycles travel with their
492/// data beside the label — `termination` for `done`, `waiting_on` for a `sub_agent_join` wait —
493/// rather than by mirroring the internal enum's shape.
494#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
495#[serde(deny_unknown_fields)]
496pub struct TaskControlState {
497    pub task_id: TaskId,
498    #[serde(default)]
499    pub parent_task_id: Option<TaskId>,
500    pub lifecycle: String,
501    /// Why a `done` task is done. `None` for every other lifecycle.
502    #[serde(default)]
503    pub termination: Option<String>,
504    #[serde(default)]
505    pub wait: Option<String>,
506    /// The children a `sub_agent_join` wait is blocked on. Empty for every other wait.
507    #[serde(default)]
508    pub waiting_on: Vec<TaskId>,
509    #[serde(default)]
510    pub capability_ids: Vec<String>,
511    /// Sub-agent process identity and join state. `None` only for the root task.
512    #[serde(default)]
513    pub process: Option<ChildProcessState>,
514    pub tokens_used: WireU64,
515    pub turns_used: u32,
516}
517
518#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
519#[serde(deny_unknown_fields)]
520pub struct ChildProcessState {
521    pub role: String,
522    pub isolation: String,
523    pub context_inheritance: String,
524    /// Present once the child has joined; the task lifecycle alone does not reproduce its output.
525    #[serde(default)]
526    pub join_result: Option<BoundedJson>,
527}
528
529#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
530#[serde(deny_unknown_fields)]
531pub struct TaskAttemptState {
532    pub task_id: TaskId,
533    pub attempt_id: AttemptId,
534}
535
536#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
537#[serde(deny_unknown_fields)]
538pub struct WorkflowGraphState {
539    pub workflow_id: WorkflowId,
540    /// The complete, index-ordered DAG and its runtime state. The semantic scheduler rebuilds its
541    /// private reverse edges, ready heap and agent lookup from this projection during restore.
542    #[serde(default)]
543    pub nodes: Vec<WorkflowNodeState>,
544}
545
546/// One workflow node as source state rather than a snapshot of `TaskGraph` internals.
547///
548/// `kind` is explicit even though the canonical ABI currently admits only `spawn`: a checkpoint
549/// must fail closed if a future producer writes a control-flow kind this revision cannot rebuild.
550#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
551#[serde(deny_unknown_fields)]
552pub struct WorkflowNodeState {
553    pub node_id: NodeId,
554    pub task: LogicalTask,
555    #[serde(default)]
556    pub depends_on: Vec<NodeId>,
557    #[serde(default)]
558    pub run_spec: Option<LogicalAgentSpec>,
559    pub kind: String,
560    pub status: String,
561    /// The deterministic child identity while this node is running.
562    #[serde(default)]
563    pub active_agent_id: Option<String>,
564    #[serde(default)]
565    pub iterations_completed: u32,
566}
567
568#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
569#[serde(deny_unknown_fields)]
570pub struct QueuedSignalState {
571    pub signal_id: SignalId,
572    pub source: String,
573    pub signal_type: String,
574    pub urgency: String,
575    pub summary: String,
576    #[serde(default)]
577    pub payload: BoundedJson,
578    #[serde(default)]
579    pub dedupe_key: Option<String>,
580    #[serde(default)]
581    pub deadline_ms: Option<WireU64>,
582    #[serde(default)]
583    pub coalesce_key: Option<String>,
584    pub coalesced_count: u32,
585    #[serde(default)]
586    pub recipient: Option<String>,
587    pub timestamp_ms: WireU64,
588    pub deadline_escalated: bool,
589    /// Every business key represented by this queue entry after coalescing.
590    #[serde(default)]
591    pub dedupe_keys: Vec<String>,
592}
593
594#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
595#[serde(deny_unknown_fields)]
596pub struct MilestoneState {
597    /// The contract whose cascade is installed. `phase_id` is unique only inside it, so the pair
598    /// is the host's complete lookup key (§7.9 note 6).
599    pub contract_id: String,
600    #[serde(default)]
601    pub phase_id: Option<String>,
602    pub complete: bool,
603    /// Consecutive blocks on the current phase — the retry budget. A restore that reset it would
604    /// hand a stalled cascade a fresh set of attempts.
605    #[serde(default)]
606    pub blocked_count: u32,
607}
608
609/// §12.1 · the P3 plane: the handle table and its allocator, skills and their leases, the
610/// knowledge slots, the signal partition and the compaction/renewal clocks.
611///
612/// What is here is everything the context VM *cannot re-derive*: identity (handle ids and the
613/// allocator that mints them), leases, pin/evict marks, the pending page-in verification targets,
614/// the clocks the decay ladders read — and, since Task 16, the **stored messages** themselves.
615///
616/// The message projection is what makes §12.2 true (adjudication §5q-2). Task 15 carried only token
617/// counts and lengths on the theory that a tail replay could rebuild the bodies; it cannot, because
618/// a tail that starts above genesis never replays the inputs that produced the older messages. And
619/// as long as the bodies were unrecoverable, acking a checkpoint and pruning the journal prefix
620/// destroyed the rendering input for good. So they are here — as [`StoredMessageState`], a *source*
621/// projection. §15.2's ban is on derived planned steps and rendered context;
622/// nothing here is either. A body that is over §7.10's inline threshold is **not** inlined: an
623/// `External`/`PagedOut` residency travels as its handle reference and digest, exactly as it does in
624/// working context.
625#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
626#[serde(deny_unknown_fields)]
627pub struct ContextVmStateV1 {
628    /// **The** home of P3 handle identity and residency.
629    #[serde(default)]
630    pub handles: Vec<HandleState>,
631    /// The monotonic allocator. Checkpointing it is what stops a restored kernel from re-issuing a
632    /// handle id that an outstanding effect still addresses.
633    pub next_handle_id: u32,
634    /// §7.10 rule 4 · the digest each pending `LoadPayload` will verify its body against. Held
635    /// here rather than re-read from the handle table at resolution time, so a residency that moved
636    /// in between cannot change what a page-in is checked against.
637    #[serde(default)]
638    pub pending_payload_loads: Vec<PendingPayloadLoadState>,
639    #[serde(default)]
640    pub active_skills: Vec<SkillLeaseState>,
641    #[serde(default)]
642    pub knowledge: Vec<KnowledgeSlotState>,
643    #[serde(default)]
644    pub signals: Vec<String>,
645    /// §5q-2 · the stored messages of the system and history partitions, in render order. **The**
646    /// home of message bodies; the knowledge partition carries its own inside
647    /// [`KnowledgeSlotState`], because a knowledge entry is an identified slot rather than a
648    /// positional message.
649    #[serde(default)]
650    pub messages: Vec<StoredMessageState>,
651    /// The durable task board (goal / plan / progress / directives). It renders into the prompt like
652    /// a message does, survives compression by construction, and is the one partition of §12.1's P3
653    /// plane that is neither a message list nor a handle.
654    pub task_state: LogicalTaskState,
655    pub partition_tokens: PartitionTokenState,
656    pub history_len: u32,
657    /// Message-count boundary projected as `frozen_prefix_len` on future provider effects.
658    #[serde(default)]
659    pub frozen_history_len: u32,
660    pub last_activity_ms: WireU64,
661    #[serde(default)]
662    pub last_compact_ms: Option<WireU64>,
663}
664
665/// Which partition a [`StoredMessageState`] belongs to.
666///
667/// Only the two positional partitions: knowledge entries are keyed slots with their own lifecycle
668/// flags, so they live in [`KnowledgeSlotState`] instead of being a third value here.
669#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
670#[serde(rename_all = "snake_case")]
671pub enum MessagePartition {
672    System,
673    History,
674}
675
676/// One stored message, projected (§12.1, adjudication §5q-2).
677///
678/// This is the *source* the renderer reads, not the rendered result: no prompt assembly, no
679/// residency projection, no salience footer. A restore rebuilds the partitions from these and then
680/// renders exactly what an uninterrupted run would have rendered.
681#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
682#[serde(deny_unknown_fields)]
683pub struct StoredMessageState {
684    pub partition: MessagePartition,
685    /// `system` | `user` | `assistant` | `tool`.
686    pub role: String,
687    pub body: StoredMessageBody,
688    /// The calls an assistant message asked for — the half of "tool association" that points
689    /// forward.
690    #[serde(default)]
691    pub tool_calls: Vec<LogicalToolCall>,
692    /// The cached token count the partition counter was built from. Carried rather than recomputed
693    /// so a restore reproduces the same budget arithmetic even if the tokenizer moved.
694    pub tokens: u32,
695}
696
697/// A message body, inline or by reference (§7.10).
698///
699/// The reference arm is the whole point: a tool result that was over the inline threshold when it
700/// was generated (`External`) or that left working context under pressure (`PagedOut`) is already
701/// represented in context by a preview plus a handle, and a checkpoint that re-inlined it would put
702/// bytes back into the journal that §7.10 spent an effect kind keeping out.
703#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
704#[serde(tag = "form", rename_all = "snake_case")]
705pub enum StoredMessageBody {
706    /// A body small enough to live in context.
707    Inline(InlineMessageBody),
708    /// A body that lives with the host. Carries the reference and the digest that verifies a
709    /// page-in, never the bytes.
710    Reference(ReferencedMessageBody),
711    /// A multimodal body the text projection cannot express (image or audio parts), carried as the
712    /// canonical JSON of its content.
713    ///
714    /// It exists so the projection is never *silently* lossy: a body that does not reduce to text
715    /// travels whole rather than being flattened to the text parts that happen to be next to it.
716    Structured(StructuredMessageBody),
717}
718
719#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
720#[serde(deny_unknown_fields)]
721pub struct InlineMessageBody {
722    pub text: String,
723    /// For a `tool` message: the call this result answers, and whether it failed.
724    #[serde(default)]
725    pub tool_call_id: Option<String>,
726    #[serde(default)]
727    pub is_error: bool,
728}
729
730#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
731#[serde(deny_unknown_fields)]
732pub struct ReferencedMessageBody {
733    /// The P3 handle that addresses the body. Its residency in [`ContextVmStateV1::handles`] is
734    /// what a page-in reads.
735    pub handle_id: u32,
736    /// The digest a page-in must reproduce (§7.10 rule 4).
737    pub digest: String,
738    /// What is actually resident: the preview the model can see. Never the whole body.
739    pub preview: String,
740    #[serde(default)]
741    pub tool_call_id: Option<String>,
742    #[serde(default)]
743    pub is_error: bool,
744}
745
746#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
747#[serde(deny_unknown_fields)]
748pub struct StructuredMessageBody {
749    /// Canonical JSON of the message's `content`.
750    pub content_json: String,
751}
752
753#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
754#[serde(deny_unknown_fields)]
755pub struct LogicalToolCall {
756    pub call_id: String,
757    pub name: String,
758    /// Canonical JSON text of the arguments. A string rather than a `Value` so the checkpoint's
759    /// canonical bytes are the arguments' canonical bytes, with no second serialiser in between.
760    pub arguments: String,
761}
762
763/// §12.1 · the durable task board, projected.
764///
765/// Explicitly re-declared rather than reusing `crate::context::task_state::TaskState`: that type is
766/// semantic-kernel state whose serde shape is free to move, and §12.1's first rule is that a field
767/// added there must not silently change the checkpoint format.
768#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
769#[serde(deny_unknown_fields)]
770pub struct LogicalTaskState {
771    #[serde(default)]
772    pub goal: String,
773    #[serde(default)]
774    pub criteria: Vec<String>,
775    #[serde(default)]
776    pub plan: Vec<LogicalPlanStep>,
777    #[serde(default)]
778    pub current_step: Option<u32>,
779    #[serde(default)]
780    pub progress: String,
781    #[serde(default)]
782    pub scratchpad: String,
783    #[serde(default)]
784    pub blocked_on: Vec<String>,
785    #[serde(default)]
786    pub directives: Vec<String>,
787    #[serde(default)]
788    pub preserved_refs: Vec<String>,
789    #[serde(default)]
790    pub recent_actions: Vec<String>,
791    #[serde(default)]
792    pub compression_log: Vec<LogicalCompressionEntry>,
793    #[serde(default)]
794    pub compression_log_dropped: WireU64,
795}
796
797#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
798#[serde(deny_unknown_fields)]
799pub struct LogicalPlanStep {
800    pub label: String,
801    pub done: bool,
802}
803
804#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
805#[serde(deny_unknown_fields)]
806pub struct LogicalCompressionEntry {
807    pub action: String,
808    pub summary: String,
809}
810
811/// One P3 handle. `residency` is the label plus the locator fields that residency carries, so the
812/// DTO neither mirrors the internal enum's shape nor loses what a page-in needs.
813#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
814#[serde(deny_unknown_fields)]
815pub struct HandleState {
816    pub handle_id: u32,
817    pub kind: String,
818    pub residency: String,
819    #[serde(default)]
820    pub payload_ref: Option<String>,
821    #[serde(default)]
822    pub digest: Option<String>,
823    #[serde(default)]
824    pub original_size: Option<WireU64>,
825    pub tokens: u32,
826    /// Link back to the source object in working context (a tool `call_id` for a tool result).
827    #[serde(default)]
828    pub source: Option<String>,
829}
830
831#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
832#[serde(deny_unknown_fields)]
833pub struct PendingPayloadLoadState {
834    pub effect_id: EffectId,
835    pub handle_id: String,
836    pub digest: String,
837    #[serde(default)]
838    pub original_size: Option<WireU64>,
839}
840
841#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
842#[serde(deny_unknown_fields)]
843pub struct SkillLeaseState {
844    pub skill: String,
845    /// `None` = permanent; otherwise the turn the lease expires on.
846    #[serde(default)]
847    pub lease_until_turn: Option<u32>,
848}
849
850/// One knowledge slot, body included.
851///
852/// A knowledge entry has identity (its key) and its own lifecycle flags, so it is not a positional
853/// [`StoredMessageState`] — but it renders into the prompt all the same, which is why Task 16 gave it
854/// the same body projection. Without it, a restore rebuilt the *shape* of the knowledge partition
855/// and none of its content.
856#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
857#[serde(deny_unknown_fields)]
858pub struct KnowledgeSlotState {
859    /// `None` = an unkeyed append; keyed entries upsert.
860    #[serde(default)]
861    pub key: Option<String>,
862    pub role: String,
863    pub body: StoredMessageBody,
864    pub tokens: u32,
865    pub pinned: bool,
866    pub evict_at_boundary: bool,
867}
868
869#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
870#[serde(deny_unknown_fields)]
871pub struct PartitionTokenState {
872    pub system: u32,
873    pub knowledge: u32,
874    pub history: u32,
875}
876
877/// What the semantic driver contributes to a checkpoint.
878///
879/// Three of the four partitions plus the two transition fields the driver — not the transaction —
880/// owns. It is a value, not a borrow of the driver: the checkpoint is built from an explicit
881/// projection, never from a live reference into the engine.
882#[derive(Debug, Clone, PartialEq)]
883pub struct LogicalStateProjection {
884    pub root_kind: Option<RootKind>,
885    pub focus: Option<ExecutionFocus>,
886    pub syscall: SyscallStateV1,
887    pub scheduler: SchedulerStateV1,
888    pub context_vm: ContextVmStateV1,
889}
890
891// ---------------------------------------------------------------------------------------------
892// §12.1 · the checkpoint
893// ---------------------------------------------------------------------------------------------
894
895/// Everything [`KernelCheckpoint::assemble`] needs. A struct rather than eight positional
896/// arguments, because two of them are step sequences and two are digests.
897#[derive(Debug, Clone, PartialEq)]
898pub struct CheckpointDraft {
899    pub operation_id: OperationId,
900    pub genesis_digest: Digest,
901    pub base_step_seq: WireU64,
902    pub base_record_digest: Digest,
903    pub through_step_seq: WireU64,
904    pub covered_transaction_head_digest: Digest,
905    pub logical_state: LogicalKernelState,
906    pub tail_inputs: Vec<CanonicalInput>,
907}
908
909/// One logical checkpoint (§12.1).
910///
911/// Fields are private and every digest is computed by [`Self::assemble`], so there is no
912/// constructor that takes a digest and "the host recomputed the hash and disagreed" is not a
913/// reachable state — the same discipline [`KernelRecord`](super::record::KernelRecord) uses.
914/// Decoding goes through the same verification, which is why a tampered blob fails at the boundary
915/// rather than half-way through a restore.
916#[derive(Debug, Clone, PartialEq, Serialize)]
917pub struct KernelCheckpoint {
918    checkpoint_version: CheckpointRevision,
919    abi_version: AbiRevision,
920    operation_id: OperationId,
921    /// The digest of the operation's genesis record — its identity. A checkpoint built from
922    /// another operation's journal therefore cannot be installed by accident.
923    genesis_digest: Digest,
924    /// The step the logical state below describes.
925    base_step_seq: WireU64,
926    /// The record digest at `base_step_seq` — the chain anchor a tail replay starts from.
927    ///
928    /// The other end of the range the header already states. Without it a rebase could be
929    /// *verified* and never *replayed*: the record before the first tail entry is exactly the one an
930    /// acked checkpoint is allowed to have pruned, so its digest has to travel with the tail that
931    /// depends on it. For a full-state checkpoint it is the covered head, because `base == through`.
932    base_record_digest: Digest,
933    /// The step the checkpoint covers once its tail is replayed.
934    through_step_seq: WireU64,
935    /// The record digest at `through_step_seq`. §12.3 rule 2: install checks that this names the
936    /// through step, **not** that it is still the current head.
937    covered_transaction_head_digest: Digest,
938    logical_state: LogicalKernelState,
939    tail_inputs: Vec<CanonicalInput>,
940    state_digest: Digest,
941    tail_digest: Digest,
942    checkpoint_digest: Digest,
943}
944
945/// The digested body: every field of a checkpoint except the digest that summarises it.
946#[derive(Serialize)]
947struct CheckpointBody<'a> {
948    checkpoint_version: CheckpointRevision,
949    abi_version: AbiRevision,
950    operation_id: &'a OperationId,
951    genesis_digest: &'a Digest,
952    base_step_seq: WireU64,
953    base_record_digest: &'a Digest,
954    through_step_seq: WireU64,
955    covered_transaction_head_digest: &'a Digest,
956    logical_state: &'a LogicalKernelState,
957    tail_inputs: &'a [CanonicalInput],
958    state_digest: &'a Digest,
959    tail_digest: &'a Digest,
960}
961
962impl KernelCheckpoint {
963    /// Build a checkpoint, computing all three digests and checking the tail covers
964    /// `(base_step_seq, through_step_seq]` exactly.
965    pub fn assemble(draft: CheckpointDraft) -> Result<Self, CheckpointError> {
966        let CheckpointDraft {
967            operation_id,
968            genesis_digest,
969            base_step_seq,
970            base_record_digest,
971            through_step_seq,
972            covered_transaction_head_digest,
973            logical_state,
974            tail_inputs,
975        } = draft;
976
977        check_tail(
978            &operation_id,
979            base_step_seq,
980            &base_record_digest,
981            through_step_seq,
982            &covered_transaction_head_digest,
983            &tail_inputs,
984        )?;
985
986        let state_digest = canonical_digest(canonical_bytes(&logical_state)?.as_slice());
987        let tail_digest = canonical_digest(canonical_bytes(&tail_inputs)?.as_slice());
988        let checkpoint_digest = Self::body_digest(&CheckpointBody {
989            checkpoint_version: CheckpointRevision::CURRENT,
990            abi_version: AbiRevision::CURRENT,
991            operation_id: &operation_id,
992            genesis_digest: &genesis_digest,
993            base_step_seq,
994            base_record_digest: &base_record_digest,
995            through_step_seq,
996            covered_transaction_head_digest: &covered_transaction_head_digest,
997            logical_state: &logical_state,
998            tail_inputs: &tail_inputs,
999            state_digest: &state_digest,
1000            tail_digest: &tail_digest,
1001        })?;
1002
1003        Ok(Self {
1004            checkpoint_version: CheckpointRevision::CURRENT,
1005            abi_version: AbiRevision::CURRENT,
1006            operation_id,
1007            genesis_digest,
1008            base_step_seq,
1009            base_record_digest,
1010            through_step_seq,
1011            covered_transaction_head_digest,
1012            logical_state,
1013            tail_inputs,
1014            state_digest,
1015            tail_digest,
1016            checkpoint_digest,
1017        })
1018    }
1019
1020    fn body_digest(body: &CheckpointBody<'_>) -> Result<Digest, CheckpointError> {
1021        Ok(canonical_digest(canonical_bytes(body)?.as_slice()))
1022    }
1023
1024    // ----- read-only accessors -----
1025
1026    pub fn checkpoint_version(&self) -> u32 {
1027        self.checkpoint_version.get()
1028    }
1029
1030    pub fn abi_version(&self) -> u32 {
1031        self.abi_version.get()
1032    }
1033
1034    pub fn operation_id(&self) -> &OperationId {
1035        &self.operation_id
1036    }
1037
1038    pub fn genesis_digest(&self) -> &Digest {
1039        &self.genesis_digest
1040    }
1041
1042    pub fn base_step_seq(&self) -> WireU64 {
1043        self.base_step_seq
1044    }
1045
1046    pub fn base_record_digest(&self) -> &Digest {
1047        &self.base_record_digest
1048    }
1049
1050    pub fn through_step_seq(&self) -> WireU64 {
1051        self.through_step_seq
1052    }
1053
1054    pub fn covered_transaction_head_digest(&self) -> &Digest {
1055        &self.covered_transaction_head_digest
1056    }
1057
1058    pub fn logical_state(&self) -> &LogicalKernelState {
1059        &self.logical_state
1060    }
1061
1062    pub fn tail_inputs(&self) -> &[CanonicalInput] {
1063        &self.tail_inputs
1064    }
1065
1066    pub fn state_digest(&self) -> &Digest {
1067        &self.state_digest
1068    }
1069
1070    pub fn tail_digest(&self) -> &Digest {
1071        &self.tail_digest
1072    }
1073
1074    pub fn checkpoint_digest(&self) -> &Digest {
1075        &self.checkpoint_digest
1076    }
1077
1078    // ----- projection -----
1079
1080    /// Canonical bytes of the whole checkpoint — the blob a host persists.
1081    pub fn checkpoint_bytes(&self) -> CanonicalBytes {
1082        canonical_bytes(self).expect("a checkpoint contains only canonical scalars")
1083    }
1084
1085    /// Decode a checkpoint from its stored bytes, verifying every digest and the tail coverage.
1086    pub fn from_checkpoint_bytes(bytes: &[u8]) -> Result<Self, CheckpointError> {
1087        let text = std::str::from_utf8(bytes).map_err(|error| {
1088            CheckpointError::NotCanonical(format!("checkpoint bytes are not UTF-8: {error}"))
1089        })?;
1090        serde_json::from_str(text).map_err(|error| decode_error(&error.to_string()))
1091    }
1092
1093    /// The prefix an ack of this checkpoint may reclaim (§12.3 rule 6).
1094    pub fn boundary(&self) -> super::transaction::CheckpointBoundary {
1095        super::transaction::CheckpointBoundary {
1096            through_step_seq: self.through_step_seq,
1097            covered_head: self.covered_transaction_head_digest.clone(),
1098        }
1099    }
1100
1101    /// The §12.3 candidate this checkpoint hands the host.
1102    pub fn into_candidate(self) -> CheckpointCandidate {
1103        let ack_token = ack_token_for(
1104            &self.operation_id,
1105            self.through_step_seq,
1106            &self.checkpoint_digest,
1107        );
1108        CheckpointCandidate {
1109            checkpoint_bytes: self.checkpoint_bytes(),
1110            through_step_seq: self.through_step_seq,
1111            covered_head: self.covered_transaction_head_digest.clone(),
1112            state_digest: self.state_digest.clone(),
1113            ack_token,
1114        }
1115    }
1116
1117    // ----- verification -----
1118
1119    /// Recompute every digest from the bytes this checkpoint carries and re-check its tail.
1120    ///
1121    /// The first two lines of §12.2's ladder. `verify_belongs_to` adds the operation/genesis half;
1122    /// the version halves are enforced by the decoder, which cannot produce a checkpoint whose
1123    /// `checkpoint_version` or `abi_version` this kernel does not read.
1124    pub fn verify(&self) -> Result<(), CheckpointError> {
1125        check_tail(
1126            &self.operation_id,
1127            self.base_step_seq,
1128            &self.base_record_digest,
1129            self.through_step_seq,
1130            &self.covered_transaction_head_digest,
1131            &self.tail_inputs,
1132        )?;
1133
1134        let state_digest = canonical_digest(canonical_bytes(&self.logical_state)?.as_slice());
1135        if state_digest != self.state_digest {
1136            return Err(CheckpointError::Corrupted(format!(
1137                "checkpoint {} through step {}: the logical state hashes to {state_digest}, \
1138                 but the checkpoint claims {}",
1139                self.operation_id, self.through_step_seq, self.state_digest
1140            )));
1141        }
1142        let tail_digest = canonical_digest(canonical_bytes(&self.tail_inputs)?.as_slice());
1143        if tail_digest != self.tail_digest {
1144            return Err(CheckpointError::Corrupted(format!(
1145                "checkpoint {} through step {}: the bounded tail hashes to {tail_digest}, \
1146                 but the checkpoint claims {}",
1147                self.operation_id, self.through_step_seq, self.tail_digest
1148            )));
1149        }
1150        let checkpoint_digest = Self::body_digest(&CheckpointBody {
1151            checkpoint_version: self.checkpoint_version,
1152            abi_version: self.abi_version,
1153            operation_id: &self.operation_id,
1154            genesis_digest: &self.genesis_digest,
1155            base_step_seq: self.base_step_seq,
1156            base_record_digest: &self.base_record_digest,
1157            through_step_seq: self.through_step_seq,
1158            covered_transaction_head_digest: &self.covered_transaction_head_digest,
1159            logical_state: &self.logical_state,
1160            tail_inputs: &self.tail_inputs,
1161            state_digest: &self.state_digest,
1162            tail_digest: &self.tail_digest,
1163        })?;
1164        if checkpoint_digest != self.checkpoint_digest {
1165            return Err(CheckpointError::Corrupted(format!(
1166                "checkpoint {} through step {}: the body hashes to {checkpoint_digest}, \
1167                 but the checkpoint claims {}",
1168                self.operation_id, self.through_step_seq, self.checkpoint_digest
1169            )));
1170        }
1171        Ok(())
1172    }
1173
1174    /// Whether this checkpoint is this operation's (§12.2 line 2).
1175    pub fn verify_belongs_to(
1176        &self,
1177        operation_id: &OperationId,
1178        genesis_digest: &Digest,
1179    ) -> Result<(), CheckpointError> {
1180        if &self.operation_id != operation_id {
1181            return Err(CheckpointError::Incompatible(format!(
1182                "checkpoint belongs to operation {}, this runtime to {operation_id}",
1183                self.operation_id
1184            )));
1185        }
1186        if &self.genesis_digest != genesis_digest {
1187            return Err(CheckpointError::Incompatible(format!(
1188                "checkpoint {operation_id} binds genesis {}, this journal's genesis is \
1189                 {genesis_digest}",
1190                self.genesis_digest
1191            )));
1192        }
1193        Ok(())
1194    }
1195}
1196
1197/// §12.1 · the bounded tail covers `(base_step_seq, through_step_seq]` exactly.
1198///
1199/// One walk catches all four failure modes the spec names: a hole (a gap in the sequence), a
1200/// duplicate (the same step twice), an out-of-range entry (before `base` or after `through`), and a
1201/// length that disagrees with the range. It also refuses a tail entry from another operation —
1202/// the cheapest way to notice a checkpoint assembled from two journals.
1203fn check_tail(
1204    operation_id: &OperationId,
1205    base_step_seq: WireU64,
1206    base_record_digest: &Digest,
1207    through_step_seq: WireU64,
1208    covered_transaction_head_digest: &Digest,
1209    tail_inputs: &[CanonicalInput],
1210) -> Result<(), CheckpointError> {
1211    if base_step_seq > through_step_seq {
1212        return Err(CheckpointError::Corrupted(format!(
1213            "checkpoint {operation_id} bases at step {base_step_seq} but covers only through \
1214             {through_step_seq}"
1215        )));
1216    }
1217    // The two ends of the range meet when the range is empty: a full-state checkpoint's base *is*
1218    // its covered head, and a header that disagreed with itself about that would hand a restore two
1219    // different anchors for one record.
1220    if base_step_seq == through_step_seq && base_record_digest != covered_transaction_head_digest {
1221        return Err(CheckpointError::Corrupted(format!(
1222            "checkpoint {operation_id} covers no tail, so its base {base_record_digest} and its \
1223             covered head {covered_transaction_head_digest} name the same record — but they differ"
1224        )));
1225    }
1226    let expected = through_step_seq.get() - base_step_seq.get();
1227    if tail_inputs.len() as u64 != expected {
1228        return Err(CheckpointError::Corrupted(format!(
1229            "checkpoint {operation_id} covers ({base_step_seq}, {through_step_seq}] — {expected} \
1230             inputs — but its bounded tail holds {}",
1231            tail_inputs.len()
1232        )));
1233    }
1234    for (offset, entry) in tail_inputs.iter().enumerate() {
1235        let want = base_step_seq.get() + offset as u64 + 1;
1236        if entry.step_seq.get() != want {
1237            return Err(CheckpointError::Corrupted(format!(
1238                "checkpoint {operation_id} bounded tail is not the contiguous range \
1239                 ({base_step_seq}, {through_step_seq}]: position {offset} is step {} where step \
1240                 {want} was due",
1241                entry.step_seq
1242            )));
1243        }
1244        if &entry.input.operation_id != operation_id {
1245            return Err(CheckpointError::Incompatible(format!(
1246                "checkpoint {operation_id} bounded tail carries an input of operation {} at step \
1247                 {}",
1248                entry.input.operation_id, entry.step_seq
1249            )));
1250        }
1251    }
1252    // The last tail entry *is* the covered head; a tail that ends somewhere else covers a different
1253    // prefix than the header claims.
1254    if let Some(last) = tail_inputs.last()
1255        && &last.record_digest != covered_transaction_head_digest
1256    {
1257        return Err(CheckpointError::Corrupted(format!(
1258            "checkpoint {operation_id} claims covered head {covered_transaction_head_digest}, but \
1259             its bounded tail ends at {} on step {}",
1260            last.record_digest, last.step_seq
1261        )));
1262    }
1263    Ok(())
1264}
1265
1266fn ack_token_for(
1267    operation_id: &OperationId,
1268    through_step_seq: WireU64,
1269    checkpoint_digest: &Digest,
1270) -> CheckpointAckToken {
1271    CheckpointAckToken::new(format!(
1272        "{operation_id}:checkpoint:{through_step_seq}:{checkpoint_digest}"
1273    ))
1274    .expect("an operation-scoped checkpoint ack token is always a legal branded ref")
1275}
1276
1277/// §12.3 · what `kernel.checkpoint_candidate()` hands the host.
1278///
1279/// Exactly the five values of the spec's arrow, and nothing that would let a host reconstruct the
1280/// checkpoint itself: `checkpoint_bytes` is opaque storage, `through_step_seq`/`covered_head` are
1281/// the install precondition, `state_digest` is what an installed blob is audited against, and
1282/// `ack_token` is the maintenance handle that closes the loop.
1283#[derive(Debug, Clone, PartialEq)]
1284pub struct CheckpointCandidate {
1285    pub checkpoint_bytes: CanonicalBytes,
1286    pub through_step_seq: WireU64,
1287    pub covered_head: Digest,
1288    pub state_digest: Digest,
1289    pub ack_token: CheckpointAckToken,
1290}
1291
1292impl CheckpointCandidate {
1293    /// The boundary this candidate would let an ack reclaim (§12.3 rule 6).
1294    pub fn boundary(&self) -> super::transaction::CheckpointBoundary {
1295        super::transaction::CheckpointBoundary {
1296            through_step_seq: self.through_step_seq,
1297            covered_head: self.covered_head.clone(),
1298        }
1299    }
1300
1301    /// Decode the blob back into a verified checkpoint — what an install path does before it
1302    /// writes anything.
1303    pub fn decode(&self) -> Result<KernelCheckpoint, CheckpointError> {
1304        KernelCheckpoint::from_checkpoint_bytes(self.checkpoint_bytes.as_slice())
1305    }
1306}
1307
1308// ---------------------------------------------------------------------------------------------
1309// decoding
1310// ---------------------------------------------------------------------------------------------
1311
1312/// Wire projection of a checkpoint, used only as the decode target, so [`KernelCheckpoint`]'s
1313/// fields stay private and every decoded checkpoint is verified before it exists.
1314#[derive(Deserialize)]
1315#[serde(deny_unknown_fields)]
1316struct CheckpointProjection {
1317    checkpoint_version: CheckpointRevision,
1318    abi_version: AbiRevision,
1319    operation_id: OperationId,
1320    genesis_digest: Digest,
1321    base_step_seq: WireU64,
1322    base_record_digest: Digest,
1323    through_step_seq: WireU64,
1324    covered_transaction_head_digest: Digest,
1325    logical_state: LogicalKernelState,
1326    tail_inputs: Vec<CanonicalInput>,
1327    state_digest: Digest,
1328    tail_digest: Digest,
1329    checkpoint_digest: Digest,
1330}
1331
1332/// Recover a rejection's class from the string `serde` hands back.
1333///
1334/// Three sources reach here: this module's own [`CheckpointError`] rendered by
1335/// [`fmt::Display`] (which names its code), the scalar layer's ABI-revision refusal, and
1336/// everything structural — an unknown field, a missing field, a value of the wrong shape.
1337fn decode_error(message: &str) -> CheckpointError {
1338    if message.contains(CHECKPOINT_ERROR_MARKER) {
1339        for code in [
1340            KernelFaultCode::CheckpointIncompatible,
1341            KernelFaultCode::CheckpointCorrupted,
1342        ] {
1343            if message.contains(&format!("{CHECKPOINT_ERROR_MARKER} ({})", code.as_str())) {
1344                return match code {
1345                    KernelFaultCode::CheckpointIncompatible => {
1346                        CheckpointError::Incompatible(message.to_string())
1347                    }
1348                    _ => CheckpointError::Corrupted(message.to_string()),
1349                };
1350            }
1351        }
1352        return CheckpointError::Corrupted(message.to_string());
1353    }
1354    if message.contains(SCALAR_ERROR_MARKER) && message.contains("ABI revision") {
1355        return CheckpointError::Incompatible(message.to_string());
1356    }
1357    CheckpointError::NotCanonical(format!("checkpoint does not decode: {message}"))
1358}
1359
1360impl<'de> Deserialize<'de> for KernelCheckpoint {
1361    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
1362        let projection = CheckpointProjection::deserialize(deserializer)?;
1363        let checkpoint = Self {
1364            checkpoint_version: projection.checkpoint_version,
1365            abi_version: projection.abi_version,
1366            operation_id: projection.operation_id,
1367            genesis_digest: projection.genesis_digest,
1368            base_step_seq: projection.base_step_seq,
1369            base_record_digest: projection.base_record_digest,
1370            through_step_seq: projection.through_step_seq,
1371            covered_transaction_head_digest: projection.covered_transaction_head_digest,
1372            logical_state: projection.logical_state,
1373            tail_inputs: projection.tail_inputs,
1374            state_digest: projection.state_digest,
1375            tail_digest: projection.tail_digest,
1376            checkpoint_digest: projection.checkpoint_digest,
1377        };
1378        checkpoint
1379            .verify()
1380            .map_err(|error| serde::de::Error::custom(error.to_string()))?;
1381        Ok(checkpoint)
1382    }
1383}
1384
1385#[cfg(test)]
1386mod tests {
1387    use std::collections::BTreeMap;
1388    use std::fs;
1389    use std::path::PathBuf;
1390
1391    use serde_json::Value;
1392
1393    use super::super::config::{ConfigDefaults, HostEffectSupport, OperationConfig};
1394    use super::super::effect::EffectKindTag;
1395    use super::super::envelope::{ConfigureOperation, KernelInput, WireEnvelope};
1396    use super::*;
1397
1398    // -----------------------------------------------------------------------------------------
1399    // helpers
1400    // -----------------------------------------------------------------------------------------
1401
1402    const OPERATION: &str = "op-checkpoint-1";
1403
1404    fn operation() -> OperationId {
1405        OperationId::new(OPERATION).unwrap()
1406    }
1407
1408    fn digest(label: &str) -> Digest {
1409        canonical_digest(label.as_bytes())
1410    }
1411
1412    fn normalized(input_id: &str, at: u64) -> NormalizedInput {
1413        let envelope = WireEnvelope::new(
1414            operation(),
1415            InputId::new(input_id).unwrap(),
1416            WireU64::new(at),
1417            KernelInput::ConfigureOperation(ConfigureOperation {
1418                config: OperationConfig {
1419                    host_effect_support: HostEffectSupport {
1420                        supported: vec![EffectKindTag::CallProvider],
1421                    },
1422                    ..OperationConfig::default()
1423                },
1424            }),
1425        );
1426        NormalizedInput::normalize(&envelope, &ConfigDefaults::default()).expect("normalizes")
1427    }
1428
1429    fn tail_entry(step_seq: u64) -> CanonicalInput {
1430        CanonicalInput {
1431            step_seq: WireU64::new(step_seq),
1432            record_digest: digest(&format!("record-{step_seq}")),
1433            input: normalized(&format!("in-{step_seq}"), 1_700_000_000_000 + step_seq),
1434        }
1435    }
1436
1437    fn resolved_config() -> ResolvedOperationConfig {
1438        OperationConfig {
1439            host_effect_support: HostEffectSupport {
1440                supported: vec![EffectKindTag::CallProvider],
1441            },
1442            ..OperationConfig::default()
1443        }
1444        .resolve(&ConfigDefaults::default())
1445        .expect("the default configuration resolves")
1446    }
1447
1448    fn logical_state() -> LogicalKernelState {
1449        LogicalKernelState {
1450            transition: TransitionStateV1 {
1451                lifecycle: OperationLifecycle::Running,
1452                resolved_config: resolved_config(),
1453                root_kind: Some(RootKind::Agent),
1454                focus: None,
1455                last_observed_at_ms: WireU64::new(1_700_000_002_000),
1456                pending_effects: Vec::new(),
1457                resolved_effects: Vec::new(),
1458                launch_tokens: Vec::new(),
1459                accepted_inputs: vec![AcceptedInputState {
1460                    input_id: InputId::new("in-configure").unwrap(),
1461                    step_seq: WireU64::ZERO,
1462                    record_digest: digest("record-0"),
1463                }],
1464                accepted_cancellation: None,
1465                terminal: None,
1466            },
1467            syscall: SyscallStateV1::default(),
1468            scheduler: SchedulerStateV1::default(),
1469            context_vm: ContextVmStateV1::default(),
1470        }
1471    }
1472
1473    fn draft(base: u64, through: u64, tail: Vec<CanonicalInput>) -> CheckpointDraft {
1474        CheckpointDraft {
1475            operation_id: operation(),
1476            genesis_digest: digest("genesis"),
1477            base_step_seq: WireU64::new(base),
1478            base_record_digest: digest(&format!("record-{base}")),
1479            through_step_seq: WireU64::new(through),
1480            covered_transaction_head_digest: digest(&format!("record-{through}")),
1481            logical_state: logical_state(),
1482            tail_inputs: tail,
1483        }
1484    }
1485
1486    fn checkpoint() -> KernelCheckpoint {
1487        KernelCheckpoint::assemble(draft(3, 3, Vec::new())).expect("assembles")
1488    }
1489
1490    /// Round-trip a checkpoint through JSON with one field rewritten — the cheapest way to test a
1491    /// tamper without a constructor that could produce one.
1492    fn tampered(edit: impl FnOnce(&mut serde_json::Map<String, Value>)) -> CheckpointError {
1493        let mut document: serde_json::Map<String, Value> =
1494            serde_json::from_slice(checkpoint().checkpoint_bytes().as_slice()).unwrap();
1495        edit(&mut document);
1496        let bytes = serde_json::to_vec(&document).unwrap();
1497        KernelCheckpoint::from_checkpoint_bytes(&bytes)
1498            .expect_err("a tampered checkpoint must not decode")
1499    }
1500
1501    // -----------------------------------------------------------------------------------------
1502    // §12.1 · shape
1503    // -----------------------------------------------------------------------------------------
1504
1505    #[test]
1506    fn a_checkpoint_states_its_own_versions_from_the_kernels_constants() {
1507        let checkpoint = checkpoint();
1508        assert_eq!(checkpoint.checkpoint_version(), KERNEL_CHECKPOINT_VERSION);
1509        assert_eq!(checkpoint.checkpoint_version(), 1, "§12.1 starts at 1");
1510        assert_eq!(
1511            checkpoint.abi_version(),
1512            super::super::KERNEL_ABI_VERSION,
1513            "DEC-6 · the ABI revision is read from core's constant, never copied"
1514        );
1515    }
1516
1517    /// The load-bearing invariant of §12.1: each piece of correctness state has exactly one home,
1518    /// and the header repeats none of it.
1519    #[test]
1520    fn single_ownership_is_structural() {
1521        let checkpoint = checkpoint();
1522        let document: Value =
1523            serde_json::from_slice(checkpoint.checkpoint_bytes().as_slice()).unwrap();
1524        let state = &document["logical_state"];
1525
1526        // 1. every owned key appears exactly once in the whole logical state document. This is a
1527        //    total scan, not a spot check: the partitions serialise their whole key set (no
1528        //    `skip_serializing_if`), so an empty vector is still a visible claim of ownership.
1529        for (owned, owner) in [
1530            ("pending_effects", "transition"),
1531            ("resolved_effects", "transition"),
1532            ("launch_tokens", "transition"),
1533            ("accepted_inputs", "transition"),
1534            ("accepted_cancellation", "transition"),
1535            ("terminal", "transition"),
1536            ("attempts", "scheduler"),
1537            ("tasks", "scheduler"),
1538            ("handles", "context_vm"),
1539            ("pending_payload_loads", "context_vm"),
1540            ("provider_calls", "syscall"),
1541        ] {
1542            let mut seen = Vec::new();
1543            for partition in ["transition", "syscall", "scheduler", "context_vm"] {
1544                if state[partition]
1545                    .as_object()
1546                    .map(|map| map.contains_key(owned))
1547                    .unwrap_or(false)
1548                {
1549                    seen.push(partition);
1550                }
1551            }
1552            assert_eq!(
1553                seen,
1554                vec![owner],
1555                "{owned} must live in exactly one partition"
1556            );
1557        }
1558
1559        // 2. the four partitions share no key at all
1560        let mut home: BTreeMap<String, &str> = BTreeMap::new();
1561        for partition in ["transition", "syscall", "scheduler", "context_vm"] {
1562            for key in state[partition]
1563                .as_object()
1564                .expect("a partition object")
1565                .keys()
1566            {
1567                if let Some(previous) = home.insert(key.clone(), partition) {
1568                    panic!("key {key} lives in both {previous} and {partition}");
1569                }
1570            }
1571        }
1572
1573        // 3. the header stores no sub-state
1574        let header: Vec<&String> = document
1575            .as_object()
1576            .unwrap()
1577            .keys()
1578            .filter(|key| home.contains_key(*key))
1579            .collect();
1580        assert!(
1581            header.is_empty(),
1582            "the checkpoint header duplicates sub-state: {header:?}"
1583        );
1584    }
1585
1586    /// The DTO must be buildable without touching the semantic engine — the whole point of the
1587    /// explicit projection. A default projection is a legal (empty) checkpoint.
1588    #[test]
1589    fn the_dto_is_constructible_without_any_state_machine() {
1590        let state = LogicalKernelState {
1591            transition: TransitionStateV1 {
1592                lifecycle: OperationLifecycle::Created,
1593                resolved_config: resolved_config(),
1594                root_kind: None,
1595                focus: None,
1596                last_observed_at_ms: WireU64::ZERO,
1597                pending_effects: Vec::new(),
1598                resolved_effects: Vec::new(),
1599                launch_tokens: Vec::new(),
1600                accepted_inputs: Vec::new(),
1601                accepted_cancellation: None,
1602                terminal: None,
1603            },
1604            syscall: SyscallStateV1::default(),
1605            scheduler: SchedulerStateV1::default(),
1606            context_vm: ContextVmStateV1::default(),
1607        };
1608        let mut draft = draft(0, 0, Vec::new());
1609        draft.logical_state = state;
1610        KernelCheckpoint::assemble(draft).expect("an empty logical state is still a checkpoint");
1611    }
1612
1613    // -----------------------------------------------------------------------------------------
1614    // §12.1 · digests
1615    // -----------------------------------------------------------------------------------------
1616
1617    #[test]
1618    fn the_three_digests_summarise_three_different_things() {
1619        let checkpoint = checkpoint();
1620        assert_eq!(
1621            checkpoint.state_digest(),
1622            &canonical_digest(
1623                canonical_bytes(checkpoint.logical_state())
1624                    .unwrap()
1625                    .as_slice()
1626            ),
1627        );
1628        assert_eq!(
1629            checkpoint.tail_digest(),
1630            &canonical_digest(
1631                canonical_bytes(checkpoint.tail_inputs())
1632                    .unwrap()
1633                    .as_slice()
1634            ),
1635        );
1636        assert_ne!(checkpoint.state_digest(), checkpoint.checkpoint_digest());
1637        assert_ne!(checkpoint.tail_digest(), checkpoint.checkpoint_digest());
1638        checkpoint
1639            .verify()
1640            .expect("a freshly built checkpoint verifies");
1641    }
1642
1643    /// The checkpoint digest covers the header too: moving `through_step_seq` without moving the
1644    /// digest is corruption, not a different-but-valid checkpoint.
1645    #[test]
1646    fn the_checkpoint_digest_covers_the_header_and_the_bounded_tail() {
1647        let with_tail = KernelCheckpoint::assemble(draft(3, 4, vec![tail_entry(4)])).unwrap();
1648        let without_tail = checkpoint();
1649        assert_eq!(
1650            with_tail.state_digest(),
1651            without_tail.state_digest(),
1652            "the same logical state digests the same either way"
1653        );
1654        assert_ne!(
1655            with_tail.checkpoint_digest(),
1656            without_tail.checkpoint_digest(),
1657            "but the checkpoint digest moves with the tail and the header"
1658        );
1659
1660        let error = tampered(|document| {
1661            document.insert("through_step_seq".to_string(), Value::String("9".into()));
1662        });
1663        assert_eq!(error.code(), KernelFaultCode::CheckpointCorrupted);
1664    }
1665
1666    #[test]
1667    fn a_checkpoint_round_trips_through_its_bytes() {
1668        let original = KernelCheckpoint::assemble(draft(2, 4, vec![tail_entry(3), tail_entry(4)]))
1669            .expect("a bounded-tail checkpoint assembles");
1670        let decoded =
1671            KernelCheckpoint::from_checkpoint_bytes(original.checkpoint_bytes().as_slice())
1672                .expect("its own bytes decode");
1673        assert_eq!(decoded, original);
1674        assert_eq!(decoded.tail_inputs().len(), 2);
1675    }
1676
1677    // -----------------------------------------------------------------------------------------
1678    // corruption and incompatibility
1679    // -----------------------------------------------------------------------------------------
1680
1681    #[test]
1682    fn a_digest_that_does_not_match_its_bytes_is_corruption() {
1683        for field in ["state_digest", "tail_digest", "checkpoint_digest"] {
1684            let error = tampered(|document| {
1685                document.insert(
1686                    field.to_string(),
1687                    Value::String(digest("bogus").to_string()),
1688                );
1689            });
1690            assert_eq!(
1691                error.code(),
1692                KernelFaultCode::CheckpointCorrupted,
1693                "{field} must fail closed"
1694            );
1695            assert!(
1696                error.to_string().contains(CHECKPOINT_ERROR_MARKER),
1697                "{field}: every rejection carries the classifier marker"
1698            );
1699        }
1700    }
1701
1702    #[test]
1703    fn a_logical_state_edited_after_the_fact_is_corruption() {
1704        let error = tampered(|document| {
1705            document["logical_state"]["transition"]["lifecycle"] = Value::String("failed".into());
1706        });
1707        assert_eq!(error.code(), KernelFaultCode::CheckpointCorrupted);
1708        assert!(error.message().contains("logical state hashes to"));
1709    }
1710
1711    #[test]
1712    fn an_unrecognised_checkpoint_version_is_incompatible() {
1713        for version in [0u64, 2, 99] {
1714            let error = tampered(|document| {
1715                document.insert("checkpoint_version".to_string(), Value::from(version));
1716            });
1717            assert_eq!(
1718                error.code(),
1719                KernelFaultCode::CheckpointIncompatible,
1720                "checkpoint version {version} must be refused, not guessed at"
1721            );
1722        }
1723    }
1724
1725    #[test]
1726    fn an_abi_revision_this_kernel_does_not_read_is_incompatible() {
1727        let error = tampered(|document| {
1728            document.insert(
1729                "abi_version".to_string(),
1730                Value::from(u64::from(super::super::KERNEL_ABI_VERSION) + 1),
1731            );
1732        });
1733        assert_eq!(error.code(), KernelFaultCode::CheckpointIncompatible);
1734    }
1735
1736    #[test]
1737    fn an_unknown_field_is_refused_rather_than_ignored() {
1738        let error = tampered(|document| {
1739            // §12.4 deleted `last_step`; a blob that still carries one is a v2 snapshot.
1740            document.insert("last_step".to_string(), Value::Null);
1741        });
1742        assert_eq!(error.code(), KernelFaultCode::MalformedEnvelope);
1743    }
1744
1745    #[test]
1746    fn a_checkpoint_from_another_operation_or_genesis_is_incompatible() {
1747        let checkpoint = checkpoint();
1748        let other = OperationId::new("op-checkpoint-2").unwrap();
1749
1750        let error = checkpoint
1751            .verify_belongs_to(&other, &digest("genesis"))
1752            .expect_err("another operation's checkpoint is not installable");
1753        assert_eq!(error.code(), KernelFaultCode::CheckpointIncompatible);
1754        assert!(error.message().contains("belongs to operation"));
1755
1756        let error = checkpoint
1757            .verify_belongs_to(&operation(), &digest("another-genesis"))
1758            .expect_err("a different genesis is a different operation");
1759        assert_eq!(error.code(), KernelFaultCode::CheckpointIncompatible);
1760        assert!(error.message().contains("binds genesis"));
1761
1762        checkpoint
1763            .verify_belongs_to(&operation(), &digest("genesis"))
1764            .expect("its own operation and genesis are accepted");
1765    }
1766
1767    // -----------------------------------------------------------------------------------------
1768    // §12.1 · the bounded tail covers (base, through] exactly
1769    // -----------------------------------------------------------------------------------------
1770
1771    #[test]
1772    fn a_tail_that_covers_the_range_exactly_is_accepted() {
1773        KernelCheckpoint::assemble(draft(0, 0, Vec::new())).expect("an empty range needs no tail");
1774        KernelCheckpoint::assemble(draft(
1775            2,
1776            5,
1777            vec![tail_entry(3), tail_entry(4), tail_entry(5)],
1778        ))
1779        .expect("(2, 5] is three contiguous inputs");
1780    }
1781
1782    #[test]
1783    fn a_tail_with_a_hole_is_refused() {
1784        let error = KernelCheckpoint::assemble(draft(
1785            2,
1786            5,
1787            vec![tail_entry(3), tail_entry(5), tail_entry(6)],
1788        ))
1789        .expect_err("step 4 is missing");
1790        assert_eq!(error.code(), KernelFaultCode::CheckpointCorrupted);
1791        assert!(error.message().contains("step 4 was due"), "{error}");
1792    }
1793
1794    #[test]
1795    fn a_tail_with_a_duplicate_is_refused() {
1796        let error = KernelCheckpoint::assemble(draft(
1797            2,
1798            5,
1799            vec![tail_entry(3), tail_entry(3), tail_entry(4)],
1800        ))
1801        .expect_err("step 3 appears twice");
1802        assert_eq!(error.code(), KernelFaultCode::CheckpointCorrupted);
1803        assert!(error.message().contains("contiguous range"), "{error}");
1804    }
1805
1806    #[test]
1807    fn a_tail_entry_outside_the_range_is_refused() {
1808        // before the base
1809        let error = KernelCheckpoint::assemble(draft(2, 4, vec![tail_entry(2), tail_entry(3)]))
1810            .expect_err("step 2 is the base, not part of (2, 4]");
1811        assert_eq!(error.code(), KernelFaultCode::CheckpointCorrupted);
1812
1813        // after the covered head
1814        let error = KernelCheckpoint::assemble(draft(2, 4, vec![tail_entry(3), tail_entry(9)]))
1815            .expect_err("step 9 is past the covered head");
1816        assert_eq!(error.code(), KernelFaultCode::CheckpointCorrupted);
1817    }
1818
1819    #[test]
1820    fn a_tail_whose_length_disagrees_with_the_range_is_refused() {
1821        let error = KernelCheckpoint::assemble(draft(2, 5, vec![tail_entry(3)]))
1822            .expect_err("(2, 5] is three inputs, not one");
1823        assert_eq!(error.code(), KernelFaultCode::CheckpointCorrupted);
1824        assert!(error.message().contains("bounded tail holds 1"), "{error}");
1825
1826        let error = KernelCheckpoint::assemble(draft(4, 2, Vec::new()))
1827            .expect_err("a base past the covered head is not a range at all");
1828        assert_eq!(error.code(), KernelFaultCode::CheckpointCorrupted);
1829    }
1830
1831    #[test]
1832    fn a_tail_input_from_another_operation_is_incompatible() {
1833        let mut foreign = tail_entry(3);
1834        foreign.input.operation_id = OperationId::new("op-checkpoint-2").unwrap();
1835        let error = KernelCheckpoint::assemble(draft(2, 3, vec![foreign]))
1836            .expect_err("a tail assembled from two journals is not a checkpoint");
1837        assert_eq!(error.code(), KernelFaultCode::CheckpointIncompatible);
1838    }
1839
1840    /// Decoding re-runs the coverage check, so a hole punched into a stored blob is caught at the
1841    /// boundary rather than half-way through a replay.
1842    #[test]
1843    fn a_tail_edited_in_storage_is_refused_at_decode() {
1844        let original = KernelCheckpoint::assemble(draft(2, 4, vec![tail_entry(3), tail_entry(4)]))
1845            .expect("assembles");
1846        let mut document: serde_json::Map<String, Value> =
1847            serde_json::from_slice(original.checkpoint_bytes().as_slice()).unwrap();
1848        let tail = document["tail_inputs"].as_array_mut().unwrap();
1849        tail.remove(0);
1850        let bytes = serde_json::to_vec(&document).unwrap();
1851        let error = KernelCheckpoint::from_checkpoint_bytes(&bytes)
1852            .expect_err("a truncated tail no longer covers its range");
1853        assert_eq!(error.code(), KernelFaultCode::CheckpointCorrupted);
1854    }
1855
1856    // -----------------------------------------------------------------------------------------
1857    // §12.3 · the candidate handle
1858    // -----------------------------------------------------------------------------------------
1859
1860    #[test]
1861    fn a_candidate_carries_the_five_values_of_the_spec_arrow() {
1862        let checkpoint = KernelCheckpoint::assemble(draft(3, 4, vec![tail_entry(4)])).unwrap();
1863        let expected_digest = checkpoint.checkpoint_digest().clone();
1864        let candidate = checkpoint.into_candidate();
1865
1866        assert_eq!(candidate.through_step_seq, WireU64::new(4));
1867        assert_eq!(candidate.covered_head, digest("record-4"));
1868        assert!(
1869            candidate.ack_token.as_str().contains(OPERATION)
1870                && candidate
1871                    .ack_token
1872                    .as_str()
1873                    .contains(expected_digest.as_str()),
1874            "the ack token names the checkpoint it acknowledges: {}",
1875            candidate.ack_token
1876        );
1877
1878        let decoded = candidate.decode().expect("the blob decodes and verifies");
1879        assert_eq!(decoded.checkpoint_digest(), &expected_digest);
1880        assert_eq!(decoded.state_digest(), &candidate.state_digest);
1881        assert_eq!(
1882            candidate.boundary().through_step_seq,
1883            candidate.through_step_seq
1884        );
1885    }
1886
1887    // -----------------------------------------------------------------------------------------
1888    // rejection fixtures
1889    // -----------------------------------------------------------------------------------------
1890
1891    fn fixture_dir() -> PathBuf {
1892        PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../tests/fixtures/kernel-wire")
1893    }
1894
1895    /// Regenerate every `reject_checkpoint_*` blob from this module's own constructors.
1896    ///
1897    /// The fixtures used to be hand-written, which made them a second, drifting copy of the
1898    /// checkpoint shape: adding one field to the DTO invalidated all eight, and each had to be
1899    /// edited by hand into a document that still failed for the *declared* reason rather than for
1900    /// "missing field". Deriving them means a shape change costs one `BLESS_KERNEL_RECORD_FIXTURES=1`
1901    /// run, and — more importantly — a fixture can never claim to test a corruption while actually
1902    /// testing a stale schema.
1903    #[test]
1904    fn bless_checkpoint_rejection_fixtures() {
1905        if std::env::var("BLESS_KERNEL_RECORD_FIXTURES").as_deref() != Ok("1") {
1906            return;
1907        }
1908        let dir = fixture_dir();
1909        for (name, expect, description, mutate) in rejection_cases() {
1910            let mut document: serde_json::Map<String, Value> =
1911                serde_json::from_slice(mutate.0.checkpoint_bytes().as_slice()).unwrap();
1912            (mutate.1)(&mut document);
1913            let fixture = serde_json::json!({
1914                "expect": expect,
1915                "description": description,
1916                "checkpoint": Value::Object(document),
1917            });
1918            let mut text = serde_json::to_string_pretty(&fixture).unwrap();
1919            text.push('\n');
1920            fs::write(dir.join(name), text).unwrap_or_else(|e| panic!("cannot bless {name}: {e}"));
1921        }
1922    }
1923
1924    #[allow(clippy::type_complexity)]
1925    fn rejection_cases() -> Vec<(
1926        &'static str,
1927        &'static str,
1928        &'static str,
1929        (
1930            KernelCheckpoint,
1931            Box<dyn Fn(&mut serde_json::Map<String, Value>)>,
1932        ),
1933    )> {
1934        let full = || checkpoint();
1935        let with_tail =
1936            || KernelCheckpoint::assemble(draft(2, 4, vec![tail_entry(3), tail_entry(4)])).unwrap();
1937        vec![
1938            (
1939                "reject_checkpoint_unknown_checkpoint_version.json",
1940                "checkpoint_incompatible",
1941                "A checkpoint of a revision this kernel does not read is refused at the decode \
1942                 boundary, never guessed at (spec 12.1, DEC-6).",
1943                (
1944                    full(),
1945                    Box::new(|d: &mut serde_json::Map<String, Value>| {
1946                        d.insert("checkpoint_version".into(), Value::from(99u64));
1947                    }) as Box<dyn Fn(&mut serde_json::Map<String, Value>)>,
1948                ),
1949            ),
1950            (
1951                "reject_checkpoint_abi_revision_future.json",
1952                "checkpoint_incompatible",
1953                "A checkpoint written against a newer wire revision is incompatible, not corrupt: \
1954                 the answer is a different checkpoint, not more tail (spec 16.2).",
1955                (
1956                    full(),
1957                    Box::new(|d: &mut serde_json::Map<String, Value>| {
1958                        d.insert(
1959                            "abi_version".into(),
1960                            Value::from(u64::from(super::super::KERNEL_ABI_VERSION) + 1),
1961                        );
1962                    }),
1963                ),
1964            ),
1965            (
1966                "reject_checkpoint_state_digest_mismatch.json",
1967                "checkpoint_corrupted",
1968                "The logical state does not hash to the digest the checkpoint claims (spec 12.1).",
1969                (
1970                    full(),
1971                    Box::new(|d: &mut serde_json::Map<String, Value>| {
1972                        d.insert(
1973                            "state_digest".into(),
1974                            Value::String(digest("bogus").to_string()),
1975                        );
1976                    }),
1977                ),
1978            ),
1979            (
1980                "reject_checkpoint_missing_field_checkpoint_digest.json",
1981                "malformed_envelope",
1982                "A structural refusal that names the field: a checkpoint without its own digest is \
1983                 not a checkpoint with an unverified digest.",
1984                (
1985                    full(),
1986                    Box::new(|d: &mut serde_json::Map<String, Value>| {
1987                        d.remove("checkpoint_digest");
1988                    }),
1989                ),
1990            ),
1991            (
1992                "reject_checkpoint_unknown_field_last_step.json",
1993                "malformed_envelope",
1994                "Spec 12.4 deleted `last_step`; a legacy blob that still carries one is refused \
1995                 rather than partially read.",
1996                (
1997                    full(),
1998                    Box::new(|d: &mut serde_json::Map<String, Value>| {
1999                        d.insert("last_step".into(), Value::Null);
2000                    }),
2001                ),
2002            ),
2003            (
2004                "reject_checkpoint_base_disagrees_with_covered_head.json",
2005                "checkpoint_corrupted",
2006                "A full-state checkpoint covers no tail, so its base and its covered head name the \
2007                 same record; a header that disagrees with itself would hand a restore two \
2008                 different chain anchors (spec 12.1, Task 16).",
2009                (
2010                    full(),
2011                    Box::new(|d: &mut serde_json::Map<String, Value>| {
2012                        d.insert(
2013                            "base_record_digest".into(),
2014                            Value::String(digest("another-record").to_string()),
2015                        );
2016                    }),
2017                ),
2018            ),
2019            (
2020                "reject_checkpoint_tail_hole.json",
2021                "checkpoint_corrupted",
2022                "The bounded tail must cover (base, through] with no hole (spec 12.1).",
2023                (
2024                    with_tail(),
2025                    Box::new(|d: &mut serde_json::Map<String, Value>| {
2026                        d["tail_inputs"].as_array_mut().unwrap().remove(0);
2027                    }),
2028                ),
2029            ),
2030            (
2031                "reject_checkpoint_tail_duplicate.json",
2032                "checkpoint_corrupted",
2033                "The bounded tail must cover (base, through] with no duplicate (spec 12.1).",
2034                (
2035                    with_tail(),
2036                    Box::new(|d: &mut serde_json::Map<String, Value>| {
2037                        let tail = d["tail_inputs"].as_array_mut().unwrap();
2038                        tail[1] = tail[0].clone();
2039                    }),
2040                ),
2041            ),
2042            (
2043                "reject_checkpoint_tail_foreign_operation.json",
2044                "checkpoint_incompatible",
2045                "A bounded tail assembled from two journals is not a checkpoint (spec 12.1).",
2046                (
2047                    with_tail(),
2048                    Box::new(|d: &mut serde_json::Map<String, Value>| {
2049                        d["tail_inputs"][0]["input"]["operation_id"] =
2050                            Value::String("op-checkpoint-2".into());
2051                    }),
2052                ),
2053            ),
2054            (
2055                "reject_checkpoint_tail_ends_off_the_covered_head.json",
2056                "checkpoint_corrupted",
2057                "The last bounded-tail entry *is* the covered head; a tail that ends somewhere \
2058                 else covers a different prefix than the header claims (spec 12.1, Task 16).",
2059                (
2060                    with_tail(),
2061                    Box::new(|d: &mut serde_json::Map<String, Value>| {
2062                        d["tail_inputs"][1]["record_digest"] =
2063                            Value::String(digest("some-other-record").to_string());
2064                    }),
2065                ),
2066            ),
2067        ]
2068    }
2069
2070    #[test]
2071    fn checkpoint_rejection_fixtures_fail_closed_with_the_declared_kind() {
2072        let dir = fixture_dir();
2073        let mut names: Vec<String> = fs::read_dir(&dir)
2074            .unwrap_or_else(|e| panic!("failed to read {}: {e}", dir.display()))
2075            .map(|entry| {
2076                entry
2077                    .expect("dir entry")
2078                    .file_name()
2079                    .to_string_lossy()
2080                    .to_string()
2081            })
2082            .filter(|name| name.starts_with("reject_checkpoint_") && name.ends_with(".json"))
2083            .collect();
2084        names.sort();
2085        assert!(
2086            names.len() >= 5,
2087            "too few checkpoint rejection fixtures: {names:?}"
2088        );
2089
2090        for name in names {
2091            let raw = fs::read_to_string(dir.join(&name)).expect("fixture reads");
2092            let fixture: Value = serde_json::from_str(&raw).expect("fixture is JSON");
2093            let expected = fixture["expect"]
2094                .as_str()
2095                .expect("every fixture declares `expect`");
2096            let bytes = serde_json::to_vec(&fixture["checkpoint"]).unwrap();
2097            let error = KernelCheckpoint::from_checkpoint_bytes(&bytes)
2098                .expect_err(&format!("{name}: expected a rejection"));
2099            assert_eq!(
2100                error.code().as_str(),
2101                expected,
2102                "{name}: {} (message: {})",
2103                error.code().as_str(),
2104                error.message()
2105            );
2106            // The `missing_field` / `unknown_field` naming convention has to mean something: both
2107            // are structural refusals, and both must name the field so a host can act on them.
2108            for (marker, needle) in [
2109                ("_missing_field_", "missing field"),
2110                ("_unknown_field_", "unknown field"),
2111            ] {
2112                if name.contains(marker) {
2113                    assert_eq!(
2114                        error.code(),
2115                        KernelFaultCode::MalformedEnvelope,
2116                        "{name}: a structural refusal is malformed_envelope"
2117                    );
2118                    assert!(
2119                        error.message().contains(needle),
2120                        "{name}: the rejection must say which field ({})",
2121                        error.message()
2122                    );
2123                }
2124            }
2125        }
2126    }
2127}