1use 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
50pub const CHECKPOINT_ERROR_MARKER: &str = "kernel checkpoint rejected";
56
57#[derive(Debug, Clone, PartialEq, Eq)]
64pub enum CheckpointError {
65 Incompatible(String),
68 Corrupted(String),
71 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 pub fn fault(&self) -> KernelFault {
94 KernelFault::new(self.code(), self.to_string())
95 }
96}
97
98impl fmt::Display for CheckpointError {
99 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#[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 CheckpointAckToken,
216 "checkpoint ack token"
217);
218
219#[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 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#[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#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
275#[serde(deny_unknown_fields)]
276pub struct TransitionStateV1 {
277 pub lifecycle: OperationLifecycle,
278 pub resolved_config: ResolvedOperationConfig,
285 #[serde(default)]
287 pub root_kind: Option<RootKind>,
288 #[serde(default)]
291 pub focus: Option<ExecutionFocus>,
292 pub last_observed_at_ms: WireU64,
295 #[serde(default)]
298 pub pending_effects: Vec<KernelEffect>,
299 #[serde(default)]
302 pub resolved_effects: Vec<ResolvedEffectState>,
303 #[serde(default)]
306 pub launch_tokens: Vec<LaunchTokenState>,
307 #[serde(default)]
310 pub accepted_inputs: Vec<AcceptedInputState>,
311 #[serde(default)]
314 pub accepted_cancellation: Option<AcceptedCancellationState>,
315 #[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#[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 pub command_digest: Digest,
356 pub input_id: InputId,
357 pub step_seq: WireU64,
358}
359
360#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
363#[serde(deny_unknown_fields)]
364pub struct SyscallStateV1 {
365 #[serde(default)]
368 pub policy_revision: Option<WireU64>,
369 #[serde(default)]
372 pub live_config: Option<ResolvedOperationConfig>,
373 #[serde(default)]
376 pub provider_calls: Vec<PendingProviderCallState>,
377 #[serde(default)]
379 pub consumed_call_ids: Vec<String>,
380 #[serde(default)]
383 pub authored_memory_writes: Vec<AuthoredMemoryWriteState>,
384 #[serde(default)]
385 pub authored_memory_queries: Vec<AuthoredMemoryQueryState>,
386 #[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 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#[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 #[serde(default)]
431 pub started_at_ms: Option<WireU64>,
432 #[serde(default)]
436 pub wall_budget_ms: Option<WireU64>,
437 #[serde(default)]
440 pub tasks: Vec<TaskControlState>,
441 #[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 #[serde(default)]
452 pub signal_dedupe_keys: Vec<String>,
453 #[serde(default)]
454 pub milestone: Option<MilestoneState>,
455 #[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 #[serde(default)]
467 pub window: Vec<EntropyTurnState>,
468 pub rollbacks_pending: u32,
470 pub disarmed: bool,
472 #[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#[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 #[serde(default)]
503 pub termination: Option<String>,
504 #[serde(default)]
505 pub wait: Option<String>,
506 #[serde(default)]
508 pub waiting_on: Vec<TaskId>,
509 #[serde(default)]
510 pub capability_ids: Vec<String>,
511 #[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 #[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 #[serde(default)]
543 pub nodes: Vec<WorkflowNodeState>,
544}
545
546#[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 #[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 #[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 pub contract_id: String,
600 #[serde(default)]
601 pub phase_id: Option<String>,
602 pub complete: bool,
603 #[serde(default)]
606 pub blocked_count: u32,
607}
608
609#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
626#[serde(deny_unknown_fields)]
627pub struct ContextVmStateV1 {
628 #[serde(default)]
630 pub handles: Vec<HandleState>,
631 pub next_handle_id: u32,
634 #[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 #[serde(default)]
650 pub messages: Vec<StoredMessageState>,
651 pub task_state: LogicalTaskState,
655 pub partition_tokens: PartitionTokenState,
656 pub history_len: u32,
657 #[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#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
670#[serde(rename_all = "snake_case")]
671pub enum MessagePartition {
672 System,
673 History,
674}
675
676#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
682#[serde(deny_unknown_fields)]
683pub struct StoredMessageState {
684 pub partition: MessagePartition,
685 pub role: String,
687 pub body: StoredMessageBody,
688 #[serde(default)]
691 pub tool_calls: Vec<LogicalToolCall>,
692 pub tokens: u32,
695}
696
697#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
704#[serde(tag = "form", rename_all = "snake_case")]
705pub enum StoredMessageBody {
706 Inline(InlineMessageBody),
708 Reference(ReferencedMessageBody),
711 Structured(StructuredMessageBody),
717}
718
719#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
720#[serde(deny_unknown_fields)]
721pub struct InlineMessageBody {
722 pub text: String,
723 #[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 pub handle_id: u32,
736 pub digest: String,
738 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 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 pub arguments: String,
761}
762
763#[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#[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 #[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 #[serde(default)]
847 pub lease_until_turn: Option<u32>,
848}
849
850#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
857#[serde(deny_unknown_fields)]
858pub struct KnowledgeSlotState {
859 #[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#[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#[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#[derive(Debug, Clone, PartialEq, Serialize)]
917pub struct KernelCheckpoint {
918 checkpoint_version: CheckpointRevision,
919 abi_version: AbiRevision,
920 operation_id: OperationId,
921 genesis_digest: Digest,
924 base_step_seq: WireU64,
926 base_record_digest: Digest,
933 through_step_seq: WireU64,
935 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#[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 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 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 pub fn checkpoint_bytes(&self) -> CanonicalBytes {
1082 canonical_bytes(self).expect("a checkpoint contains only canonical scalars")
1083 }
1084
1085 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 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 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 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 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
1197fn 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 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 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#[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 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 pub fn decode(&self) -> Result<KernelCheckpoint, CheckpointError> {
1304 KernelCheckpoint::from_checkpoint_bytes(self.checkpoint_bytes.as_slice())
1305 }
1306}
1307
1308#[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
1332fn 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 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 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 #[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 #[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 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 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 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 #[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 #[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 #[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 #[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 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 #[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 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 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 #[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 #[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 fn fixture_dir() -> PathBuf {
1892 PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../tests/fixtures/kernel-wire")
1893 }
1894
1895 #[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 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}