1use thiserror::Error;
49
50use crate::event::{Event, EventEnvelope, SCHEMA_VERSION};
51use crate::id::{RunId, SequenceNumber};
52
53#[derive(Debug, Clone, PartialEq, Eq, Error)]
58pub enum ValidationError {
59 #[error("candidate run id {} does not match the log's run id {}", .found.as_uuid(), .expected.as_uuid())]
61 RunIdMismatch {
62 expected: RunId,
64 found: RunId,
66 },
67 #[error("candidate seq {found} is not the expected next position {expected}")]
69 NonContiguousSeq {
70 expected: SequenceNumber,
72 found: SequenceNumber,
74 },
75 #[error("candidate schema_version {version} is out of range 1..={max}")]
77 BadSchemaVersion {
78 version: u32,
80 max: u32,
82 },
83 #[error("a run log must start with RunStarted, candidate is {found}")]
85 ExpectedRunStarted {
86 found: &'static str,
88 },
89 #[error("a run head may only be the first event; the log already has history")]
92 DuplicateRunStarted,
93 #[error("no event may follow the terminal {terminal}")]
95 AfterTerminal {
96 terminal: &'static str,
98 },
99 #[error(
103 "the intent at seq {intent_seq} awaits its completion; candidate {found} cannot follow it"
104 )]
105 ExpectedCompletion {
106 intent_seq: SequenceNumber,
108 found: &'static str,
110 },
111 #[error("completion correlates to seq {found} but the pending intent is at seq {expected}")]
114 MiscorrelatedCompletion {
115 expected: SequenceNumber,
117 found: SequenceNumber,
119 },
120 #[error("candidate {found} is a completion with no pending intent to correlate to")]
123 UncorrelatedCompletion {
124 found: &'static str,
126 },
127 #[error("intent at envelope seq {envelope_seq} carries a mismatched inner seq {inner_seq}")]
130 IntentSeqMismatch {
131 envelope_seq: SequenceNumber,
133 inner_seq: SequenceNumber,
135 },
136}
137
138pub fn validate_next(
150 log: &[EventEnvelope],
151 candidate: &EventEnvelope,
152) -> Result<(), ValidationError> {
153 if candidate.schema_version == 0 || candidate.schema_version > SCHEMA_VERSION {
155 return Err(ValidationError::BadSchemaVersion {
156 version: candidate.schema_version,
157 max: SCHEMA_VERSION,
158 });
159 }
160 let expected_seq = log
161 .last()
162 .map_or(SequenceNumber::new(0), |last| last.seq.next());
163 if candidate.seq != expected_seq {
164 return Err(ValidationError::NonContiguousSeq {
165 expected: expected_seq,
166 found: candidate.seq,
167 });
168 }
169
170 let Some(last) = log.last() else {
171 return match &candidate.event {
176 Event::RunStarted { .. } | Event::GraphRunStarted { .. } => Ok(()),
177 other => Err(ValidationError::ExpectedRunStarted {
178 found: kind_name(other),
179 }),
180 };
181 };
182
183 if candidate.run_id != last.run_id {
185 return Err(ValidationError::RunIdMismatch {
186 expected: last.run_id,
187 found: candidate.run_id,
188 });
189 }
190
191 if matches!(
194 candidate.event,
195 Event::RunStarted { .. } | Event::GraphRunStarted { .. }
196 ) {
197 return Err(ValidationError::DuplicateRunStarted);
198 }
199
200 if matches!(
204 last.event,
205 Event::RunCompleted { .. } | Event::RunFailed { .. } | Event::RunAbandoned { .. }
206 ) {
207 return Err(ValidationError::AfterTerminal {
208 terminal: kind_name(&last.event),
209 });
210 }
211
212 match &last.event {
217 Event::ModelCallRequested {
218 seq: intent_seq, ..
219 } => check_completes(*intent_seq, &candidate.event, CompletionKind::Model),
220 Event::ToolCallRequested {
221 seq: intent_seq, ..
222 } => check_completes(*intent_seq, &candidate.event, CompletionKind::Tool),
223 _ => match &candidate.event {
226 Event::ModelCallCompleted { .. } | Event::ToolCallCompleted { .. } => {
227 Err(ValidationError::UncorrelatedCompletion {
228 found: kind_name(&candidate.event),
229 })
230 }
231 Event::ModelCallRequested { seq, .. } | Event::ToolCallRequested { seq, .. } => {
232 if *seq == candidate.seq {
235 Ok(())
236 } else {
237 Err(ValidationError::IntentSeqMismatch {
238 envelope_seq: candidate.seq,
239 inner_seq: *seq,
240 })
241 }
242 }
243 _ => Ok(()),
244 },
245 }
246}
247
248#[derive(Clone, Copy)]
250enum CompletionKind {
251 Model,
252 Tool,
253}
254
255fn check_completes(
258 intent_seq: SequenceNumber,
259 candidate: &Event,
260 awaited: CompletionKind,
261) -> Result<(), ValidationError> {
262 let found = kind_name(candidate);
263 match (awaited, candidate) {
264 (CompletionKind::Model, Event::ModelCallCompleted { seq, .. })
265 | (CompletionKind::Tool, Event::ToolCallCompleted { seq, .. }) => {
266 if *seq == intent_seq {
267 Ok(())
268 } else {
269 Err(ValidationError::MiscorrelatedCompletion {
270 expected: intent_seq,
271 found: *seq,
272 })
273 }
274 }
275 _ => Err(ValidationError::ExpectedCompletion { intent_seq, found }),
276 }
277}
278
279#[derive(Debug, Clone)]
287pub struct LogValidator {
288 log: Vec<EventEnvelope>,
289}
290
291impl LogValidator {
292 #[must_use]
294 pub fn new(log: Vec<EventEnvelope>) -> Self {
295 Self { log }
296 }
297
298 #[must_use]
300 pub fn next_seq(&self) -> SequenceNumber {
301 self.log
302 .last()
303 .map_or(SequenceNumber::new(0), |last| last.seq.next())
304 }
305
306 #[must_use]
308 pub fn log(&self) -> &[EventEnvelope] {
309 &self.log
310 }
311
312 pub fn validate(&self, candidate: &EventEnvelope) -> Result<(), ValidationError> {
318 validate_next(&self.log, candidate)
319 }
320
321 pub fn push(&mut self, candidate: EventEnvelope) -> Result<(), ValidationError> {
328 self.validate(&candidate)?;
329 self.log.push(candidate);
330 Ok(())
331 }
332}
333
334fn kind_name(event: &Event) -> &'static str {
337 match event {
338 Event::RunStarted { .. } => "RunStarted",
339 Event::ModelCallRequested { .. } => "ModelCallRequested",
340 Event::ModelCallCompleted { .. } => "ModelCallCompleted",
341 Event::ToolCallRequested { .. } => "ToolCallRequested",
342 Event::ToolCallCompleted { .. } => "ToolCallCompleted",
343 Event::NowObserved { .. } => "NowObserved",
344 Event::RandomObserved { .. } => "RandomObserved",
345 Event::Suspended { .. } => "Suspended",
346 Event::Resumed { .. } => "Resumed",
347 Event::BudgetExceeded { .. } => "BudgetExceeded",
348 Event::RunCompleted { .. } => "RunCompleted",
349 Event::RunFailed { .. } => "RunFailed",
350 Event::RunAbandoned { .. } => "RunAbandoned",
351 Event::GraphRunStarted { .. } => "GraphRunStarted",
352 Event::NodeEntered { .. } => "NodeEntered",
353 Event::NodeExited { .. } => "NodeExited",
354 Event::NodeSkipped { .. } => "NodeSkipped",
355 Event::BranchTaken { .. } => "BranchTaken",
356 Event::MapFannedOut { .. } => "MapFannedOut",
357 Event::MapIterationStarted { .. } => "MapIterationStarted",
358 Event::MapIterationJoined { .. } => "MapIterationJoined",
359 Event::FoldIterationStarted { .. } => "FoldIterationStarted",
360 Event::FoldIterationJoined { .. } => "FoldIterationJoined",
361 Event::FoldConverged { .. } => "FoldConverged",
362 }
363}
364
365#[cfg(test)]
366mod tests {
367 use super::*;
368 use crate::effect::Effect;
369 use crate::event::{Budget, BudgetKind, TokenUsage};
370 use time::OffsetDateTime;
371 use time::macros::datetime;
372 use uuid::Uuid;
373
374 fn run_a() -> RunId {
375 RunId::from_uuid(Uuid::parse_str("00000000-0000-4000-8000-00000000000a").unwrap())
376 }
377
378 fn run_b() -> RunId {
379 RunId::from_uuid(Uuid::parse_str("00000000-0000-4000-8000-00000000000b").unwrap())
380 }
381
382 fn ts() -> OffsetDateTime {
383 datetime!(2026-07-11 12:00:00 UTC)
384 }
385
386 fn env(seq: u64, event: Event) -> EventEnvelope {
388 EventEnvelope::new(run_a(), SequenceNumber::new(seq), ts(), event)
389 }
390
391 fn started() -> Event {
392 Event::RunStarted {
393 agent_def_hash: "sha256:agent".into(),
394 input: serde_json::json!({"topic": "otters"}),
395 labels: None,
396 }
397 }
398
399 fn model_intent(seq: u64) -> Event {
400 Event::ModelCallRequested {
401 seq: SequenceNumber::new(seq),
402 request_hash: "sha256:req".into(),
403 request_body: None,
404 }
405 }
406
407 fn model_done(seq: u64) -> Event {
408 Event::ModelCallCompleted {
409 seq: SequenceNumber::new(seq),
410 response: serde_json::json!({"text": "hi"}),
411 usage: TokenUsage {
412 input_tokens: 1,
413 output_tokens: 1,
414 },
415 }
416 }
417
418 fn tool_intent(seq: u64, effect: Effect) -> Event {
419 Event::ToolCallRequested {
420 seq: SequenceNumber::new(seq),
421 tool: "render".into(),
422 input: serde_json::json!({"src": "x"}),
423 effect,
424 idempotency_key: None,
425 }
426 }
427
428 fn tool_done(seq: u64) -> Event {
429 Event::ToolCallCompleted {
430 seq: SequenceNumber::new(seq),
431 output: serde_json::json!({"ok": true}),
432 }
433 }
434
435 #[test]
438 fn a_legal_sequence_validates_event_by_event() {
439 let sequence = vec![
440 started(),
441 Event::NowObserved { now: ts() },
442 Event::RandomObserved { value: 7 },
443 Event::Suspended {
444 reason: "approval".into(),
445 input_schema: serde_json::json!({"type": "object"}),
446 },
447 Event::Resumed {
448 input: serde_json::json!({"approved": true}),
449 },
450 Event::BudgetExceeded {
451 budget: Budget {
452 kind: BudgetKind::Tokens,
453 limit: 100.0,
454 },
455 observed: 101.0,
456 },
457 Event::RunCompleted {
458 output: serde_json::json!({"done": true}),
459 },
460 ];
461 let mut validator = LogValidator::new(vec![]);
462 for (seq, event) in sequence.into_iter().enumerate() {
463 validator
464 .push(env(seq as u64, event))
465 .expect("each event is the legal next one");
466 }
467 }
468
469 #[test]
472 fn model_intent_then_correlated_completion_is_legal() {
473 let mut v = LogValidator::new(vec![]);
474 v.push(env(0, started())).unwrap();
475 v.push(env(1, model_intent(1))).unwrap();
476 v.push(env(2, model_done(1)))
477 .expect("the completion correlates to the intent at seq 1");
478 }
479
480 #[test]
484 fn write_intent_completion_is_well_formed() {
485 let mut v = LogValidator::new(vec![]);
486 v.push(env(0, started())).unwrap();
487 v.push(env(1, tool_intent(1, Effect::Write))).unwrap();
488 v.push(env(2, tool_done(1)))
489 .expect("a completion after a write intent is well formed");
490 }
491
492 fn graph_started() -> Event {
493 Event::GraphRunStarted {
494 graph_hash: "sha256:graph".into(),
495 input: serde_json::json!({"topic": "otters"}),
496 labels: None,
497 forked_from: None,
498 }
499 }
500
501 #[test]
505 fn graph_run_head_and_markers_validate() {
506 let mut v = LogValidator::new(vec![]);
507 v.push(env(0, graph_started()))
508 .expect("a graph run head opens a fresh log");
509 v.push(env(
510 1,
511 Event::NodeEntered {
512 node: "research".into(),
513 },
514 ))
515 .expect("a node marker is a legal free-standing event");
516 v.push(env(
517 2,
518 Event::BranchTaken {
519 node: "gate".into(),
520 case: "approved".into(),
521 },
522 ))
523 .expect("a branch marker is a legal free-standing event");
524 v.push(env(
525 3,
526 Event::NodeExited {
527 node: "research".into(),
528 },
529 ))
530 .expect("a node marker is a legal free-standing event");
531 }
532
533 #[test]
536 fn duplicate_graph_run_head_is_rejected() {
537 let log = vec![env(0, graph_started())];
538 let err = validate_next(&log, &env(1, graph_started())).unwrap_err();
539 assert_eq!(err, ValidationError::DuplicateRunStarted);
540 }
541
542 #[test]
545 fn graph_marker_after_intent_is_rejected() {
546 let log = vec![
547 env(0, graph_started()),
548 env(1, Event::NodeEntered { node: "n".into() }),
549 env(2, model_intent(2)),
550 ];
551 let err = validate_next(&log, &env(3, Event::NodeExited { node: "n".into() })).unwrap_err();
552 assert_eq!(
553 err,
554 ValidationError::ExpectedCompletion {
555 intent_seq: SequenceNumber::new(2),
556 found: "NodeExited",
557 }
558 );
559 }
560
561 #[test]
563 fn empty_log_rejects_non_run_started() {
564 let err = validate_next(&[], &env(0, Event::NowObserved { now: ts() })).unwrap_err();
565 assert_eq!(
566 err,
567 ValidationError::ExpectedRunStarted {
568 found: "NowObserved"
569 }
570 );
571 }
572
573 #[test]
575 fn duplicate_run_started_is_rejected() {
576 let log = vec![env(0, started())];
577 let err = validate_next(&log, &env(1, started())).unwrap_err();
578 assert_eq!(err, ValidationError::DuplicateRunStarted);
579 }
580
581 #[test]
584 fn non_contiguous_seq_is_rejected() {
585 let log = vec![env(0, started())];
586 let err = validate_next(&log, &env(5, Event::NowObserved { now: ts() })).unwrap_err();
587 assert_eq!(
588 err,
589 ValidationError::NonContiguousSeq {
590 expected: SequenceNumber::new(1),
591 found: SequenceNumber::new(5),
592 }
593 );
594 }
595
596 #[test]
598 fn wrong_run_id_is_rejected() {
599 let log = vec![env(0, started())];
600 let foreign = EventEnvelope::new(
601 run_b(),
602 SequenceNumber::new(1),
603 ts(),
604 Event::NowObserved { now: ts() },
605 );
606 let err = validate_next(&log, &foreign).unwrap_err();
607 assert_eq!(
608 err,
609 ValidationError::RunIdMismatch {
610 expected: run_a(),
611 found: run_b(),
612 }
613 );
614 }
615
616 #[test]
618 fn uncorrelated_completion_is_rejected() {
619 let log = vec![env(0, started())];
620 let err = validate_next(&log, &env(1, model_done(1))).unwrap_err();
621 assert_eq!(
622 err,
623 ValidationError::UncorrelatedCompletion {
624 found: "ModelCallCompleted"
625 }
626 );
627 }
628
629 #[test]
632 fn two_pending_intents_are_rejected() {
633 let log = vec![env(0, started()), env(1, model_intent(1))];
634 let err = validate_next(&log, &env(2, model_intent(2))).unwrap_err();
635 assert_eq!(
636 err,
637 ValidationError::ExpectedCompletion {
638 intent_seq: SequenceNumber::new(1),
639 found: "ModelCallRequested",
640 }
641 );
642 }
643
644 #[test]
647 fn bad_correlation_completion_is_rejected() {
648 let log = vec![env(0, started()), env(1, model_intent(1))];
649 let err = validate_next(&log, &env(2, model_done(9))).unwrap_err();
652 assert_eq!(
653 err,
654 ValidationError::MiscorrelatedCompletion {
655 expected: SequenceNumber::new(1),
656 found: SequenceNumber::new(9),
657 }
658 );
659 }
660
661 #[test]
664 fn context_event_after_intent_is_rejected() {
665 let log = vec![env(0, started()), env(1, model_intent(1))];
666 let err = validate_next(&log, &env(2, Event::NowObserved { now: ts() })).unwrap_err();
667 assert_eq!(
668 err,
669 ValidationError::ExpectedCompletion {
670 intent_seq: SequenceNumber::new(1),
671 found: "NowObserved",
672 }
673 );
674 }
675
676 #[test]
678 fn event_after_terminal_is_rejected() {
679 let log = vec![
680 env(0, started()),
681 env(
682 1,
683 Event::RunCompleted {
684 output: serde_json::json!({"done": true}),
685 },
686 ),
687 ];
688 let err = validate_next(&log, &env(2, Event::NowObserved { now: ts() })).unwrap_err();
689 assert_eq!(
690 err,
691 ValidationError::AfterTerminal {
692 terminal: "RunCompleted"
693 }
694 );
695 }
696
697 #[test]
701 fn event_after_abandoned_is_rejected() {
702 let log = vec![
703 env(0, started()),
704 env(
705 1,
706 Event::RunAbandoned {
707 reason: Some("husk is dead forever".into()),
708 unresolved_write: None,
709 },
710 ),
711 ];
712 let err = validate_next(&log, &env(2, Event::NowObserved { now: ts() })).unwrap_err();
713 assert_eq!(
714 err,
715 ValidationError::AfterTerminal {
716 terminal: "RunAbandoned"
717 }
718 );
719 }
720
721 #[test]
724 fn intent_inner_seq_must_match_envelope() {
725 let log = vec![env(0, started())];
726 let err = validate_next(&log, &env(1, model_intent(4))).unwrap_err();
728 assert_eq!(
729 err,
730 ValidationError::IntentSeqMismatch {
731 envelope_seq: SequenceNumber::new(1),
732 inner_seq: SequenceNumber::new(4),
733 }
734 );
735 }
736
737 #[test]
739 fn bad_schema_version_is_rejected() {
740 let mut e = env(0, started());
741 e.schema_version = 0;
742 assert_eq!(
743 validate_next(&[], &e).unwrap_err(),
744 ValidationError::BadSchemaVersion {
745 version: 0,
746 max: SCHEMA_VERSION,
747 }
748 );
749 let mut future = env(0, started());
750 future.schema_version = SCHEMA_VERSION + 1;
751 assert_eq!(
752 validate_next(&[], &future).unwrap_err(),
753 ValidationError::BadSchemaVersion {
754 version: SCHEMA_VERSION + 1,
755 max: SCHEMA_VERSION,
756 }
757 );
758 }
759}