1use std::fmt;
23use std::fmt::Write as _;
24
25use serde::de::Deserializer;
26use serde::{Deserialize, Serialize};
27use sha2::{Digest as _, Sha256};
28
29use super::config::{ConfigDefaults, ResolvedOperationConfig};
30use super::effect::Digest;
31use super::envelope::{
32 AbiRevision, DeliverExternalEvent, HostControl, KernelInput, OperationLifecycle, ResolveEffect,
33 StartOperation, WireEnvelope, WireRejection,
34};
35use super::fault::KernelFaultCode;
36use super::scalar::{CanonicalBytes, InputId, JS_SAFE_INTEGER_MAX, OperationId, WireU64};
37
38pub const RECORD_ERROR_MARKER: &str = "kernel record rejected";
44
45#[derive(Debug, Clone, PartialEq, Eq)]
47pub enum RecordError {
48 NotCanonical(String),
51 ChainBroken(String),
54 DigestMismatch(String),
57}
58
59impl RecordError {
60 pub fn message(&self) -> &str {
61 match self {
62 Self::NotCanonical(message)
63 | Self::ChainBroken(message)
64 | Self::DigestMismatch(message) => message,
65 }
66 }
67
68 pub fn code(&self) -> KernelFaultCode {
80 match self {
81 Self::NotCanonical(_) => KernelFaultCode::MalformedEnvelope,
82 Self::ChainBroken(_) => KernelFaultCode::TransactionConflict,
83 Self::DigestMismatch(_) => KernelFaultCode::RecordCorrupted,
84 }
85 }
86}
87
88impl fmt::Display for RecordError {
89 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
90 write!(f, "{RECORD_ERROR_MARKER}: {}", self.message())
91 }
92}
93
94impl std::error::Error for RecordError {}
95
96pub const CANONICAL_MAX_DEPTH: usize = 128;
103
104pub const DIGEST_ALGORITHM: &str = "sha256";
107
108pub fn canonical_bytes<T: Serialize + ?Sized>(value: &T) -> Result<CanonicalBytes, RecordError> {
129 let value = serde_json::to_value(value).map_err(|error| {
130 RecordError::NotCanonical(format!("value is not serialisable: {error}"))
131 })?;
132 let mut out = String::new();
133 write_canonical(&value, 1, &mut out)?;
134 Ok(CanonicalBytes::new(out.into_bytes()))
135}
136
137pub fn canonical_digest(bytes: &[u8]) -> Digest {
139 let hash = Sha256::digest(bytes);
140 let mut text = String::with_capacity(DIGEST_ALGORITHM.len() + 1 + hash.len() * 2);
141 text.push_str(DIGEST_ALGORITHM);
142 text.push(':');
143 for byte in hash {
144 write!(text, "{byte:02x}").expect("writing to a String cannot fail");
145 }
146 Digest::new(text).expect("an algorithm-prefixed hex digest is always a legal branded ref")
147}
148
149fn write_canonical(
150 value: &serde_json::Value,
151 depth: usize,
152 out: &mut String,
153) -> Result<(), RecordError> {
154 if depth > CANONICAL_MAX_DEPTH {
155 return Err(RecordError::NotCanonical(format!(
156 "value nests deeper than {CANONICAL_MAX_DEPTH}"
157 )));
158 }
159 match value {
160 serde_json::Value::Null => out.push_str("null"),
161 serde_json::Value::Bool(true) => out.push_str("true"),
162 serde_json::Value::Bool(false) => out.push_str("false"),
163 serde_json::Value::Number(number) => write_canonical_number(number, out)?,
164 serde_json::Value::String(text) => write_canonical_string(text, out),
165 serde_json::Value::Array(items) => {
166 out.push('[');
167 for (index, item) in items.iter().enumerate() {
168 if index > 0 {
169 out.push(',');
170 }
171 write_canonical(item, depth + 1, out)?;
172 }
173 out.push(']');
174 }
175 serde_json::Value::Object(map) => {
176 let mut keys: Vec<&String> = map.keys().collect();
177 keys.sort_unstable();
178 out.push('{');
179 for (index, key) in keys.into_iter().enumerate() {
180 if index > 0 {
181 out.push(',');
182 }
183 write_canonical_string(key, out);
184 out.push(':');
185 write_canonical(&map[key], depth + 1, out)?;
186 }
187 out.push('}');
188 }
189 }
190 Ok(())
191}
192
193fn write_canonical_string(text: &str, out: &mut String) {
194 let encoded = serde_json::to_string(text).expect("a Rust string is always JSON-encodable");
195 out.push_str(&encoded);
196}
197
198fn write_canonical_number(
199 number: &serde_json::Number,
200 out: &mut String,
201) -> Result<(), RecordError> {
202 const SAFE: i128 = JS_SAFE_INTEGER_MAX as i128;
203
204 let unsafe_integer = |value: i128| {
205 RecordError::NotCanonical(format!(
206 "integer {value} exceeds the cross-language exact-integer range \
207 (±{JS_SAFE_INTEGER_MAX}); logical u64 travels as a decimal string"
208 ))
209 };
210
211 if let Some(value) = number.as_u64() {
212 if i128::from(value) > SAFE {
213 return Err(unsafe_integer(i128::from(value)));
214 }
215 write!(out, "{value}").expect("writing to a String cannot fail");
216 return Ok(());
217 }
218 if let Some(value) = number.as_i64() {
219 if i128::from(value) < -SAFE {
220 return Err(unsafe_integer(i128::from(value)));
221 }
222 write!(out, "{value}").expect("writing to a String cannot fail");
223 return Ok(());
224 }
225
226 let float = number
227 .as_f64()
228 .filter(|value| value.is_finite())
229 .ok_or_else(|| RecordError::NotCanonical(format!("number {number} is not finite")))?;
230 if float.fract() == 0.0 && float.abs() <= JS_SAFE_INTEGER_MAX as f64 {
233 write!(out, "{}", float as i64).expect("writing to a String cannot fail");
234 } else {
235 let encoded = serde_json::to_string(&float).expect("a finite f64 is always JSON-encodable");
236 out.push_str(&encoded);
237 }
238 Ok(())
239}
240
241#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
258#[serde(deny_unknown_fields)]
259pub struct NormalizedInput {
260 pub abi_version: AbiRevision,
261 pub operation_id: OperationId,
262 pub input_id: InputId,
263 pub observed_at_ms: WireU64,
264 pub input: NormalizedPayload,
265}
266
267#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
270#[serde(tag = "kind", rename_all = "snake_case")]
271pub enum NormalizedPayload {
272 ConfigureOperation(ResolvedConfiguration),
274 StartOperation(StartOperation),
275 ResolveEffect(ResolveEffect),
276 DeliverExternalEvent(DeliverExternalEvent),
277 HostControl(HostControl),
278}
279
280#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
281#[serde(deny_unknown_fields)]
282pub struct ResolvedConfiguration {
283 pub config: ResolvedOperationConfig,
284}
285
286impl NormalizedPayload {
287 pub fn is_genesis(&self) -> bool {
289 matches!(self, Self::ConfigureOperation(_))
290 }
291
292 pub fn kind(&self) -> &'static str {
293 match self {
294 Self::ConfigureOperation(_) => "configure_operation",
295 Self::StartOperation(_) => "start_operation",
296 Self::ResolveEffect(_) => "resolve_effect",
297 Self::DeliverExternalEvent(_) => "deliver_external_event",
298 Self::HostControl(_) => "host_control",
299 }
300 }
301
302 pub fn admissible_lifecycles(&self) -> &'static [OperationLifecycle] {
309 match self {
310 Self::ConfigureOperation(_) => &[OperationLifecycle::Created],
311 Self::StartOperation(_) => &[OperationLifecycle::Configured],
312 Self::ResolveEffect(_) | Self::DeliverExternalEvent(_) => {
313 &[OperationLifecycle::Running, OperationLifecycle::Suspended]
314 }
315 Self::HostControl(_) => &[
316 OperationLifecycle::Configured,
317 OperationLifecycle::Running,
318 OperationLifecycle::Suspended,
319 ],
320 }
321 }
322}
323
324impl NormalizedInput {
325 pub fn normalize(
329 envelope: &WireEnvelope,
330 defaults: &ConfigDefaults,
331 ) -> Result<Self, WireRejection> {
332 let input = match &envelope.input {
333 KernelInput::ConfigureOperation(configure) => {
334 NormalizedPayload::ConfigureOperation(ResolvedConfiguration {
335 config: configure.config.resolve(defaults)?,
336 })
337 }
338 KernelInput::StartOperation(start) => NormalizedPayload::StartOperation(start.clone()),
339 KernelInput::ResolveEffect(resolve) => {
340 NormalizedPayload::ResolveEffect(resolve.clone())
341 }
342 KernelInput::DeliverExternalEvent(event) => {
343 NormalizedPayload::DeliverExternalEvent(event.clone())
344 }
345 KernelInput::HostControl(control) => NormalizedPayload::HostControl(control.clone()),
346 };
347 Ok(Self {
348 abi_version: envelope.abi_version,
349 operation_id: envelope.operation_id.clone(),
350 input_id: envelope.input_id.clone(),
351 observed_at_ms: envelope.observed_at_ms,
352 input,
353 })
354 }
355
356 pub fn is_genesis(&self) -> bool {
357 self.input.is_genesis()
358 }
359
360 pub fn resolved_config(&self) -> Option<&ResolvedOperationConfig> {
362 match &self.input {
363 NormalizedPayload::ConfigureOperation(configure) => Some(&configure.config),
364 _ => None,
365 }
366 }
367}
368
369#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
381pub struct KernelRecord {
382 abi_version: AbiRevision,
383 operation_id: OperationId,
384 input_id: InputId,
385 step_seq: WireU64,
386 previous_record_digest: Option<Digest>,
389 canonical_input: CanonicalBytes,
392 input_digest: Digest,
393 step_digest: Digest,
395 record_digest: Digest,
396}
397
398#[derive(Debug, Clone, PartialEq, Eq)]
404pub struct ChainAnchor {
405 pub operation_id: OperationId,
406 pub step_seq: WireU64,
407 pub record_digest: Digest,
408}
409
410#[derive(Serialize)]
412struct RecordBody<'a> {
413 abi_version: AbiRevision,
414 operation_id: &'a OperationId,
415 input_id: &'a InputId,
416 step_seq: WireU64,
417 previous_record_digest: Option<&'a Digest>,
418 canonical_input: &'a CanonicalBytes,
419 input_digest: &'a Digest,
420 step_digest: &'a Digest,
421}
422
423impl KernelRecord {
424 pub fn chain<S: Serialize + ?Sized>(
432 previous: Option<&Self>,
433 input: &NormalizedInput,
434 planned_step: &S,
435 ) -> Result<Self, RecordError> {
436 Self::chain_after(previous.map(Self::anchor).as_ref(), input, planned_step)
437 }
438
439 pub fn anchor(&self) -> ChainAnchor {
441 ChainAnchor {
442 operation_id: self.operation_id.clone(),
443 step_seq: self.step_seq,
444 record_digest: self.record_digest.clone(),
445 }
446 }
447
448 pub fn chain_after<S: Serialize + ?Sized>(
455 previous: Option<&ChainAnchor>,
456 input: &NormalizedInput,
457 planned_step: &S,
458 ) -> Result<Self, RecordError> {
459 let (step_seq, previous_record_digest) = match previous {
460 None => {
461 if !input.is_genesis() {
462 return Err(RecordError::ChainBroken(format!(
463 "an operation's first record must be its resolved configuration, got {}",
464 input.input.kind()
465 )));
466 }
467 (WireU64::ZERO, None)
468 }
469 Some(previous) => {
470 if input.is_genesis() {
471 return Err(RecordError::ChainBroken(
472 "an operation is configured exactly once; a second configure_operation \
473 has no position in the chain"
474 .to_string(),
475 ));
476 }
477 if previous.operation_id != input.operation_id {
478 return Err(RecordError::ChainBroken(format!(
479 "input belongs to operation {}, but the chain head belongs to {}",
480 input.operation_id, previous.operation_id
481 )));
482 }
483 let next = previous.step_seq.get().checked_add(1).ok_or_else(|| {
484 RecordError::ChainBroken("step sequence overflowed u64".to_string())
485 })?;
486 (WireU64::new(next), Some(previous.record_digest.clone()))
487 }
488 };
489
490 let canonical_input = canonical_bytes(input)?;
491 let input_digest = canonical_digest(canonical_input.as_slice());
492 let step_digest = canonical_digest(canonical_bytes(planned_step)?.as_slice());
493 let record_digest = Self::body_digest(
494 &input.operation_id,
495 &input.input_id,
496 step_seq,
497 previous_record_digest.as_ref(),
498 &canonical_input,
499 &input_digest,
500 &step_digest,
501 )?;
502
503 Ok(Self {
504 abi_version: AbiRevision::CURRENT,
505 operation_id: input.operation_id.clone(),
506 input_id: input.input_id.clone(),
507 step_seq,
508 previous_record_digest,
509 canonical_input,
510 input_digest,
511 step_digest,
512 record_digest,
513 })
514 }
515
516 #[allow(clippy::too_many_arguments)]
517 fn body_digest(
518 operation_id: &OperationId,
519 input_id: &InputId,
520 step_seq: WireU64,
521 previous_record_digest: Option<&Digest>,
522 canonical_input: &CanonicalBytes,
523 input_digest: &Digest,
524 step_digest: &Digest,
525 ) -> Result<Digest, RecordError> {
526 let body = RecordBody {
527 abi_version: AbiRevision::CURRENT,
528 operation_id,
529 input_id,
530 step_seq,
531 previous_record_digest,
532 canonical_input,
533 input_digest,
534 step_digest,
535 };
536 Ok(canonical_digest(canonical_bytes(&body)?.as_slice()))
537 }
538
539 pub fn abi_version(&self) -> u32 {
542 self.abi_version.get()
543 }
544
545 pub fn operation_id(&self) -> &OperationId {
546 &self.operation_id
547 }
548
549 pub fn input_id(&self) -> &InputId {
550 &self.input_id
551 }
552
553 pub fn step_seq(&self) -> WireU64 {
554 self.step_seq
555 }
556
557 pub fn previous_record_digest(&self) -> Option<&Digest> {
558 self.previous_record_digest.as_ref()
559 }
560
561 pub fn expected_head(&self) -> Option<&Digest> {
564 self.previous_record_digest.as_ref()
565 }
566
567 pub fn canonical_input(&self) -> &CanonicalBytes {
568 &self.canonical_input
569 }
570
571 pub fn input_digest(&self) -> &Digest {
572 &self.input_digest
573 }
574
575 pub fn step_digest(&self) -> &Digest {
576 &self.step_digest
577 }
578
579 pub fn record_digest(&self) -> &Digest {
580 &self.record_digest
581 }
582
583 pub fn is_genesis(&self) -> bool {
584 self.previous_record_digest.is_none()
585 }
586
587 pub fn record_bytes(&self) -> CanonicalBytes {
592 canonical_bytes(self).expect("a record contains only canonical scalars")
593 }
594
595 pub fn from_record_bytes(bytes: &[u8]) -> Result<Self, RecordError> {
597 let text = std::str::from_utf8(bytes).map_err(|error| {
598 RecordError::NotCanonical(format!("record bytes are not UTF-8: {error}"))
599 })?;
600 serde_json::from_str(text).map_err(|error| decode_error(&error.to_string()))
601 }
602
603 pub fn normalized_input(&self) -> Result<NormalizedInput, RecordError> {
605 let text = std::str::from_utf8(self.canonical_input.as_slice()).map_err(|error| {
606 RecordError::NotCanonical(format!("canonical input is not UTF-8: {error}"))
607 })?;
608 serde_json::from_str(text).map_err(|error| {
609 RecordError::NotCanonical(format!("canonical input does not decode: {error}"))
610 })
611 }
612
613 pub fn verify(&self) -> Result<(), RecordError> {
617 let input_digest = canonical_digest(self.canonical_input.as_slice());
618 if input_digest != self.input_digest {
619 return Err(RecordError::DigestMismatch(format!(
620 "record {} step {}: canonical input hashes to {input_digest}, \
621 but the record claims {}",
622 self.operation_id, self.step_seq, self.input_digest
623 )));
624 }
625 let record_digest = Self::body_digest(
626 &self.operation_id,
627 &self.input_id,
628 self.step_seq,
629 self.previous_record_digest.as_ref(),
630 &self.canonical_input,
631 &self.input_digest,
632 &self.step_digest,
633 )?;
634 if record_digest != self.record_digest {
635 return Err(RecordError::DigestMismatch(format!(
636 "record {} step {}: body hashes to {record_digest}, but the record claims {}",
637 self.operation_id, self.step_seq, self.record_digest
638 )));
639 }
640 Ok(())
641 }
642
643 pub fn verify_step<S: Serialize + ?Sized>(&self, planned_step: &S) -> Result<(), RecordError> {
645 let digest = canonical_digest(canonical_bytes(planned_step)?.as_slice());
646 if digest != self.step_digest {
647 return Err(RecordError::DigestMismatch(format!(
648 "record {} step {}: the rebuilt step hashes to {digest}, \
649 but the record froze {}",
650 self.operation_id, self.step_seq, self.step_digest
651 )));
652 }
653 Ok(())
654 }
655
656 pub fn verify_follows(&self, previous: Option<&Self>) -> Result<(), RecordError> {
658 self.verify()?;
659 match (previous, self.previous_record_digest.as_ref()) {
660 (None, None) => {
661 if self.step_seq != WireU64::ZERO {
662 return Err(RecordError::ChainBroken(format!(
663 "a genesis record is step 0, got step {}",
664 self.step_seq
665 )));
666 }
667 if self.normalized_input()?.is_genesis() {
668 Ok(())
669 } else {
670 Err(RecordError::ChainBroken(
671 "a genesis record must carry the resolved configuration".to_string(),
672 ))
673 }
674 }
675 (None, Some(digest)) => Err(RecordError::ChainBroken(format!(
676 "record {} step {} expects head {digest}, but the operation has no head",
677 self.operation_id, self.step_seq
678 ))),
679 (Some(previous), None) => Err(RecordError::ChainBroken(format!(
680 "record {} step {} claims to be a genesis, but the operation head is {}",
681 self.operation_id, self.step_seq, previous.record_digest
682 ))),
683 (Some(previous), Some(digest)) => {
684 if previous.operation_id != self.operation_id {
685 return Err(RecordError::ChainBroken(format!(
686 "record belongs to operation {}, its predecessor to {}",
687 self.operation_id, previous.operation_id
688 )));
689 }
690 if digest != &previous.record_digest {
691 return Err(RecordError::ChainBroken(format!(
692 "record {} step {} expects head {digest}, but the head is {}",
693 self.operation_id, self.step_seq, previous.record_digest
694 )));
695 }
696 if previous.step_seq.get().checked_add(1) != Some(self.step_seq.get()) {
697 return Err(RecordError::ChainBroken(format!(
698 "record {} is step {}, but its predecessor is step {}",
699 self.operation_id, self.step_seq, previous.step_seq
700 )));
701 }
702 Ok(())
703 }
704 }
705 }
706}
707
708pub type RecordPreparation<Step> = super::fault::KernelPreparation<KernelRecord, Step>;
711
712pub fn verify_record_chain(records: &[KernelRecord]) -> Result<&Digest, RecordError> {
718 let Some(genesis) = records.first() else {
719 return Err(RecordError::ChainBroken(
720 "an operation chain starts at its genesis record; this one is empty".to_string(),
721 ));
722 };
723 genesis.verify_follows(None)?;
724 for pair in records.windows(2) {
725 pair[1].verify_follows(Some(&pair[0]))?;
726 }
727 Ok(&genesis.record_digest)
728}
729
730#[derive(Deserialize)]
737#[serde(deny_unknown_fields)]
738struct RecordProjection {
739 abi_version: AbiRevision,
740 operation_id: OperationId,
741 input_id: InputId,
742 step_seq: WireU64,
743 previous_record_digest: Option<Digest>,
744 canonical_input: CanonicalBytes,
745 input_digest: Digest,
746 step_digest: Digest,
747 record_digest: Digest,
748}
749
750fn decode_error(message: &str) -> RecordError {
751 if message.contains(RECORD_ERROR_MARKER) {
752 RecordError::DigestMismatch(message.to_string())
753 } else {
754 RecordError::NotCanonical(format!("record does not decode: {message}"))
755 }
756}
757
758impl<'de> Deserialize<'de> for KernelRecord {
759 fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
760 let projection = RecordProjection::deserialize(deserializer)?;
761 let record = Self {
762 abi_version: projection.abi_version,
763 operation_id: projection.operation_id,
764 input_id: projection.input_id,
765 step_seq: projection.step_seq,
766 previous_record_digest: projection.previous_record_digest,
767 canonical_input: projection.canonical_input,
768 input_digest: projection.input_digest,
769 step_digest: projection.step_digest,
770 record_digest: projection.record_digest,
771 };
772 record
773 .verify()
774 .map_err(|error| serde::de::Error::custom(error.to_string()))?;
775 Ok(record)
776 }
777}
778
779#[cfg(test)]
780mod tests {
781 use std::collections::BTreeSet;
782 use std::fs;
783 use std::path::PathBuf;
784
785 use serde_json::{Value, json};
786
787 use super::super::*;
788
789 fn operation() -> OperationId {
794 OperationId::new("op-record-1").unwrap()
795 }
796
797 fn input_id(seq: u32) -> InputId {
798 InputId::new(format!("in-{seq}")).unwrap()
799 }
800
801 fn boot_config() -> OperationConfig {
802 OperationConfig {
803 execution_policy: Some(ExecutionPolicy {
804 max_turns: Some(12),
805 ..ExecutionPolicy::default()
806 }),
807 host_effect_support: HostEffectSupport::new([
808 EffectKindTag::CallProvider,
809 EffectKindTag::ExecuteTools,
810 ]),
811 ..OperationConfig::default()
812 }
813 }
814
815 fn envelope(seq: u32, observed_at_ms: u64, input: KernelInput) -> WireEnvelope {
816 WireEnvelope::new(
817 operation(),
818 input_id(seq),
819 WireU64::new(observed_at_ms),
820 input,
821 )
822 }
823
824 fn configure_envelope() -> WireEnvelope {
825 envelope(
826 0,
827 1_700_000_000_000,
828 KernelInput::ConfigureOperation(ConfigureOperation {
829 config: boot_config(),
830 }),
831 )
832 }
833
834 fn start_envelope() -> WireEnvelope {
835 envelope(
836 1,
837 1_700_000_000_500,
838 KernelInput::StartOperation(StartOperation {
839 entry: RootEntry::Agent(RootAgentEntry {
840 task: LogicalTask::new("write the brief"),
841 run_spec: None,
842 }),
843 initial_context: InitialContext::default(),
844 }),
845 )
846 }
847
848 fn resolve_envelope() -> WireEnvelope {
849 envelope(
850 2,
851 1_700_000_001_000,
852 KernelInput::ResolveEffect(ResolveEffect {
853 effect_id: EffectId::new("op-record-1:step:1:effect:0").unwrap(),
854 outcome: EffectOutcome::Failed(EffectFailed {
855 failure: HostEffectFailure {
856 kind: HostEffectFailureKind::ProtocolError,
857 message: "provider refused the request".to_string(),
858 retryable: Some(false),
859 },
860 }),
861 }),
862 )
863 }
864
865 fn cancel_envelope() -> WireEnvelope {
866 envelope(
867 3,
868 1_700_000_002_000,
869 KernelInput::HostControl(HostControl {
870 command: HostCommand::Cancel(CancelCommand {
871 reason: CancellationReason::User,
872 pending_call_ids: vec![],
873 }),
874 }),
875 )
876 }
877
878 fn normalize(envelope: &WireEnvelope) -> NormalizedInput {
879 NormalizedInput::normalize(envelope, &ConfigDefaults::default())
880 .expect("the sample envelope normalises")
881 }
882
883 fn step(name: &str) -> Value {
884 json!({ "planned": name, "effects": [{ "kind": "call_provider" }] })
885 }
886
887 fn genesis_record() -> KernelRecord {
888 KernelRecord::chain(None, &normalize(&configure_envelope()), &step("configure")).unwrap()
889 }
890
891 fn sample_chain() -> Vec<KernelRecord> {
893 let genesis = genesis_record();
894 let started = KernelRecord::chain(
895 Some(&genesis),
896 &normalize(&start_envelope()),
897 &step("start"),
898 )
899 .unwrap();
900 let resolved = KernelRecord::chain(
901 Some(&started),
902 &normalize(&resolve_envelope()),
903 &step("resolve"),
904 )
905 .unwrap();
906 vec![genesis, started, resolved]
907 }
908
909 fn canonical_text<T: Serialize + ?Sized>(value: &T) -> String {
910 String::from_utf8(canonical_bytes(value).unwrap().into_vec()).unwrap()
911 }
912
913 fn fixture_dir() -> PathBuf {
914 PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../tests/fixtures/kernel-wire")
915 }
916
917 fn golden(name: &str, produced: &Value) -> Value {
922 let path = fixture_dir().join(name);
923 if std::env::var("BLESS_KERNEL_RECORD_FIXTURES").as_deref() == Ok("1") {
924 let mut text = serde_json::to_string_pretty(produced).unwrap();
925 text.push('\n');
926 fs::write(&path, text).unwrap_or_else(|e| panic!("cannot bless {name}: {e}"));
927 return produced.clone();
928 }
929 let raw = fs::read_to_string(&path).unwrap_or_else(|e| {
930 panic!("missing golden {name} ({e}); re-bless with BLESS_KERNEL_RECORD_FIXTURES=1")
931 });
932 serde_json::from_str(&raw).unwrap_or_else(|e| panic!("{name} is not JSON: {e}"))
933 }
934
935 #[test]
940 fn canonical_bytes_sort_keys_and_carry_no_whitespace() {
941 assert_eq!(
942 canonical_text(&json!({ "b": 1, "a": { "d": [1, 2], "c": true }, "": null })),
943 r#"{"":null,"a":{"c":true,"d":[1,2]},"b":1}"#
944 );
945 assert_eq!(canonical_text(&json!([])), "[]");
946 assert_eq!(canonical_text(&json!({})), "{}");
947 assert_eq!(
948 canonical_text(&json!("quote \" backslash \\ newline \n tab \t")),
949 r#""quote \" backslash \\ newline \n tab \t""#
950 );
951 assert_eq!(canonical_text(&json!("\u{1}")), "\"\\u0001\"");
952 }
953
954 #[test]
955 fn canonical_object_keys_sort_by_code_point_not_utf16() {
956 assert_eq!(
960 canonical_text(&json!({ "\u{10000}": 1, "\u{fffd}": 2, "z": 3 })),
961 "{\"z\":3,\"\u{fffd}\":2,\"\u{10000}\":1}"
962 );
963 }
964
965 #[test]
966 fn canonical_numbers_are_language_neutral() {
967 assert_eq!(canonical_text(&json!(0)), "0");
968 assert_eq!(canonical_text(&json!(-0.0)), "0", "-0 and 0 are one value");
969 assert_eq!(canonical_text(&json!(2.0)), "2", "2 and 2.0 are one value");
970 assert_eq!(canonical_text(&json!(-17)), "-17");
971 assert_eq!(canonical_text(&json!(0.5)), "0.5");
972 assert_eq!(
973 canonical_text(&json!(JS_SAFE_INTEGER_MAX)),
974 "9007199254740991"
975 );
976 }
977
978 #[test]
979 fn canonical_bytes_reject_what_no_host_can_read_back() {
980 let too_large = serde_json::from_str::<Value>("9007199254740992").unwrap();
981 let error = canonical_bytes(&too_large).expect_err("beyond the exact-integer range");
982 assert!(matches!(error, RecordError::NotCanonical(_)), "{error}");
983
984 let too_negative = serde_json::from_str::<Value>("-9007199254740992").unwrap();
985 assert!(canonical_bytes(&too_negative).is_err());
986
987 let mut deep = json!(0);
988 for _ in 0..(CANONICAL_MAX_DEPTH + 2) {
989 deep = Value::Array(vec![deep]);
990 }
991 assert!(canonical_bytes(&deep).is_err(), "recursion must be bounded");
992 }
993
994 #[test]
995 fn canonical_bytes_are_byte_identical_across_repeated_runs() {
996 let input = normalize(&configure_envelope());
997 let first = canonical_bytes(&input).unwrap();
998 for _ in 0..8 {
999 assert_eq!(canonical_bytes(&input).unwrap(), first);
1000 }
1001
1002 let one = canonical_bytes(&json!({ "a": 1, "b": 2 })).unwrap();
1004 let other = canonical_bytes(&json!({ "b": 2, "a": 1 })).unwrap();
1005 assert_eq!(one, other);
1006 }
1007
1008 #[test]
1009 fn the_digest_is_sha256_over_exactly_the_canonical_bytes() {
1010 assert_eq!(
1012 canonical_digest(b"").as_str(),
1013 "sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
1014 );
1015 assert_eq!(
1016 canonical_digest(b"abc").as_str(),
1017 "sha256:ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad"
1018 );
1019 assert_eq!(
1020 canonical_digest(canonical_text(&json!({ "a": 1 })).as_bytes()).as_str(),
1021 canonical_digest(br#"{"a":1}"#).as_str()
1022 );
1023 }
1024
1025 #[test]
1030 fn genesis_binds_the_resolved_config_and_an_empty_previous_head() {
1031 let record = genesis_record();
1032
1033 assert!(record.is_genesis());
1034 assert_eq!(record.step_seq(), WireU64::ZERO);
1035 assert_eq!(record.previous_record_digest(), None);
1036 assert_eq!(
1037 record.expected_head(),
1038 None,
1039 "genesis appends against an empty head"
1040 );
1041
1042 let stored = record.normalized_input().unwrap();
1043 let resolved = stored
1044 .resolved_config()
1045 .expect("genesis stores a resolved config");
1046 assert_eq!(
1047 resolved,
1048 &boot_config().resolve(&ConfigDefaults::default()).unwrap(),
1049 "the genesis record freezes the resolved configuration, not the sparse one"
1050 );
1051 assert_eq!(resolved.execution_policy.max_turns, 12);
1052
1053 let value = serde_json::to_value(resolved).unwrap();
1056 for required in [
1057 "execution_policy",
1058 "governance_policy",
1059 "scheduler_policy",
1060 "resource_quota",
1061 "signal_policy",
1062 "context_policy",
1063 "recovery_policy",
1064 "payload_policy",
1065 "kernel_limits",
1066 "memory_policy",
1067 "feature_policy",
1068 "host_effect_support",
1069 ] {
1070 assert!(
1071 value.get(required).is_some(),
1072 "resolved config lacks {required}"
1073 );
1074 }
1075 }
1076
1077 #[test]
1078 fn the_genesis_digest_is_the_operations_identity() {
1079 let chain = sample_chain();
1080 let genesis_digest = verify_record_chain(&chain).unwrap();
1081 assert_eq!(genesis_digest, chain[0].record_digest());
1082 assert_eq!(
1083 chain[1].previous_record_digest(),
1084 Some(genesis_digest),
1085 "the first transition links straight to the genesis digest"
1086 );
1087 }
1088
1089 #[test]
1090 fn a_genesis_replay_survives_kernel_default_drift() {
1091 let mut drifted = ConfigDefaults::default();
1093 drifted.baseline.execution_policy.max_context_tokens = 999;
1094 drifted.baseline.recovery_policy.provider_recovery_attempts = 9;
1095
1096 let record = genesis_record();
1097 let replayed = record.normalized_input().unwrap();
1100 assert_eq!(
1101 replayed
1102 .resolved_config()
1103 .unwrap()
1104 .execution_policy
1105 .max_context_tokens,
1106 ConfigDefaults::default()
1107 .baseline
1108 .execution_policy
1109 .max_context_tokens
1110 );
1111 assert_ne!(
1112 drifted.baseline.execution_policy.max_context_tokens,
1113 replayed
1114 .resolved_config()
1115 .unwrap()
1116 .execution_policy
1117 .max_context_tokens,
1118 "the fixture must actually drift, or this test proves nothing"
1119 );
1120 assert_eq!(
1121 canonical_bytes(&replayed).unwrap(),
1122 *record.canonical_input()
1123 );
1124 record
1125 .verify_step(&step("configure"))
1126 .expect("the frozen step still verifies");
1127 }
1128
1129 #[test]
1134 fn each_record_links_to_its_predecessor() {
1135 let chain = sample_chain();
1136 for (index, record) in chain.iter().enumerate() {
1137 assert_eq!(record.step_seq(), WireU64::new(index as u64));
1138 assert_eq!(record.operation_id(), &operation());
1139 if index == 0 {
1140 assert!(record.previous_record_digest().is_none());
1141 } else {
1142 assert_eq!(
1143 record.previous_record_digest(),
1144 Some(chain[index - 1].record_digest())
1145 );
1146 }
1147 }
1148 verify_record_chain(&chain).unwrap();
1149 }
1150
1151 #[test]
1152 fn the_chain_refuses_a_genesis_in_the_wrong_position() {
1153 let error = KernelRecord::chain(None, &normalize(&start_envelope()), &step("start"))
1155 .expect_err("only a resolved configuration opens a chain");
1156 assert!(matches!(error, RecordError::ChainBroken(_)), "{error}");
1157
1158 let genesis = genesis_record();
1160 let error = KernelRecord::chain(
1161 Some(&genesis),
1162 &normalize(&configure_envelope()),
1163 &step("configure again"),
1164 )
1165 .expect_err("an operation is configured exactly once");
1166 assert!(matches!(error, RecordError::ChainBroken(_)), "{error}");
1167 }
1168
1169 #[test]
1170 fn a_record_cannot_chain_onto_another_operation() {
1171 let genesis = genesis_record();
1172 let mut foreign = normalize(&start_envelope());
1173 foreign.operation_id = OperationId::new("op-other").unwrap();
1174 let error = KernelRecord::chain(Some(&genesis), &foreign, &step("start"))
1175 .expect_err("operations do not share a chain");
1176 assert!(matches!(error, RecordError::ChainBroken(_)), "{error}");
1177 }
1178
1179 #[test]
1180 fn tampering_with_any_link_is_detected() {
1181 let chain = sample_chain();
1182
1183 for field in [
1185 "operation_id",
1186 "input_id",
1187 "step_seq",
1188 "input_digest",
1189 "step_digest",
1190 "previous_record_digest",
1191 ] {
1192 let mut value = serde_json::to_value(&chain[2]).unwrap();
1193 value[field] = match field {
1194 "step_seq" => json!("99"),
1195 "previous_record_digest" | "input_digest" | "step_digest" => {
1196 json!(canonical_digest(b"forged").as_str())
1197 }
1198 _ => json!("forged"),
1199 };
1200 let error = serde_json::from_value::<KernelRecord>(value)
1201 .expect_err(&format!("editing {field} must be detected"));
1202 assert!(
1203 error.to_string().contains(RECORD_ERROR_MARKER),
1204 "{field}: {error}"
1205 );
1206 }
1207
1208 let mut value = serde_json::to_value(&chain[1]).unwrap();
1210 value["canonical_input"]["data"] =
1211 serde_json::to_value(CanonicalBytes::new(b"{}".to_vec())).unwrap()["data"].clone();
1212 assert!(serde_json::from_value::<KernelRecord>(value).is_err());
1213
1214 let forged = KernelRecord::chain(
1216 Some(&chain[0]),
1217 &normalize(&cancel_envelope()),
1218 &step("forged"),
1219 )
1220 .unwrap();
1221 let mut broken = chain.clone();
1222 broken[1] = forged;
1223 let error = verify_record_chain(&broken).expect_err("link 2 no longer follows link 1");
1224 assert!(matches!(error, RecordError::ChainBroken(_)), "{error}");
1225
1226 let gapped = vec![chain[0].clone(), chain[2].clone()];
1228 assert!(verify_record_chain(&gapped).is_err());
1229
1230 assert!(verify_record_chain(&[]).is_err());
1232 }
1233
1234 #[test]
1235 fn a_record_verifies_the_step_a_rebuild_recomputes() {
1236 let chain = sample_chain();
1237 chain[1]
1238 .verify_step(&step("start"))
1239 .expect("the same step verifies");
1240 let error = chain[1]
1241 .verify_step(&step("something else"))
1242 .expect_err("a different step must not verify");
1243 assert!(matches!(error, RecordError::DigestMismatch(_)), "{error}");
1244 }
1245
1246 #[test]
1251 fn the_record_is_exactly_the_nine_declared_fields() {
1252 let value = serde_json::to_value(genesis_record()).unwrap();
1253 let keys: BTreeSet<String> = value.as_object().unwrap().keys().cloned().collect();
1254 assert_eq!(
1255 keys,
1256 BTreeSet::from([
1257 "abi_version".to_string(),
1258 "operation_id".to_string(),
1259 "input_id".to_string(),
1260 "step_seq".to_string(),
1261 "previous_record_digest".to_string(),
1262 "canonical_input".to_string(),
1263 "input_digest".to_string(),
1264 "step_digest".to_string(),
1265 "record_digest".to_string(),
1266 ])
1267 );
1268 }
1269
1270 #[test]
1271 fn no_planned_step_survives_into_the_durable_record() {
1272 const BANNED: [&str; 8] = [
1273 "step",
1274 "planned_step",
1275 "committed_step",
1276 "actions",
1277 "effects",
1278 "rendered_context",
1279 "messages",
1280 "faults",
1281 ];
1282
1283 let record = KernelRecord::chain(
1284 None,
1285 &normalize(&configure_envelope()),
1286 &json!({
1287 "actions": [{ "kind": "call_provider" }],
1288 "rendered_context": { "messages": [{ "role": "user", "content": "x" }] },
1289 "effects": [{ "effect_id": "op-record-1:step:0:effect:0" }],
1290 }),
1291 )
1292 .unwrap();
1293
1294 let text = String::from_utf8(record.record_bytes().into_vec()).unwrap();
1295 let value: Value = serde_json::from_str(&text).unwrap();
1296 let mut keys = BTreeSet::new();
1297 collect_keys(&value, &mut keys);
1298 for banned in BANNED {
1299 assert!(
1300 !keys.contains(banned),
1301 "the record leaked the step key {banned:?}"
1302 );
1303 }
1304 assert!(!text.contains("rendered_context"));
1305 assert!(!text.contains("call_provider"));
1306 }
1307
1308 #[test]
1309 fn record_size_is_a_function_of_the_input_not_of_the_step() {
1310 let input = normalize(&resolve_envelope());
1311 let genesis = genesis_record();
1312
1313 let tiny = KernelRecord::chain(Some(&genesis), &input, &json!({})).unwrap();
1314 let huge = KernelRecord::chain(
1315 Some(&genesis),
1316 &input,
1317 &json!({ "rendered_context": "x".repeat(200_000) }),
1318 )
1319 .unwrap();
1320
1321 assert_eq!(
1322 tiny.record_bytes().len(),
1323 huge.record_bytes().len(),
1324 "a 200 KiB step must not grow the journal record by one byte"
1325 );
1326 assert_ne!(tiny.step_digest(), huge.step_digest());
1327 assert_eq!(tiny.input_digest(), huge.input_digest());
1328 }
1329
1330 fn collect_keys(value: &Value, out: &mut BTreeSet<String>) {
1331 match value {
1332 Value::Object(map) => {
1333 for (key, item) in map {
1334 out.insert(key.clone());
1335 collect_keys(item, out);
1336 }
1337 }
1338 Value::Array(items) => items.iter().for_each(|item| collect_keys(item, out)),
1339 _ => {}
1340 }
1341 }
1342
1343 #[test]
1348 fn records_round_trip_through_their_journal_bytes() {
1349 for record in sample_chain() {
1350 let bytes = record.record_bytes();
1351 let back = KernelRecord::from_record_bytes(bytes.as_slice()).unwrap();
1352 assert_eq!(back, record);
1353 assert_eq!(back.record_bytes(), bytes, "journal bytes are stable");
1354
1355 let input = record.normalized_input().unwrap();
1357 assert_eq!(canonical_bytes(&input).unwrap(), *record.canonical_input());
1358 }
1359 }
1360
1361 #[test]
1362 fn a_record_from_another_abi_revision_fails_closed() {
1363 let mut value = serde_json::to_value(genesis_record()).unwrap();
1364 value["abi_version"] = json!(KERNEL_ABI_VERSION + 1);
1365 assert!(serde_json::from_value::<KernelRecord>(value).is_err());
1366
1367 let mut value = serde_json::to_value(genesis_record()).unwrap();
1368 value["surprise"] = json!(true);
1369 assert!(
1370 serde_json::from_value::<KernelRecord>(value).is_err(),
1371 "a record with an unknown field is not a record"
1372 );
1373 }
1374
1375 #[test]
1376 fn the_record_is_the_durable_half_of_a_preparation() {
1377 let record = genesis_record();
1378 let preparation: RecordPreparation<Value> =
1379 KernelPreparation::Prepared(PreparedTransition {
1380 token: PrepareToken::new("prepare-1").unwrap(),
1381 record: record.clone(),
1382 planned_step: step("configure"),
1383 });
1384
1385 assert_eq!(preparation.record(), Some(&record));
1386 assert!(
1387 preparation.token().is_some(),
1388 "only a prepared transition commits"
1389 );
1390
1391 let text = serde_json::to_string(&preparation).unwrap();
1392 let back: RecordPreparation<Value> = serde_json::from_str(&text).unwrap();
1393 assert_eq!(back, preparation);
1394
1395 record
1397 .verify_step(preparation.step().unwrap())
1398 .expect("the preparation's step is the one the record froze");
1399 }
1400
1401 #[test]
1406 fn golden_canonical_bytes_vectors() {
1407 let vectors = vec![
1408 ("empty_object", json!({})),
1409 ("empty_array", json!([])),
1410 ("key_order", json!({ "b": 1, "A": 2, "a": 3, "": 4 })),
1411 (
1412 "code_point_key_order",
1413 json!({ "\u{10000}": 1, "\u{fffd}": 2, "z": 3 }),
1414 ),
1415 ("escapes", json!("\" \\ \n \t \u{1} \u{e9} \u{1f600}")),
1416 (
1417 "numbers",
1418 json!([0, -0.0, 2.0, -17, 0.5, 9007199254740991u64]),
1419 ),
1420 (
1421 "null_and_bools",
1422 json!({ "a": null, "b": true, "c": false }),
1423 ),
1424 (
1425 "nested",
1426 json!({ "outer": { "inner": [1, { "deep": "value" }] } }),
1427 ),
1428 (
1429 "wire_u64_is_a_string",
1430 json!({ "step_seq": "18446744073709551615" }),
1431 ),
1432 ];
1433
1434 let produced = json!({
1435 "description":
1436 "Canonical byte vectors (spec 7.1.1). `canonical` is the exact UTF-8 byte string \
1437 core produces; `digest` is SHA-256 over those bytes, hex, sha256-prefixed. \
1438 Object keys sort by Unicode code point.",
1439 "vectors": vectors
1440 .iter()
1441 .map(|(name, value)| {
1442 let canonical = canonical_text(value);
1443 json!({
1444 "name": name,
1445 "value": value,
1446 "canonical": canonical,
1447 "digest": canonical_digest(canonical.as_bytes()).as_str(),
1448 })
1449 })
1450 .collect::<Vec<_>>(),
1451 "rejected": [
1452 { "name": "integer_beyond_exact_range", "value": 9007199254740992u64 },
1453 { "name": "negative_integer_beyond_exact_range", "value": -9007199254740992i64 },
1454 ],
1455 });
1456
1457 let expected = golden("golden_record_canonical_bytes.json", &produced);
1458 assert_eq!(produced, expected, "canonical byte vectors drifted");
1459
1460 for rejected in expected["rejected"].as_array().unwrap() {
1461 assert!(
1462 canonical_bytes(&rejected["value"]).is_err(),
1463 "{} must have no canonical form",
1464 rejected["name"]
1465 );
1466 }
1467 }
1468
1469 #[test]
1470 fn golden_genesis_record() {
1471 let envelope = configure_envelope();
1472 let input = normalize(&envelope);
1473 let planned = step("configure");
1474 let record = KernelRecord::chain(None, &input, &planned).unwrap();
1475
1476 let produced = json!({
1477 "description":
1478 "Genesis record (spec 8.1, 12.1). `canonical_input` holds the resolved \
1479 configuration, `previous_record_digest` is null, and `record_digest` is the \
1480 operation's genesis_digest.",
1481 "envelope": serde_json::to_value(&envelope).unwrap(),
1482 "step": planned,
1483 "normalized_input": serde_json::to_value(&input).unwrap(),
1484 "canonical_input": canonical_text(&input),
1485 "record": serde_json::to_value(&record).unwrap(),
1486 "record_bytes": String::from_utf8(record.record_bytes().into_vec()).unwrap(),
1487 "genesis_digest": record.record_digest().as_str(),
1488 });
1489
1490 let expected = golden("golden_record_genesis.json", &produced);
1491 assert_eq!(produced, expected, "the genesis record drifted");
1492 assert_eq!(expected["record"]["previous_record_digest"], Value::Null);
1493 }
1494
1495 #[test]
1496 fn golden_transition_record() {
1497 let genesis = genesis_record();
1498 let envelope = resolve_envelope();
1499 let input = normalize(&envelope);
1500 let planned = step("resolve");
1501 let record = KernelRecord::chain(Some(&genesis), &input, &planned).unwrap();
1502
1503 let produced = json!({
1504 "description":
1505 "A non-genesis record (spec 8.1). It stores the normalised envelope plus a step \
1506 digest, never the planned step, and links to `previous_record`'s digest.",
1507 "previous_record": serde_json::to_value(&genesis).unwrap(),
1508 "envelope": serde_json::to_value(&envelope).unwrap(),
1509 "step": planned,
1510 "normalized_input": serde_json::to_value(&input).unwrap(),
1511 "canonical_input": canonical_text(&input),
1512 "record": serde_json::to_value(&record).unwrap(),
1513 "record_bytes": String::from_utf8(record.record_bytes().into_vec()).unwrap(),
1514 });
1515
1516 let expected = golden("golden_record_transition.json", &produced);
1517 assert_eq!(produced, expected, "the transition record drifted");
1518
1519 let previous: KernelRecord =
1521 serde_json::from_value(expected["previous_record"].clone()).unwrap();
1522 let decoded: KernelRecord = serde_json::from_value(expected["record"].clone()).unwrap();
1523 decoded.verify_follows(Some(&previous)).unwrap();
1524 }
1525
1526 #[test]
1527 fn golden_record_chain_of_three() {
1528 let envelopes = [configure_envelope(), start_envelope(), resolve_envelope()];
1529 let steps = [step("configure"), step("start"), step("resolve")];
1530
1531 let mut records: Vec<KernelRecord> = Vec::new();
1532 let mut links = Vec::new();
1533 for (envelope, planned) in envelopes.iter().zip(steps.iter()) {
1534 let input = normalize(envelope);
1535 let record = KernelRecord::chain(records.last(), &input, planned).unwrap();
1536 links.push(json!({
1537 "envelope": serde_json::to_value(envelope).unwrap(),
1538 "step": planned,
1539 "record": serde_json::to_value(&record).unwrap(),
1540 }));
1541 records.push(record);
1542 }
1543
1544 let produced = json!({
1545 "description":
1546 "Three-link record chain (spec 8.1, 12.1). Replaying the envelopes through \
1547 normalisation and KernelRecord::chain must reproduce every record byte for \
1548 byte; `genesis_digest` is the operation's identity and `head_digest` its CAS head.",
1549 "genesis_digest": records[0].record_digest().as_str(),
1550 "head_digest": records[2].record_digest().as_str(),
1551 "links": links,
1552 });
1553
1554 let expected = golden("golden_record_chain.json", &produced);
1555 assert_eq!(produced, expected, "the record chain drifted");
1556
1557 let decoded: Vec<KernelRecord> = expected["links"]
1558 .as_array()
1559 .unwrap()
1560 .iter()
1561 .map(|link| serde_json::from_value(link["record"].clone()).unwrap())
1562 .collect();
1563 assert_eq!(
1564 verify_record_chain(&decoded).unwrap().as_str(),
1565 expected["genesis_digest"].as_str().unwrap()
1566 );
1567 }
1568}