1use std::collections::{BTreeMap, BTreeSet};
8
9use serde::{Deserialize, Serialize};
10use time::OffsetDateTime;
11
12use super::{
13 ceremony_definition::CeremonyDefinition, CeremonyEvidencePack, CeremonyIntervention,
14 PublishedCeremonyDefinition,
15};
16use crate::error::DomainError;
17use crate::ports::CeremonyEvidenceRequest;
18use crate::value_objects::{
19 AuditActorKind, CeremonyContext, CeremonyDefinitionDigest, CeremonyDefinitionDigestMigration,
20 CeremonyEvidenceSourceId, CeremonyGuardApproval, CeremonyGuardDeferral,
21 CeremonyGuardDeferralContent, CeremonyId, CeremonyInterventionContent, CeremonyInterventionId,
22 CeremonyInterventionKind, CeremonyInterventionProvenance, CeremonyInterventionResponse,
23 CeremonyInterventionTarget, CeremonyName, CeremonyParticipantBinding, CeremonyReason,
24 CeremonyReasonKind, CeremonyRecordRef, CeremonyTransitionRecord, CeremonyVersion,
25 GuardCondition, GuardName, IdempotencyKey, MemoryConfidence, ReasonAsserter, RoleAction,
26 RoleId, Specialty, StateId, StepAttempt, StepExecutionRecord, StepId, StepLease, StepResult,
27 StepStatus, TransitionTrigger,
28};
29
30mod guard_decisions;
31mod interventions;
32mod invariants;
33mod participant_bindings;
34mod step_execution;
35mod transitions;
36
37#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
38pub struct CeremonyInstance {
39 id: CeremonyId,
40 definition_name: CeremonyName,
41 definition_version: CeremonyVersion,
42 current_state: StateId,
43 step_records: BTreeMap<StepId, StepExecutionRecord>,
44 #[serde(default)]
51 step_record_history: BTreeMap<StepId, Vec<StepExecutionRecord>>,
52 #[serde(default)]
53 interventions: Vec<CeremonyIntervention>,
54 #[serde(default)]
55 guard_deferrals: Vec<CeremonyGuardDeferral>,
56 #[serde(default)]
66 guard_approvals: Vec<CeremonyGuardApproval>,
67 #[serde(default)]
74 transitions: Vec<CeremonyTransitionRecord>,
75 #[serde(default)]
82 reasons: Vec<CeremonyReason>,
83 #[serde(default)]
87 participant_bindings: BTreeMap<RoleId, CeremonyParticipantBinding>,
88 context: CeremonyContext,
89 idempotency_keys: BTreeSet<IdempotencyKey>,
90 #[serde(with = "time::serde::rfc3339")]
91 created_at: OffsetDateTime,
92 #[serde(with = "time::serde::rfc3339")]
93 updated_at: OffsetDateTime,
94 #[serde(with = "time::serde::rfc3339::option")]
95 completed_at: Option<OffsetDateTime>,
96 #[serde(default)]
107 bound_definition: Option<CeremonyDefinitionDigest>,
108}
109
110impl CeremonyInstance {
111 #[must_use]
116 pub fn start(
117 id: CeremonyId,
118 definition: &CeremonyDefinition,
119 context: CeremonyContext,
120 now: OffsetDateTime,
121 ) -> Self {
122 Self::open(id, definition, context, now, None)
123 }
124
125 #[must_use]
131 pub fn start_bound(
132 id: CeremonyId,
133 published: &PublishedCeremonyDefinition,
134 context: CeremonyContext,
135 now: OffsetDateTime,
136 ) -> Self {
137 Self::open(
138 id,
139 published.definition(),
140 context,
141 now,
142 Some(published.digest()),
143 )
144 }
145
146 fn open(
147 id: CeremonyId,
148 definition: &CeremonyDefinition,
149 context: CeremonyContext,
150 now: OffsetDateTime,
151 bound_definition: Option<CeremonyDefinitionDigest>,
152 ) -> Self {
153 let step_records = definition
154 .steps()
155 .keys()
156 .map(|step_id| (step_id.clone(), StepExecutionRecord::pending()))
157 .collect();
158
159 Self {
160 id,
161 definition_name: definition.name().clone(),
162 definition_version: definition.version().clone(),
163 current_state: definition.initial_state_id().clone(),
164 step_records,
165 step_record_history: BTreeMap::new(),
166 interventions: Vec::new(),
167 guard_deferrals: Vec::new(),
168 guard_approvals: Vec::new(),
169 transitions: Vec::new(),
170 reasons: Vec::new(),
171 participant_bindings: BTreeMap::new(),
172 context,
173 idempotency_keys: BTreeSet::new(),
174 created_at: now,
175 updated_at: now,
176 completed_at: None,
177 bound_definition,
178 }
179 }
180
181 #[must_use]
184 pub fn bound_definition(&self) -> Option<CeremonyDefinitionDigest> {
185 self.bound_definition
186 }
187
188 #[must_use]
191 pub fn is_bound_to_a_published_definition(&self) -> bool {
192 self.bound_definition.is_some()
193 }
194
195 pub fn migrate_definition_binding(
203 &mut self,
204 migration: &CeremonyDefinitionDigestMigration,
205 ) -> Result<bool, DomainError> {
206 if self.definition_name != *migration.definition_name()
207 || self.definition_version != *migration.definition_version()
208 {
209 return Err(DomainError::InvariantViolated {
210 reason: "a definition binding migration cannot change name or version",
211 });
212 }
213
214 match self.bound_definition {
215 Some(current) if current == migration.destination() => Ok(false),
216 Some(current) if current == migration.source() => {
217 self.bound_definition = Some(migration.destination());
218 Ok(true)
219 }
220 _ => Err(DomainError::InvariantViolated {
221 reason: "a definition binding migration did not match the stored identity",
222 }),
223 }
224 }
225
226 #[must_use]
227 pub fn id(&self) -> &CeremonyId {
228 &self.id
229 }
230
231 #[must_use]
232 pub fn definition_name(&self) -> &CeremonyName {
233 &self.definition_name
234 }
235
236 #[must_use]
237 pub fn definition_version(&self) -> &CeremonyVersion {
238 &self.definition_version
239 }
240
241 #[must_use]
242 pub fn current_state(&self) -> &StateId {
243 &self.current_state
244 }
245
246 #[must_use]
247 pub fn step_records(&self) -> &BTreeMap<StepId, StepExecutionRecord> {
248 &self.step_records
249 }
250
251 #[must_use]
252 pub fn step_record(&self, step_id: &StepId) -> Option<&StepExecutionRecord> {
253 self.step_records.get(step_id)
254 }
255
256 #[must_use]
258 pub fn step_record_history(&self, step_id: &StepId) -> &[StepExecutionRecord] {
259 self.step_record_history
260 .get(step_id)
261 .map(Vec::as_slice)
262 .unwrap_or_default()
263 }
264
265 #[must_use]
268 pub fn step_repeat_limit_reached(
269 &self,
270 definition: &CeremonyDefinition,
271 step_id: &StepId,
272 ) -> bool {
273 let Some(step) = definition.step(step_id) else {
274 return false;
275 };
276 let Some(policy) = step.repeat_policy() else {
277 return false;
278 };
279 let Some(record) = self.step_record(step_id) else {
280 return false;
281 };
282 record.status().is_success()
283 && !policy.is_satisfied(record.output())
284 && !policy.permits_another_iteration(record.iteration())
285 }
286
287 #[must_use]
288 pub fn interventions(&self) -> &[CeremonyIntervention] {
289 &self.interventions
290 }
291
292 #[must_use]
293 pub fn guard_deferrals(&self) -> &[CeremonyGuardDeferral] {
294 &self.guard_deferrals
295 }
296
297 #[must_use]
303 pub fn guard_approvals(&self) -> &[CeremonyGuardApproval] {
304 &self.guard_approvals
305 }
306
307 #[must_use]
309 pub fn transitions(&self) -> &[CeremonyTransitionRecord] {
310 &self.transitions
311 }
312
313 #[must_use]
315 pub fn reasons(&self) -> &[CeremonyReason] {
316 &self.reasons
317 }
318
319 #[must_use]
320 pub fn intervention(
321 &self,
322 intervention_id: &CeremonyInterventionId,
323 ) -> Option<&CeremonyIntervention> {
324 self.interventions
325 .iter()
326 .find(|intervention| intervention.id() == intervention_id)
327 }
328
329 #[must_use]
330 pub fn context(&self) -> &CeremonyContext {
331 &self.context
332 }
333
334 #[must_use]
335 pub fn idempotency_keys(&self) -> &BTreeSet<IdempotencyKey> {
336 &self.idempotency_keys
337 }
338
339 #[must_use]
340 pub fn created_at(&self) -> OffsetDateTime {
341 self.created_at
342 }
343
344 #[must_use]
345 pub fn updated_at(&self) -> OffsetDateTime {
346 self.updated_at
347 }
348
349 #[must_use]
350 pub fn completed_at(&self) -> Option<OffsetDateTime> {
351 self.completed_at
352 }
353
354 #[must_use]
355 pub fn is_terminal(&self, definition: &CeremonyDefinition) -> bool {
356 self.matches_definition(definition) && definition.is_terminal_state(&self.current_state)
357 }
358
359 #[must_use]
360 pub fn is_completed(&self, definition: &CeremonyDefinition) -> bool {
361 self.is_terminal(definition) && self.completed_at.is_some()
362 }
363}
364
365#[cfg(test)]
366mod tests {
367 use super::*;
368 use crate::value_objects::{
369 Attributes, CeremonyGuard, CeremonyState, CeremonyStep, CeremonyTransition, GuardCondition,
370 GuardName, LeaseOwnerId, RepeatUntilCondition, RetryPolicy, StepHandlerConfig,
371 StepHandlerKind, StepIteration, StepOutput, StepOutputField, StepRepeatPolicy,
372 };
373 use serde_json::json;
374 use time::macros::datetime;
375
376 fn now() -> OffsetDateTime {
377 datetime!(2026-06-06 12:00:00 UTC)
378 }
379
380 fn state_id(raw: &str) -> StateId {
381 StateId::new(raw).unwrap()
382 }
383
384 fn step_id(raw: &str) -> StepId {
385 StepId::new(raw).unwrap()
386 }
387
388 fn trigger(raw: &str) -> TransitionTrigger {
389 TransitionTrigger::new(raw).unwrap()
390 }
391
392 fn role_id(raw: &str) -> RoleId {
393 RoleId::new(raw).unwrap()
394 }
395
396 fn guard_name(raw: &str) -> GuardName {
397 GuardName::new(raw).unwrap()
398 }
399
400 fn handler_kind() -> StepHandlerKind {
401 StepHandlerKind::new("multiagent_round").unwrap()
402 }
403
404 fn retrying_step(raw_step_id: &str, raw_state_id: &str) -> CeremonyStep {
405 CeremonyStep::new(
406 step_id(raw_step_id),
407 state_id(raw_state_id),
408 handler_kind(),
409 StepHandlerConfig::empty(),
410 RetryPolicy::new(
411 StepAttempt::new(3).unwrap(),
412 crate::value_objects::DurationMs::ZERO,
413 ),
414 None,
415 )
416 }
417
418 fn single_attempt_step(raw_step_id: &str, raw_state_id: &str) -> CeremonyStep {
419 CeremonyStep::new(
420 step_id(raw_step_id),
421 state_id(raw_state_id),
422 handler_kind(),
423 StepHandlerConfig::empty(),
424 RetryPolicy::single_attempt(),
425 None,
426 )
427 }
428
429 fn repeating_plan(max_iterations: u32) -> CeremonyStep {
430 retrying_step("plan", "drafting").with_repeat_policy(StepRepeatPolicy::new(
431 RepeatUntilCondition::output_field_equals(
432 StepOutputField::new("ready").unwrap(),
433 json!(true),
434 ),
435 StepIteration::new(max_iterations).unwrap(),
436 ))
437 }
438
439 fn readiness_output(ready: bool) -> StepOutput {
440 StepOutput::new(
441 Attributes::new(std::collections::BTreeMap::from([(
442 "ready".to_owned(),
443 json!(ready),
444 )]))
445 .unwrap(),
446 )
447 }
448
449 fn lease(
450 raw_owner_id: &str,
451 raw_key: &str,
452 acquired_at: OffsetDateTime,
453 expires_at: OffsetDateTime,
454 ) -> StepLease {
455 StepLease::new(
456 LeaseOwnerId::new(raw_owner_id).unwrap(),
457 IdempotencyKey::new(raw_key).unwrap(),
458 acquired_at,
459 expires_at,
460 )
461 .unwrap()
462 }
463
464 fn role(actions: Vec<RoleAction>) -> crate::value_objects::CeremonyRole {
465 crate::value_objects::CeremonyRole::new(role_id("facilitator"), actions).unwrap()
466 }
467
468 fn definition_with_steps(steps: Vec<CeremonyStep>) -> CeremonyDefinition {
469 let plan_done = CeremonyGuard::new(
470 guard_name("plan_done"),
471 GuardCondition::StepStatus {
472 step_id: step_id("plan"),
473 status: StepStatus::Completed,
474 },
475 );
476 let finish = CeremonyTransition::new(
477 state_id("drafting"),
478 state_id("done"),
479 trigger("finish"),
480 vec![plan_done.name().clone()],
481 )
482 .unwrap();
483 let role = role(vec![
484 RoleAction::step(step_id("plan")),
485 RoleAction::transition(finish.trigger().clone()),
486 RoleAction::request_intervention(),
487 ]);
488 let observer = crate::value_objects::CeremonyRole::new(
489 role_id("observer"),
490 vec![RoleAction::respond_to_intervention()],
491 )
492 .unwrap();
493
494 CeremonyDefinition::new(
495 crate::value_objects::CeremonyName::new("planning_ceremony").unwrap(),
496 CeremonyVersion::v1(),
497 None,
498 Vec::new(),
499 Vec::new(),
500 vec![
501 CeremonyState::initial(state_id("drafting")),
502 CeremonyState::intermediate(state_id("review")),
503 CeremonyState::terminal(state_id("done")),
504 ],
505 vec![finish],
506 steps,
507 vec![plan_done],
508 vec![role, observer],
509 )
510 .unwrap()
511 }
512
513 fn definition() -> CeremonyDefinition {
514 definition_with_steps(vec![
515 retrying_step("plan", "drafting"),
516 single_attempt_step("review_step", "review"),
517 ])
518 }
519
520 #[test]
521 fn a_verified_digest_migration_rebinds_only_its_exact_definition() {
522 let definition = definition();
523 let published = PublishedCeremonyDefinition::seal(definition.clone()).unwrap();
524 let migration = definition.choreographer_v1_digest_migration().unwrap();
525 let mut value = serde_json::to_value(CeremonyInstance::start_bound(
526 CeremonyId::new("legacy-bound").unwrap(),
527 &published,
528 CeremonyContext::empty(),
529 now(),
530 ))
531 .unwrap();
532 value["bound_definition"] = serde_json::to_value(migration.source()).unwrap();
533 let mut instance: CeremonyInstance = serde_json::from_value(value).unwrap();
534
535 assert!(instance.migrate_definition_binding(&migration).unwrap());
536 assert_eq!(instance.bound_definition(), Some(migration.destination()));
537 assert!(!instance.migrate_definition_binding(&migration).unwrap());
538 }
539
540 #[test]
541 fn a_digest_migration_for_another_definition_is_rejected() {
542 let definition = definition();
543 let published = PublishedCeremonyDefinition::seal(definition.clone()).unwrap();
544 let mut instance = CeremonyInstance::start_bound(
545 CeremonyId::new("still-bound").unwrap(),
546 &published,
547 CeremonyContext::empty(),
548 now(),
549 );
550 let other = CeremonyDefinition::new(
551 CeremonyName::new("another_ceremony").unwrap(),
552 CeremonyVersion::v1(),
553 None,
554 [],
555 [],
556 [CeremonyState::initial(state_id("OPEN"))],
557 [],
558 [],
559 [],
560 [],
561 )
562 .unwrap()
563 .choreographer_v1_digest_migration()
564 .unwrap();
565
566 assert!(instance.migrate_definition_binding(&other).is_err());
567 assert_eq!(instance.bound_definition(), Some(published.digest()));
568 }
569
570 fn definition_with_human_guard(approval: &CeremonyGuard) -> CeremonyDefinition {
573 let finish = CeremonyTransition::new(
574 state_id("drafting"),
575 state_id("done"),
576 trigger("approve"),
577 vec![approval.name().clone()],
578 )
579 .unwrap();
580 CeremonyDefinition::new(
581 crate::value_objects::CeremonyName::new("approval_ceremony").unwrap(),
582 CeremonyVersion::v1(),
583 None,
584 Vec::new(),
585 Vec::new(),
586 vec![
587 CeremonyState::initial(state_id("drafting")),
588 CeremonyState::terminal(state_id("done")),
589 ],
590 vec![finish.clone()],
591 Vec::new(),
592 vec![approval.clone()],
593 vec![role(vec![RoleAction::transition(finish.trigger().clone())])],
594 )
595 .unwrap()
596 }
597
598 fn instance(definition: &CeremonyDefinition) -> CeremonyInstance {
599 CeremonyInstance::start(
600 CeremonyId::new("ceremony-1").unwrap(),
601 definition,
602 CeremonyContext::empty(),
603 now(),
604 )
605 }
606
607 #[test]
608 fn starts_in_initial_state_with_pending_records() {
609 let definition = definition();
610 let instance = instance(&definition);
611
612 assert_eq!(instance.current_state(), &state_id("drafting"));
613 assert_eq!(
614 instance.step_record(&step_id("plan")).unwrap().status(),
615 StepStatus::Pending
616 );
617 assert_eq!(
618 instance
619 .step_record(&step_id("review_step"))
620 .unwrap()
621 .status(),
622 StepStatus::Pending
623 );
624 }
625
626 #[test]
627 fn instances_without_iteration_fields_load_as_the_first_iteration() {
628 let definition = definition();
629 let mut value = serde_json::to_value(instance(&definition)).unwrap();
630 value.as_object_mut().unwrap().remove("step_record_history");
631 for record in value["step_records"].as_object_mut().unwrap().values_mut() {
632 record.as_object_mut().unwrap().remove("iteration");
633 }
634
635 let restored: CeremonyInstance = serde_json::from_value(value).unwrap();
636
637 assert!(restored.step_record_history(&step_id("plan")).is_empty());
638 assert_eq!(
639 restored.step_record(&step_id("plan")).unwrap().iteration(),
640 StepIteration::FIRST
641 );
642 }
643
644 #[test]
645 fn dynamic_intervention_collects_role_scoped_response_and_requester_closes_it() {
646 let definition = definition();
647 let mut instance = instance(&definition);
648 let intervention_id = CeremonyInterventionId::new("queue-check").unwrap();
649 let facilitator = role_id("facilitator");
650 let observer = role_id("observer");
651
652 instance
653 .request_intervention_as(
654 &definition,
655 intervention_id.clone(),
656 facilitator.clone(),
657 CeremonyInterventionKind::Investigation,
658 CeremonyInterventionTarget::roles([observer.clone()]).unwrap(),
659 CeremonyInterventionContent::new(
660 "Inspect the queue without consuming messages.",
661 Attributes::empty(),
662 )
663 .unwrap(),
664 now(),
665 )
666 .unwrap();
667 instance
668 .respond_to_intervention_as(
669 &definition,
670 &intervention_id,
671 observer.clone(),
672 CeremonyInterventionContent::new("Queue depth is stable.", Attributes::empty())
673 .unwrap(),
674 now(),
675 )
676 .unwrap();
677 let selected_intervention_id = CeremonyInterventionId::new("selected-check").unwrap();
678 instance
679 .request_intervention_with_provenance_as(
680 &definition,
681 selected_intervention_id.clone(),
682 facilitator.clone(),
683 CeremonyInterventionKind::Investigation,
684 CeremonyInterventionTarget::roles([observer.clone()]).unwrap(),
685 CeremonyInterventionContent::new(
686 "Inspect the proposed signal.",
687 Attributes::empty(),
688 )
689 .unwrap(),
690 Some(CeremonyInterventionProvenance::selected_from(
691 intervention_id.clone(),
692 observer.clone(),
693 observer.clone(),
694 )),
695 now(),
696 )
697 .unwrap();
698 instance
699 .close_intervention_as(&definition, &intervention_id, &facilitator, now())
700 .unwrap();
701
702 let intervention = instance.intervention(&intervention_id).unwrap();
703 assert_eq!(intervention.responses().len(), 1);
704 assert_eq!(
705 intervention.status(),
706 crate::value_objects::CeremonyInterventionStatus::Closed
707 );
708 let provenance = instance
709 .intervention(&selected_intervention_id)
710 .unwrap()
711 .provenance()
712 .unwrap();
713 assert_eq!(provenance.source_intervention_id(), &intervention_id);
714 assert_eq!(provenance.selected_role_id(), &observer);
715 }
716
717 #[test]
718 fn intervention_rejects_roles_without_the_required_capability() {
719 let definition = definition();
720 let mut instance = instance(&definition);
721
722 let error = instance
723 .request_intervention_as(
724 &definition,
725 CeremonyInterventionId::new("not-allowed").unwrap(),
726 role_id("observer"),
727 CeremonyInterventionKind::Opinion,
728 CeremonyInterventionTarget::table(),
729 CeremonyInterventionContent::new("What do you think?", Attributes::empty())
730 .unwrap(),
731 now(),
732 )
733 .unwrap_err();
734
735 assert!(matches!(error, DomainError::InvariantViolated { .. }));
736 }
737
738 #[test]
739 fn rejects_step_execution_outside_current_state() {
740 let definition = definition();
741 let mut instance = instance(&definition);
742
743 let err = instance
744 .start_step(
745 &definition,
746 &step_id("review_step"),
747 lease(
748 "runner-1",
749 "key-1",
750 now(),
751 datetime!(2026-06-06 12:05:00 UTC),
752 ),
753 now(),
754 )
755 .unwrap_err();
756
757 assert!(matches!(err, DomainError::InvalidTransition { .. }));
758 }
759
760 #[test]
761 fn completed_step_unlocks_guarded_transition() {
762 let definition = definition();
763 let mut instance = instance(&definition);
764
765 instance
766 .start_step_as(
767 &definition,
768 &role_id("facilitator"),
769 &step_id("plan"),
770 lease(
771 "runner-1",
772 "key-1",
773 now(),
774 datetime!(2026-06-06 12:05:00 UTC),
775 ),
776 now(),
777 )
778 .unwrap();
779 instance
780 .apply_step_result(
781 &definition,
782 &step_id("plan"),
783 StepResult::completed(StepOutput::empty()).unwrap(),
784 datetime!(2026-06-06 12:01:00 UTC),
785 )
786 .unwrap();
787 let state = instance
788 .apply_transition_as(
789 &definition,
790 &role_id("facilitator"),
791 &trigger("finish"),
792 datetime!(2026-06-06 12:02:00 UTC),
793 )
794 .unwrap();
795
796 assert_eq!(state, state_id("done"));
797 assert!(instance.is_completed(&definition));
798 }
799
800 #[test]
801 fn false_repeat_condition_archives_iteration_and_schedules_the_next() {
802 let definition = definition_with_steps(vec![repeating_plan(3)]);
803 let mut instance = instance(&definition);
804
805 instance
806 .start_step(
807 &definition,
808 &step_id("plan"),
809 lease(
810 "runner-1",
811 "repeat-1",
812 now(),
813 datetime!(2026-06-06 12:05:00 UTC),
814 ),
815 now(),
816 )
817 .unwrap();
818 instance
819 .apply_step_result(
820 &definition,
821 &step_id("plan"),
822 StepResult::completed(readiness_output(false)).unwrap(),
823 datetime!(2026-06-06 12:01:00 UTC),
824 )
825 .unwrap();
826
827 let current = instance.step_record(&step_id("plan")).unwrap();
828 assert_eq!(current.status(), StepStatus::Pending);
829 assert_eq!(current.iteration().get(), 2);
830 assert_eq!(current.attempt(), StepAttempt::FIRST);
831 let history = instance.step_record_history(&step_id("plan"));
832 assert_eq!(history.len(), 1);
833 assert_eq!(history[0].iteration(), StepIteration::FIRST);
834 assert_eq!(history[0].output(), &readiness_output(false));
835 assert!(instance
836 .apply_transition(&definition, &trigger("finish"), now())
837 .is_err());
838
839 instance
840 .start_step(
841 &definition,
842 &step_id("plan"),
843 lease(
844 "runner-1",
845 "repeat-2",
846 datetime!(2026-06-06 12:02:00 UTC),
847 datetime!(2026-06-06 12:07:00 UTC),
848 ),
849 datetime!(2026-06-06 12:02:00 UTC),
850 )
851 .unwrap();
852 instance
853 .apply_step_result(
854 &definition,
855 &step_id("plan"),
856 StepResult::completed(readiness_output(true)).unwrap(),
857 datetime!(2026-06-06 12:03:00 UTC),
858 )
859 .unwrap();
860
861 let current = instance.step_record(&step_id("plan")).unwrap();
862 assert_eq!(current.status(), StepStatus::Completed);
863 assert_eq!(current.iteration().get(), 2);
864 assert!(!instance.step_repeat_limit_reached(&definition, &step_id("plan")));
865 assert_eq!(
866 instance
867 .apply_transition(&definition, &trigger("finish"), now())
868 .unwrap(),
869 state_id("done")
870 );
871 }
872
873 #[test]
874 fn repeat_limit_is_terminal_for_the_step_and_blocks_transition() {
875 let definition = definition_with_steps(vec![repeating_plan(2)]);
876 let mut instance = instance(&definition);
877
878 for iteration in 1..=2 {
879 instance
880 .start_step(
881 &definition,
882 &step_id("plan"),
883 lease(
884 "runner-1",
885 &format!("limit-{iteration}"),
886 now(),
887 datetime!(2026-06-06 12:05:00 UTC),
888 ),
889 now(),
890 )
891 .unwrap();
892 instance
893 .apply_step_result(
894 &definition,
895 &step_id("plan"),
896 StepResult::completed(readiness_output(false)).unwrap(),
897 now(),
898 )
899 .unwrap();
900 }
901
902 assert!(instance.step_repeat_limit_reached(&definition, &step_id("plan")));
903 assert_eq!(
904 instance
905 .step_record(&step_id("plan"))
906 .unwrap()
907 .iteration()
908 .get(),
909 2
910 );
911 assert_eq!(instance.step_record_history(&step_id("plan")).len(), 1);
912 assert!(instance
913 .apply_transition(&definition, &trigger("finish"), now())
914 .is_err());
915 assert!(instance
916 .start_step(
917 &definition,
918 &step_id("plan"),
919 lease(
920 "runner-1",
921 "limit-3",
922 now(),
923 datetime!(2026-06-06 12:05:00 UTC),
924 ),
925 now(),
926 )
927 .is_err());
928 }
929
930 #[test]
931 fn active_lease_blocks_failover_takeover() {
932 let definition = definition();
933 let mut instance = instance(&definition);
934
935 instance
936 .start_step(
937 &definition,
938 &step_id("plan"),
939 lease(
940 "runner-1",
941 "key-1",
942 now(),
943 datetime!(2026-06-06 12:05:00 UTC),
944 ),
945 now(),
946 )
947 .unwrap();
948 let err = instance
949 .start_step(
950 &definition,
951 &step_id("plan"),
952 lease(
953 "runner-2",
954 "key-2",
955 datetime!(2026-06-06 12:01:00 UTC),
956 datetime!(2026-06-06 12:06:00 UTC),
957 ),
958 datetime!(2026-06-06 12:01:00 UTC),
959 )
960 .unwrap_err();
961
962 assert!(matches!(err, DomainError::InvariantViolated { .. }));
963 assert_eq!(
964 instance
965 .step_record(&step_id("plan"))
966 .unwrap()
967 .lease()
968 .unwrap()
969 .owner_id()
970 .as_str(),
971 "runner-1"
972 );
973 }
974
975 #[test]
976 fn expired_lease_allows_failover_takeover_with_next_attempt() {
977 let definition = definition();
978 let mut instance = instance(&definition);
979
980 instance
981 .start_step(
982 &definition,
983 &step_id("plan"),
984 lease(
985 "runner-1",
986 "key-1",
987 now(),
988 datetime!(2026-06-06 12:05:00 UTC),
989 ),
990 now(),
991 )
992 .unwrap();
993 let attempt = instance
994 .start_step(
995 &definition,
996 &step_id("plan"),
997 lease(
998 "runner-2",
999 "key-2",
1000 datetime!(2026-06-06 12:06:00 UTC),
1001 datetime!(2026-06-06 12:11:00 UTC),
1002 ),
1003 datetime!(2026-06-06 12:06:00 UTC),
1004 )
1005 .unwrap();
1006
1007 assert_eq!(attempt, StepAttempt::new(2).unwrap());
1008 let record = instance.step_record(&step_id("plan")).unwrap();
1009 assert_eq!(record.attempt(), StepAttempt::new(2).unwrap());
1010 assert_eq!(record.lease().unwrap().owner_id().as_str(), "runner-2");
1011 }
1012
1013 #[test]
1014 fn approving_a_guard_the_ceremony_never_declared_is_refused() {
1015 let approval =
1016 CeremonyGuard::new(guard_name("human_approved"), GuardCondition::HumanApproval);
1017 let finish = CeremonyTransition::new(
1018 state_id("drafting"),
1019 state_id("done"),
1020 trigger("approve"),
1021 vec![approval.name().clone()],
1022 )
1023 .unwrap();
1024 let definition = CeremonyDefinition::new(
1025 crate::value_objects::CeremonyName::new("approval_ceremony").unwrap(),
1026 CeremonyVersion::v1(),
1027 None,
1028 Vec::new(),
1029 Vec::new(),
1030 vec![
1031 CeremonyState::initial(state_id("drafting")),
1032 CeremonyState::terminal(state_id("done")),
1033 ],
1034 vec![finish.clone()],
1035 Vec::new(),
1036 vec![approval],
1037 vec![role(vec![RoleAction::transition(finish.trigger().clone())])],
1038 )
1039 .unwrap();
1040 let mut instance = instance(&definition);
1041
1042 assert!(matches!(
1047 instance.approve_guard(
1048 &definition,
1049 &guard_name("not_a_guard"),
1050 role_id("facilitator"),
1051 AuditActorKind::Human,
1052 now()
1053 ),
1054 Err(DomainError::NotFound {
1055 what: "ceremony_guard"
1056 })
1057 ));
1058 assert!(!instance
1059 .context()
1060 .is_guard_approved(&guard_name("not_a_guard")));
1061 }
1062
1063 #[test]
1064 fn human_approval_guard_uses_typed_context() {
1065 let approval =
1066 CeremonyGuard::new(guard_name("human_approved"), GuardCondition::HumanApproval);
1067 let finish = CeremonyTransition::new(
1068 state_id("drafting"),
1069 state_id("done"),
1070 trigger("approve"),
1071 vec![approval.name().clone()],
1072 )
1073 .unwrap();
1074 let definition = CeremonyDefinition::new(
1075 crate::value_objects::CeremonyName::new("approval_ceremony").unwrap(),
1076 CeremonyVersion::v1(),
1077 None,
1078 Vec::new(),
1079 Vec::new(),
1080 vec![
1081 CeremonyState::initial(state_id("drafting")),
1082 CeremonyState::terminal(state_id("done")),
1083 ],
1084 vec![finish.clone()],
1085 Vec::new(),
1086 vec![approval.clone()],
1087 vec![role(vec![RoleAction::transition(finish.trigger().clone())])],
1088 )
1089 .unwrap();
1090 let mut instance = instance(&definition);
1091
1092 assert!(matches!(
1093 instance.apply_transition(&definition, &trigger("approve"), now()),
1094 Err(DomainError::InvariantViolated { .. })
1095 ));
1096 instance
1097 .approve_guard(
1098 &definition,
1099 approval.name(),
1100 role_id("facilitator"),
1101 AuditActorKind::Human,
1102 datetime!(2026-06-06 12:01:00 UTC),
1103 )
1104 .unwrap();
1105 instance
1106 .apply_transition(
1107 &definition,
1108 &trigger("approve"),
1109 datetime!(2026-06-06 12:02:00 UTC),
1110 )
1111 .unwrap();
1112
1113 assert!(instance.is_completed(&definition));
1114 }
1115
1116 #[test]
1117 fn human_guard_deferral_preserves_uncertainty_without_approving() {
1118 let approval =
1119 CeremonyGuard::new(guard_name("human_approved"), GuardCondition::HumanApproval);
1120 let finish = CeremonyTransition::new(
1121 state_id("drafting"),
1122 state_id("done"),
1123 trigger("approve"),
1124 vec![approval.name().clone()],
1125 )
1126 .unwrap();
1127 let definition = CeremonyDefinition::new(
1128 crate::value_objects::CeremonyName::new("deferral_ceremony").unwrap(),
1129 CeremonyVersion::v1(),
1130 None,
1131 Vec::new(),
1132 Vec::new(),
1133 vec![
1134 CeremonyState::initial(state_id("drafting")),
1135 CeremonyState::terminal(state_id("done")),
1136 ],
1137 vec![finish.clone()],
1138 Vec::new(),
1139 vec![approval.clone()],
1140 vec![role(vec![RoleAction::transition(finish.trigger().clone())])],
1141 )
1142 .unwrap();
1143 let mut instance = instance(&definition);
1144
1145 instance
1146 .defer_guard(
1147 &definition,
1148 approval.name().clone(),
1149 CeremonyGuardDeferralContent::new(
1150 "I do not know.",
1151 "I cannot explain how the issue was resolved.",
1152 vec!["New evidence explains the resolution.".to_owned()],
1153 )
1154 .unwrap(),
1155 role_id("facilitator"),
1156 AuditActorKind::Human,
1157 datetime!(2026-06-06 12:01:00 UTC),
1158 )
1159 .unwrap();
1160
1161 assert!(!instance.context().is_guard_approved(approval.name()));
1162 assert!(instance
1163 .apply_transition(&definition, &trigger("approve"), now())
1164 .is_err());
1165 let deferral = &instance.guard_deferrals()[0];
1166 assert_eq!(deferral.guard_name(), approval.name());
1167 assert_eq!(deferral.content().statement(), "I do not know.");
1168 }
1169 #[test]
1172 fn approving_a_human_guard_records_the_seat_that_did_it() {
1173 let approval =
1174 CeremonyGuard::new(guard_name("human_approved"), GuardCondition::HumanApproval);
1175 let definition = definition_with_human_guard(&approval);
1176 let mut instance = instance(&definition);
1177
1178 instance
1179 .approve_guard(
1180 &definition,
1181 approval.name(),
1182 role_id("facilitator"),
1183 AuditActorKind::Human,
1184 datetime!(2026-06-06 12:01:00 UTC),
1185 )
1186 .unwrap();
1187
1188 let [recorded] = instance.guard_approvals() else {
1189 panic!(
1190 "expected one approval, got {:?}",
1191 instance.guard_approvals()
1192 );
1193 };
1194 assert_eq!(recorded.guard_name(), approval.name());
1195 assert_eq!(recorded.approved_by(), &role_id("facilitator"));
1196 assert_eq!(recorded.approved_at(), datetime!(2026-06-06 12:01:00 UTC));
1197 assert!(instance.context().is_guard_approved(approval.name()));
1198 }
1199
1200 #[test]
1204 fn a_seat_the_definition_does_not_declare_cannot_approve() {
1205 let approval =
1206 CeremonyGuard::new(guard_name("human_approved"), GuardCondition::HumanApproval);
1207 let definition = definition_with_human_guard(&approval);
1208 let mut instance = instance(&definition);
1209
1210 let outcome = instance.approve_guard(
1211 &definition,
1212 approval.name(),
1213 role_id("someone-who-is-not-here"),
1214 AuditActorKind::Human,
1215 now(),
1216 );
1217
1218 assert!(matches!(
1219 outcome,
1220 Err(DomainError::NotFound {
1221 what: "ceremony_role"
1222 })
1223 ));
1224 assert!(instance.guard_approvals().is_empty());
1225 assert!(!instance.context().is_guard_approved(approval.name()));
1226 }
1227 fn session_with_a_contribution(
1230 definition: &CeremonyDefinition,
1231 ) -> (CeremonyInstance, CeremonyInterventionId) {
1232 let mut instance = instance(definition);
1233 let agenda_item = CeremonyInterventionId::new("queue-check").unwrap();
1234 instance
1235 .request_intervention_as(
1236 definition,
1237 agenda_item.clone(),
1238 role_id("facilitator"),
1239 CeremonyInterventionKind::Investigation,
1240 CeremonyInterventionTarget::roles([role_id("observer")]).unwrap(),
1241 CeremonyInterventionContent::new("Inspect the queue.", Attributes::empty())
1242 .unwrap(),
1243 now(),
1244 )
1245 .unwrap();
1246 instance
1247 .respond_to_intervention_as(
1248 definition,
1249 &agenda_item,
1250 role_id("observer"),
1251 CeremonyInterventionContent::new("Queue depth is stable.", Attributes::empty())
1252 .unwrap(),
1253 now(),
1254 )
1255 .unwrap();
1256 (instance, agenda_item)
1257 }
1258
1259 #[test]
1262 fn a_contribution_is_recorded_as_answering_its_agenda_item() {
1263 let definition = definition();
1264 let (instance, agenda_item) = session_with_a_contribution(&definition);
1265
1266 let [answered] = instance.reasons() else {
1267 panic!("expected exactly one reason, got {:?}", instance.reasons());
1268 };
1269 assert_eq!(answered.kind(), CeremonyReasonKind::Answers);
1270 assert_eq!(
1271 answered.from(),
1272 &CeremonyRecordRef::contribution(agenda_item.clone(), 0)
1273 );
1274 assert_eq!(answered.to(), &CeremonyRecordRef::agenda_item(agenda_item));
1275 assert_eq!(
1276 answered.asserted_by(),
1277 None,
1278 "the engine observed it; naming a seat would be inventing one"
1279 );
1280 }
1281
1282 #[test]
1285 fn a_seat_cannot_assert_what_only_the_engine_observes() {
1286 let definition = definition();
1287 let (mut instance, agenda_item) = session_with_a_contribution(&definition);
1288
1289 let outcome = instance.assert_reason_as(
1290 &definition,
1291 role_id("observer"),
1292 CeremonyRecordRef::contribution(agenda_item.clone(), 0),
1293 CeremonyRecordRef::agenda_item(agenda_item),
1294 CeremonyReasonKind::Answers,
1295 "because I say it does",
1296 MemoryConfidence::High,
1297 now(),
1298 );
1299
1300 assert!(matches!(
1301 outcome,
1302 Err(DomainError::InvariantViolated { .. })
1303 ));
1304 }
1305
1306 #[test]
1309 fn only_whoever_contributed_may_say_why_they_did() {
1310 let definition = definition();
1311 let (mut instance, agenda_item) = session_with_a_contribution(&definition);
1312 let contribution = CeremonyRecordRef::contribution(agenda_item.clone(), 0);
1313 let item = CeremonyRecordRef::agenda_item(agenda_item);
1314
1315 let by_someone_else = instance.assert_reason_as(
1316 &definition,
1317 role_id("facilitator"),
1318 contribution.clone(),
1319 item.clone(),
1320 CeremonyReasonKind::ChosenBecause,
1321 "they must have thought the queue mattered",
1322 MemoryConfidence::Low,
1323 now(),
1324 );
1325 assert!(matches!(
1326 by_someone_else,
1327 Err(DomainError::InvariantViolated { .. })
1328 ));
1329
1330 instance
1331 .assert_reason_as(
1332 &definition,
1333 role_id("observer"),
1334 contribution,
1335 item,
1336 CeremonyReasonKind::ChosenBecause,
1337 "the depth graph had been flat for an hour",
1338 MemoryConfidence::High,
1339 now(),
1340 )
1341 .expect("its author may say why");
1342 assert_eq!(instance.reasons().len(), 2);
1343 }
1344
1345 #[test]
1348 fn any_seat_may_claim_that_one_thing_came_from_another() {
1349 let definition = definition();
1350 let (mut instance, agenda_item) = session_with_a_contribution(&definition);
1351
1352 instance
1353 .assert_reason_as(
1354 &definition,
1355 role_id("facilitator"),
1356 CeremonyRecordRef::agenda_item(agenda_item.clone()),
1357 CeremonyRecordRef::contribution(agenda_item, 0),
1358 CeremonyReasonKind::FollowsFrom,
1359 "the item stayed open because the answer raised a new question",
1360 MemoryConfidence::Medium,
1361 now(),
1362 )
1363 .expect("a claim about the world is open to any seat");
1364
1365 let asserted = instance.reasons().last().unwrap();
1366 assert_eq!(asserted.confidence(), MemoryConfidence::Medium);
1367 assert_eq!(asserted.asserted_by(), Some(&role_id("facilitator")));
1368 }
1369
1370 #[test]
1373 fn a_reason_cannot_cite_something_that_never_happened() {
1374 let definition = definition();
1375 let (mut instance, agenda_item) = session_with_a_contribution(&definition);
1376
1377 let outcome = instance.assert_reason_as(
1378 &definition,
1379 role_id("observer"),
1380 CeremonyRecordRef::contribution(agenda_item.clone(), 7),
1381 CeremonyRecordRef::agenda_item(agenda_item),
1382 CeremonyReasonKind::FollowsFrom,
1383 "a contribution nobody made",
1384 MemoryConfidence::Low,
1385 now(),
1386 );
1387
1388 assert!(matches!(
1389 outcome,
1390 Err(DomainError::NotFound {
1391 what: "ceremony_record"
1392 })
1393 ));
1394 }
1395
1396 #[test]
1399 fn a_move_is_recorded_with_whoever_made_it() {
1400 let approval =
1401 CeremonyGuard::new(guard_name("human_approved"), GuardCondition::HumanApproval);
1402 let definition = definition_with_human_guard(&approval);
1403 let mut instance = instance(&definition);
1404 instance
1405 .approve_guard(
1406 &definition,
1407 approval.name(),
1408 role_id("facilitator"),
1409 AuditActorKind::Human,
1410 now(),
1411 )
1412 .unwrap();
1413
1414 instance
1415 .apply_transition_as(
1416 &definition,
1417 &role_id("facilitator"),
1418 &trigger("approve"),
1419 datetime!(2026-06-06 12:05:00 UTC),
1420 )
1421 .unwrap();
1422
1423 let [moved] = instance.transitions() else {
1424 panic!("expected one move, got {:?}", instance.transitions());
1425 };
1426 assert_eq!(moved.trigger(), &trigger("approve"));
1427 assert_eq!(moved.from_state(), &state_id("drafting"));
1428 assert_eq!(moved.to_state(), &state_id("done"));
1429 assert_eq!(moved.applied_by(), Some(&role_id("facilitator")));
1430 }
1431
1432 #[test]
1436 fn a_move_the_engine_took_names_nobody() {
1437 let approval =
1438 CeremonyGuard::new(guard_name("human_approved"), GuardCondition::HumanApproval);
1439 let definition = definition_with_human_guard(&approval);
1440 let mut instance = instance(&definition);
1441 instance
1442 .approve_guard(
1443 &definition,
1444 approval.name(),
1445 role_id("facilitator"),
1446 AuditActorKind::Human,
1447 now(),
1448 )
1449 .unwrap();
1450
1451 instance
1452 .apply_transition(&definition, &trigger("approve"), now())
1453 .unwrap();
1454
1455 assert_eq!(instance.transitions()[0].applied_by(), None);
1456 }
1457 #[test]
1467 fn an_approval_records_the_kind_it_was_told_not_the_one_the_guard_wanted() {
1468 let approval =
1469 CeremonyGuard::new(guard_name("human_approved"), GuardCondition::HumanApproval);
1470 let definition = definition_with_human_guard(&approval);
1471 let mut instance = instance(&definition);
1472
1473 instance
1474 .approve_guard(
1475 &definition,
1476 approval.name(),
1477 role_id("facilitator"),
1478 AuditActorKind::Agent,
1479 now(),
1480 )
1481 .unwrap();
1482
1483 let [recorded] = instance.guard_approvals() else {
1484 panic!("expected one approval");
1485 };
1486 assert_eq!(
1487 recorded.approved_by_kind(),
1488 AuditActorKind::Agent,
1489 "the guard asked for a human and an agent answered; saying otherwise \
1490 would be the engine vouching for something it cannot see"
1491 );
1492 assert!(instance.context().is_guard_approved(approval.name()));
1493 }
1494}