use std::fmt;
use serde::de::{self, Deserializer, Visitor};
use serde::{Deserialize, Serialize, Serializer};
use super::config::ResolvedOperationConfig;
use super::effect::{Digest, KernelEffect, LaunchToken, wire_opaque_ref};
use super::envelope::OperationLifecycle;
use super::fault::{KernelFault, KernelFaultCode};
use super::record::{NormalizedInput, RecordError, canonical_bytes, canonical_digest};
use super::root::{ExecutionFocus, LogicalAgentSpec, LogicalTask, RootKind};
use super::scalar::{
AttemptId, BoundedJson, CanonicalBytes, EffectId, InputId, MemoryBindingId, NodeId,
OperationId, SCALAR_ERROR_MARKER, SignalId, TaskId, WireScalarError, WireU64, WorkflowId,
};
use super::syscall::MemoryKind;
use super::terminal::KernelTerminal;
pub const CHECKPOINT_ERROR_MARKER: &str = "kernel checkpoint rejected";
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CheckpointError {
Incompatible(String),
Corrupted(String),
NotCanonical(String),
}
impl CheckpointError {
pub fn message(&self) -> &str {
match self {
Self::Incompatible(message)
| Self::Corrupted(message)
| Self::NotCanonical(message) => message,
}
}
pub fn code(&self) -> KernelFaultCode {
match self {
Self::Incompatible(_) => KernelFaultCode::CheckpointIncompatible,
Self::Corrupted(_) => KernelFaultCode::CheckpointCorrupted,
Self::NotCanonical(_) => KernelFaultCode::MalformedEnvelope,
}
}
pub fn fault(&self) -> KernelFault {
KernelFault::new(self.code(), self.to_string())
}
}
impl fmt::Display for CheckpointError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(
f,
"{CHECKPOINT_ERROR_MARKER} ({}): {}",
self.code().as_str(),
self.message()
)
}
}
impl std::error::Error for CheckpointError {}
impl From<RecordError> for CheckpointError {
fn from(error: RecordError) -> Self {
Self::NotCanonical(error.message().to_string())
}
}
wire_opaque_ref!(
CheckpointAckToken,
"checkpoint ack token"
);
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct CanonicalInput {
pub step_seq: WireU64,
pub record_digest: Digest,
pub input: NormalizedInput,
}
impl CanonicalInput {
pub fn from_record(record: &super::record::KernelRecord) -> Result<Self, CheckpointError> {
Ok(Self {
step_seq: record.step_seq(),
record_digest: record.record_digest().clone(),
input: record.normalized_input()?,
})
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct LogicalKernelState {
pub transition: TransitionState,
pub syscall: SyscallState,
pub scheduler: SchedulerState,
pub context_vm: ContextVmState,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TransitionState {
pub lifecycle: OperationLifecycle,
pub resolved_config: ResolvedOperationConfig,
#[serde(default)]
pub root_kind: Option<RootKind>,
#[serde(default)]
pub focus: Option<ExecutionFocus>,
pub last_observed_at_ms: WireU64,
#[serde(default)]
pub pending_effects: Vec<KernelEffect>,
#[serde(default)]
pub resolved_effects: Vec<ResolvedEffectState>,
#[serde(default)]
pub launch_tokens: Vec<LaunchTokenState>,
#[serde(default)]
pub accepted_inputs: Vec<AcceptedInputState>,
#[serde(default)]
pub accepted_cancellation: Option<AcceptedCancellationState>,
#[serde(default)]
pub terminal: Option<KernelTerminal>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ResolvedEffectState {
pub effect_id: EffectId,
pub outcome_digest: Digest,
pub input_id: InputId,
pub step_seq: WireU64,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct LaunchTokenState {
pub launch_token: LaunchToken,
pub step_seq: WireU64,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct AcceptedInputState {
pub input_id: InputId,
pub step_seq: WireU64,
pub record_digest: Digest,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct AcceptedCancellationState {
pub command_digest: Digest,
pub input_id: InputId,
pub step_seq: WireU64,
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SyscallState {
#[serde(default)]
pub policy_revision: Option<WireU64>,
#[serde(default)]
pub live_config: Option<ResolvedOperationConfig>,
#[serde(default)]
pub provider_calls: Vec<PendingProviderCallState>,
#[serde(default)]
pub consumed_call_ids: Vec<String>,
#[serde(default)]
pub authored_memory_writes: Vec<AuthoredMemoryWriteState>,
#[serde(default)]
pub authored_memory_queries: Vec<AuthoredMemoryQueryState>,
#[serde(default)]
pub memory_write_window_ms: Vec<WireU64>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct PendingProviderCallState {
pub effect_id: EffectId,
pub task_id: TaskId,
pub exposed_tools: Vec<String>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct AuthoredMemoryWriteState {
pub effect_id: EffectId,
pub binding_id: MemoryBindingId,
pub name: String,
pub kind: MemoryKind,
pub size_bytes: u32,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct AuthoredMemoryQueryState {
pub effect_id: EffectId,
pub binding_id: MemoryBindingId,
pub text: String,
pub requested_k: u32,
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SchedulerState {
pub run_spec: Option<LogicalAgentSpec>,
pub advertised_tool_ids: Option<Vec<String>>,
pub turn: u32,
pub total_tokens: WireU64,
pub rounds_completed: u32,
pub subagents_spawned: u32,
#[serde(default)]
pub started_at_ms: Option<WireU64>,
#[serde(default)]
pub wall_budget_ms: Option<WireU64>,
#[serde(default)]
pub tasks: Vec<TaskControlState>,
#[serde(default)]
pub attempts: Vec<TaskAttemptState>,
#[serde(default)]
pub workflow: Option<WorkflowGraphState>,
#[serde(default)]
pub queued_signals: Vec<QueuedSignalState>,
#[serde(default)]
pub signal_dedupe_keys: Vec<String>,
#[serde(default)]
pub milestone: Option<MilestoneState>,
#[serde(default)]
pub entropy: EntropyState,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub channels: Vec<LocalChannelState>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub objects: Vec<crate::mm::handle::ObjectDescriptor>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct LocalChannelState {
pub channel_id: String,
pub channel: crate::scheduler::mailbox::Channel,
}
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EntropyState {
#[serde(default)]
pub window: Vec<EntropyTurnState>,
pub rollbacks_pending: u32,
pub disarmed: bool,
#[serde(default)]
pub last_alert_turn: Option<u32>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EntropyTurnState {
pub errored_results: u32,
pub total_results: u32,
pub rollbacks: u32,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TaskControlState {
pub task_id: TaskId,
#[serde(default)]
pub parent_task_id: Option<TaskId>,
pub lifecycle: String,
#[serde(default, skip_serializing_if = "is_nested_runnable_cause")]
pub runnable_cause: crate::scheduler::tcb::RunnableCause,
#[serde(default)]
pub termination: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub wait_set: Option<TaskWaitSetState>,
#[serde(default)]
pub capability_ids: Vec<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub capabilities: Vec<crate::types::capability::Capability>,
#[serde(default)]
pub process: Option<ChildProcessState>,
#[serde(default, skip_serializing_if = "is_default_supervision")]
pub supervision: crate::scheduler::tcb::SupervisionPolicy,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub supervision_events: Vec<crate::scheduler::tcb::SupervisionEvent>,
pub tokens_used: WireU64,
pub turns_used: u32,
#[serde(default)]
pub child_budget_remaining: Option<crate::scheduler::budget_grant::ResourceBudget>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub budget_grant: Option<crate::scheduler::budget_grant::BudgetGrant>,
#[serde(
default,
skip_serializing_if = "crate::scheduler::mailbox::Mailbox::is_empty"
)]
pub mailbox: crate::scheduler::mailbox::Mailbox,
}
fn is_default_supervision(value: &crate::scheduler::tcb::SupervisionPolicy) -> bool {
value == &crate::scheduler::tcb::SupervisionPolicy::default()
}
fn is_nested_runnable_cause(value: &crate::scheduler::tcb::RunnableCause) -> bool {
*value == crate::scheduler::tcb::RunnableCause::NestedTask
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TaskWaitSetState {
pub mode: String,
pub conditions: Vec<TaskWaitConditionState>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub satisfied: Vec<u32>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum TaskWaitConditionState {
Effect { effect_id: EffectId },
Child { task_id: TaskId },
Children { task_ids: Vec<TaskId> },
Approval { approval_id: String },
Signal { filter: String },
Timer { deadline_ms: WireU64 },
Channel { channel_id: String },
Resource { resource_key: String },
External { subscription_id: String },
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ChildProcessState {
pub role: String,
pub isolation: String,
pub context_inheritance: String,
#[serde(default)]
pub join_result: Option<BoundedJson>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TaskAttemptState {
pub task_id: TaskId,
pub attempt_id: AttemptId,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WorkflowGraphState {
pub workflow_id: WorkflowId,
#[serde(default)]
pub nodes: Vec<WorkflowNodeState>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WorkflowNodeState {
pub node_id: NodeId,
pub task: LogicalTask,
#[serde(default)]
pub depends_on: Vec<NodeId>,
#[serde(default)]
pub run_spec: Option<LogicalAgentSpec>,
pub kind: String,
pub status: String,
#[serde(default)]
pub active_agent_id: Option<String>,
#[serde(default)]
pub iterations_completed: u32,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct QueuedSignalState {
pub signal_id: SignalId,
pub source: String,
pub signal_type: String,
pub urgency: String,
pub summary: String,
#[serde(default)]
pub payload: BoundedJson,
#[serde(default)]
pub dedupe_key: Option<String>,
#[serde(default)]
pub deadline_ms: Option<WireU64>,
#[serde(default)]
pub coalesce_key: Option<String>,
pub coalesced_count: u32,
#[serde(default)]
pub recipient: Option<String>,
pub timestamp_ms: WireU64,
pub deadline_escalated: bool,
#[serde(default)]
pub dedupe_keys: Vec<String>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct MilestoneState {
pub contract_id: String,
#[serde(default)]
pub phase_id: Option<String>,
pub complete: bool,
#[serde(default)]
pub blocked_count: u32,
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ContextVmState {
#[serde(default)]
pub handles: Vec<HandleState>,
pub next_handle_id: u32,
#[serde(default)]
pub pending_payload_loads: Vec<PendingPayloadLoadState>,
#[serde(default)]
pub active_skills: Vec<SkillLeaseState>,
#[serde(default)]
pub knowledge: Vec<KnowledgeSlotState>,
#[serde(default)]
pub signals: Vec<String>,
#[serde(default)]
pub messages: Vec<StoredMessageState>,
pub task_state: LogicalTaskState,
pub partition_tokens: PartitionTokenState,
pub history_len: u32,
#[serde(default)]
pub frozen_history_len: u32,
pub last_activity_ms: WireU64,
#[serde(default)]
pub last_compact_ms: Option<WireU64>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum MessagePartition {
System,
History,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct StoredMessageState {
pub partition: MessagePartition,
pub role: String,
pub body: StoredMessageBody,
#[serde(default)]
pub tool_calls: Vec<LogicalToolCall>,
pub tokens: u32,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "form", rename_all = "snake_case")]
pub enum StoredMessageBody {
Inline(InlineMessageBody),
Reference(ReferencedMessageBody),
Structured(StructuredMessageBody),
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct InlineMessageBody {
pub text: String,
#[serde(default)]
pub tool_call_id: Option<String>,
#[serde(default)]
pub is_error: bool,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ReferencedMessageBody {
pub handle_id: u32,
pub digest: String,
pub preview: String,
#[serde(default)]
pub tool_call_id: Option<String>,
#[serde(default)]
pub is_error: bool,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct StructuredMessageBody {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub durable_content: Option<crate::types::durable_content::DurableContent>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub durable_tool_results: Vec<crate::types::durable_content::DurableToolResult>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct LogicalToolCall {
pub call_id: String,
pub name: String,
pub arguments: String,
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct LogicalTaskState {
#[serde(default)]
pub goal: String,
#[serde(default)]
pub criteria: Vec<String>,
#[serde(default)]
pub plan: Vec<LogicalPlanStep>,
#[serde(default)]
pub current_step: Option<u32>,
#[serde(default)]
pub progress: String,
#[serde(default)]
pub scratchpad: String,
#[serde(default)]
pub blocked_on: Vec<String>,
#[serde(default)]
pub directives: Vec<String>,
#[serde(default)]
pub preserved_refs: Vec<String>,
#[serde(default)]
pub recent_actions: Vec<String>,
#[serde(default)]
pub compression_log: Vec<LogicalCompressionEntry>,
#[serde(default)]
pub compression_log_dropped: WireU64,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct LogicalPlanStep {
pub label: String,
pub done: bool,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct LogicalCompressionEntry {
pub action: String,
pub summary: String,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct HandleState {
pub handle_id: u32,
pub kind: String,
pub residency: String,
#[serde(default)]
pub payload_ref: Option<String>,
#[serde(default)]
pub digest: Option<String>,
#[serde(default)]
pub original_size: Option<WireU64>,
pub tokens: u32,
#[serde(default)]
pub source: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct PendingPayloadLoadState {
pub effect_id: EffectId,
pub handle_id: String,
pub digest: String,
#[serde(default)]
pub original_size: Option<WireU64>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SkillLeaseState {
pub skill: String,
#[serde(default)]
pub lease_until_turn: Option<u32>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct KnowledgeSlotState {
#[serde(default)]
pub key: Option<String>,
pub role: String,
pub body: StoredMessageBody,
pub tokens: u32,
pub pinned: bool,
pub evict_at_boundary: bool,
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct PartitionTokenState {
pub system: u32,
pub knowledge: u32,
pub history: u32,
}
#[derive(Debug, Clone, PartialEq)]
pub struct LogicalStateProjection {
pub root_kind: Option<RootKind>,
pub focus: Option<ExecutionFocus>,
pub syscall: SyscallState,
pub scheduler: SchedulerState,
pub context_vm: ContextVmState,
}
#[derive(Debug, Clone, PartialEq)]
pub struct CheckpointDraft {
pub operation_id: OperationId,
pub genesis_digest: Digest,
pub base_step_seq: WireU64,
pub base_record_digest: Digest,
pub through_step_seq: WireU64,
pub covered_transaction_head_digest: Digest,
pub logical_state: LogicalKernelState,
pub tail_inputs: Vec<CanonicalInput>,
}
#[derive(Debug, Clone, PartialEq, Serialize)]
pub struct KernelCheckpoint {
operation_id: OperationId,
genesis_digest: Digest,
base_step_seq: WireU64,
base_record_digest: Digest,
through_step_seq: WireU64,
covered_transaction_head_digest: Digest,
logical_state: LogicalKernelState,
tail_inputs: Vec<CanonicalInput>,
state_digest: Digest,
tail_digest: Digest,
checkpoint_digest: Digest,
}
#[derive(Serialize)]
struct CheckpointBody<'a> {
operation_id: &'a OperationId,
genesis_digest: &'a Digest,
base_step_seq: WireU64,
base_record_digest: &'a Digest,
through_step_seq: WireU64,
covered_transaction_head_digest: &'a Digest,
logical_state: &'a LogicalKernelState,
tail_inputs: &'a [CanonicalInput],
state_digest: &'a Digest,
tail_digest: &'a Digest,
}
impl KernelCheckpoint {
pub fn assemble(draft: CheckpointDraft) -> Result<Self, CheckpointError> {
let CheckpointDraft {
operation_id,
genesis_digest,
base_step_seq,
base_record_digest,
through_step_seq,
covered_transaction_head_digest,
logical_state,
tail_inputs,
} = draft;
validate_durable_message_bodies(&logical_state.context_vm)?;
check_tail(
&operation_id,
base_step_seq,
&base_record_digest,
through_step_seq,
&covered_transaction_head_digest,
&tail_inputs,
)?;
let state_digest = canonical_digest(canonical_bytes(&logical_state)?.as_slice());
let tail_digest = canonical_digest(canonical_bytes(&tail_inputs)?.as_slice());
let checkpoint_digest = Self::body_digest(&CheckpointBody {
operation_id: &operation_id,
genesis_digest: &genesis_digest,
base_step_seq,
base_record_digest: &base_record_digest,
through_step_seq,
covered_transaction_head_digest: &covered_transaction_head_digest,
logical_state: &logical_state,
tail_inputs: &tail_inputs,
state_digest: &state_digest,
tail_digest: &tail_digest,
})?;
Ok(Self {
operation_id,
genesis_digest,
base_step_seq,
base_record_digest,
through_step_seq,
covered_transaction_head_digest,
logical_state,
tail_inputs,
state_digest,
tail_digest,
checkpoint_digest,
})
}
fn body_digest(body: &CheckpointBody<'_>) -> Result<Digest, CheckpointError> {
Ok(canonical_digest(canonical_bytes(body)?.as_slice()))
}
pub fn operation_id(&self) -> &OperationId {
&self.operation_id
}
pub fn genesis_digest(&self) -> &Digest {
&self.genesis_digest
}
pub fn base_step_seq(&self) -> WireU64 {
self.base_step_seq
}
pub fn base_record_digest(&self) -> &Digest {
&self.base_record_digest
}
pub fn through_step_seq(&self) -> WireU64 {
self.through_step_seq
}
pub fn covered_transaction_head_digest(&self) -> &Digest {
&self.covered_transaction_head_digest
}
pub fn logical_state(&self) -> &LogicalKernelState {
&self.logical_state
}
pub fn tail_inputs(&self) -> &[CanonicalInput] {
&self.tail_inputs
}
pub fn state_digest(&self) -> &Digest {
&self.state_digest
}
pub fn tail_digest(&self) -> &Digest {
&self.tail_digest
}
pub fn checkpoint_digest(&self) -> &Digest {
&self.checkpoint_digest
}
pub fn checkpoint_bytes(&self) -> CanonicalBytes {
canonical_bytes(self).expect("a checkpoint contains only canonical scalars")
}
pub fn from_checkpoint_bytes(bytes: &[u8]) -> Result<Self, CheckpointError> {
let text = std::str::from_utf8(bytes).map_err(|error| {
CheckpointError::NotCanonical(format!("checkpoint bytes are not UTF-8: {error}"))
})?;
let document =
serde_json::from_str(text).map_err(|error| decode_error(&error.to_string()))?;
decode_checkpoint_value(document)
}
pub fn boundary(&self) -> super::transaction::CheckpointBoundary {
super::transaction::CheckpointBoundary {
through_step_seq: self.through_step_seq,
covered_head: self.covered_transaction_head_digest.clone(),
}
}
pub fn into_candidate(self) -> CheckpointCandidate {
let ack_token = ack_token_for(
&self.operation_id,
self.through_step_seq,
&self.checkpoint_digest,
);
CheckpointCandidate {
checkpoint_bytes: self.checkpoint_bytes(),
through_step_seq: self.through_step_seq,
covered_head: self.covered_transaction_head_digest.clone(),
state_digest: self.state_digest.clone(),
ack_token,
}
}
pub fn verify(&self) -> Result<(), CheckpointError> {
validate_durable_message_bodies(&self.logical_state.context_vm)?;
check_tail(
&self.operation_id,
self.base_step_seq,
&self.base_record_digest,
self.through_step_seq,
&self.covered_transaction_head_digest,
&self.tail_inputs,
)?;
let state_digest = canonical_digest(canonical_bytes(&self.logical_state)?.as_slice());
if state_digest != self.state_digest {
return Err(CheckpointError::Corrupted(format!(
"checkpoint {} through step {}: the logical state hashes to {state_digest}, \
but the checkpoint claims {}",
self.operation_id, self.through_step_seq, self.state_digest
)));
}
let tail_digest = canonical_digest(canonical_bytes(&self.tail_inputs)?.as_slice());
if tail_digest != self.tail_digest {
return Err(CheckpointError::Corrupted(format!(
"checkpoint {} through step {}: the bounded tail hashes to {tail_digest}, \
but the checkpoint claims {}",
self.operation_id, self.through_step_seq, self.tail_digest
)));
}
let checkpoint_digest = Self::body_digest(&CheckpointBody {
operation_id: &self.operation_id,
genesis_digest: &self.genesis_digest,
base_step_seq: self.base_step_seq,
base_record_digest: &self.base_record_digest,
through_step_seq: self.through_step_seq,
covered_transaction_head_digest: &self.covered_transaction_head_digest,
logical_state: &self.logical_state,
tail_inputs: &self.tail_inputs,
state_digest: &self.state_digest,
tail_digest: &self.tail_digest,
})?;
if checkpoint_digest != self.checkpoint_digest {
return Err(CheckpointError::Corrupted(format!(
"checkpoint {} through step {}: the body hashes to {checkpoint_digest}, \
but the checkpoint claims {}",
self.operation_id, self.through_step_seq, self.checkpoint_digest
)));
}
Ok(())
}
pub fn verify_belongs_to(
&self,
operation_id: &OperationId,
genesis_digest: &Digest,
) -> Result<(), CheckpointError> {
if &self.operation_id != operation_id {
return Err(CheckpointError::Incompatible(format!(
"checkpoint belongs to operation {}, this runtime to {operation_id}",
self.operation_id
)));
}
if &self.genesis_digest != genesis_digest {
return Err(CheckpointError::Incompatible(format!(
"checkpoint {operation_id} binds genesis {}, this journal's genesis is \
{genesis_digest}",
self.genesis_digest
)));
}
Ok(())
}
}
fn validate_durable_message_bodies(context: &ContextVmState) -> Result<(), CheckpointError> {
let bodies = context
.messages
.iter()
.map(|message| &message.body)
.chain(context.knowledge.iter().map(|slot| &slot.body));
for body in bodies {
let StoredMessageBody::Structured(structured) = body else {
continue;
};
let body_forms = usize::from(structured.durable_content.is_some())
+ usize::from(!structured.durable_tool_results.is_empty());
if body_forms > 1 {
return Err(CheckpointError::Incompatible(
"structured message carries more than one durable body form".into(),
));
}
if !structured.durable_tool_results.is_empty() {
for result in &structured.durable_tool_results {
result.validate().map_err(|error| {
CheckpointError::Incompatible(format!(
"structured message carries invalid durable tool result: {error}"
))
})?;
}
} else if let Some(content) = &structured.durable_content {
content.validate().map_err(|error| {
CheckpointError::Incompatible(format!(
"structured message carries invalid durable content: {error}"
))
})?;
} else {
return Err(CheckpointError::Incompatible(
"structured message carries no durable content".into(),
));
}
}
Ok(())
}
fn check_tail(
operation_id: &OperationId,
base_step_seq: WireU64,
base_record_digest: &Digest,
through_step_seq: WireU64,
covered_transaction_head_digest: &Digest,
tail_inputs: &[CanonicalInput],
) -> Result<(), CheckpointError> {
if base_step_seq > through_step_seq {
return Err(CheckpointError::Corrupted(format!(
"checkpoint {operation_id} bases at step {base_step_seq} but covers only through \
{through_step_seq}"
)));
}
if base_step_seq == through_step_seq && base_record_digest != covered_transaction_head_digest {
return Err(CheckpointError::Corrupted(format!(
"checkpoint {operation_id} covers no tail, so its base {base_record_digest} and its \
covered head {covered_transaction_head_digest} name the same record — but they differ"
)));
}
let expected = through_step_seq.get() - base_step_seq.get();
if tail_inputs.len() as u64 != expected {
return Err(CheckpointError::Corrupted(format!(
"checkpoint {operation_id} covers ({base_step_seq}, {through_step_seq}] — {expected} \
inputs — but its bounded tail holds {}",
tail_inputs.len()
)));
}
for (offset, entry) in tail_inputs.iter().enumerate() {
let want = base_step_seq.get() + offset as u64 + 1;
if entry.step_seq.get() != want {
return Err(CheckpointError::Corrupted(format!(
"checkpoint {operation_id} bounded tail is not the contiguous range \
({base_step_seq}, {through_step_seq}]: position {offset} is step {} where step \
{want} was due",
entry.step_seq
)));
}
if &entry.input.operation_id != operation_id {
return Err(CheckpointError::Incompatible(format!(
"checkpoint {operation_id} bounded tail carries an input of operation {} at step \
{}",
entry.input.operation_id, entry.step_seq
)));
}
}
if let Some(last) = tail_inputs.last()
&& &last.record_digest != covered_transaction_head_digest
{
return Err(CheckpointError::Corrupted(format!(
"checkpoint {operation_id} claims covered head {covered_transaction_head_digest}, but \
its bounded tail ends at {} on step {}",
last.record_digest, last.step_seq
)));
}
Ok(())
}
fn ack_token_for(
operation_id: &OperationId,
through_step_seq: WireU64,
checkpoint_digest: &Digest,
) -> CheckpointAckToken {
CheckpointAckToken::new(format!(
"{operation_id}:checkpoint:{through_step_seq}:{checkpoint_digest}"
))
.expect("an operation-scoped checkpoint ack token is always a legal branded ref")
}
#[derive(Debug, Clone, PartialEq)]
pub struct CheckpointCandidate {
pub checkpoint_bytes: CanonicalBytes,
pub through_step_seq: WireU64,
pub covered_head: Digest,
pub state_digest: Digest,
pub ack_token: CheckpointAckToken,
}
impl CheckpointCandidate {
pub fn boundary(&self) -> super::transaction::CheckpointBoundary {
super::transaction::CheckpointBoundary {
through_step_seq: self.through_step_seq,
covered_head: self.covered_head.clone(),
}
}
pub fn decode(&self) -> Result<KernelCheckpoint, CheckpointError> {
KernelCheckpoint::from_checkpoint_bytes(self.checkpoint_bytes.as_slice())
}
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct CheckpointProjection {
operation_id: OperationId,
genesis_digest: Digest,
base_step_seq: WireU64,
base_record_digest: Digest,
through_step_seq: WireU64,
covered_transaction_head_digest: Digest,
logical_state: LogicalKernelState,
tail_inputs: Vec<CanonicalInput>,
state_digest: Digest,
tail_digest: Digest,
checkpoint_digest: Digest,
}
fn decode_checkpoint_value(
document: serde_json::Value,
) -> Result<KernelCheckpoint, CheckpointError> {
decode_current_checkpoint(document)
}
fn decode_current_checkpoint(
document: serde_json::Value,
) -> Result<KernelCheckpoint, CheckpointError> {
let projection = serde_json::from_value::<CheckpointProjection>(document)
.map_err(|error| decode_error(&error.to_string()))?;
let checkpoint = KernelCheckpoint {
operation_id: projection.operation_id,
genesis_digest: projection.genesis_digest,
base_step_seq: projection.base_step_seq,
base_record_digest: projection.base_record_digest,
through_step_seq: projection.through_step_seq,
covered_transaction_head_digest: projection.covered_transaction_head_digest,
logical_state: projection.logical_state,
tail_inputs: projection.tail_inputs,
state_digest: projection.state_digest,
tail_digest: projection.tail_digest,
checkpoint_digest: projection.checkpoint_digest,
};
checkpoint.verify()?;
Ok(checkpoint)
}
fn decode_error(message: &str) -> CheckpointError {
if message.contains(CHECKPOINT_ERROR_MARKER) {
for code in [
KernelFaultCode::CheckpointIncompatible,
KernelFaultCode::CheckpointCorrupted,
] {
if message.contains(&format!("{CHECKPOINT_ERROR_MARKER} ({})", code.as_str())) {
return match code {
KernelFaultCode::CheckpointIncompatible => {
CheckpointError::Incompatible(message.to_string())
}
_ => CheckpointError::Corrupted(message.to_string()),
};
}
}
return CheckpointError::Corrupted(message.to_string());
}
if message.contains(SCALAR_ERROR_MARKER) && message.contains("ABI revision") {
return CheckpointError::Incompatible(message.to_string());
}
CheckpointError::NotCanonical(format!("checkpoint does not decode: {message}"))
}
impl<'de> Deserialize<'de> for KernelCheckpoint {
fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
let document = serde_json::Value::deserialize(deserializer)?;
decode_checkpoint_value(document)
.map_err(|error| serde::de::Error::custom(error.to_string()))
}
}
#[cfg(test)]
mod tests {
use std::collections::BTreeMap;
use std::fs;
use std::path::PathBuf;
use serde_json::Value;
use super::super::config::{ConfigDefaults, HostEffectSupport, OperationConfig};
use super::super::effect::EffectKindTag;
use super::super::envelope::{ConfigureOperation, KernelInput, WireEnvelope};
use super::*;
const OPERATION: &str = "op-checkpoint-1";
fn operation() -> OperationId {
OperationId::new(OPERATION).unwrap()
}
fn digest(label: &str) -> Digest {
canonical_digest(label.as_bytes())
}
fn normalized(input_id: &str, at: u64) -> NormalizedInput {
let envelope = WireEnvelope::new(
operation(),
InputId::new(input_id).unwrap(),
WireU64::new(at),
KernelInput::ConfigureOperation(ConfigureOperation {
config: OperationConfig {
host_effect_support: HostEffectSupport {
supported: vec![EffectKindTag::CallProvider],
},
..OperationConfig::default()
},
}),
);
NormalizedInput::normalize(&envelope, &ConfigDefaults::default()).expect("normalizes")
}
fn tail_entry(step_seq: u64) -> CanonicalInput {
CanonicalInput {
step_seq: WireU64::new(step_seq),
record_digest: digest(&format!("record-{step_seq}")),
input: normalized(&format!("in-{step_seq}"), 1_700_000_000_000 + step_seq),
}
}
fn resolved_config() -> ResolvedOperationConfig {
OperationConfig {
host_effect_support: HostEffectSupport {
supported: vec![EffectKindTag::CallProvider],
},
..OperationConfig::default()
}
.resolve(&ConfigDefaults::default())
.expect("the default configuration resolves")
}
fn logical_state() -> LogicalKernelState {
LogicalKernelState {
transition: TransitionState {
lifecycle: OperationLifecycle::Running,
resolved_config: resolved_config(),
root_kind: Some(RootKind::Agent),
focus: None,
last_observed_at_ms: WireU64::new(1_700_000_002_000),
pending_effects: Vec::new(),
resolved_effects: Vec::new(),
launch_tokens: Vec::new(),
accepted_inputs: vec![AcceptedInputState {
input_id: InputId::new("in-configure").unwrap(),
step_seq: WireU64::ZERO,
record_digest: digest("record-0"),
}],
accepted_cancellation: None,
terminal: None,
},
syscall: SyscallState::default(),
scheduler: SchedulerState::default(),
context_vm: ContextVmState::default(),
}
}
fn draft(base: u64, through: u64, tail: Vec<CanonicalInput>) -> CheckpointDraft {
CheckpointDraft {
operation_id: operation(),
genesis_digest: digest("genesis"),
base_step_seq: WireU64::new(base),
base_record_digest: digest(&format!("record-{base}")),
through_step_seq: WireU64::new(through),
covered_transaction_head_digest: digest(&format!("record-{through}")),
logical_state: logical_state(),
tail_inputs: tail,
}
}
fn checkpoint() -> KernelCheckpoint {
KernelCheckpoint::assemble(draft(3, 3, Vec::new())).expect("assembles")
}
fn tampered(edit: impl FnOnce(&mut serde_json::Map<String, Value>)) -> CheckpointError {
let mut document: serde_json::Map<String, Value> =
serde_json::from_slice(checkpoint().checkpoint_bytes().as_slice()).unwrap();
edit(&mut document);
let bytes = serde_json::to_vec(&document).unwrap();
KernelCheckpoint::from_checkpoint_bytes(&bytes)
.expect_err("a tampered checkpoint must not decode")
}
#[test]
fn a_checkpoint_has_no_version_axis() {
let document: Value =
serde_json::from_slice(checkpoint().checkpoint_bytes().as_slice()).unwrap();
assert!(document.get("checkpoint_version").is_none());
assert!(document.get("abi_version").is_none());
}
#[test]
fn single_ownership_is_structural() {
let checkpoint = checkpoint();
let document: Value =
serde_json::from_slice(checkpoint.checkpoint_bytes().as_slice()).unwrap();
let state = &document["logical_state"];
for (owned, owner) in [
("pending_effects", "transition"),
("resolved_effects", "transition"),
("launch_tokens", "transition"),
("accepted_inputs", "transition"),
("accepted_cancellation", "transition"),
("terminal", "transition"),
("attempts", "scheduler"),
("tasks", "scheduler"),
("handles", "context_vm"),
("pending_payload_loads", "context_vm"),
("provider_calls", "syscall"),
] {
let mut seen = Vec::new();
for partition in ["transition", "syscall", "scheduler", "context_vm"] {
if state[partition]
.as_object()
.map(|map| map.contains_key(owned))
.unwrap_or(false)
{
seen.push(partition);
}
}
assert_eq!(
seen,
vec![owner],
"{owned} must live in exactly one partition"
);
}
let mut home: BTreeMap<String, &str> = BTreeMap::new();
for partition in ["transition", "syscall", "scheduler", "context_vm"] {
for key in state[partition]
.as_object()
.expect("a partition object")
.keys()
{
if let Some(previous) = home.insert(key.clone(), partition) {
panic!("key {key} lives in both {previous} and {partition}");
}
}
}
let header: Vec<&String> = document
.as_object()
.unwrap()
.keys()
.filter(|key| home.contains_key(*key))
.collect();
assert!(
header.is_empty(),
"the checkpoint header duplicates sub-state: {header:?}"
);
}
#[test]
fn the_dto_is_constructible_without_any_state_machine() {
let state = LogicalKernelState {
transition: TransitionState {
lifecycle: OperationLifecycle::Created,
resolved_config: resolved_config(),
root_kind: None,
focus: None,
last_observed_at_ms: WireU64::ZERO,
pending_effects: Vec::new(),
resolved_effects: Vec::new(),
launch_tokens: Vec::new(),
accepted_inputs: Vec::new(),
accepted_cancellation: None,
terminal: None,
},
syscall: SyscallState::default(),
scheduler: SchedulerState::default(),
context_vm: ContextVmState::default(),
};
let mut draft = draft(0, 0, Vec::new());
draft.logical_state = state;
KernelCheckpoint::assemble(draft).expect("an empty logical state is still a checkpoint");
}
#[test]
fn the_three_digests_summarise_three_different_things() {
let checkpoint = checkpoint();
assert_eq!(
checkpoint.state_digest(),
&canonical_digest(
canonical_bytes(checkpoint.logical_state())
.unwrap()
.as_slice()
),
);
assert_eq!(
checkpoint.tail_digest(),
&canonical_digest(
canonical_bytes(checkpoint.tail_inputs())
.unwrap()
.as_slice()
),
);
assert_ne!(checkpoint.state_digest(), checkpoint.checkpoint_digest());
assert_ne!(checkpoint.tail_digest(), checkpoint.checkpoint_digest());
checkpoint
.verify()
.expect("a freshly built checkpoint verifies");
}
#[test]
fn the_checkpoint_digest_covers_the_header_and_the_bounded_tail() {
let with_tail = KernelCheckpoint::assemble(draft(3, 4, vec![tail_entry(4)])).unwrap();
let without_tail = checkpoint();
assert_eq!(
with_tail.state_digest(),
without_tail.state_digest(),
"the same logical state digests the same either way"
);
assert_ne!(
with_tail.checkpoint_digest(),
without_tail.checkpoint_digest(),
"but the checkpoint digest moves with the tail and the header"
);
let error = tampered(|document| {
document.insert("through_step_seq".to_string(), Value::String("9".into()));
});
assert_eq!(error.code(), KernelFaultCode::CheckpointCorrupted);
}
#[test]
fn a_checkpoint_round_trips_through_its_bytes() {
let original = KernelCheckpoint::assemble(draft(2, 4, vec![tail_entry(3), tail_entry(4)]))
.expect("a bounded-tail checkpoint assembles");
let decoded =
KernelCheckpoint::from_checkpoint_bytes(original.checkpoint_bytes().as_slice())
.expect("its own bytes decode");
assert_eq!(decoded, original);
assert_eq!(decoded.tail_inputs().len(), 2);
}
#[test]
fn structured_message_body_rejects_removed_body_forms() {
assert!(
serde_json::from_value::<StructuredMessageBody>(serde_json::json!({
"content_json": "{\\\"Text\\\":\\\"hello\\\"}"
}))
.is_err()
);
assert!(
serde_json::from_value::<StructuredMessageBody>(serde_json::json!({
"schema_version": 1,
"durable_content": {"blocks": []}
}))
.is_err()
);
}
#[test]
fn structured_message_body_rejects_unknown_fields() {
assert!(
serde_json::from_value::<StructuredMessageBody>(serde_json::json!({
"durable_content": {"blocks": []},
"unknown": true,
}))
.is_err()
);
}
#[test]
fn removed_durable_content_schema_field_is_not_readable() {
assert!(
serde_json::from_value::<StructuredMessageBody>(serde_json::json!({
"durable_content": {"schema_version": 1, "blocks": []}
}))
.is_err()
);
}
#[test]
fn checkpoint_rejects_durable_tool_result_with_a_second_body_form() {
let mut draft = draft(3, 3, Vec::new());
draft
.logical_state
.context_vm
.messages
.push(StoredMessageState {
partition: MessagePartition::History,
role: "tool".into(),
body: StoredMessageBody::Structured(StructuredMessageBody {
durable_content: Some(crate::types::durable_content::DurableContent::text(
"wrong",
)),
durable_tool_results: vec![
crate::types::durable_content::DurableToolResult::text(
"call-1",
"also wrong",
false,
),
],
}),
tool_calls: Vec::new(),
tokens: 0,
});
assert!(matches!(
KernelCheckpoint::assemble(draft),
Err(CheckpointError::Incompatible(_))
));
}
#[test]
fn a_digest_that_does_not_match_its_bytes_is_corruption() {
for field in ["state_digest", "tail_digest", "checkpoint_digest"] {
let error = tampered(|document| {
document.insert(
field.to_string(),
Value::String(digest("bogus").to_string()),
);
});
assert_eq!(
error.code(),
KernelFaultCode::CheckpointCorrupted,
"{field} must fail closed"
);
assert!(
error.to_string().contains(CHECKPOINT_ERROR_MARKER),
"{field}: every rejection carries the classifier marker"
);
}
}
#[test]
fn a_logical_state_edited_after_the_fact_is_corruption() {
let error = tampered(|document| {
document["logical_state"]["transition"]["lifecycle"] = Value::String("failed".into());
});
assert_eq!(error.code(), KernelFaultCode::CheckpointCorrupted);
assert!(error.message().contains("logical state hashes to"));
}
#[test]
fn removed_version_fields_are_malformed() {
for field in ["checkpoint_version", "abi_version"] {
let error = tampered(|document| {
document.insert(field.to_string(), Value::from(1));
});
assert_eq!(error.code(), KernelFaultCode::MalformedEnvelope);
}
}
#[test]
fn an_unknown_field_is_refused_rather_than_ignored() {
let error = tampered(|document| {
document.insert("last_step".to_string(), Value::Null);
});
assert_eq!(error.code(), KernelFaultCode::MalformedEnvelope);
}
#[test]
fn a_checkpoint_from_another_operation_or_genesis_is_incompatible() {
let checkpoint = checkpoint();
let other = OperationId::new("op-checkpoint-2").unwrap();
let error = checkpoint
.verify_belongs_to(&other, &digest("genesis"))
.expect_err("another operation's checkpoint is not installable");
assert_eq!(error.code(), KernelFaultCode::CheckpointIncompatible);
assert!(error.message().contains("belongs to operation"));
let error = checkpoint
.verify_belongs_to(&operation(), &digest("another-genesis"))
.expect_err("a different genesis is a different operation");
assert_eq!(error.code(), KernelFaultCode::CheckpointIncompatible);
assert!(error.message().contains("binds genesis"));
checkpoint
.verify_belongs_to(&operation(), &digest("genesis"))
.expect("its own operation and genesis are accepted");
}
#[test]
fn a_tail_that_covers_the_range_exactly_is_accepted() {
KernelCheckpoint::assemble(draft(0, 0, Vec::new())).expect("an empty range needs no tail");
KernelCheckpoint::assemble(draft(
2,
5,
vec![tail_entry(3), tail_entry(4), tail_entry(5)],
))
.expect("(2, 5] is three contiguous inputs");
}
#[test]
fn a_tail_with_a_hole_is_refused() {
let error = KernelCheckpoint::assemble(draft(
2,
5,
vec![tail_entry(3), tail_entry(5), tail_entry(6)],
))
.expect_err("step 4 is missing");
assert_eq!(error.code(), KernelFaultCode::CheckpointCorrupted);
assert!(error.message().contains("step 4 was due"), "{error}");
}
#[test]
fn a_tail_with_a_duplicate_is_refused() {
let error = KernelCheckpoint::assemble(draft(
2,
5,
vec![tail_entry(3), tail_entry(3), tail_entry(4)],
))
.expect_err("step 3 appears twice");
assert_eq!(error.code(), KernelFaultCode::CheckpointCorrupted);
assert!(error.message().contains("contiguous range"), "{error}");
}
#[test]
fn a_tail_entry_outside_the_range_is_refused() {
let error = KernelCheckpoint::assemble(draft(2, 4, vec![tail_entry(2), tail_entry(3)]))
.expect_err("step 2 is the base, not part of (2, 4]");
assert_eq!(error.code(), KernelFaultCode::CheckpointCorrupted);
let error = KernelCheckpoint::assemble(draft(2, 4, vec![tail_entry(3), tail_entry(9)]))
.expect_err("step 9 is past the covered head");
assert_eq!(error.code(), KernelFaultCode::CheckpointCorrupted);
}
#[test]
fn a_tail_whose_length_disagrees_with_the_range_is_refused() {
let error = KernelCheckpoint::assemble(draft(2, 5, vec![tail_entry(3)]))
.expect_err("(2, 5] is three inputs, not one");
assert_eq!(error.code(), KernelFaultCode::CheckpointCorrupted);
assert!(error.message().contains("bounded tail holds 1"), "{error}");
let error = KernelCheckpoint::assemble(draft(4, 2, Vec::new()))
.expect_err("a base past the covered head is not a range at all");
assert_eq!(error.code(), KernelFaultCode::CheckpointCorrupted);
}
#[test]
fn a_tail_input_from_another_operation_is_incompatible() {
let mut foreign = tail_entry(3);
foreign.input.operation_id = OperationId::new("op-checkpoint-2").unwrap();
let error = KernelCheckpoint::assemble(draft(2, 3, vec![foreign]))
.expect_err("a tail assembled from two journals is not a checkpoint");
assert_eq!(error.code(), KernelFaultCode::CheckpointIncompatible);
}
#[test]
fn a_tail_edited_in_storage_is_refused_at_decode() {
let original = KernelCheckpoint::assemble(draft(2, 4, vec![tail_entry(3), tail_entry(4)]))
.expect("assembles");
let mut document: serde_json::Map<String, Value> =
serde_json::from_slice(original.checkpoint_bytes().as_slice()).unwrap();
let tail = document["tail_inputs"].as_array_mut().unwrap();
tail.remove(0);
let bytes = serde_json::to_vec(&document).unwrap();
let error = KernelCheckpoint::from_checkpoint_bytes(&bytes)
.expect_err("a truncated tail no longer covers its range");
assert_eq!(error.code(), KernelFaultCode::CheckpointCorrupted);
}
#[test]
fn a_candidate_carries_the_five_values_of_the_spec_arrow() {
let checkpoint = KernelCheckpoint::assemble(draft(3, 4, vec![tail_entry(4)])).unwrap();
let expected_digest = checkpoint.checkpoint_digest().clone();
let candidate = checkpoint.into_candidate();
assert_eq!(candidate.through_step_seq, WireU64::new(4));
assert_eq!(candidate.covered_head, digest("record-4"));
assert!(
candidate.ack_token.as_str().contains(OPERATION)
&& candidate
.ack_token
.as_str()
.contains(expected_digest.as_str()),
"the ack token names the checkpoint it acknowledges: {}",
candidate.ack_token
);
let decoded = candidate.decode().expect("the blob decodes and verifies");
assert_eq!(decoded.checkpoint_digest(), &expected_digest);
assert_eq!(decoded.state_digest(), &candidate.state_digest);
assert_eq!(
candidate.boundary().through_step_seq,
candidate.through_step_seq
);
}
fn fixture_dir() -> PathBuf {
PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../tests/fixtures/kernel-wire")
}
#[test]
fn bless_checkpoint_rejection_fixtures() {
if std::env::var("BLESS_KERNEL_RECORD_FIXTURES").as_deref() != Ok("1") {
return;
}
let dir = fixture_dir();
for (name, expect, description, mutate) in rejection_cases() {
let mut document: serde_json::Map<String, Value> =
serde_json::from_slice(mutate.0.checkpoint_bytes().as_slice()).unwrap();
(mutate.1)(&mut document);
let fixture = serde_json::json!({
"expect": expect,
"description": description,
"checkpoint": Value::Object(document),
});
let mut text = serde_json::to_string_pretty(&fixture).unwrap();
text.push('\n');
fs::write(dir.join(name), text).unwrap_or_else(|e| panic!("cannot bless {name}: {e}"));
}
}
#[allow(clippy::type_complexity)]
fn rejection_cases() -> Vec<(
&'static str,
&'static str,
&'static str,
(
KernelCheckpoint,
Box<dyn Fn(&mut serde_json::Map<String, Value>)>,
),
)> {
let full = || checkpoint();
let with_tail =
|| KernelCheckpoint::assemble(draft(2, 4, vec![tail_entry(3), tail_entry(4)])).unwrap();
vec![
(
"reject_checkpoint_removed_checkpoint_version.json",
"malformed_envelope",
"A checkpoint carrying the removed checkpoint version field is refused at the \
strict decode boundary.",
(
full(),
Box::new(|d: &mut serde_json::Map<String, Value>| {
d.insert("checkpoint_version".into(), Value::from(99u64));
}) as Box<dyn Fn(&mut serde_json::Map<String, Value>)>,
),
),
(
"reject_checkpoint_removed_abi_version.json",
"malformed_envelope",
"A checkpoint carrying the removed ABI version field is refused at the strict \
decode boundary.",
(
full(),
Box::new(|d: &mut serde_json::Map<String, Value>| {
d.insert("abi_version".into(), Value::from(1));
}),
),
),
(
"reject_checkpoint_state_digest_mismatch.json",
"checkpoint_corrupted",
"The logical state does not hash to the digest the checkpoint claims (spec 12.1).",
(
full(),
Box::new(|d: &mut serde_json::Map<String, Value>| {
d.insert(
"state_digest".into(),
Value::String(digest("bogus").to_string()),
);
}),
),
),
(
"reject_checkpoint_missing_field_checkpoint_digest.json",
"malformed_envelope",
"A structural refusal that names the field: a checkpoint without its own digest is \
not a checkpoint with an unverified digest.",
(
full(),
Box::new(|d: &mut serde_json::Map<String, Value>| {
d.remove("checkpoint_digest");
}),
),
),
(
"reject_checkpoint_unknown_field_last_step.json",
"malformed_envelope",
"Spec 12.4 deleted `last_step`; a blob that still carries one is refused \
rather than partially read.",
(
full(),
Box::new(|d: &mut serde_json::Map<String, Value>| {
d.insert("last_step".into(), Value::Null);
}),
),
),
(
"reject_checkpoint_base_disagrees_with_covered_head.json",
"checkpoint_corrupted",
"A full-state checkpoint covers no tail, so its base and its covered head name the \
same record; a header that disagrees with itself would hand a restore two \
different chain anchors (spec 12.1, Task 16).",
(
full(),
Box::new(|d: &mut serde_json::Map<String, Value>| {
d.insert(
"base_record_digest".into(),
Value::String(digest("another-record").to_string()),
);
}),
),
),
(
"reject_checkpoint_tail_hole.json",
"checkpoint_corrupted",
"The bounded tail must cover (base, through] with no hole (spec 12.1).",
(
with_tail(),
Box::new(|d: &mut serde_json::Map<String, Value>| {
d["tail_inputs"].as_array_mut().unwrap().remove(0);
}),
),
),
(
"reject_checkpoint_tail_duplicate.json",
"checkpoint_corrupted",
"The bounded tail must cover (base, through] with no duplicate (spec 12.1).",
(
with_tail(),
Box::new(|d: &mut serde_json::Map<String, Value>| {
let tail = d["tail_inputs"].as_array_mut().unwrap();
tail[1] = tail[0].clone();
}),
),
),
(
"reject_checkpoint_tail_foreign_operation.json",
"checkpoint_incompatible",
"A bounded tail assembled from two journals is not a checkpoint (spec 12.1).",
(
with_tail(),
Box::new(|d: &mut serde_json::Map<String, Value>| {
d["tail_inputs"][0]["input"]["operation_id"] =
Value::String("op-checkpoint-2".into());
}),
),
),
(
"reject_checkpoint_tail_ends_off_the_covered_head.json",
"checkpoint_corrupted",
"The last bounded-tail entry *is* the covered head; a tail that ends somewhere \
else covers a different prefix than the header claims (spec 12.1, Task 16).",
(
with_tail(),
Box::new(|d: &mut serde_json::Map<String, Value>| {
d["tail_inputs"][1]["record_digest"] =
Value::String(digest("some-other-record").to_string());
}),
),
),
]
}
#[test]
fn checkpoint_rejection_fixtures_fail_closed_with_the_declared_kind() {
let dir = fixture_dir();
let mut names: Vec<String> = fs::read_dir(&dir)
.unwrap_or_else(|e| panic!("failed to read {}: {e}", dir.display()))
.map(|entry| {
entry
.expect("dir entry")
.file_name()
.to_string_lossy()
.to_string()
})
.filter(|name| name.starts_with("reject_checkpoint_") && name.ends_with(".json"))
.collect();
names.sort();
assert!(
names.len() >= 5,
"too few checkpoint rejection fixtures: {names:?}"
);
for name in names {
let raw = fs::read_to_string(dir.join(&name)).expect("fixture reads");
let fixture: Value = serde_json::from_str(&raw).expect("fixture is JSON");
let expected = fixture["expect"]
.as_str()
.expect("every fixture declares `expect`");
let bytes = serde_json::to_vec(&fixture["checkpoint"]).unwrap();
let error = KernelCheckpoint::from_checkpoint_bytes(&bytes)
.expect_err(&format!("{name}: expected a rejection"));
assert_eq!(
error.code().as_str(),
expected,
"{name}: {} (message: {})",
error.code().as_str(),
error.message()
);
for (marker, needle) in [
("_missing_field_", "missing field"),
("_unknown_field_", "unknown field"),
] {
if name.contains(marker) {
assert_eq!(
error.code(),
KernelFaultCode::MalformedEnvelope,
"{name}: a structural refusal is malformed_envelope"
);
assert!(
error.message().contains(needle),
"{name}: the rejection must say which field ({})",
error.message()
);
}
}
}
}
}