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