1use std::collections::HashMap;
80
81use serde::Serialize;
82
83use crate::runtime::kernel::wire::ConfigDefaults;
84use crate::runtime::kernel::wire::checkpoint::KernelCheckpoint;
85use crate::runtime::kernel::wire::effect::{
86 EffectKind, EffectOutcome, EffectSuccess, ProviderOutcome, SpawnTasksEffect,
87};
88use crate::runtime::kernel::wire::record::{
89 KernelRecord, NormalizedPayload, RecordError, verify_record_chain,
90};
91use crate::runtime::kernel::wire::restore::{RestoredOperation, restore_operation};
92use crate::runtime::kernel::wire::transaction::InMemoryRecordIndex;
93
94pub const UNATTRIBUTED_SEGMENT: &str = "(unattributed)";
97
98const DEFERRED_ALWAYS: &str = "c4.parent_chain: parent links are not journaled; an orphan spawn \
101 cannot resolve (no outstanding effect), which C3's re-plan enforces structurally";
102
103const DEFERRED_WITHOUT_CHECKPOINT: &str = "c5b.launch_token_ledger: the durable LaunchToken \
107 ledger lives in checkpoints; provide --checkpoint to check token reuse across TaskLaunch \
108 payloads (the journal-direct shadow — (task_id, attempt_id) pair uniqueness — is checked \
109 under C4)";
110
111#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
113pub struct RuleReport {
114 pub rule: String,
116 pub verdict: Verdict,
117 pub detail: String,
119}
120
121#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
122#[serde(rename_all = "snake_case")]
123pub enum Verdict {
124 Pass,
125 Fail,
126 Degraded,
128}
129
130#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
132pub struct DegradedHop {
133 pub ordinal: usize,
135 pub step_seq: Option<u64>,
136 pub reason: String,
138}
139
140#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
142pub struct SegmentReport {
143 pub operation_id: String,
144 pub hops: usize,
145 pub degraded_hops: Vec<DegradedHop>,
146 pub rules: Vec<RuleReport>,
147}
148
149#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
150pub struct ValidationReport {
151 pub segments: Vec<SegmentReport>,
152 pub unparseable_records: usize,
154 #[serde(default)]
157 pub cross_checks: Vec<RuleReport>,
158 #[serde(default)]
161 pub unparseable_events: usize,
162 #[serde(default)]
166 pub session_events: Option<usize>,
167 #[serde(default)]
170 pub checkpoint_checks: Vec<RuleReport>,
171 #[serde(default)]
174 pub unparseable_checkpoints: usize,
175 #[serde(default)]
179 pub checkpoints: Option<usize>,
180 pub deferred: Vec<String>,
182}
183
184impl ValidationReport {
185 pub fn has_violations(&self) -> bool {
186 self.segments
187 .iter()
188 .flat_map(|segment| segment.rules.iter())
189 .chain(self.cross_checks.iter())
190 .chain(self.checkpoint_checks.iter())
191 .any(|rule| rule.verdict == Verdict::Fail)
192 }
193
194 pub fn exit_code(&self) -> i32 {
200 if self.has_violations() {
201 1
202 } else if self.unparseable_records > 0
203 || self.unparseable_events > 0
204 || self.unparseable_checkpoints > 0
205 || self.segments.is_empty()
206 || matches!(self.session_events, Some(0))
207 || matches!(self.checkpoints, Some(0))
208 {
209 2
210 } else {
211 0
212 }
213 }
214}
215
216enum Hop {
219 Complete(KernelRecord),
220 Degraded(DegradedRecord),
221}
222
223struct DegradedRecord {
224 ordinal: usize,
225 operation_id: Option<String>,
226 input_id: Option<String>,
227 step_seq: Option<u64>,
228 previous_record_digest: Option<String>,
229 record_digest: Option<String>,
230 reason: String,
231 integrity_failure: bool,
232}
233
234impl DegradedRecord {
235 fn marking(&self) -> DegradedHop {
236 DegradedHop {
237 ordinal: self.ordinal,
238 step_seq: self.step_seq,
239 reason: self.reason.clone(),
240 }
241 }
242}
243
244impl Hop {
245 fn operation_id(&self) -> Option<&str> {
246 match self {
247 Self::Complete(record) => Some(record.operation_id().as_str()),
248 Self::Degraded(degraded) => degraded.operation_id.as_deref(),
249 }
250 }
251
252 fn step_seq(&self) -> Option<u64> {
253 match self {
254 Self::Complete(record) => Some(record.step_seq().get()),
255 Self::Degraded(degraded) => degraded.step_seq,
256 }
257 }
258}
259
260pub fn validate_journal<B: AsRef<[u8]>>(blobs: &[B]) -> ValidationReport {
264 validate_with_session_log(blobs, &[] as &[Vec<Vec<u8>>])
265}
266
267pub fn validate_with_session_log<J, S>(
272 journal_blobs: &[J],
273 session_streams: &[Vec<S>],
274) -> ValidationReport
275where
276 J: AsRef<[u8]>,
277 S: AsRef<[u8]>,
278{
279 validate_with_checkpoint(journal_blobs, session_streams, &[] as &[Vec<u8>], false)
280}
281
282pub fn validate_with_checkpoint<J, S, C>(
289 journal_blobs: &[J],
290 session_streams: &[Vec<S>],
291 checkpoint_blobs: &[C],
292 strict: bool,
293) -> ValidationReport
294where
295 J: AsRef<[u8]>,
296 S: AsRef<[u8]>,
297 C: AsRef<[u8]>,
298{
299 let (outcomes, unparseable_records, segment_records) = validate_journal_plane(journal_blobs);
300
301 let mut streams: Vec<SessionStream> = Vec::with_capacity(session_streams.len());
302 for stream_blobs in session_streams {
303 let mut events = Vec::with_capacity(stream_blobs.len());
304 let mut unparseable_events = 0;
305 for blob in stream_blobs {
306 match classify_session_event(blob.as_ref()) {
307 Some(event) => events.push(event),
308 None => unparseable_events += 1,
309 }
310 }
311 streams.push(SessionStream {
312 events,
313 unparseable_events,
314 });
315 }
316
317 let session_plane_provided = !session_streams.is_empty();
318 let session_events =
319 session_plane_provided.then(|| streams.iter().map(|stream| stream.events.len()).sum());
320 let unparseable_events = streams.iter().map(|stream| stream.unparseable_events).sum();
321
322 let cross_checks = if session_plane_provided {
324 let mut checks = check_c6(&streams, &outcomes);
325 checks.push(check_c8(&streams, &outcomes));
326 checks
327 } else {
328 Vec::new()
329 };
330
331 let checkpoint_plane_provided = !checkpoint_blobs.is_empty();
333 let (checkpoint_checks, unparseable_checkpoints, checkpoints) = if checkpoint_plane_provided {
334 let (checks, unparseable, parsed) = check_c5(&segment_records, checkpoint_blobs, strict);
335 (checks, unparseable, Some(parsed))
336 } else {
337 (Vec::new(), 0, None)
338 };
339
340 let mut deferred = vec![DEFERRED_ALWAYS.to_string()];
341 if !checkpoint_plane_provided {
342 deferred.push(DEFERRED_WITHOUT_CHECKPOINT.to_string());
343 }
344
345 ValidationReport {
346 segments: outcomes.into_iter().map(|outcome| outcome.report).collect(),
347 unparseable_records,
348 cross_checks,
349 unparseable_events,
350 session_events,
351 checkpoint_checks,
352 unparseable_checkpoints,
353 checkpoints,
354 deferred,
355 }
356}
357
358fn validate_journal_plane<B: AsRef<[u8]>>(
362 blobs: &[B],
363) -> (
364 Vec<SegmentOutcome>,
365 usize,
366 HashMap<String, Vec<KernelRecord>>,
367) {
368 let mut hops: Vec<Hop> = Vec::with_capacity(blobs.len());
369 let mut unparseable_records = 0;
370 for (ordinal, blob) in blobs.iter().enumerate() {
371 match classify(ordinal, blob.as_ref()) {
372 Some(hop) => hops.push(hop),
373 None => unparseable_records += 1,
374 }
375 }
376
377 let mut segments: HashMap<String, Vec<Hop>> = HashMap::new();
378 let mut complete: HashMap<String, Vec<KernelRecord>> = HashMap::new();
379 for hop in hops {
380 let key = hop
381 .operation_id()
382 .map(str::to_string)
383 .unwrap_or_else(|| UNATTRIBUTED_SEGMENT.to_string());
384 if let Hop::Complete(record) = &hop {
385 complete
386 .entry(key.clone())
387 .or_default()
388 .push(record.clone());
389 }
390 segments.entry(key).or_default().push(hop);
391 }
392
393 let mut keys: Vec<String> = segments.keys().cloned().collect();
394 keys.sort();
395 let reports = keys
396 .iter()
397 .map(|key| validate_segment(key, segments.remove(key).unwrap_or_default()))
398 .collect();
399 (reports, unparseable_records, complete)
400}
401
402pub struct SessionStream {
409 pub events: Vec<EvidenceEvent>,
410 pub unparseable_events: usize,
411}
412
413#[derive(Debug, Clone, PartialEq, Eq)]
416pub enum EvidenceEvent {
417 RunStarted { route_id: Option<String> },
420 ProviderAttempt {
422 effect_id: Option<String>,
423 request_fingerprint: Option<String>,
424 route_id: Option<String>,
425 status: Option<String>,
426 },
427 PromptMeasured {
430 effect_id: Option<String>,
431 request_fingerprint: Option<String>,
432 },
433 LlmCompleted {
436 effect_id: Option<String>,
437 invocation_id: Option<String>,
438 },
439 Other,
441}
442
443fn classify_session_event(bytes: &[u8]) -> Option<EvidenceEvent> {
446 let value: serde_json::Value = serde_json::from_slice(bytes).ok()?;
447 let object = value.as_object()?;
448 let string = |key: &str| {
449 object
450 .get(key)
451 .and_then(serde_json::Value::as_str)
452 .map(str::to_string)
453 };
454 let kind = string("kind");
455 let event = match kind.as_deref() {
456 Some("run_started") => EvidenceEvent::RunStarted {
457 route_id: route_id_of(&value),
458 },
459 Some("provider_attempt") => EvidenceEvent::ProviderAttempt {
460 effect_id: string("effect_id"),
461 request_fingerprint: string("request_fingerprint"),
462 route_id: route_id_of(&value),
463 status: string("status"),
464 },
465 Some("prompt_measured") => EvidenceEvent::PromptMeasured {
466 effect_id: string("effect_id"),
467 request_fingerprint: object
470 .get("measurement")
471 .and_then(|measurement| {
472 measurement
473 .get("requestFingerprint")
474 .or_else(|| measurement.get("request_fingerprint"))
475 })
476 .and_then(serde_json::Value::as_str)
477 .map(str::to_string),
478 },
479 Some("llm_completed") => EvidenceEvent::LlmCompleted {
480 effect_id: string("effect_id"),
481 invocation_id: string("invocation_id"),
482 },
483 _ => EvidenceEvent::Other,
484 };
485 Some(event)
486}
487
488fn route_id_of(event: &serde_json::Value) -> Option<String> {
490 let route = event.get("route")?;
491 route
492 .get("routeId")
493 .or_else(|| route.get("route_id"))
494 .and_then(serde_json::Value::as_str)
495 .map(str::to_string)
496}
497
498fn check_c6(streams: &[SessionStream], outcomes: &[SegmentOutcome]) -> Vec<RuleReport> {
527 let segments: HashMap<&str, &SegmentOutcome> = outcomes
528 .iter()
529 .filter(|outcome| outcome.report.operation_id != UNATTRIBUTED_SEGMENT)
530 .map(|outcome| (outcome.report.operation_id.as_str(), outcome))
531 .collect();
532
533 let mut correspondence = ClauseAccumulator::default();
534 let mut fingerprints = ClauseAccumulator::default();
535 let mut routes = ClauseAccumulator::default();
536 let mut total_attempts = 0usize;
537
538 for (index, stream) in streams.iter().enumerate() {
539 let measured: std::collections::HashSet<&str> = stream
540 .events
541 .iter()
542 .filter_map(|event| match event {
543 EvidenceEvent::PromptMeasured {
544 request_fingerprint,
545 ..
546 } => request_fingerprint.as_deref(),
547 _ => None,
548 })
549 .collect();
550
551 let mut baseline: Option<&str> = None;
554 let mut last_pinned: Option<&str> = None;
555
556 for event in &stream.events {
557 match event {
558 EvidenceEvent::RunStarted { route_id } => {
559 if let Some(new_route) = route_id.as_deref() {
560 if let Some(previous) = last_pinned
561 && previous != new_route
562 {
563 routes.degraded(format!(
564 "stream #{index}: run resumed on route {new_route} (was \
565 {previous}) — a cross-resume change, degraded per Q3"
566 ));
567 }
568 baseline = Some(new_route);
569 last_pinned = Some(new_route);
570 } else {
571 baseline = None;
574 }
575 }
576 EvidenceEvent::ProviderAttempt {
577 effect_id,
578 request_fingerprint,
579 route_id,
580 ..
581 } => {
582 total_attempts += 1;
583 let label = effect_id.as_deref().unwrap_or("(no effect_id)");
584
585 match effect_id.as_deref() {
587 None => correspondence.violation(format!(
588 "stream #{index}: provider_attempt without effect_id — the writers \
589 mint it from the kernel effect, so a keyless attempt is forged \
590 evidence"
591 )),
592 Some(effect) => match parse_effect_step(effect) {
593 None => correspondence.violation(format!(
594 "stream #{index}: attempt names {effect}, which is not in the \
595 `operation:step:N:effect:M` vocabulary"
596 )),
597 Some((operation, step)) => match segments.get(operation) {
598 None => correspondence.degraded(format!(
599 "stream #{index}: attempt names {effect}, but operation \
600 {operation} has no journal segment (the journal may cover \
601 a subset of the session)"
602 )),
603 Some(outcome) => match &outcome.effects {
604 None => correspondence.degraded(format!(
605 "stream #{index}: segment {operation} could not be \
606 re-planned, so {effect}'s publication is unverifiable"
607 )),
608 Some(effects) if effects.published.contains(effect) => {
609 correspondence.checked += 1;
610 }
611 Some(effects) if step > effects.max_step => {
612 correspondence.degraded(format!(
613 "stream #{index}: attempt names {effect} at step \
614 {step}, past the journal's tip (step {}) — a \
615 prefix cannot disprove it",
616 effects.max_step
617 ));
618 }
619 Some(_) => correspondence.violation(format!(
620 "stream #{index}: attempt names {effect}, but the \
621 deterministic re-plan of {operation} never published \
622 it — the attempt is unmoored from the journal"
623 )),
624 },
625 },
626 },
627 }
628
629 match request_fingerprint.as_deref() {
631 None => fingerprints.violation(format!(
632 "stream #{index}: provider_attempt {label} without \
633 request_fingerprint — the writers require it (G2)"
634 )),
635 Some(fingerprint) if measured.contains(fingerprint) => {
636 fingerprints.checked += 1;
637 }
638 Some(fingerprint) => fingerprints.violation(format!(
639 "stream #{index}: attempt {label} carries fingerprint \
640 {fingerprint}, but no prompt_measured in this session carries it"
641 )),
642 }
643
644 match (route_id.as_deref(), baseline) {
646 (Some(route), Some(pinned)) if route != pinned => {
647 routes.violation(format!(
648 "stream #{index}: attempt {label} ran on route {route} inside \
649 a run pinned to {pinned} — an in-run route change is a \
650 violation (Q3)"
651 ))
652 }
653 (Some(_), Some(_)) => routes.checked += 1,
654 _ => routes.unverifiable += 1,
656 }
657 }
658 _ => {}
659 }
660 }
661 }
662
663 vec![
664 correspondence.report(
665 "C6.1",
666 total_attempts,
667 "attempt↔journal effect correspondence",
668 ),
669 fingerprints.report(
670 "C6.2",
671 total_attempts,
672 "attempt fingerprint↔prompt_measured join",
673 ),
674 routes.report("C6.3", total_attempts, "in-run route stability"),
675 ]
676}
677
678#[derive(Default)]
681struct ClauseAccumulator {
682 checked: usize,
683 unverifiable: usize,
684 violations: Vec<String>,
685 degraded_notes: Vec<String>,
686}
687
688impl ClauseAccumulator {
689 fn violation(&mut self, detail: String) {
690 self.violations.push(detail);
691 }
692
693 fn degraded(&mut self, note: String) {
694 self.degraded_notes.push(note);
695 }
696
697 fn report(self, rule: &str, total_attempts: usize, what: &str) -> RuleReport {
698 let rule = rule.to_string();
699 if !self.violations.is_empty() {
700 return RuleReport {
701 rule,
702 verdict: Verdict::Fail,
703 detail: self.violations.join("; "),
704 };
705 }
706 if total_attempts == 0 {
707 return RuleReport {
708 rule,
709 verdict: Verdict::Degraded,
710 detail: format!(
711 "no provider_attempt events in any stream — a pre-0.2.63 log carries none \
712 (C7); {what} unchecked"
713 ),
714 };
715 }
716 if !self.degraded_notes.is_empty() || self.unverifiable > 0 {
717 let mut detail = self.degraded_notes.join("; ");
718 if self.unverifiable > 0 {
719 if !detail.is_empty() {
720 detail.push_str("; ");
721 }
722 detail.push_str(&format!(
723 "{} attempt(s) unverifiable (no pinned run route)",
724 self.unverifiable
725 ));
726 }
727 return RuleReport {
728 rule,
729 verdict: Verdict::Degraded,
730 detail: format!("{} attempt(s) verified for {what}; {detail}", self.checked),
731 };
732 }
733 RuleReport {
734 rule,
735 verdict: Verdict::Pass,
736 detail: format!(
737 "{} attempt(s) verified — {what} holds across every stream",
738 self.checked
739 ),
740 }
741 }
742}
743
744fn check_c8(streams: &[SessionStream], outcomes: &[SegmentOutcome]) -> RuleReport {
778 let rule = "C8".to_string();
779 let segments: HashMap<&str, &SegmentOutcome> = outcomes
780 .iter()
781 .filter(|outcome| outcome.report.operation_id != UNATTRIBUTED_SEGMENT)
782 .map(|outcome| (outcome.report.operation_id.as_str(), outcome))
783 .collect();
784
785 let mut llm_completed_events = 0usize;
786 let mut checked = 0usize;
787 let mut trivial = 0usize;
788 let mut unverifiable = 0usize;
789 let mut violations: Vec<String> = Vec::new();
790 let mut degraded_notes: Vec<String> = Vec::new();
791
792 for (index, stream) in streams.iter().enumerate() {
793 for event in &stream.events {
794 let EvidenceEvent::LlmCompleted {
795 effect_id,
796 invocation_id,
797 } = event
798 else {
799 continue;
800 };
801 llm_completed_events += 1;
802 let (Some(head), Some(selected)) = (invocation_id.as_deref(), effect_id.as_deref())
803 else {
804 unverifiable += 1;
807 continue;
808 };
809 if head == selected {
810 trivial += 1;
811 continue;
812 }
813
814 let (Some((head_op, head_step)), Some((selected_op, selected_step))) =
815 (parse_effect_step(head), parse_effect_step(selected))
816 else {
817 violations.push(format!(
818 "stream #{index}: llm_completed claims invocation {head} → {selected}, but \
819 one of the pair is not in the `operation:step:N:effect:M` vocabulary"
820 ));
821 continue;
822 };
823 if head_op != selected_op {
824 violations.push(format!(
825 "stream #{index}: llm_completed claims invocation {head} → {selected} — an \
826 invocation chain cannot cross operations"
827 ));
828 continue;
829 }
830 if selected_step <= head_step {
831 violations.push(format!(
832 "stream #{index}: llm_completed claims invocation {head} → {selected}, but \
833 the selected effect does not follow the chain head"
834 ));
835 continue;
836 }
837 let Some(outcome) = segments.get(head_op) else {
838 degraded_notes.push(format!(
839 "stream #{index}: operation {head_op} has no journal segment (the journal \
840 may cover a subset of the session)"
841 ));
842 continue;
843 };
844 let Some(effects) = &outcome.effects else {
845 degraded_notes.push(format!(
846 "stream #{index}: segment {head_op} could not be re-planned, so the \
847 invocation {head} → {selected} is unverifiable"
848 ));
849 continue;
850 };
851
852 match effects.resolutions.get(head) {
854 Some(ResolutionFact::Completed) | Some(ResolutionFact::Other) => {
855 violations.push(format!(
856 "stream #{index}: llm_completed claims invocation {head} → {selected}, \
857 but {head} resolved to completion — a completed effect closes its \
858 invocation; nothing chains from it"
859 ));
860 continue;
861 }
862 Some(ResolutionFact::Overflow) | Some(ResolutionFact::Failed) => {}
863 None if effects.published.contains(head) => degraded_notes.push(format!(
864 "stream #{index}: chain head {head} is published but its resolution has \
865 not landed in the journal (the planes are not synchronised)"
866 )),
867 None if head_step > effects.max_step => degraded_notes.push(format!(
868 "stream #{index}: chain head {head} claims step {head_step}, past the \
869 journal's tip (step {})",
870 effects.max_step
871 )),
872 None => {
873 violations.push(format!(
874 "stream #{index}: llm_completed claims invocation head {head}, but the \
875 deterministic re-plan of {head_op} never published it"
876 ));
877 continue;
878 }
879 }
880
881 if effects.resolutions.contains_key(selected) {
883 checked += 1;
884 } else if effects.published.contains(selected) || selected_step > effects.max_step {
885 degraded_notes.push(format!(
886 "stream #{index}: selected effect {selected} is not resolved in the \
887 journal (the planes are not synchronised)"
888 ));
889 } else {
890 violations.push(format!(
891 "stream #{index}: llm_completed selects {selected}, but the deterministic \
892 re-plan of {selected_op} never published it"
893 ));
894 }
895 }
896 }
897
898 if !violations.is_empty() {
899 return RuleReport {
900 rule,
901 verdict: Verdict::Fail,
902 detail: violations.join("; "),
903 };
904 }
905 if llm_completed_events == 0 {
906 return RuleReport {
907 rule,
908 verdict: Verdict::Degraded,
909 detail: "no llm_completed events in any stream — invocation adjacency unchecked"
910 .to_string(),
911 };
912 }
913 if !degraded_notes.is_empty() || unverifiable > 0 {
914 let mut detail = degraded_notes.join("; ");
915 if unverifiable > 0 {
916 if !detail.is_empty() {
917 detail.push_str("; ");
918 }
919 detail.push_str(&format!(
920 "{unverifiable} llm_completed event(s) without invocation_id/effect_id \
921 (pre-0.2.63 fields — C7)"
922 ));
923 }
924 if !detail.is_empty() {
925 return RuleReport {
926 rule,
927 verdict: Verdict::Degraded,
928 detail: format!(
929 "{checked} retried invocation(s) verified, {trivial} first-try chain(s) \
930 closed; {detail}"
931 ),
932 };
933 }
934 }
935 RuleReport {
936 rule,
937 verdict: Verdict::Pass,
938 detail: format!(
939 "{checked} retried invocation(s) verified end-to-end, {trivial} first-try chain(s) \
940 closed"
941 ),
942 }
943}
944
945fn classify(ordinal: usize, bytes: &[u8]) -> Option<Hop> {
949 let error = match KernelRecord::from_record_bytes(bytes) {
950 Ok(record) => return Some(Hop::Complete(record)),
951 Err(error) => error,
952 };
953 let value: serde_json::Value = serde_json::from_slice(bytes).ok()?;
954 let object = value.as_object()?;
955 let string = |key: &str| {
956 object
957 .get(key)
958 .and_then(serde_json::Value::as_str)
959 .map(str::to_string)
960 };
961 let step_seq = object.get("step_seq").and_then(|value| {
964 value
965 .as_u64()
966 .or_else(|| value.as_str().and_then(|text| text.parse().ok()))
967 });
968 let degraded = DegradedRecord {
969 ordinal,
970 operation_id: string("operation_id"),
971 input_id: string("input_id"),
972 step_seq,
973 previous_record_digest: string("previous_record_digest"),
974 record_digest: string("record_digest"),
975 reason: format!("{}: {}", error.code().as_str(), error.message()),
976 integrity_failure: matches!(error, RecordError::DigestMismatch(_)),
977 };
978 if degraded.operation_id.is_some() || degraded.step_seq.is_some() {
981 Some(Hop::Degraded(degraded))
982 } else {
983 None
984 }
985}
986
987fn validate_segment(operation_id: &str, mut hops: Vec<Hop>) -> SegmentOutcome {
988 hops.sort_by_key(|hop| {
991 (
992 hop.step_seq().unwrap_or(u64::MAX),
993 match hop {
994 Hop::Complete(_) => 0usize,
995 Hop::Degraded(degraded) => degraded.ordinal,
996 },
997 )
998 });
999
1000 let degraded_hops: Vec<DegradedHop> = hops
1001 .iter()
1002 .filter_map(|hop| match hop {
1003 Hop::Degraded(degraded) => Some(degraded.marking()),
1004 Hop::Complete(_) => None,
1005 })
1006 .collect();
1007 let hop_count = hops.len();
1008
1009 let c1 = check_c1(&hops);
1010 let replan = replan_segment(&hops, &c1);
1011 let c2 = check_c2(&hops);
1012 let c3 = render_c3(&replan);
1013 let c4 = check_c4(&hops, operation_id);
1014
1015 let effects = match &replan {
1018 Replan::Restored(restored) => {
1019 let mut published: std::collections::HashSet<String> = std::collections::HashSet::new();
1020 let mut resolutions: HashMap<String, ResolutionFact> = HashMap::new();
1021 for hop in &hops {
1022 if let Some((effect_id, fact)) = resolution_of(hop) {
1023 published.insert(effect_id.clone());
1024 resolutions.insert(effect_id, fact);
1025 }
1026 }
1027 published.extend(
1028 restored
1029 .transaction
1030 .pending_effects()
1031 .map(|effect| effect.effect_id.as_str().to_string()),
1032 );
1033 let max_step = hops
1034 .iter()
1035 .filter_map(|hop| hop.step_seq())
1036 .max()
1037 .unwrap_or(0);
1038 Some(SegmentEffects {
1039 max_step,
1040 published,
1041 resolutions,
1042 })
1043 }
1044 _ => None,
1045 };
1046
1047 SegmentOutcome {
1048 report: SegmentReport {
1049 operation_id: operation_id.to_string(),
1050 hops: hop_count,
1051 degraded_hops,
1052 rules: vec![c1, c2, c3, c4],
1053 },
1054 effects,
1055 }
1056}
1057
1058struct SegmentOutcome {
1060 report: SegmentReport,
1061 effects: Option<SegmentEffects>,
1063}
1064
1065struct SegmentEffects {
1066 max_step: u64,
1069 published: std::collections::HashSet<String>,
1071 resolutions: HashMap<String, ResolutionFact>,
1073}
1074
1075#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1078enum ResolutionFact {
1079 Completed,
1081 Overflow,
1084 Failed,
1088 Other,
1090}
1091
1092fn resolution_of(hop: &Hop) -> Option<(String, ResolutionFact)> {
1095 let Hop::Complete(record) = hop else {
1096 return None;
1097 };
1098 let input = record.normalized_input().ok()?;
1099 let NormalizedPayload::ResolveEffect(resolve) = &input.input else {
1100 return None;
1101 };
1102 let fact = match &resolve.outcome {
1103 EffectOutcome::Failed(_) => ResolutionFact::Failed,
1104 EffectOutcome::Succeeded(success) => match &success.result {
1105 EffectSuccess::Provider(provider) => match &provider.outcome {
1106 ProviderOutcome::Completed(_) => ResolutionFact::Completed,
1107 ProviderOutcome::ContextOverflow(_) => ResolutionFact::Overflow,
1108 },
1109 _ => ResolutionFact::Other,
1110 },
1111 };
1112 Some((resolve.effect_id.as_str().to_string(), fact))
1113}
1114
1115fn check_c1(hops: &[Hop]) -> RuleReport {
1117 let rule = "C1".to_string();
1118 if hops.is_empty() {
1119 return RuleReport {
1120 rule,
1121 verdict: Verdict::Degraded,
1122 detail: "no records in this segment".to_string(),
1123 };
1124 }
1125 let all_complete = hops.iter().all(|hop| matches!(hop, Hop::Complete(_)));
1126 if all_complete {
1127 let records: Vec<KernelRecord> = hops
1128 .iter()
1129 .filter_map(|hop| match hop {
1130 Hop::Complete(record) => Some(record.clone()),
1131 Hop::Degraded(_) => None,
1132 })
1133 .collect();
1134 return match verify_record_chain(&records) {
1135 Ok(genesis_digest) => RuleReport {
1136 rule,
1137 verdict: Verdict::Pass,
1138 detail: format!(
1139 "{} record(s), genesis {genesis_digest}, every link verified",
1140 records.len()
1141 ),
1142 },
1143 Err(error) => RuleReport {
1144 rule,
1145 verdict: Verdict::Fail,
1146 detail: format!("{}: {}", error.code().as_str(), error.message()),
1147 },
1148 };
1149 }
1150
1151 let mut broken: Vec<String> = hops
1154 .iter()
1155 .filter_map(|hop| match hop {
1156 Hop::Degraded(record) if record.integrity_failure => Some(record.reason.clone()),
1157 _ => None,
1158 })
1159 .collect();
1160 let mut unverifiable_links = 0usize;
1161 let mut previous: Option<(&Hop, Option<&KernelRecord>)> = None;
1162 for hop in hops {
1163 let step = hop.step_seq();
1164 let (prev_digest, _) = digests_of(hop);
1165 if let Some((previous_hop, previous_complete)) = previous {
1166 let previous_step = previous_hop.step_seq();
1167 let previous_digest = digests_of(previous_hop).1;
1168 match (prev_digest, previous_digest) {
1169 (Some(expected), Some(actual)) if expected != actual => broken.push(format!(
1170 "hop at step {} expects head {expected}, but its predecessor's digest is \
1171 {actual}",
1172 step.map_or("?".to_string(), |seq| seq.to_string()),
1173 )),
1174 (None, _) => unverifiable_links += 1,
1175 (_, None) => unverifiable_links += 1,
1176 _ => {}
1177 }
1178 match (step, previous_step) {
1179 (Some(step), Some(previous_step)) if step != previous_step + 1 => broken.push(
1180 format!("hop is step {step}, but its predecessor is step {previous_step}"),
1181 ),
1182 (Some(_), Some(_)) => {}
1183 _ => unverifiable_links += 1,
1184 }
1185 if let (Hop::Complete(record), Some(previous_record)) = (hop, previous_complete)
1188 && let Err(error) = record.verify_follows(Some(previous_record))
1189 {
1190 broken.push(format!("{}: {}", error.code().as_str(), error.message()));
1191 }
1192 } else if let Hop::Complete(record) = hop
1193 && let Err(error) = record.verify_follows(None)
1194 {
1195 broken.push(format!("{}: {}", error.code().as_str(), error.message()));
1196 }
1197 previous = Some((
1198 hop,
1199 match hop {
1200 Hop::Complete(record) => Some(record),
1201 Hop::Degraded(_) => None,
1202 },
1203 ));
1204 }
1205
1206 if !broken.is_empty() {
1207 return RuleReport {
1208 rule,
1209 verdict: Verdict::Fail,
1210 detail: broken.join("; "),
1211 };
1212 }
1213 RuleReport {
1214 rule,
1215 verdict: Verdict::Degraded,
1216 detail: format!(
1217 "partial chain: every surviving link verified, {unverifiable_links} link(s) \
1218 unverifiable across degraded hop(s)"
1219 ),
1220 }
1221}
1222
1223fn check_c2(hops: &[Hop]) -> RuleReport {
1225 let rule = "C2".to_string();
1226 let mut by_input: HashMap<&str, &str> = HashMap::new();
1227 let mut conflicts: Vec<String> = Vec::new();
1228 let mut retries = 0usize;
1229 let mut unverifiable = 0usize;
1230 for hop in hops {
1231 let (input_id, record_digest) = match hop {
1232 Hop::Complete(record) => (
1233 Some(record.input_id().as_str()),
1234 Some(record.record_digest().as_str()),
1235 ),
1236 Hop::Degraded(degraded) => (
1237 degraded.input_id.as_deref(),
1238 degraded.record_digest.as_deref(),
1239 ),
1240 };
1241 let Some(input_id) = input_id else { continue };
1242 let Some(digest) = record_digest else {
1243 unverifiable += 1;
1244 continue;
1245 };
1246 match by_input.get(input_id) {
1247 Some(existing) if *existing != digest => conflicts.push(format!(
1248 "input {input_id} produced two different records ({existing} and {digest}); a \
1249 retry must reach the same record"
1250 )),
1251 Some(_) => retries += 1,
1252 None => {
1253 by_input.insert(input_id, digest);
1254 }
1255 }
1256 }
1257 if !conflicts.is_empty() {
1258 return RuleReport {
1259 rule,
1260 verdict: Verdict::Fail,
1261 detail: conflicts.join("; "),
1262 };
1263 }
1264 if unverifiable > 0 {
1265 return RuleReport {
1266 rule,
1267 verdict: Verdict::Degraded,
1268 detail: format!(
1269 "{} unique input(s), {retries} idempotent retry hit(s); {unverifiable} degraded \
1270 hop(s) could not be compared",
1271 by_input.len(),
1272 ),
1273 };
1274 }
1275 RuleReport {
1276 rule,
1277 verdict: Verdict::Pass,
1278 detail: format!(
1279 "{} unique input(s), {retries} idempotent retry hit(s), no divergent duplicates",
1280 by_input.len(),
1281 ),
1282 }
1283}
1284
1285enum Replan {
1290 Unavailable(&'static str),
1292 Failed(String),
1294 Restored(RestoredOperation),
1295}
1296
1297fn replan_segment(hops: &[Hop], c1: &RuleReport) -> Replan {
1298 if hops.iter().any(|hop| matches!(hop, Hop::Degraded(_))) {
1299 return Replan::Unavailable(
1300 "re-plan requires complete records; this segment has degraded hops",
1301 );
1302 }
1303 if c1.verdict == Verdict::Fail {
1304 return Replan::Unavailable(
1305 "C1 failed; a re-plan over a broken chain would only re-report that break",
1306 );
1307 }
1308 let records: Vec<KernelRecord> = hops
1309 .iter()
1310 .filter_map(|hop| match hop {
1311 Hop::Complete(record) => Some(record.clone()),
1312 Hop::Degraded(_) => None,
1313 })
1314 .collect();
1315 if records.is_empty() {
1316 return Replan::Unavailable("no records in this segment");
1317 }
1318 match restore_operation(
1319 None,
1320 &records,
1321 ConfigDefaults::default(),
1322 InMemoryRecordIndex::from_records(&records),
1323 ) {
1324 Ok(restored) => Replan::Restored(restored),
1325 Err(fault) => Replan::Failed(format!("{}: {}", fault.code.as_str(), fault.message)),
1326 }
1327}
1328
1329fn render_c3(replan: &Replan) -> RuleReport {
1330 let rule = "C3".to_string();
1331 match replan {
1332 Replan::Unavailable(reason) => RuleReport {
1333 rule,
1334 verdict: Verdict::Degraded,
1335 detail: (*reason).to_string(),
1336 },
1337 Replan::Failed(fault) => RuleReport {
1338 rule,
1339 verdict: Verdict::Fail,
1340 detail: fault.clone(),
1341 },
1342 Replan::Restored(restored) => RuleReport {
1343 rule,
1344 verdict: Verdict::Pass,
1345 detail: format!(
1346 "re-planned {} record(s) from genesis; every durable record digest reproduced",
1347 restored.cost.records_before_checkpoint
1348 ),
1349 },
1350 }
1351}
1352
1353fn check_c4(hops: &[Hop], operation_id: &str) -> RuleReport {
1355 let rule = "C4".to_string();
1356 struct LaunchFact {
1357 task_id: String,
1358 attempt_id: String,
1359 step_seq: u64,
1360 effect_id: String,
1361 }
1362
1363 let mut launches: Vec<LaunchFact> = Vec::new();
1364 let mut unreadable_inputs = 0usize;
1365 for hop in hops {
1366 let Hop::Complete(record) = hop else { continue };
1367 let input = match record.normalized_input() {
1368 Ok(input) => input,
1369 Err(_) => {
1370 unreadable_inputs += 1;
1371 continue;
1372 }
1373 };
1374 let NormalizedPayload::ResolveEffect(resolve) = &input.input else {
1375 continue;
1376 };
1377 let EffectOutcome::Succeeded(success) = &resolve.outcome else {
1378 continue;
1379 };
1380 let EffectSuccess::TasksSpawned(spawned) = &success.result else {
1381 continue;
1382 };
1383 for attempt in &spawned.attempts {
1384 launches.push(LaunchFact {
1385 task_id: attempt.task_id.as_str().to_string(),
1386 attempt_id: attempt.attempt_id.as_str().to_string(),
1387 step_seq: record.step_seq().get(),
1388 effect_id: resolve.effect_id.as_str().to_string(),
1389 });
1390 }
1391 }
1392
1393 let mut violations: Vec<String> = Vec::new();
1394 let mut seen: HashMap<(&str, &str), u64> = HashMap::new();
1395 for fact in &launches {
1396 let pair = (fact.task_id.as_str(), fact.attempt_id.as_str());
1397 if let Some(first_step) = seen.insert(pair, fact.step_seq) {
1398 violations.push(format!(
1399 "task {} attempt {} launched at steps {first_step} and {}; the launch token is \
1400 derived from that pair, so a repeated pair is a reused LaunchToken",
1401 fact.task_id, fact.attempt_id, fact.step_seq,
1402 ));
1403 }
1404 match parse_effect_step(&fact.effect_id) {
1405 Some((effect_operation, effect_step)) => {
1406 if effect_operation != operation_id {
1407 violations.push(format!(
1408 "task {} launch at step {} resolves effect {} of another operation — \
1409 causation cannot cross operations",
1410 fact.task_id, fact.step_seq, fact.effect_id,
1411 ));
1412 } else if effect_step >= fact.step_seq {
1413 violations.push(format!(
1414 "task {} launch resolved at step {} names an effect published at step \
1415 {effect_step} — the resolution precedes the publication",
1416 fact.task_id, fact.step_seq,
1417 ));
1418 }
1419 }
1420 None => violations.push(format!(
1421 "task {} launch at step {} names effect {}, which is not in the \
1422 `operation:step:N:effect:M` vocabulary",
1423 fact.task_id, fact.step_seq, fact.effect_id,
1424 )),
1425 }
1426 }
1427
1428 if !violations.is_empty() {
1429 return RuleReport {
1430 rule,
1431 verdict: Verdict::Fail,
1432 detail: violations.join("; "),
1433 };
1434 }
1435 let degraded_hops = hops
1436 .iter()
1437 .filter(|hop| matches!(hop, Hop::Degraded(_)))
1438 .count();
1439 if degraded_hops > 0 || unreadable_inputs > 0 {
1440 return RuleReport {
1441 rule,
1442 verdict: Verdict::Degraded,
1443 detail: format!(
1444 "{} launch(es) checked; {degraded_hops} degraded hop(s) and \
1445 {unreadable_inputs} unreadable input(s) could hide further launches",
1446 launches.len(),
1447 ),
1448 };
1449 }
1450 RuleReport {
1451 rule,
1452 verdict: Verdict::Pass,
1453 detail: format!(
1454 "{} launch(es), every (task_id, attempt_id) pair unique, every spawn resolution \
1455 names an earlier step of this operation",
1456 launches.len(),
1457 ),
1458 }
1459}
1460
1461fn parse_effect_step(effect_id: &str) -> Option<(&str, u64)> {
1464 let (before_effect, _) = effect_id.rsplit_once(":effect:")?;
1465 let (operation, step) = before_effect.rsplit_once(":step:")?;
1466 Some((operation, step.parse().ok()?))
1467}
1468
1469fn digests_of(hop: &Hop) -> (Option<&str>, Option<&str>) {
1470 match hop {
1471 Hop::Complete(record) => (
1472 record
1473 .previous_record_digest()
1474 .map(|digest| digest.as_str()),
1475 Some(record.record_digest().as_str()),
1476 ),
1477 Hop::Degraded(degraded) => (
1478 degraded.previous_record_digest.as_deref(),
1479 degraded.record_digest.as_deref(),
1480 ),
1481 }
1482}
1483
1484struct C5Clauses {
1491 violations: Vec<String>,
1492 degradations: Vec<String>,
1493 confirmations: Vec<String>,
1494}
1495
1496impl C5Clauses {
1497 fn new() -> Self {
1498 Self {
1499 violations: Vec::new(),
1500 degradations: Vec::new(),
1501 confirmations: Vec::new(),
1502 }
1503 }
1504
1505 fn violation(&mut self, detail: String) {
1506 self.violations.push(detail);
1507 }
1508
1509 fn degraded(&mut self, detail: String) {
1510 self.degradations.push(detail);
1511 }
1512
1513 fn confirms(&mut self, detail: String) {
1514 self.confirmations.push(detail);
1515 }
1516
1517 fn report(self, rule: &str) -> RuleReport {
1518 let rule = rule.to_string();
1519 if !self.violations.is_empty() {
1520 return RuleReport {
1521 rule,
1522 verdict: Verdict::Fail,
1523 detail: self.violations.join("; "),
1524 };
1525 }
1526 if !self.degradations.is_empty() {
1527 let mut detail = self.degradations.join("; ");
1528 if !self.confirmations.is_empty() {
1529 detail.push_str("; ");
1530 detail.push_str(&self.confirmations.join("; "));
1531 }
1532 return RuleReport {
1533 rule,
1534 verdict: Verdict::Degraded,
1535 detail,
1536 };
1537 }
1538 RuleReport {
1539 rule,
1540 verdict: Verdict::Pass,
1541 detail: if self.confirmations.is_empty() {
1542 "nothing to anchor".to_string()
1543 } else {
1544 self.confirmations.join("; ")
1545 },
1546 }
1547 }
1548}
1549
1550fn check_c5<C: AsRef<[u8]>>(
1553 segments: &HashMap<String, Vec<KernelRecord>>,
1554 blobs: &[C],
1555 strict: bool,
1556) -> (Vec<RuleReport>, usize, usize) {
1557 let mut checks = Vec::new();
1558 let mut unparseable = 0usize;
1559 let mut parsed = 0usize;
1560 for blob in blobs {
1561 match KernelCheckpoint::from_checkpoint_bytes(blob.as_ref()) {
1562 Ok(checkpoint) => {
1563 parsed += 1;
1564 let records = segments.get(checkpoint.operation_id().as_str());
1565 let replay = strict.then(|| strict_replay(&checkpoint, records.map(Vec::as_slice)));
1567 checks.push(check_c5a(
1568 &checkpoint,
1569 records.map(Vec::as_slice),
1570 replay.as_ref(),
1571 ));
1572 checks.push(check_c5b(&checkpoint, replay.as_ref()));
1573 }
1574 Err(error) => {
1575 unparseable += 1;
1576 checks.push(RuleReport {
1577 rule: "C5a".to_string(),
1578 verdict: Verdict::Degraded,
1579 detail: format!(
1580 "checkpoint blob did not decode — its claims are unverifiable (C7): {}",
1581 error.message()
1582 ),
1583 });
1584 }
1585 }
1586 }
1587 (checks, unparseable, parsed)
1588}
1589
1590enum StrictReplay {
1597 Skipped(String),
1599 Faulted(String),
1601 Done {
1602 at_base: KernelCheckpoint,
1608 ladder: Result<(), String>,
1609 above_records: usize,
1610 },
1611}
1612
1613fn strict_replay(checkpoint: &KernelCheckpoint, records: Option<&[KernelRecord]>) -> StrictReplay {
1614 let through = checkpoint.through_step_seq().get();
1615 let base = checkpoint.base_step_seq().get();
1616 let Some(records) = records else {
1617 return StrictReplay::Skipped(
1618 "the journal holds no segment for this operation".to_string(),
1619 );
1620 };
1621
1622 let mut steps: Vec<u64> = records
1626 .iter()
1627 .map(|record| record.step_seq().get())
1628 .filter(|step| *step <= through)
1629 .collect();
1630 steps.sort_unstable();
1631 steps.dedup();
1632 if steps != (0..=through).collect::<Vec<u64>>() {
1633 return StrictReplay::Skipped(format!(
1634 "the journal does not hold an unbroken record run from step 0 through {through} \
1635 (pruned prefix or partial copy); the re-plan replay cannot start at genesis"
1636 ));
1637 }
1638
1639 let fold_to_base = || -> Result<KernelCheckpoint, String> {
1643 let mut prefix: Vec<&KernelRecord> = records
1644 .iter()
1645 .filter(|record| record.step_seq().get() <= base)
1646 .collect();
1647 prefix.sort_by_key(|record| record.step_seq().get());
1648 let prefix: Vec<KernelRecord> = prefix.into_iter().cloned().collect();
1649 let folded = restore_operation(
1650 None,
1651 &prefix,
1652 ConfigDefaults::default(),
1653 InMemoryRecordIndex::from_records(&prefix),
1654 )
1655 .map_err(|fault| format!("{}: {}", fault.code.as_str(), fault.message))?;
1656 folded
1657 .transaction
1658 .checkpoint_candidate(folded.driver.project_logical_state())
1659 .map_err(|fault| format!("{}: {}", fault.code.as_str(), fault.message))?
1660 .decode()
1661 .map_err(|error| {
1662 format!(
1663 "the re-derived checkpoint does not decode: {}",
1664 error.message()
1665 )
1666 })
1667 };
1668 let at_base = match fold_to_base() {
1669 Ok(checkpoint) => checkpoint,
1670 Err(fault) => return StrictReplay::Faulted(fault),
1671 };
1672
1673 let mut above: Vec<&KernelRecord> = records
1677 .iter()
1678 .filter(|record| record.step_seq().get() > through)
1679 .collect();
1680 above.sort_by_key(|record| record.step_seq().get());
1681 let above_records = above.len();
1682 let above: Vec<KernelRecord> = above.into_iter().cloned().collect();
1683 let ladder = restore_operation(
1684 Some(checkpoint),
1685 &above,
1686 ConfigDefaults::default(),
1687 InMemoryRecordIndex::from_records(&above),
1688 )
1689 .map(|_: RestoredOperation<InMemoryRecordIndex>| ())
1690 .map_err(|fault| format!("{}: {}", fault.code.as_str(), fault.message));
1691
1692 StrictReplay::Done {
1693 at_base,
1694 ladder,
1695 above_records,
1696 }
1697}
1698
1699fn check_c5a(
1703 checkpoint: &KernelCheckpoint,
1704 records: Option<&[KernelRecord]>,
1705 replay: Option<&StrictReplay>,
1706) -> RuleReport {
1707 let rule = "C5a";
1708 let mut clauses = C5Clauses::new();
1709 let Some(records) = records else {
1710 return RuleReport {
1711 rule: rule.to_string(),
1712 verdict: Verdict::Degraded,
1713 detail: format!(
1714 "checkpoint names operation {}; the journal holds no segment for it, so no \
1715 anchor can be checked",
1716 checkpoint.operation_id()
1717 ),
1718 };
1719 };
1720 let through = checkpoint.through_step_seq().get();
1721
1722 match record_at(records, 0) {
1724 Some(genesis)
1725 if genesis.record_digest().as_str() != checkpoint.genesis_digest().as_str() =>
1726 {
1727 clauses.violation(format!(
1728 "the journal's genesis record hashes to {}, but the checkpoint binds genesis {} — \
1729 this checkpoint was captured on another chain",
1730 genesis.record_digest(),
1731 checkpoint.genesis_digest()
1732 ));
1733 }
1734 Some(_) => clauses.confirms("genesis digest anchored".to_string()),
1735 None => clauses.degraded(
1736 "the genesis record is not in the journal (pruned prefix); the identity anchor is \
1737 unverifiable"
1738 .to_string(),
1739 ),
1740 }
1741
1742 match record_at(records, through) {
1745 Some(record)
1746 if record.record_digest().as_str()
1747 != checkpoint.covered_transaction_head_digest().as_str() =>
1748 {
1749 clauses.violation(format!(
1750 "the journal record at the covered step {through} hashes to {}, but the \
1751 checkpoint's covered head is {}",
1752 record.record_digest(),
1753 checkpoint.covered_transaction_head_digest()
1754 ));
1755 }
1756 Some(_) => clauses.confirms(format!("covered head anchored at step {through}")),
1757 None => clauses.degraded(format!(
1758 "no journal record at the covered step {through}; the covered head is unverifiable"
1759 )),
1760 }
1761
1762 let base = checkpoint.base_step_seq().get();
1765 if base != through {
1766 match record_at(records, base) {
1767 Some(record)
1768 if record.record_digest().as_str() != checkpoint.base_record_digest().as_str() =>
1769 {
1770 clauses.violation(format!(
1771 "the journal record at the tail base step {base} hashes to {}, but the \
1772 checkpoint anchors its tail on {}",
1773 record.record_digest(),
1774 checkpoint.base_record_digest()
1775 ));
1776 }
1777 _ => {}
1778 }
1779 }
1780 let mut reconciled = 0usize;
1781 let mut pruned = 0usize;
1782 for entry in checkpoint.tail_inputs() {
1783 match record_at(records, entry.step_seq.get()) {
1784 Some(record) if record.record_digest().as_str() != entry.record_digest.as_str() => {
1785 clauses.violation(format!(
1786 "the journal record at step {} disagrees with the checkpoint's bounded tail \
1787 (journal {}, checkpoint {})",
1788 entry.step_seq.get(),
1789 record.record_digest(),
1790 entry.record_digest
1791 ));
1792 }
1793 Some(_) => reconciled += 1,
1794 None => pruned += 1,
1795 }
1796 }
1797 if reconciled > 0 {
1798 clauses.confirms(format!(
1799 "{reconciled} bounded-tail entries reconcile with the journal"
1800 ));
1801 }
1802 if pruned > 0 {
1803 clauses.degraded(format!(
1804 "{pruned} bounded-tail entries have no journal record (pruned interval)"
1805 ));
1806 }
1807
1808 match replay {
1809 Some(StrictReplay::Skipped(reason)) => {
1810 clauses.degraded(format!("strict replay skipped: {reason}"));
1811 }
1812 Some(StrictReplay::Faulted(fault)) => {
1813 clauses.violation(format!("strict replay faulted: {fault}"));
1814 }
1815 Some(StrictReplay::Done {
1816 at_base,
1817 ladder,
1818 above_records,
1819 }) => {
1820 let base = checkpoint.base_step_seq().get();
1821 if at_base.state_digest() != checkpoint.state_digest() {
1822 clauses.violation(format!(
1823 "strict replay folds the journal to state digest {} at the checkpoint's \
1824 base step {base}, but the checkpoint captured {} there",
1825 at_base.state_digest(),
1826 checkpoint.state_digest()
1827 ));
1828 } else {
1829 clauses.confirms(format!(
1830 "strict replay reproduces the captured state digest at step {base}"
1831 ));
1832 }
1833 match ladder {
1834 Ok(()) => clauses.confirms(format!(
1835 "the checkpoint+tail restore ladder holds against the {above_records} \
1836 journal record(s) above the covered step"
1837 )),
1838 Err(fault) => clauses.violation(format!(
1839 "the checkpoint+tail restore ladder faults against this journal: {fault}"
1840 )),
1841 }
1842 }
1843 None => {}
1844 }
1845
1846 clauses.report(rule)
1847}
1848
1849fn check_c5b(checkpoint: &KernelCheckpoint, replay: Option<&StrictReplay>) -> RuleReport {
1855 let rule = "C5b";
1856 let mut clauses = C5Clauses::new();
1857 let transition = &checkpoint.logical_state().transition;
1858 let through = checkpoint.through_step_seq().get();
1859
1860 let mut mints: HashMap<&str, u64> = HashMap::new();
1861 for entry in &transition.launch_tokens {
1862 let token = entry.launch_token.as_str();
1863 let step = entry.step_seq.get();
1864 match mints.get(token) {
1865 Some(previous) if *previous != step => clauses.violation(format!(
1866 "launch token {token} is minted at step {step} and step {previous} — reuse \
1867 across TaskLaunch payloads"
1868 )),
1869 Some(_) => clauses.violation(format!(
1870 "launch token {token} is registered twice at step {step} — a duplicated ledger \
1871 entry"
1872 )),
1873 None => {
1874 mints.insert(token, step);
1875 }
1876 }
1877 if step > through {
1878 clauses.violation(format!(
1879 "launch token {token} is minted at step {step}, beyond the covered boundary \
1880 {through}"
1881 ));
1882 }
1883 }
1884
1885 for effect in &transition.pending_effects {
1886 let EffectKind::SpawnTasks(spawn) = &effect.effect else {
1887 continue;
1888 };
1889 let SpawnTasksEffect { tasks, .. } = spawn;
1890 let effect_step = parse_effect_step(effect.effect_id.as_str()).map(|(_, step)| step);
1891 for launch in tasks {
1892 let token = launch.launch_token.as_str();
1893 match (mints.get(token), effect_step) {
1894 (None, _) => clauses.violation(format!(
1895 "pending effect {} carries launch token {token} the ledger never registered",
1896 effect.effect_id
1897 )),
1898 (Some(&minted), Some(step)) if minted != step => clauses.violation(format!(
1899 "pending effect {} carries launch token {token} minted at step {minted}, not \
1900 at the effect's own step {step}",
1901 effect.effect_id
1902 )),
1903 _ => {}
1904 }
1905 }
1906 }
1907
1908 if !transition.launch_tokens.is_empty() {
1909 clauses.confirms(format!(
1910 "{} launch token(s) anchored; no reuse across TaskLaunch payloads",
1911 transition.launch_tokens.len()
1912 ));
1913 }
1914
1915 match replay {
1916 Some(StrictReplay::Skipped(reason)) => {
1917 clauses.degraded(format!("strict replay skipped: {reason}"));
1918 }
1919 Some(StrictReplay::Faulted(fault)) => {
1920 clauses.violation(format!("strict replay faulted: {fault}"));
1921 }
1922 Some(StrictReplay::Done { at_base, .. }) => {
1923 let replayed_ledger = &at_base.logical_state().transition.launch_tokens;
1924 if replayed_ledger != &transition.launch_tokens {
1925 clauses.violation(format!(
1926 "strict replay mints a different launch-token ledger: the journal fold \
1927 registers {} token(s), the checkpoint carries {}",
1928 replayed_ledger.len(),
1929 transition.launch_tokens.len()
1930 ));
1931 } else {
1932 clauses.confirms(
1933 "strict replay reproduces the launch-token ledger exactly".to_string(),
1934 );
1935 }
1936 }
1937 None => {}
1938 }
1939
1940 clauses.report(rule)
1941}
1942
1943fn record_at(records: &[KernelRecord], step_seq: u64) -> Option<&KernelRecord> {
1945 records
1946 .iter()
1947 .find(|record| record.step_seq().get() == step_seq)
1948}
1949
1950#[cfg(test)]
1955mod tests {
1956 use serde_json::json;
1957
1958 use super::*;
1959 use crate::runtime::kernel::wire::config::{
1960 ConfigDefaults, ExecutionPolicy, HostEffectSupport, OperationConfig,
1961 };
1962 use crate::runtime::kernel::wire::driver::CanonicalOperationDriver;
1963 use crate::runtime::kernel::wire::effect::{
1964 EffectKindTag,
1965 EffectSucceeded,
1966 ProviderCompleted,
1967 ProviderContextOverflow,
1968 ProviderMessage,
1969 ProviderOutcome,
1970 ProviderSuccess,
1971 TaskLaunchOutcome,
1972 TaskLaunchStarted,
1973 TaskLaunchStatus,
1974 TasksSpawnedSuccess,
1975 ToolCall as WireToolCall,
1978 };
1979 use crate::runtime::kernel::wire::envelope::{
1980 ConfigureOperation, KernelInput, ResolveEffect, StartOperation, WireEnvelope,
1981 };
1982 use crate::runtime::kernel::wire::record::{KernelRecord, NormalizedInput};
1983 use crate::runtime::kernel::wire::root::{
1984 InitialContext, LogicalAgentSpec, LogicalMessage, LogicalTask, MessageRole, RootAgentEntry,
1985 RootEntry, RootWorkflowEntry, WorkflowNode as WireWorkflowNode,
1986 WorkflowSpec as WireWorkflowSpec,
1987 };
1988 use crate::runtime::kernel::wire::scalar::{
1989 AttemptId, BoundedJson, CallId, EffectId, InputId, NodeId, OperationId, TaskId, WireU64,
1990 };
1991 use crate::runtime::kernel::wire::transaction::{InMemoryRecordIndex, KernelTransaction};
1992
1993 fn operation(id: &str) -> OperationId {
1998 OperationId::new(id).unwrap()
1999 }
2000
2001 fn envelope(op: &OperationId, id: &str, at: u64, input: KernelInput) -> WireEnvelope {
2002 WireEnvelope::new(
2003 op.clone(),
2004 InputId::new(id).unwrap(),
2005 WireU64::new(at),
2006 input,
2007 )
2008 }
2009
2010 fn configure_envelope(op: &OperationId) -> WireEnvelope {
2011 envelope(
2012 op,
2013 "in-configure",
2014 1_700_000_000_000,
2015 KernelInput::ConfigureOperation(ConfigureOperation {
2016 config: OperationConfig {
2017 execution_policy: Some(ExecutionPolicy {
2018 max_turns: Some(12),
2019 ..ExecutionPolicy::default()
2020 }),
2021 host_effect_support: HostEffectSupport::new([
2022 EffectKindTag::CallProvider,
2023 EffectKindTag::SpawnTasks,
2024 ]),
2025 ..OperationConfig::default()
2026 },
2027 }),
2028 )
2029 }
2030
2031 fn agent_start_envelope(op: &OperationId) -> WireEnvelope {
2032 envelope(
2033 op,
2034 "in-start",
2035 1_700_000_001_000,
2036 KernelInput::StartOperation(StartOperation {
2037 entry: RootEntry::Agent(RootAgentEntry {
2038 task: LogicalTask::new("write the brief"),
2039 run_spec: Some(LogicalAgentSpec::new("write the brief")),
2040 }),
2041 initial_context: InitialContext::default(),
2042 }),
2043 )
2044 }
2045
2046 fn agent_start_with_history_envelope(op: &OperationId, messages: usize) -> WireEnvelope {
2050 envelope(
2051 op,
2052 "in-start",
2053 1_700_000_001_000,
2054 KernelInput::StartOperation(StartOperation {
2055 entry: RootEntry::Agent(RootAgentEntry {
2056 task: LogicalTask::new("write the brief"),
2057 run_spec: Some(LogicalAgentSpec::new("write the brief")),
2058 }),
2059 initial_context: InitialContext {
2060 messages: (0..messages)
2061 .map(|index| LogicalMessage {
2062 role: if index % 2 == 0 {
2063 MessageRole::User
2064 } else {
2065 MessageRole::Assistant
2066 },
2067 content: format!(
2068 "turn {index}: a long enough body that compaction has \
2069 something to reclaim when the prompt stops fitting"
2070 ),
2071 tokens: Some(64),
2072 tool_call_id: None,
2073 })
2074 .collect(),
2075 ..InitialContext::default()
2076 },
2077 }),
2078 )
2079 }
2080
2081 fn workflow_start_envelope(op: &OperationId) -> WireEnvelope {
2082 envelope(
2083 op,
2084 "in-start",
2085 1_700_000_001_000,
2086 KernelInput::StartOperation(StartOperation {
2087 entry: RootEntry::Workflow(RootWorkflowEntry {
2088 spec: WireWorkflowSpec {
2089 name: "brief".to_string(),
2090 nodes: vec![
2091 WireWorkflowNode {
2092 node_id: NodeId::new("collect").unwrap(),
2093 task: LogicalTask::new("collect the sources"),
2094 depends_on: vec![],
2095 run_spec: Some(LogicalAgentSpec::new("collect the sources")),
2096 },
2097 WireWorkflowNode {
2098 node_id: NodeId::new("write").unwrap(),
2099 task: LogicalTask::new("write the brief"),
2100 depends_on: vec![NodeId::new("collect").unwrap()],
2101 run_spec: Some(LogicalAgentSpec::new("write the brief")),
2102 },
2103 ],
2104 },
2105 }),
2106 initial_context: InitialContext::default(),
2107 }),
2108 )
2109 }
2110
2111 fn resolve_overflow_envelope(op: &OperationId, effect_step: u64) -> WireEnvelope {
2112 envelope(
2113 op,
2114 "in-resolve",
2115 1_700_000_002_000,
2116 KernelInput::ResolveEffect(ResolveEffect {
2117 effect_id: EffectId::new(format!("{op}:step:{effect_step}:effect:0")).unwrap(),
2118 outcome: EffectOutcome::Succeeded(EffectSucceeded {
2119 result: EffectSuccess::Provider(ProviderSuccess {
2120 outcome: ProviderOutcome::ContextOverflow(
2121 ProviderContextOverflow::default(),
2122 ),
2123 }),
2124 }),
2125 }),
2126 )
2127 }
2128
2129 fn resolve_completed_envelope(
2132 op: &OperationId,
2133 id: &str,
2134 at: u64,
2135 effect_step: u64,
2136 with_tool_call: bool,
2137 ) -> WireEnvelope {
2138 envelope(
2139 op,
2140 id,
2141 at,
2142 KernelInput::ResolveEffect(ResolveEffect {
2143 effect_id: EffectId::new(format!("{op}:step:{effect_step}:effect:0")).unwrap(),
2144 outcome: EffectOutcome::Succeeded(EffectSucceeded {
2145 result: EffectSuccess::Provider(ProviderSuccess {
2146 outcome: ProviderOutcome::Completed(ProviderCompleted {
2147 message: ProviderMessage {
2148 role: MessageRole::Assistant,
2149 content: "done".to_string(),
2150 tool_calls: if with_tool_call {
2151 vec![WireToolCall {
2152 call_id: CallId::new("call-1").unwrap(),
2153 name: "read_file".to_string(),
2154 arguments: BoundedJson::new(json!({})).unwrap(),
2155 }]
2156 } else {
2157 Vec::new()
2158 },
2159 tool_call_id: None,
2160 tokens: None,
2161 },
2162 observed_input_tokens: None,
2163 observed_output_tokens: None,
2164 stop_reason: None,
2165 }),
2166 }),
2167 }),
2168 }),
2169 )
2170 }
2171
2172 fn resolve_spawn_envelope(
2173 op: &OperationId,
2174 id: &str,
2175 at: u64,
2176 effect_id: &str,
2177 tasks: &[(&str, &str)],
2178 ) -> WireEnvelope {
2179 envelope(
2180 op,
2181 id,
2182 at,
2183 KernelInput::ResolveEffect(ResolveEffect {
2184 effect_id: EffectId::new(effect_id).unwrap(),
2185 outcome: EffectOutcome::Succeeded(EffectSucceeded {
2186 result: EffectSuccess::TasksSpawned(TasksSpawnedSuccess {
2187 attempts: tasks
2188 .iter()
2189 .map(|(task, attempt)| TaskLaunchOutcome {
2190 task_id: TaskId::new(*task).unwrap(),
2191 attempt_id: AttemptId::new(*attempt).unwrap(),
2192 outcome: TaskLaunchStatus::Started(TaskLaunchStarted {}),
2193 })
2194 .collect(),
2195 }),
2196 }),
2197 }),
2198 )
2199 }
2200
2201 fn live_chain(envelopes: &[WireEnvelope]) -> Vec<KernelRecord> {
2208 let mut tx = KernelTransaction::new(ConfigDefaults::default(), InMemoryRecordIndex::new());
2209 let mut driver = CanonicalOperationDriver::new();
2210 let mut journal = Vec::new();
2211 for envelope in envelopes {
2212 let preparation = tx.prepare(envelope, |context| driver.plan(context));
2213 let token = preparation
2214 .token()
2215 .unwrap_or_else(|| {
2216 panic!("expected a prepared step, got {:?}", preparation.fault())
2217 })
2218 .clone();
2219 let head = preparation.record().unwrap().record_digest().clone();
2220 let committed = tx.commit(&token, &head).expect("commit must succeed");
2221 journal.push(committed.record.clone());
2222 driver
2223 .note_committed(committed.step_seq)
2224 .expect("the driver folds the step it planned");
2225 }
2226 journal
2227 }
2228
2229 fn hand_chain(envelopes: &[WireEnvelope]) -> Vec<KernelRecord> {
2233 let mut records: Vec<KernelRecord> = Vec::new();
2234 for (index, envelope) in envelopes.iter().enumerate() {
2235 let input = NormalizedInput::normalize(envelope, &ConfigDefaults::default())
2236 .expect("the envelope normalises");
2237 let step = json!({ "planned": format!("step-{index}"), "effects": [] });
2238 let record =
2239 KernelRecord::chain(records.last(), &input, &step).expect("the record chains");
2240 records.push(record);
2241 }
2242 records
2243 }
2244
2245 fn blobs(records: &[KernelRecord]) -> Vec<Vec<u8>> {
2246 records
2247 .iter()
2248 .map(|record| record.record_bytes().into_vec())
2249 .collect()
2250 }
2251
2252 fn rule<'a>(report: &'a ValidationReport, segment: usize, id: &str) -> &'a RuleReport {
2253 report.segments[segment]
2254 .rules
2255 .iter()
2256 .find(|rule| rule.rule == id)
2257 .unwrap_or_else(|| panic!("segment {segment} has no {id} verdict"))
2258 }
2259
2260 fn cross<'a>(report: &'a ValidationReport, id: &str) -> &'a RuleReport {
2261 report
2262 .cross_checks
2263 .iter()
2264 .find(|rule| rule.rule == id)
2265 .unwrap_or_else(|| panic!("the report has no {id} cross-check"))
2266 }
2267
2268 #[test]
2273 fn a_green_agent_chain_passes_every_rule() {
2274 let op = operation("op-green-agent");
2275 let chain = live_chain(&[
2276 configure_envelope(&op),
2277 agent_start_envelope(&op),
2278 resolve_overflow_envelope(&op, 1),
2279 ]);
2280 let report = validate_journal(&blobs(&chain));
2281 assert_eq!(report.segments.len(), 1);
2282 for id in ["C1", "C2", "C3", "C4"] {
2283 assert_eq!(
2284 rule(&report, 0, id).verdict,
2285 Verdict::Pass,
2286 "{id}: {}",
2287 rule(&report, 0, id).detail
2288 );
2289 }
2290 assert!(
2291 rule(&report, 0, "C3")
2292 .detail
2293 .contains("every durable record digest reproduced"),
2294 "C3 proves the re-plan: {}",
2295 rule(&report, 0, "C3").detail
2296 );
2297 assert_eq!(report.exit_code(), 0);
2298 assert_eq!(report.unparseable_records, 0);
2299 }
2300
2301 #[test]
2302 fn a_green_workflow_chain_passes_c4_with_real_launches() {
2303 let op = operation("op-green-workflow");
2304 let chain = live_chain(&[
2305 configure_envelope(&op),
2306 workflow_start_envelope(&op),
2307 resolve_spawn_envelope(
2308 &op,
2309 "in-ack-1",
2310 1_700_000_002_000,
2311 "op-green-workflow:step:1:effect:0",
2312 &[("wf-node0", "wf-node0:attempt:1")],
2313 ),
2314 ]);
2315 let report = validate_journal(&blobs(&chain));
2316 assert_eq!(report.segments.len(), 1);
2317 for id in ["C1", "C2", "C3", "C4"] {
2318 assert_eq!(
2319 rule(&report, 0, id).verdict,
2320 Verdict::Pass,
2321 "{id}: {}",
2322 rule(&report, 0, id).detail
2323 );
2324 }
2325 assert!(
2326 rule(&report, 0, "C4").detail.contains("1 launch(es)"),
2327 "{}",
2328 rule(&report, 0, "C4").detail
2329 );
2330 assert_eq!(report.deferred.len(), 2, "batch-1 scope limits are named");
2331 assert_eq!(report.exit_code(), 0);
2332 }
2333
2334 #[test]
2335 fn input_order_is_a_storage_detail() {
2336 let op = operation("op-shuffled");
2337 let chain = live_chain(&[
2338 configure_envelope(&op),
2339 agent_start_envelope(&op),
2340 resolve_overflow_envelope(&op, 1),
2341 ]);
2342 let mut shuffled = blobs(&chain);
2343 shuffled.reverse();
2344 let report = validate_journal(&shuffled);
2345 assert_eq!(
2346 report.exit_code(),
2347 0,
2348 "the chain's own links define the order"
2349 );
2350 }
2351
2352 #[test]
2353 fn two_operations_validate_as_independent_segments() {
2354 let op_a = operation("op-seg-a");
2355 let op_b = operation("op-seg-b");
2356 let chain_a = live_chain(&[configure_envelope(&op_a), agent_start_envelope(&op_a)]);
2357 let chain_b = live_chain(&[configure_envelope(&op_b), agent_start_envelope(&op_b)]);
2358 let mut mixed = Vec::new();
2360 for index in 0..2 {
2361 mixed.push(chain_a[index].record_bytes().into_vec());
2362 mixed.push(chain_b[index].record_bytes().into_vec());
2363 }
2364 let report = validate_journal(&mixed);
2365 assert_eq!(report.segments.len(), 2);
2366 assert_eq!(report.exit_code(), 0);
2367 }
2368
2369 #[test]
2374 fn a_gap_in_the_chain_fails_c1_and_degrades_c3() {
2375 let op = operation("op-gapped");
2376 let chain = live_chain(&[
2377 configure_envelope(&op),
2378 agent_start_envelope(&op),
2379 resolve_overflow_envelope(&op, 1),
2380 ]);
2381 let gapped = blobs(&[chain[0].clone(), chain[2].clone()]);
2382 let report = validate_journal(&gapped);
2383 assert_eq!(rule(&report, 0, "C1").verdict, Verdict::Fail);
2384 assert_eq!(
2385 rule(&report, 0, "C3").verdict,
2386 Verdict::Degraded,
2387 "a re-plan over a broken chain would only re-report the C1 break"
2388 );
2389 assert_eq!(report.exit_code(), 1);
2390 }
2391
2392 #[test]
2397 fn two_different_records_for_one_input_fail_c2() {
2398 let op = operation("op-dup-input");
2399 let chain_a = hand_chain(&[configure_envelope(&op), agent_start_envelope(&op)]);
2402 let mut later_start = agent_start_envelope(&op);
2403 later_start.observed_at_ms = WireU64::new(1_700_000_001_500);
2404 let chain_b = hand_chain(&[configure_envelope(&op), later_start]);
2405 assert_ne!(
2406 chain_a[1].record_digest(),
2407 chain_b[1].record_digest(),
2408 "the fixture must produce two different records for one input id"
2409 );
2410 let report = validate_journal(&blobs(&[
2411 chain_a[0].clone(),
2412 chain_a[1].clone(),
2413 chain_b[1].clone(),
2414 ]));
2415 assert_eq!(rule(&report, 0, "C2").verdict, Verdict::Fail);
2416 assert!(
2417 rule(&report, 0, "C2").detail.contains("in-start"),
2418 "{}",
2419 rule(&report, 0, "C2").detail
2420 );
2421 assert_eq!(report.exit_code(), 1);
2422 }
2423
2424 #[test]
2429 fn a_repeated_attempt_pair_is_a_reused_launch_token() {
2430 let op = operation("op-dup-launch");
2431 let chain = hand_chain(&[
2432 configure_envelope(&op),
2433 resolve_spawn_envelope(
2434 &op,
2435 "in-ack-1",
2436 1_700_000_001_000,
2437 "op-dup-launch:step:0:effect:0",
2438 &[("writer", "writer:attempt:1")],
2439 ),
2440 resolve_spawn_envelope(
2441 &op,
2442 "in-ack-2",
2443 1_700_000_002_000,
2444 "op-dup-launch:step:0:effect:0",
2445 &[("writer", "writer:attempt:1")],
2446 ),
2447 ]);
2448 let report = validate_journal(&blobs(&chain));
2449 assert_eq!(rule(&report, 0, "C1").verdict, Verdict::Pass);
2450 assert_eq!(rule(&report, 0, "C4").verdict, Verdict::Fail);
2451 assert!(
2452 rule(&report, 0, "C4").detail.contains("LaunchToken"),
2453 "the verdict names the token reuse: {}",
2454 rule(&report, 0, "C4").detail
2455 );
2456 assert_eq!(report.exit_code(), 1);
2457 }
2458
2459 #[test]
2460 fn a_resolution_naming_a_future_step_fails_c4() {
2461 let op = operation("op-future-effect");
2462 let chain = hand_chain(&[
2463 configure_envelope(&op),
2464 resolve_spawn_envelope(
2465 &op,
2466 "in-ack-1",
2467 1_700_000_001_000,
2468 "op-future-effect:step:5:effect:0",
2469 &[("writer", "writer:attempt:1")],
2470 ),
2471 ]);
2472 let report = validate_journal(&blobs(&chain));
2473 assert_eq!(rule(&report, 0, "C4").verdict, Verdict::Fail);
2474 assert!(
2475 rule(&report, 0, "C4")
2476 .detail
2477 .contains("precedes the publication"),
2478 "{}",
2479 rule(&report, 0, "C4").detail
2480 );
2481 }
2482
2483 #[test]
2484 fn a_resolution_naming_another_operation_fails_c4() {
2485 let op = operation("op-foreign-effect");
2486 let chain = hand_chain(&[
2487 configure_envelope(&op),
2488 resolve_spawn_envelope(
2489 &op,
2490 "in-ack-1",
2491 1_700_000_001_000,
2492 "op-somewhere-else:step:0:effect:0",
2493 &[("writer", "writer:attempt:1")],
2494 ),
2495 ]);
2496 let report = validate_journal(&blobs(&chain));
2497 assert_eq!(rule(&report, 0, "C4").verdict, Verdict::Fail);
2498 assert!(
2499 rule(&report, 0, "C4").detail.contains("another operation"),
2500 "{}",
2501 rule(&report, 0, "C4").detail
2502 );
2503 }
2504
2505 #[test]
2510 fn a_tampered_hop_fails_integrity_validation() {
2511 let op = operation("op-tampered");
2512 let chain = live_chain(&[
2513 configure_envelope(&op),
2514 agent_start_envelope(&op),
2515 resolve_overflow_envelope(&op, 1),
2516 ]);
2517 let mut input = blobs(&chain);
2518 let mut forged: serde_json::Value = serde_json::from_slice(&input[1]).unwrap();
2521 forged["step_digest"] = serde_json::Value::String(chain[0].record_digest().to_string());
2522 input[1] = serde_json::to_vec(&forged).unwrap();
2523
2524 let report = validate_journal(&input);
2525 assert_eq!(report.segments.len(), 1);
2526 assert_eq!(report.segments[0].degraded_hops.len(), 1);
2527 assert_eq!(
2528 rule(&report, 0, "C1").verdict,
2529 Verdict::Fail,
2530 "a digest mismatch must fail C1: {}",
2531 rule(&report, 0, "C1").detail
2532 );
2533 assert_eq!(rule(&report, 0, "C3").verdict, Verdict::Degraded);
2534 assert_eq!(rule(&report, 0, "C4").verdict, Verdict::Degraded);
2535 assert_eq!(
2536 report.exit_code(),
2537 1,
2538 "proven digest corruption must fail the validator"
2539 );
2540 }
2541
2542 #[test]
2543 fn missing_legacy_digest_degrades_without_claiming_corruption() {
2544 let op = operation("op-legacy");
2545 let chain = live_chain(&[configure_envelope(&op)]);
2546 let mut legacy: serde_json::Value = serde_json::from_slice(&blobs(&chain)[0]).unwrap();
2547 legacy.as_object_mut().unwrap().remove("step_digest");
2548 let report = validate_journal(&[serde_json::to_vec(&legacy).unwrap()]);
2549 assert_eq!(rule(&report, 0, "C1").verdict, Verdict::Degraded);
2550 assert_eq!(report.exit_code(), 0);
2551 }
2552
2553 #[test]
2554 fn unparseable_input_is_evidence_insufficient_not_guilty() {
2555 let report = validate_journal(&[b"this is not a record".to_vec()]);
2556 assert!(report.segments.is_empty());
2557 assert_eq!(report.unparseable_records, 1);
2558 assert_eq!(report.exit_code(), 2);
2559 }
2560
2561 #[test]
2562 fn garbage_beside_a_green_chain_stays_exit_2_without_a_violation() {
2563 let op = operation("op-plus-garbage");
2564 let chain = live_chain(&[configure_envelope(&op), agent_start_envelope(&op)]);
2565 let mut input = blobs(&chain);
2566 input.push(b"this is not a record".to_vec());
2567 let report = validate_journal(&input);
2568 assert_eq!(report.segments.len(), 1);
2569 assert_eq!(rule(&report, 0, "C1").verdict, Verdict::Pass);
2570 assert_eq!(report.unparseable_records, 1);
2571 assert_eq!(
2572 report.exit_code(),
2573 2,
2574 "no violation was proven, but the evidence was partially unreadable"
2575 );
2576 }
2577
2578 #[test]
2579 fn an_empty_journal_is_evidence_insufficient() {
2580 let report = validate_journal::<Vec<u8>>(&[]);
2581 assert_eq!(report.exit_code(), 2);
2582 }
2583
2584 fn session_event(value: serde_json::Value) -> Vec<u8> {
2589 serde_json::to_vec(&value).unwrap()
2590 }
2591
2592 #[test]
2593 fn session_events_classify_leniently_across_host_spellings() {
2594 let node_attempt = session_event(json!({
2595 "kind": "provider_attempt",
2596 "effect_id": "op:step:1:effect:0",
2597 "request_fingerprint": "fp-1",
2598 "route": { "routeId": "route-a", "provider": "p" },
2599 "status": "success"
2600 }));
2601 let py_attempt = session_event(json!({
2602 "kind": "provider_attempt",
2603 "effect_id": "op:step:2:effect:0",
2604 "route": { "route_id": "route-b" }
2605 }));
2606 let run_started = session_event(json!({
2607 "kind": "run_started",
2608 "run_id": "run-1",
2609 "route": { "routeId": "route-a" }
2610 }));
2611 let node_measured = session_event(json!({
2612 "kind": "prompt_measured",
2613 "turn": 1,
2614 "effect_id": "op:step:1:effect:0",
2615 "measurement": { "requestFingerprint": "fp-1", "inputTokens": 10 }
2616 }));
2617 let py_measured = session_event(json!({
2618 "kind": "prompt_measured",
2619 "measurement": { "request_fingerprint": "fp-2" }
2620 }));
2621 let llm_completed = session_event(json!({
2622 "kind": "llm_completed",
2623 "effect_id": "op:step:2:effect:0",
2624 "invocation_id": "op:step:1:effect:0"
2625 }));
2626 let unknown_kind = session_event(json!({ "kind": "compressed", "turn": 3 }));
2627 let kindless = session_event(json!({ "turn": 3 }));
2628
2629 assert_eq!(
2630 classify_session_event(&node_attempt),
2631 Some(EvidenceEvent::ProviderAttempt {
2632 effect_id: Some("op:step:1:effect:0".to_string()),
2633 request_fingerprint: Some("fp-1".to_string()),
2634 route_id: Some("route-a".to_string()),
2635 status: Some("success".to_string()),
2636 })
2637 );
2638 assert_eq!(
2639 classify_session_event(&py_attempt),
2640 Some(EvidenceEvent::ProviderAttempt {
2641 effect_id: Some("op:step:2:effect:0".to_string()),
2642 request_fingerprint: None,
2643 route_id: Some("route-b".to_string()),
2644 status: None,
2645 })
2646 );
2647 assert_eq!(
2648 classify_session_event(&run_started),
2649 Some(EvidenceEvent::RunStarted {
2650 route_id: Some("route-a".to_string())
2651 })
2652 );
2653 assert_eq!(
2654 classify_session_event(&node_measured),
2655 Some(EvidenceEvent::PromptMeasured {
2656 effect_id: Some("op:step:1:effect:0".to_string()),
2657 request_fingerprint: Some("fp-1".to_string()),
2658 })
2659 );
2660 assert_eq!(
2661 classify_session_event(&py_measured),
2662 Some(EvidenceEvent::PromptMeasured {
2663 effect_id: None,
2664 request_fingerprint: Some("fp-2".to_string()),
2665 })
2666 );
2667 assert_eq!(
2668 classify_session_event(&llm_completed),
2669 Some(EvidenceEvent::LlmCompleted {
2670 effect_id: Some("op:step:2:effect:0".to_string()),
2671 invocation_id: Some("op:step:1:effect:0".to_string()),
2672 })
2673 );
2674 assert_eq!(
2675 classify_session_event(&unknown_kind),
2676 Some(EvidenceEvent::Other),
2677 "unknown kinds are parseable but ignored — the vocabulary evolves"
2678 );
2679 assert_eq!(
2680 classify_session_event(&kindless),
2681 Some(EvidenceEvent::Other)
2682 );
2683 assert_eq!(
2684 classify_session_event(b"not json"),
2685 None,
2686 "a non-object event blob is unparseable input, never a violation"
2687 );
2688 }
2689
2690 #[test]
2691 fn dual_input_with_a_green_journal_and_real_events_stays_green() {
2692 let op = operation("op-dual-green");
2693 let chain = live_chain(&[configure_envelope(&op), agent_start_envelope(&op)]);
2694 let stream = vec![
2695 session_event(json!({
2696 "kind": "run_started",
2697 "run_id": "r1",
2698 "route": { "routeId": "route-a" }
2699 })),
2700 session_event(json!({
2701 "kind": "prompt_measured",
2702 "turn": 1,
2703 "effect_id": "op-dual-green:step:1:effect:0",
2704 "measurement": { "requestFingerprint": "fp-1", "inputTokens": 10 }
2705 })),
2706 session_event(json!({
2707 "kind": "provider_attempt",
2708 "effect_id": "op-dual-green:step:1:effect:0",
2709 "request_fingerprint": "fp-1",
2710 "route": { "routeId": "route-a" },
2711 "status": "success"
2712 })),
2713 session_event(json!({
2714 "kind": "llm_completed",
2715 "turn": 1,
2716 "effect_id": "op-dual-green:step:1:effect:0",
2717 "invocation_id": "op-dual-green:step:1:effect:0"
2718 })),
2719 ];
2720 let report = validate_with_session_log(&blobs(&chain), &[stream]);
2721 assert_eq!(report.session_events, Some(4));
2722 assert_eq!(report.unparseable_events, 0);
2723 for id in ["C6.1", "C6.2", "C6.3", "C8"] {
2724 assert_eq!(
2725 cross(&report, id).verdict,
2726 Verdict::Pass,
2727 "{id}: {}",
2728 cross(&report, id).detail
2729 );
2730 }
2731 assert!(
2732 !report
2733 .deferred
2734 .iter()
2735 .any(|line| line.starts_with("c6.") || line.starts_with("c8.")),
2736 "C6/C8 are implemented — the interim scope notes are gone"
2737 );
2738 assert_eq!(report.exit_code(), 0);
2739 }
2740
2741 #[test]
2742 fn journal_only_validation_carries_no_session_plane() {
2743 let op = operation("op-journal-only");
2744 let chain = live_chain(&[configure_envelope(&op), agent_start_envelope(&op)]);
2745 let report = validate_journal(&blobs(&chain));
2746 assert_eq!(report.session_events, None);
2747 assert_eq!(report.unparseable_events, 0);
2748 assert_eq!(
2749 report.deferred.len(),
2750 2,
2751 "batch-1 deferred scope is unchanged"
2752 );
2753 assert_eq!(report.exit_code(), 0);
2754 }
2755
2756 #[test]
2757 fn an_empty_session_plane_is_evidence_insufficient() {
2758 let op = operation("op-empty-session");
2759 let chain = live_chain(&[configure_envelope(&op), agent_start_envelope(&op)]);
2760 let report = validate_with_session_log(&blobs(&chain), &[Vec::<Vec<u8>>::new()]);
2761 assert_eq!(report.session_events, Some(0));
2762 assert!(
2763 !report.has_violations(),
2764 "an empty log proves nothing either way"
2765 );
2766 assert_eq!(report.exit_code(), 2);
2767 }
2768
2769 #[test]
2770 fn garbage_session_events_are_evidence_insufficient_not_guilty() {
2771 let op = operation("op-garbage-session");
2772 let chain = live_chain(&[configure_envelope(&op), agent_start_envelope(&op)]);
2773 let stream = vec![
2774 session_event(json!({ "kind": "run_started", "run_id": "r1" })),
2775 b"this is not an event".to_vec(),
2776 ];
2777 let report = validate_with_session_log(&blobs(&chain), &[stream]);
2778 assert_eq!(report.session_events, Some(1));
2779 assert_eq!(report.unparseable_events, 1);
2780 assert!(!report.has_violations());
2781 assert_eq!(report.exit_code(), 2);
2782 }
2783
2784 fn honest_dual_input(op_name: &str) -> (Vec<KernelRecord>, Vec<Vec<u8>>) {
2791 let op = operation(op_name);
2792 let chain = live_chain(&[
2793 configure_envelope(&op),
2794 agent_start_envelope(&op),
2795 resolve_overflow_envelope(&op, 1),
2796 ]);
2797 let effect = format!("{op_name}:step:1:effect:0");
2798 let stream = vec![
2799 session_event(json!({
2800 "kind": "run_started",
2801 "run_id": "r1",
2802 "route": { "routeId": "route-a" }
2803 })),
2804 session_event(json!({
2805 "kind": "prompt_measured",
2806 "turn": 1,
2807 "effect_id": effect,
2808 "measurement": { "requestFingerprint": "fp-1", "inputTokens": 10 }
2809 })),
2810 session_event(json!({
2811 "kind": "provider_attempt",
2812 "effect_id": effect,
2813 "request_fingerprint": "fp-1",
2814 "route": { "routeId": "route-a" },
2815 "status": "success"
2816 })),
2817 ];
2818 (chain, stream)
2819 }
2820
2821 #[test]
2822 fn an_attempt_naming_an_effect_the_replan_never_published_fails_c6() {
2823 let (chain, mut stream) = honest_dual_input("op-forged-effect");
2824 stream[2] = session_event(json!({
2825 "kind": "provider_attempt",
2826 "effect_id": "op-forged-effect:step:1:effect:7",
2827 "request_fingerprint": "fp-1",
2828 "route": { "routeId": "route-a" },
2829 "status": "success"
2830 }));
2831 let report = validate_with_session_log(&blobs(&chain), &[stream]);
2832 assert_eq!(cross(&report, "C6.1").verdict, Verdict::Fail);
2833 assert!(
2834 cross(&report, "C6.1").detail.contains("never published"),
2835 "{}",
2836 cross(&report, "C6.1").detail
2837 );
2838 assert_eq!(
2839 report.exit_code(),
2840 1,
2841 "the forged attempt turns the run red"
2842 );
2843 }
2844
2845 #[test]
2846 fn an_attempt_without_an_effect_id_fails_c6_as_forged_evidence() {
2847 let (chain, mut stream) = honest_dual_input("op-keyless-attempt");
2848 stream[2] = session_event(json!({
2849 "kind": "provider_attempt",
2850 "request_fingerprint": "fp-1",
2851 "route": { "routeId": "route-a" },
2852 "status": "success"
2853 }));
2854 let report = validate_with_session_log(&blobs(&chain), &[stream]);
2855 assert_eq!(cross(&report, "C6.1").verdict, Verdict::Fail);
2856 assert!(
2857 cross(&report, "C6.1").detail.contains("without effect_id"),
2858 "{}",
2859 cross(&report, "C6.1").detail
2860 );
2861 assert_eq!(report.exit_code(), 1);
2862 }
2863
2864 #[test]
2865 fn an_attempt_past_the_journal_tip_degrades_c6_instead_of_failing() {
2866 let (chain, mut stream) = honest_dual_input("op-prefix-attempt");
2867 stream[2] = session_event(json!({
2868 "kind": "provider_attempt",
2869 "effect_id": "op-prefix-attempt:step:9:effect:0",
2870 "request_fingerprint": "fp-1",
2871 "route": { "routeId": "route-a" },
2872 "status": "success"
2873 }));
2874 let report = validate_with_session_log(&blobs(&chain), &[stream]);
2875 assert_eq!(
2876 cross(&report, "C6.1").verdict,
2877 Verdict::Degraded,
2878 "a journal prefix cannot disprove an effect past its tip: {}",
2879 cross(&report, "C6.1").detail
2880 );
2881 assert_eq!(report.exit_code(), 0, "degradation never turns the run red");
2882 }
2883
2884 #[test]
2885 fn an_attempt_on_an_operation_without_a_segment_degrades_c6() {
2886 let (chain, mut stream) = honest_dual_input("op-subset-journal");
2887 stream[2] = session_event(json!({
2888 "kind": "provider_attempt",
2889 "effect_id": "op-elsewhere:step:1:effect:0",
2890 "request_fingerprint": "fp-1",
2891 "route": { "routeId": "route-a" },
2892 "status": "success"
2893 }));
2894 let report = validate_with_session_log(&blobs(&chain), &[stream]);
2895 assert_eq!(cross(&report, "C6.1").verdict, Verdict::Degraded);
2896 assert!(
2897 cross(&report, "C6.1").detail.contains("no journal segment"),
2898 "{}",
2899 cross(&report, "C6.1").detail
2900 );
2901 }
2902
2903 #[test]
2904 fn an_orphan_fingerprint_fails_c6() {
2905 let (chain, mut stream) = honest_dual_input("op-orphan-fp");
2906 stream.remove(1); let report = validate_with_session_log(&blobs(&chain), &[stream]);
2908 assert_eq!(cross(&report, "C6.2").verdict, Verdict::Fail);
2909 assert!(
2910 cross(&report, "C6.2").detail.contains("fp-1"),
2911 "{}",
2912 cross(&report, "C6.2").detail
2913 );
2914 assert_eq!(report.exit_code(), 1);
2915 }
2916
2917 #[test]
2918 fn an_attempt_without_a_fingerprint_fails_c6() {
2919 let (chain, mut stream) = honest_dual_input("op-fpless-attempt");
2920 stream[2] = session_event(json!({
2921 "kind": "provider_attempt",
2922 "effect_id": "op-fpless-attempt:step:1:effect:0",
2923 "route": { "routeId": "route-a" },
2924 "status": "success"
2925 }));
2926 let report = validate_with_session_log(&blobs(&chain), &[stream]);
2927 assert_eq!(cross(&report, "C6.2").verdict, Verdict::Fail);
2928 assert!(
2929 cross(&report, "C6.2")
2930 .detail
2931 .contains("without request_fingerprint"),
2932 "{}",
2933 cross(&report, "C6.2").detail
2934 );
2935 }
2936
2937 #[test]
2938 fn an_in_run_route_change_fails_c6() {
2939 let (chain, mut stream) = honest_dual_input("op-route-flip");
2940 stream[2] = session_event(json!({
2941 "kind": "provider_attempt",
2942 "effect_id": "op-route-flip:step:1:effect:0",
2943 "request_fingerprint": "fp-1",
2944 "route": { "routeId": "route-b" },
2945 "status": "success"
2946 }));
2947 let report = validate_with_session_log(&blobs(&chain), &[stream]);
2948 assert_eq!(cross(&report, "C6.3").verdict, Verdict::Fail);
2949 assert!(
2950 cross(&report, "C6.3")
2951 .detail
2952 .contains("in-run route change"),
2953 "{}",
2954 cross(&report, "C6.3").detail
2955 );
2956 assert_eq!(report.exit_code(), 1);
2957 }
2958
2959 #[test]
2960 fn a_cross_resume_route_change_degrades_c6_per_q3() {
2961 let (chain, mut stream) = honest_dual_input("op-route-resume");
2962 stream.push(session_event(json!({
2965 "kind": "run_started",
2966 "run_id": "r1",
2967 "route": { "routeId": "route-b" }
2968 })));
2969 stream.push(session_event(json!({
2970 "kind": "prompt_measured",
2971 "turn": 2,
2972 "effect_id": "op-route-resume:step:1:effect:0",
2973 "measurement": { "requestFingerprint": "fp-2", "inputTokens": 11 }
2974 })));
2975 stream.push(session_event(json!({
2976 "kind": "provider_attempt",
2977 "effect_id": "op-route-resume:step:1:effect:0",
2978 "request_fingerprint": "fp-2",
2979 "route": { "routeId": "route-b" },
2980 "status": "success"
2981 })));
2982 let report = validate_with_session_log(&blobs(&chain), &[stream]);
2983 assert_eq!(
2984 cross(&report, "C6.3").verdict,
2985 Verdict::Degraded,
2986 "cross-resume route changes mark, they do not fail: {}",
2987 cross(&report, "C6.3").detail
2988 );
2989 assert!(
2990 cross(&report, "C6.3").detail.contains("cross-resume"),
2991 "{}",
2992 cross(&report, "C6.3").detail
2993 );
2994 assert_eq!(report.exit_code(), 0);
2995 }
2996
2997 #[test]
2998 fn a_pre_0_2_63_log_without_attempts_degrades_every_c6_clause() {
2999 let op = operation("op-old-log");
3000 let chain = live_chain(&[configure_envelope(&op), agent_start_envelope(&op)]);
3001 let stream = vec![
3002 session_event(json!({ "kind": "run_started", "run_id": "r1" })),
3003 session_event(json!({ "kind": "llm_completed", "turn": 1, "content": "done" })),
3004 ];
3005 let report = validate_with_session_log(&blobs(&chain), &[stream]);
3006 for id in ["C6.1", "C6.2", "C6.3", "C8"] {
3007 assert_eq!(
3008 cross(&report, id).verdict,
3009 Verdict::Degraded,
3010 "{id}: old logs degrade (C7), never fail — {}",
3011 cross(&report, id).detail
3012 );
3013 }
3014 assert_eq!(report.exit_code(), 0);
3015 }
3016
3017 fn overflow_retry_chain(op_name: &str) -> Vec<KernelRecord> {
3027 let op = operation(op_name);
3028 live_chain(&[
3029 configure_envelope(&op),
3030 agent_start_with_history_envelope(&op, 14),
3031 resolve_overflow_envelope(&op, 1),
3032 resolve_completed_envelope(&op, "in-resolve-2", 1_700_000_003_000, 2, false),
3033 ])
3034 }
3035
3036 #[test]
3037 fn an_honest_overflow_retry_invocation_passes_c8() {
3038 let chain = overflow_retry_chain("op-c8-green");
3039 let stream = vec![
3040 session_event(json!({ "kind": "run_started", "run_id": "r1" })),
3041 session_event(json!({
3042 "kind": "llm_completed",
3043 "turn": 1,
3044 "effect_id": "op-c8-green:step:2:effect:0",
3045 "invocation_id": "op-c8-green:step:1:effect:0"
3046 })),
3047 ];
3048 let report = validate_with_session_log(&blobs(&chain), &[stream]);
3049 assert_eq!(
3050 cross(&report, "C8").verdict,
3051 Verdict::Pass,
3052 "{}",
3053 cross(&report, "C8").detail
3054 );
3055 assert!(
3056 cross(&report, "C8")
3057 .detail
3058 .contains("1 retried invocation(s)"),
3059 "{}",
3060 cross(&report, "C8").detail
3061 );
3062 assert_eq!(report.exit_code(), 0);
3063 }
3064
3065 #[test]
3066 fn a_completed_chain_head_is_the_merge_forgery() {
3067 let op = operation("op-c8-merged");
3071 let chain = live_chain(&[
3072 configure_envelope(&op),
3073 agent_start_envelope(&op),
3074 resolve_completed_envelope(&op, "in-resolve-1", 1_700_000_002_000, 1, true),
3075 ]);
3076 let stream = vec![session_event(json!({
3077 "kind": "llm_completed",
3078 "turn": 1,
3079 "effect_id": "op-c8-merged:step:2:effect:0",
3080 "invocation_id": "op-c8-merged:step:1:effect:0"
3081 }))];
3082 let report = validate_with_session_log(&blobs(&chain), &[stream]);
3083 assert_eq!(cross(&report, "C8").verdict, Verdict::Fail);
3084 assert!(
3085 cross(&report, "C8")
3086 .detail
3087 .contains("closes its invocation"),
3088 "{}",
3089 cross(&report, "C8").detail
3090 );
3091 assert_eq!(report.exit_code(), 1);
3092 }
3093
3094 #[test]
3095 fn a_selected_effect_preceding_the_chain_head_fails_c8() {
3096 let chain = overflow_retry_chain("op-c8-backwards");
3097 let stream = vec![session_event(json!({
3098 "kind": "llm_completed",
3099 "turn": 1,
3100 "effect_id": "op-c8-backwards:step:1:effect:0",
3101 "invocation_id": "op-c8-backwards:step:2:effect:0"
3102 }))];
3103 let report = validate_with_session_log(&blobs(&chain), &[stream]);
3104 assert_eq!(cross(&report, "C8").verdict, Verdict::Fail);
3105 assert!(
3106 cross(&report, "C8")
3107 .detail
3108 .contains("does not follow the chain head"),
3109 "{}",
3110 cross(&report, "C8").detail
3111 );
3112 }
3113
3114 #[test]
3115 fn a_selected_effect_the_replan_never_published_fails_c8() {
3116 let chain = overflow_retry_chain("op-c8-phantom");
3117 let stream = vec![session_event(json!({
3118 "kind": "llm_completed",
3119 "turn": 1,
3120 "effect_id": "op-c8-phantom:step:2:effect:9",
3121 "invocation_id": "op-c8-phantom:step:1:effect:0"
3122 }))];
3123 let report = validate_with_session_log(&blobs(&chain), &[stream]);
3124 assert_eq!(cross(&report, "C8").verdict, Verdict::Fail);
3125 assert!(
3126 cross(&report, "C8").detail.contains("never published"),
3127 "{}",
3128 cross(&report, "C8").detail
3129 );
3130 assert_eq!(report.exit_code(), 1);
3131 }
3132
3133 #[test]
3134 fn a_first_try_invocation_has_no_adjacency_to_prove() {
3135 let op = operation("op-c8-first-try");
3136 let chain = live_chain(&[
3137 configure_envelope(&op),
3138 agent_start_envelope(&op),
3139 resolve_completed_envelope(&op, "in-resolve-1", 1_700_000_002_000, 1, false),
3140 ]);
3141 let stream = vec![session_event(json!({
3142 "kind": "llm_completed",
3143 "turn": 1,
3144 "effect_id": "op-c8-first-try:step:1:effect:0",
3145 "invocation_id": "op-c8-first-try:step:1:effect:0"
3146 }))];
3147 let report = validate_with_session_log(&blobs(&chain), &[stream]);
3148 assert_eq!(
3149 cross(&report, "C8").verdict,
3150 Verdict::Pass,
3151 "{}",
3152 cross(&report, "C8").detail
3153 );
3154 assert!(
3155 cross(&report, "C8").detail.contains("1 first-try"),
3156 "{}",
3157 cross(&report, "C8").detail
3158 );
3159 assert_eq!(report.exit_code(), 0);
3160 }
3161
3162 #[test]
3163 fn an_invocation_past_the_journal_tip_degrades_c8() {
3164 let op = operation("op-c8-lag");
3167 let chain = live_chain(&[configure_envelope(&op), agent_start_envelope(&op)]);
3168 let stream = vec![session_event(json!({
3169 "kind": "llm_completed",
3170 "turn": 1,
3171 "effect_id": "op-c8-lag:step:2:effect:0",
3172 "invocation_id": "op-c8-lag:step:1:effect:0"
3173 }))];
3174 let report = validate_with_session_log(&blobs(&chain), &[stream]);
3175 assert_eq!(
3176 cross(&report, "C8").verdict,
3177 Verdict::Degraded,
3178 "{}",
3179 cross(&report, "C8").detail
3180 );
3181 assert_eq!(report.exit_code(), 0);
3182 }
3183
3184 use crate::runtime::kernel::wire::checkpoint::{
3189 CheckpointDraft, KernelCheckpoint, LaunchTokenState,
3190 };
3191 use crate::runtime::kernel::wire::driver::PlannedStep;
3192 use crate::runtime::kernel::wire::transaction::CheckpointBoundary;
3193 use std::path::PathBuf;
3194
3195 fn live_runtime(
3198 envelopes: &[WireEnvelope],
3199 ) -> (
3200 Vec<KernelRecord>,
3201 KernelTransaction<PlannedStep, InMemoryRecordIndex>,
3202 CanonicalOperationDriver,
3203 ) {
3204 let mut tx = KernelTransaction::new(ConfigDefaults::default(), InMemoryRecordIndex::new());
3205 let mut driver = CanonicalOperationDriver::new();
3206 let mut journal = Vec::new();
3207 for envelope in envelopes {
3208 let preparation = tx.prepare(envelope, |context| driver.plan(context));
3209 let token = preparation
3210 .token()
3211 .unwrap_or_else(|| {
3212 panic!("expected a prepared step, got {:?}", preparation.fault())
3213 })
3214 .clone();
3215 let head = preparation.record().unwrap().record_digest().clone();
3216 let committed = tx.commit(&token, &head).expect("commit must succeed");
3217 journal.push(committed.record.clone());
3218 driver
3219 .note_committed(committed.step_seq)
3220 .expect("the driver folds the step it planned");
3221 }
3222 (journal, tx, driver)
3223 }
3224
3225 fn checkpoint_at_head(
3226 tx: &KernelTransaction<PlannedStep, InMemoryRecordIndex>,
3227 driver: &CanonicalOperationDriver,
3228 ) -> KernelCheckpoint {
3229 tx.checkpoint_candidate(driver.project_logical_state())
3230 .expect("the head checkpoints")
3231 .decode()
3232 .expect("the candidate decodes")
3233 }
3234
3235 fn checkpoint_blob(checkpoint: &KernelCheckpoint) -> Vec<u8> {
3236 checkpoint.checkpoint_bytes().into_vec()
3237 }
3238
3239 fn checkpoint_check<'a>(report: &'a ValidationReport, id: &str) -> &'a RuleReport {
3240 report
3241 .checkpoint_checks
3242 .iter()
3243 .find(|rule| rule.rule == id)
3244 .unwrap_or_else(|| panic!("the report has no {id} checkpoint-check"))
3245 }
3246
3247 fn no_streams() -> Vec<Vec<Vec<u8>>> {
3248 Vec::new()
3249 }
3250
3251 #[test]
3252 fn without_a_checkpoint_plane_c5_is_deferred_not_red() {
3253 let op = operation("op-c5-deferred");
3254 let chain = live_chain(&[configure_envelope(&op), agent_start_envelope(&op)]);
3255 let report = validate_journal(&blobs(&chain));
3256 assert!(report.checkpoint_checks.is_empty());
3257 assert_eq!(report.checkpoints, None);
3258 assert_eq!(report.unparseable_checkpoints, 0);
3259 assert_eq!(
3260 report.deferred.len(),
3261 2,
3262 "the checkpoint plane was never offered"
3263 );
3264 assert!(
3265 report.deferred[1].contains("c5b.launch_token_ledger"),
3266 "{}",
3267 report.deferred[1]
3268 );
3269 assert_eq!(report.exit_code(), 0);
3270 }
3271
3272 #[test]
3273 fn a_real_checkpoint_anchors_its_journal() {
3274 let op = operation("op-c5-green");
3275 let (chain, tx, driver) = live_runtime(&[
3276 configure_envelope(&op),
3277 agent_start_envelope(&op),
3278 resolve_overflow_envelope(&op, 1),
3279 ]);
3280 let checkpoint = checkpoint_at_head(&tx, &driver);
3281 let report = validate_with_checkpoint(
3282 &blobs(&chain),
3283 &no_streams(),
3284 &[checkpoint_blob(&checkpoint)],
3285 false,
3286 );
3287 assert_eq!(report.checkpoints, Some(1));
3288 assert_eq!(report.unparseable_checkpoints, 0);
3289 for id in ["C5a", "C5b"] {
3290 assert_eq!(
3291 checkpoint_check(&report, id).verdict,
3292 Verdict::Pass,
3293 "{id}: {}",
3294 checkpoint_check(&report, id).detail
3295 );
3296 }
3297 assert!(
3298 checkpoint_check(&report, "C5a")
3299 .detail
3300 .contains("covered head anchored at step"),
3301 "{}",
3302 checkpoint_check(&report, "C5a").detail
3303 );
3304 assert_eq!(
3305 report.deferred.len(),
3306 1,
3307 "the checkpoint plane retires the c5b deferral"
3308 );
3309 assert_eq!(report.exit_code(), 0);
3310 }
3311
3312 #[test]
3313 fn strict_replay_reproduces_the_covered_state_and_the_ladder_holds() {
3314 let op = operation("op-c5-strict");
3315 let (chain, tx, driver) = live_runtime(&[
3316 configure_envelope(&op),
3317 agent_start_envelope(&op),
3318 resolve_overflow_envelope(&op, 1),
3319 ]);
3320 let checkpoint = checkpoint_at_head(&tx, &driver);
3321 let report = validate_with_checkpoint(
3322 &blobs(&chain),
3323 &no_streams(),
3324 &[checkpoint_blob(&checkpoint)],
3325 true,
3326 );
3327 let c5a = checkpoint_check(&report, "C5a");
3328 assert_eq!(c5a.verdict, Verdict::Pass, "{}", c5a.detail);
3329 assert!(
3330 c5a.detail
3331 .contains("strict replay reproduces the captured state digest"),
3332 "{}",
3333 c5a.detail
3334 );
3335 assert!(
3336 c5a.detail.contains("restore ladder holds"),
3337 "{}",
3338 c5a.detail
3339 );
3340 let c5b = checkpoint_check(&report, "C5b");
3341 assert_eq!(c5b.verdict, Verdict::Pass, "{}", c5b.detail);
3342 assert_eq!(report.exit_code(), 0);
3343 }
3344
3345 #[test]
3346 fn strict_replay_with_tail_records_drives_the_whole_ladder() {
3347 let op = operation("op-c5-strict-tail");
3348 let (mut chain, mut tx, mut driver) =
3349 live_runtime(&[configure_envelope(&op), agent_start_envelope(&op)]);
3350 let checkpoint = checkpoint_at_head(&tx, &driver);
3353 let preparation = tx.prepare(&resolve_overflow_envelope(&op, 1), |context| {
3354 driver.plan(context)
3355 });
3356 let token = preparation.token().expect("the tail step prepares").clone();
3357 let head = preparation.record().unwrap().record_digest().clone();
3358 let committed = tx.commit(&token, &head).expect("the tail step commits");
3359 chain.push(committed.record.clone());
3360 driver
3361 .note_committed(committed.step_seq)
3362 .expect("the driver folds the tail step");
3363
3364 let report = validate_with_checkpoint(
3365 &blobs(&chain),
3366 &no_streams(),
3367 &[checkpoint_blob(&checkpoint)],
3368 true,
3369 );
3370 let c5a = checkpoint_check(&report, "C5a");
3371 assert_eq!(c5a.verdict, Verdict::Pass, "{}", c5a.detail);
3372 assert!(
3373 c5a.detail.contains("against the 1 journal record(s) above"),
3374 "{}",
3375 c5a.detail
3376 );
3377 assert_eq!(report.exit_code(), 0);
3378 }
3379
3380 #[test]
3381 fn a_bounded_tail_checkpoint_reconciles_with_the_journal() {
3382 let op = operation("op-c5-bounded");
3383 let (mut chain, mut tx, mut driver) =
3384 live_runtime(&[configure_envelope(&op), agent_start_envelope(&op)]);
3385 let candidate = tx
3388 .checkpoint_candidate(driver.project_logical_state())
3389 .expect("the base checkpoints");
3390 let boundary: CheckpointBoundary = candidate.boundary();
3391 let base_state = candidate
3392 .decode()
3393 .expect("the base decodes")
3394 .logical_state()
3395 .clone();
3396 let preparation = tx.prepare(&resolve_overflow_envelope(&op, 1), |context| {
3397 driver.plan(context)
3398 });
3399 let token = preparation.token().expect("the tail step prepares").clone();
3400 let head = preparation.record().unwrap().record_digest().clone();
3401 let committed = tx.commit(&token, &head).expect("the tail step commits");
3402 chain.push(committed.record.clone());
3403 driver
3404 .note_committed(committed.step_seq)
3405 .expect("the driver folds the tail step");
3406
3407 let rebased = tx
3408 .checkpoint_rebase(&boundary, base_state)
3409 .expect("the window rebases")
3410 .decode()
3411 .expect("the rebase decodes");
3412 assert_eq!(rebased.base_step_seq().get(), 1);
3413 assert_eq!(rebased.through_step_seq().get(), 2);
3414 assert_eq!(rebased.tail_inputs().len(), 1);
3415
3416 let report = validate_with_checkpoint(
3420 &blobs(&chain),
3421 &no_streams(),
3422 &[checkpoint_blob(&rebased)],
3423 true,
3424 );
3425 let c5a = checkpoint_check(&report, "C5a");
3426 assert_eq!(c5a.verdict, Verdict::Pass, "{}", c5a.detail);
3427 assert!(
3428 c5a.detail.contains("1 bounded-tail entries reconcile"),
3429 "{}",
3430 c5a.detail
3431 );
3432 assert!(
3433 c5a.detail.contains("the captured state digest at step 1"),
3434 "{}",
3435 c5a.detail
3436 );
3437 assert_eq!(report.exit_code(), 0);
3438 }
3439
3440 #[test]
3441 fn a_checkpoint_from_another_chain_fails_c5a() {
3442 let op = operation("op-c5-foreign");
3443 let (chain, _tx, _driver) =
3446 live_runtime(&[configure_envelope(&op), agent_start_envelope(&op)]);
3447 let foreign_config = |max_turns: u32| {
3448 envelope(
3449 &op,
3450 "in-configure",
3451 1_700_000_000_000,
3452 KernelInput::ConfigureOperation(ConfigureOperation {
3453 config: OperationConfig {
3454 execution_policy: Some(ExecutionPolicy {
3455 max_turns: Some(max_turns),
3456 ..ExecutionPolicy::default()
3457 }),
3458 host_effect_support: HostEffectSupport::new([
3459 EffectKindTag::CallProvider,
3460 EffectKindTag::SpawnTasks,
3461 ]),
3462 ..OperationConfig::default()
3463 },
3464 }),
3465 )
3466 };
3467 let (_chain_other, other_tx, other_driver) =
3468 live_runtime(&[foreign_config(24), agent_start_envelope(&op)]);
3469 let foreign = checkpoint_at_head(&other_tx, &other_driver);
3470
3471 let report = validate_with_checkpoint(
3472 &blobs(&chain),
3473 &no_streams(),
3474 &[checkpoint_blob(&foreign)],
3475 false,
3476 );
3477 let c5a = checkpoint_check(&report, "C5a");
3478 assert_eq!(c5a.verdict, Verdict::Fail, "{}", c5a.detail);
3479 assert!(
3480 c5a.detail.contains("captured on another chain"),
3481 "{}",
3482 c5a.detail
3483 );
3484 assert_eq!(report.exit_code(), 1);
3485 }
3486
3487 #[test]
3488 fn a_reused_launch_token_fails_c5b() {
3489 let op = operation("op-c5-tokens");
3490 let (chain, tx, driver) = live_runtime(&[
3491 configure_envelope(&op),
3492 workflow_start_envelope(&op),
3493 resolve_spawn_envelope(
3494 &op,
3495 "in-ack-1",
3496 1_700_000_002_000,
3497 "op-c5-tokens:step:1:effect:0",
3498 &[("wf-node0", "wf-node0:attempt:1")],
3499 ),
3500 ]);
3501 let checkpoint = checkpoint_at_head(&tx, &driver);
3502 assert_eq!(
3503 checkpoint.logical_state().transition.launch_tokens.len(),
3504 1,
3505 "the workflow start minted one launch token"
3506 );
3507
3508 let mut state = checkpoint.logical_state().clone();
3511 let minted = state.transition.launch_tokens[0].clone();
3512 state.transition.launch_tokens.push(LaunchTokenState {
3513 launch_token: minted.launch_token,
3514 step_seq: WireU64::new(2),
3515 });
3516 let forged = KernelCheckpoint::assemble(CheckpointDraft {
3517 operation_id: checkpoint.operation_id().clone(),
3518 genesis_digest: checkpoint.genesis_digest().clone(),
3519 base_step_seq: checkpoint.base_step_seq(),
3520 base_record_digest: checkpoint.base_record_digest().clone(),
3521 through_step_seq: checkpoint.through_step_seq(),
3522 covered_transaction_head_digest: checkpoint.covered_transaction_head_digest().clone(),
3523 logical_state: state,
3524 tail_inputs: checkpoint.tail_inputs().to_vec(),
3525 })
3526 .expect("the forged draft assembles");
3527
3528 let report = validate_with_checkpoint(
3529 &blobs(&chain),
3530 &no_streams(),
3531 &[checkpoint_blob(&forged)],
3532 false,
3533 );
3534 let c5b = checkpoint_check(&report, "C5b");
3535 assert_eq!(c5b.verdict, Verdict::Fail, "{}", c5b.detail);
3536 assert!(
3537 c5b.detail.contains("reuse across TaskLaunch payloads"),
3538 "{}",
3539 c5b.detail
3540 );
3541 assert_eq!(report.exit_code(), 1);
3542 }
3543
3544 #[test]
3545 fn strict_replay_catches_a_ledger_the_journal_never_minted() {
3546 let op = operation("op-c5-moved-mint");
3547 let (chain, tx, driver) = live_runtime(&[
3548 configure_envelope(&op),
3549 workflow_start_envelope(&op),
3550 resolve_spawn_envelope(
3551 &op,
3552 "in-ack-1",
3553 1_700_000_002_000,
3554 "op-c5-moved-mint:step:1:effect:0",
3555 &[("wf-node0", "wf-node0:attempt:1")],
3556 ),
3557 ]);
3558 let checkpoint = checkpoint_at_head(&tx, &driver);
3559
3560 let mut state = checkpoint.logical_state().clone();
3564 state.transition.launch_tokens[0].step_seq = WireU64::new(2);
3565 let forged = KernelCheckpoint::assemble(CheckpointDraft {
3566 operation_id: checkpoint.operation_id().clone(),
3567 genesis_digest: checkpoint.genesis_digest().clone(),
3568 base_step_seq: checkpoint.base_step_seq(),
3569 base_record_digest: checkpoint.base_record_digest().clone(),
3570 through_step_seq: checkpoint.through_step_seq(),
3571 covered_transaction_head_digest: checkpoint.covered_transaction_head_digest().clone(),
3572 logical_state: state,
3573 tail_inputs: checkpoint.tail_inputs().to_vec(),
3574 })
3575 .expect("the forged draft assembles");
3576
3577 let default_report = validate_with_checkpoint(
3578 &blobs(&chain),
3579 &no_streams(),
3580 &[checkpoint_blob(&forged)],
3581 false,
3582 );
3583 assert_eq!(
3584 checkpoint_check(&default_report, "C5b").verdict,
3585 Verdict::Pass,
3586 "the default plane cannot see a moved mint: {}",
3587 checkpoint_check(&default_report, "C5b").detail
3588 );
3589
3590 let strict_report = validate_with_checkpoint(
3591 &blobs(&chain),
3592 &no_streams(),
3593 &[checkpoint_blob(&forged)],
3594 true,
3595 );
3596 let c5b = checkpoint_check(&strict_report, "C5b");
3597 assert_eq!(c5b.verdict, Verdict::Fail, "{}", c5b.detail);
3598 assert!(
3599 c5b.detail.contains("different launch-token ledger"),
3600 "{}",
3601 c5b.detail
3602 );
3603 assert_eq!(
3604 checkpoint_check(&strict_report, "C5a").verdict,
3605 Verdict::Fail,
3606 "the ledger is part of the state, so the state digest moves too"
3607 );
3608 assert_eq!(strict_report.exit_code(), 1);
3609 }
3610
3611 #[test]
3612 fn a_tail_disconnected_from_the_journal_fails_c5a() {
3613 let op = operation("op-c5-spliced");
3614 let (chain_a, tx, driver) = live_runtime(&[
3615 configure_envelope(&op),
3616 agent_start_envelope(&op),
3617 resolve_overflow_envelope(&op, 1),
3618 ]);
3619 let checkpoint = checkpoint_at_head(&tx, &driver);
3620
3621 let other_resolution = envelope(
3625 &op,
3626 "in-resolve-other",
3627 1_700_000_002_000,
3628 KernelInput::ResolveEffect(ResolveEffect {
3629 effect_id: EffectId::new("op-c5-spliced:step:1:effect:0").unwrap(),
3630 outcome: EffectOutcome::Succeeded(EffectSucceeded {
3631 result: EffectSuccess::Provider(ProviderSuccess {
3632 outcome: ProviderOutcome::ContextOverflow(
3633 ProviderContextOverflow::default(),
3634 ),
3635 }),
3636 }),
3637 }),
3638 );
3639 let (chain_b, _tx_b, _driver_b) = live_runtime(&[
3640 configure_envelope(&op),
3641 agent_start_envelope(&op),
3642 other_resolution,
3643 ]);
3644 let mut spliced = chain_a[..2].to_vec();
3645 spliced.push(chain_b[2].clone());
3646
3647 let report = validate_with_checkpoint(
3648 &blobs(&spliced),
3649 &no_streams(),
3650 &[checkpoint_blob(&checkpoint)],
3651 false,
3652 );
3653 assert_eq!(
3654 rule(&report, 0, "C1").verdict,
3655 Verdict::Pass,
3656 "the splice chains cleanly — C1 is not the witness here"
3657 );
3658 let c5a = checkpoint_check(&report, "C5a");
3659 assert_eq!(c5a.verdict, Verdict::Fail, "{}", c5a.detail);
3660 assert!(c5a.detail.contains("covered step 2"), "{}", c5a.detail);
3661 assert_eq!(report.exit_code(), 1);
3662 }
3663
3664 #[test]
3665 fn a_pruned_journal_degrades_the_c5_anchors_and_c1_keeps_its_verdict() {
3666 let op = operation("op-c5-pruned");
3667 let (chain, tx, driver) = live_runtime(&[
3668 configure_envelope(&op),
3669 agent_start_envelope(&op),
3670 resolve_overflow_envelope(&op, 1),
3671 ]);
3672 let checkpoint = checkpoint_at_head(&tx, &driver);
3673 let pruned = chain[2..].to_vec();
3675
3676 let report = validate_with_checkpoint(
3677 &blobs(&pruned),
3678 &no_streams(),
3679 &[checkpoint_blob(&checkpoint)],
3680 true,
3681 );
3682 let c5a = checkpoint_check(&report, "C5a");
3685 assert_eq!(c5a.verdict, Verdict::Degraded, "{}", c5a.detail);
3686 assert!(
3687 c5a.detail.contains("identity anchor is unverifiable"),
3688 "{}",
3689 c5a.detail
3690 );
3691 assert!(
3692 c5a.detail.contains("covered head anchored at step 2"),
3693 "{}",
3694 c5a.detail
3695 );
3696 assert!(
3697 c5a.detail.contains("strict replay skipped"),
3698 "{}",
3699 c5a.detail
3700 );
3701 assert_eq!(rule(&report, 0, "C1").verdict, Verdict::Fail);
3705 assert!(report.has_violations());
3706 assert_eq!(report.exit_code(), 1);
3707 }
3708
3709 #[test]
3710 fn an_unparseable_checkpoint_leaves_evidence_insufficient() {
3711 let op = operation("op-c5-junk");
3712 let chain = live_chain(&[configure_envelope(&op), agent_start_envelope(&op)]);
3713 let report = validate_with_checkpoint(
3714 &blobs(&chain),
3715 &no_streams(),
3716 &[b"{not a checkpoint".to_vec()],
3717 false,
3718 );
3719 assert_eq!(report.unparseable_checkpoints, 1);
3720 assert_eq!(report.checkpoints, Some(0));
3721 assert_eq!(
3722 checkpoint_check(&report, "C5a").verdict,
3723 Verdict::Degraded,
3724 "{}",
3725 checkpoint_check(&report, "C5a").detail
3726 );
3727 assert_eq!(report.exit_code(), 2);
3728 }
3729
3730 #[test]
3731 fn the_published_golden_checkpoints_decode_through_the_checkpoint_plane() {
3732 let fixture_dir =
3733 PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../tests/fixtures/kernel-wire");
3734 let mut blobs = Vec::new();
3735 for name in [
3736 "golden_checkpoint_agent_turn",
3737 "golden_checkpoint_bounded_tail",
3738 ] {
3739 let wrapper: serde_json::Value = serde_json::from_str(
3740 &std::fs::read_to_string(fixture_dir.join(format!("{name}.json")))
3741 .expect("fixture readable"),
3742 )
3743 .expect("fixture json");
3744 blobs.push(
3745 serde_json::to_vec(&wrapper["checkpoint"]).expect("the nested checkpoint writes"),
3746 );
3747 }
3748 let report = validate_with_checkpoint(&[] as &[Vec<u8>], &no_streams(), &blobs, false);
3752 assert_eq!(report.checkpoints, Some(2));
3753 assert_eq!(report.unparseable_checkpoints, 0);
3754 let c5a = checkpoint_check(&report, "C5a");
3755 assert_eq!(c5a.verdict, Verdict::Degraded, "{}", c5a.detail);
3756 assert!(c5a.detail.contains("holds no segment"), "{}", c5a.detail);
3757 let c5b = checkpoint_check(&report, "C5b");
3758 assert_ne!(c5b.verdict, Verdict::Fail, "{}", c5b.detail);
3759 assert!(!report.has_violations());
3760 }
3761}