made_core/entities/

ceremony_instance.rs

//! [`CeremonyInstance`] aggregate.
//!
//! Runtime state for a single ceremony execution. The aggregate owns
//! step leases, retry attempts, idempotency keys and state transitions,
//! so failover remains a domain rule instead of adapter glue.

use std::collections::{BTreeMap, BTreeSet};

use serde::{Deserialize, Serialize};
use time::OffsetDateTime;

use super::{
    ceremony_definition::CeremonyDefinition, CeremonyIntervention, PublishedCeremonyDefinition,
};
use crate::error::DomainError;
use crate::ports::CeremonyEvidenceRequest;
use crate::value_objects::{
    AuditActorKind, CeremonyContext, CeremonyDefinitionDigest, CeremonyDefinitionDigestMigration,
    CeremonyEvidenceSourceId, CeremonyGuardApproval, CeremonyGuardDeferral,
    CeremonyGuardDeferralContent, CeremonyId, CeremonyInterventionContent, CeremonyInterventionId,
    CeremonyInterventionKind, CeremonyInterventionProvenance, CeremonyInterventionResponse,
    CeremonyInterventionTarget, CeremonyName, CeremonyParticipantBinding, CeremonyReason,
    CeremonyReasonKind, CeremonyRecordRef, CeremonyTransitionRecord, CeremonyVersion,
    GuardCondition, GuardName, IdempotencyKey, MemoryConfidence, ReasonAsserter, RoleAction,
    RoleId, Specialty, StateId, StepAttempt, StepExecutionRecord, StepId, StepLease, StepResult,
    StepStatus, TransitionTrigger,
};

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CeremonyInstance {
    id: CeremonyId,
    definition_name: CeremonyName,
    definition_version: CeremonyVersion,
    current_state: StateId,
    step_records: BTreeMap<StepId, StepExecutionRecord>,
    /// Finished semantic iterations preceding each step's current record.
    ///
    /// Technical retries remain represented by their attempt number and the
    /// audit journal. Semantic repetition needs its own durable history: one
    /// successful iteration must not overwrite the output that made MADE run
    /// the next one.
    #[serde(default)]
    step_record_history: BTreeMap<StepId, Vec<StepExecutionRecord>>,
    #[serde(default)]
    interventions: Vec<CeremonyIntervention>,
    #[serde(default)]
    guard_deferrals: Vec<CeremonyGuardDeferral>,
    /// Who let each human guard through.
    ///
    /// Kept beside the context rather than inside it. The context is
    /// what a guard is evaluated against and already holds "this one
    /// is approved"; sessions written before this existed carry that
    /// and nothing else, and moving approval out of the context would
    /// have made every one of them unapproved on the next read. So the
    /// context stays the state, and this is the event that produced
    /// it.
    #[serde(default)]
    guard_approvals: Vec<CeremonyGuardApproval>,
    /// Every move this session made, in order.
    ///
    /// The current state says where a session is; this says how it got
    /// there. Without it nothing could point at a move, so nothing
    /// could say why one happened — and "why did this resolve" is the
    /// question the whole thing is for.
    #[serde(default)]
    transitions: Vec<CeremonyTransitionRecord>,
    /// Why one thing here led to another.
    ///
    /// Kept apart from the records rather than inside them, because a
    /// reason is an edge and not a field: it belongs to the pair, and
    /// putting it on either end would make it readable but not
    /// followable.
    #[serde(default)]
    reasons: Vec<CeremonyReason>,
    /// Who sits in each seat for this session, where anyone was
    /// seated. A role with no binding is played the way the definition
    /// says, which is the usual case and not a lesser one.
    #[serde(default)]
    participant_bindings: BTreeMap<RoleId, CeremonyParticipantBinding>,
    context: CeremonyContext,
    idempotency_keys: BTreeSet<IdempotencyKey>,
    #[serde(with = "time::serde::rfc3339")]
    created_at: OffsetDateTime,
    #[serde(with = "time::serde::rfc3339")]
    updated_at: OffsetDateTime,
    #[serde(with = "time::serde::rfc3339::option")]
    completed_at: Option<OffsetDateTime>,
    /// The published definition this instance is bound to, when it was
    /// started from one.
    ///
    /// Absent for an instance started from a definition handed in at
    /// the time — which is a real and useful way to work, and not the
    /// same thing. Recording which of the two happened is the point: a
    /// name and a version identify a published definition only while
    /// publication is immutable, and an instance that also carries the
    /// digest can be checked against the definition rather than trusted
    /// to have run it.
    #[serde(default)]
    bound_definition: Option<CeremonyDefinitionDigest>,
}

impl CeremonyInstance {
    /// Start from a definition supplied for this run.
    ///
    /// Nothing binds the instance to a definition that can be looked up
    /// later; that is what [`Self::start_bound`] is for.
    #[must_use]
    pub fn start(
        id: CeremonyId,
        definition: &CeremonyDefinition,
        context: CeremonyContext,
        now: OffsetDateTime,
    ) -> Self {
        Self::open(id, definition, context, now, None)
    }

    /// Start from a published definition, recording its digest.
    ///
    /// The digest travels with the instance so a later reader can
    /// verify which definition ran instead of taking the name and
    /// version on trust.
    #[must_use]
    pub fn start_bound(
        id: CeremonyId,
        published: &PublishedCeremonyDefinition,
        context: CeremonyContext,
        now: OffsetDateTime,
    ) -> Self {
        Self::open(
            id,
            published.definition(),
            context,
            now,
            Some(published.digest()),
        )
    }

    fn open(
        id: CeremonyId,
        definition: &CeremonyDefinition,
        context: CeremonyContext,
        now: OffsetDateTime,
        bound_definition: Option<CeremonyDefinitionDigest>,
    ) -> Self {
        let step_records = definition
            .steps()
            .keys()
            .map(|step_id| (step_id.clone(), StepExecutionRecord::pending()))
            .collect();

        Self {
            id,
            definition_name: definition.name().clone(),
            definition_version: definition.version().clone(),
            current_state: definition.initial_state_id().clone(),
            step_records,
            step_record_history: BTreeMap::new(),
            interventions: Vec::new(),
            guard_deferrals: Vec::new(),
            guard_approvals: Vec::new(),
            transitions: Vec::new(),
            reasons: Vec::new(),
            participant_bindings: BTreeMap::new(),
            context,
            idempotency_keys: BTreeSet::new(),
            created_at: now,
            updated_at: now,
            completed_at: None,
            bound_definition,
        }
    }

    /// The digest of the published definition this instance runs, if it
    /// was started from one.
    #[must_use]
    pub fn bound_definition(&self) -> Option<CeremonyDefinitionDigest> {
        self.bound_definition
    }

    /// Whether this instance runs a definition that can be looked up
    /// and checked, rather than one supplied for the run.
    #[must_use]
    pub fn is_bound_to_a_published_definition(&self) -> bool {
        self.bound_definition.is_some()
    }

    /// Replace a legacy publication identity after the same definition has
    /// been verified under a successor digest scheme.
    ///
    /// This is deliberately narrower than a general rebind operation. A
    /// running ceremony cannot be moved to different content, name or
    /// version. Storage migrations may only replace the expected legacy
    /// identity with the identity of the already verified publication.
    pub fn migrate_definition_binding(
        &mut self,
        migration: &CeremonyDefinitionDigestMigration,
    ) -> Result<bool, DomainError> {
        if self.definition_name != *migration.definition_name()
            || self.definition_version != *migration.definition_version()
        {
            return Err(DomainError::InvariantViolated {
                reason: "a definition binding migration cannot change name or version",
            });
        }

        match self.bound_definition {
            Some(current) if current == migration.destination() => Ok(false),
            Some(current) if current == migration.source() => {
                self.bound_definition = Some(migration.destination());
                Ok(true)
            }
            _ => Err(DomainError::InvariantViolated {
                reason: "a definition binding migration did not match the stored identity",
            }),
        }
    }

    #[must_use]
    pub fn id(&self) -> &CeremonyId {
        &self.id
    }

    #[must_use]
    pub fn definition_name(&self) -> &CeremonyName {
        &self.definition_name
    }

    #[must_use]
    pub fn definition_version(&self) -> &CeremonyVersion {
        &self.definition_version
    }

    #[must_use]
    pub fn current_state(&self) -> &StateId {
        &self.current_state
    }

    #[must_use]
    pub fn step_records(&self) -> &BTreeMap<StepId, StepExecutionRecord> {
        &self.step_records
    }

    #[must_use]
    pub fn step_record(&self, step_id: &StepId) -> Option<&StepExecutionRecord> {
        self.step_records.get(step_id)
    }

    /// Finished iterations before the current record, in execution order.
    #[must_use]
    pub fn step_record_history(&self, step_id: &StepId) -> &[StepExecutionRecord] {
        self.step_record_history
            .get(step_id)
            .map(Vec::as_slice)
            .unwrap_or_default()
    }

    /// Whether a repeating step consumed its last permitted iteration without
    /// satisfying its declared stop condition.
    #[must_use]
    pub fn step_repeat_limit_reached(
        &self,
        definition: &CeremonyDefinition,
        step_id: &StepId,
    ) -> bool {
        let Some(step) = definition.step(step_id) else {
            return false;
        };
        let Some(policy) = step.repeat_policy() else {
            return false;
        };
        let Some(record) = self.step_record(step_id) else {
            return false;
        };
        record.status().is_success()
            && !policy.is_satisfied(record.output())
            && !policy.permits_another_iteration(record.iteration())
    }

    #[must_use]
    pub fn interventions(&self) -> &[CeremonyIntervention] {
        &self.interventions
    }

    #[must_use]
    pub fn guard_deferrals(&self) -> &[CeremonyGuardDeferral] {
        &self.guard_deferrals
    }

    /// Who let each human guard through, in the order they did.
    ///
    /// Empty for a session written before approvals recorded an
    /// approver, which is the truth about those sessions rather than a
    /// gap to paper over.
    #[must_use]
    pub fn guard_approvals(&self) -> &[CeremonyGuardApproval] {
        &self.guard_approvals
    }

    /// Every move this session made, in the order it made them.
    #[must_use]
    pub fn transitions(&self) -> &[CeremonyTransitionRecord] {
        &self.transitions
    }

    /// Why one thing here led to another.
    #[must_use]
    pub fn reasons(&self) -> &[CeremonyReason] {
        &self.reasons
    }

    #[must_use]
    pub fn intervention(
        &self,
        intervention_id: &CeremonyInterventionId,
    ) -> Option<&CeremonyIntervention> {
        self.interventions
            .iter()
            .find(|intervention| intervention.id() == intervention_id)
    }

    #[must_use]
    pub fn context(&self) -> &CeremonyContext {
        &self.context
    }

    #[must_use]
    pub fn idempotency_keys(&self) -> &BTreeSet<IdempotencyKey> {
        &self.idempotency_keys
    }

    #[must_use]
    pub fn created_at(&self) -> OffsetDateTime {
        self.created_at
    }

    #[must_use]
    pub fn updated_at(&self) -> OffsetDateTime {
        self.updated_at
    }

    #[must_use]
    pub fn completed_at(&self) -> Option<OffsetDateTime> {
        self.completed_at
    }

    #[must_use]
    pub fn is_terminal(&self, definition: &CeremonyDefinition) -> bool {
        self.matches_definition(definition) && definition.is_terminal_state(&self.current_state)
    }

    #[must_use]
    pub fn is_completed(&self, definition: &CeremonyDefinition) -> bool {
        self.is_terminal(definition) && self.completed_at.is_some()
    }

    pub fn start_step_as(
        &mut self,
        definition: &CeremonyDefinition,
        role_id: &RoleId,
        step_id: &StepId,
        lease: StepLease,
        now: OffsetDateTime,
    ) -> Result<StepAttempt, DomainError> {
        self.require_role(definition, role_id, &RoleAction::step(step_id.clone()))?;
        self.start_step(definition, step_id, lease, now)
    }

    pub fn start_step(
        &mut self,
        definition: &CeremonyDefinition,
        step_id: &StepId,
        lease: StepLease,
        now: OffsetDateTime,
    ) -> Result<StepAttempt, DomainError> {
        self.require_definition(definition)?;
        if self.is_terminal(definition) {
            return Err(DomainError::InvariantViolated {
                reason: "terminal ceremony instances cannot start steps",
            });
        }

        let step = definition.step(step_id).ok_or(DomainError::NotFound {
            what: "ceremony_instance.step",
        })?;
        if step.state_id() != &self.current_state {
            return Err(DomainError::InvalidTransition {
                from: "ceremony_instance.current_state",
                to: "ceremony_step.state",
            });
        }

        let record = self
            .step_records
            .get(step_id)
            .cloned()
            .ok_or(DomainError::NotFound {
                what: "ceremony_instance.step_record",
            })?;
        if !record.can_be_started_at(now) {
            return Err(DomainError::InvariantViolated {
                reason: "step lease is still active",
            });
        }

        let next_attempt = next_attempt_for_start(&record)?;
        if !step.retry_policy().allows_attempt(next_attempt) {
            return Err(DomainError::InvariantViolated {
                reason: "step retry policy exhausted",
            });
        }
        if !self
            .idempotency_keys
            .insert(lease.idempotency_key().clone())
        {
            return Err(DomainError::AlreadyExists {
                what: "ceremony_instance.idempotency_key",
            });
        }

        self.step_records
            .insert(step_id.clone(), record.with_started(lease, next_attempt));
        self.updated_at = now;
        Ok(next_attempt)
    }

    pub fn apply_step_result(
        &mut self,
        definition: &CeremonyDefinition,
        step_id: &StepId,
        result: StepResult,
        now: OffsetDateTime,
    ) -> Result<(), DomainError> {
        self.require_definition(definition)?;
        let step = definition.step(step_id).ok_or(DomainError::NotFound {
            what: "ceremony_instance.step",
        })?;
        if step.state_id() != &self.current_state {
            return Err(DomainError::InvalidTransition {
                from: "ceremony_instance.current_state",
                to: "ceremony_step.state",
            });
        }

        let record = self
            .step_records
            .get(step_id)
            .cloned()
            .ok_or(DomainError::NotFound {
                what: "ceremony_instance.step_record",
            })?;
        if record.status() != StepStatus::InProgress {
            return Err(DomainError::InvariantViolated {
                reason: "step result requires an in-progress step",
            });
        }

        let finished = record.with_result(result);
        let repeat = step.repeat_policy().filter(|policy| {
            finished.status().is_success() && !policy.is_satisfied(finished.output())
        });
        if repeat.is_some_and(|policy| policy.permits_another_iteration(finished.iteration())) {
            let next_iteration = finished.iteration().next()?;
            self.step_record_history
                .entry(step_id.clone())
                .or_default()
                .push(finished);
            self.step_records.insert(
                step_id.clone(),
                StepExecutionRecord::pending_iteration(next_iteration),
            );
        } else {
            self.step_records.insert(step_id.clone(), finished);
        }
        self.updated_at = now;
        Ok(())
    }

    /// Approving is checked the way deferring is. It used to take no
    /// definition at all, so any name at all could be "approved" —
    /// which wrote that name into the session context, told the caller
    /// it had succeeded, and left a session that would never move.
    ///
    /// Approving ahead of time is still allowed: unlike a deferral,
    /// which answers a decision being asked for now, a person may
    /// settle a guard before the work leading up to it is finished.
    pub fn approve_guard(
        &mut self,
        definition: &CeremonyDefinition,
        guard_name: &GuardName,
        approved_by: RoleId,
        approved_by_kind: AuditActorKind,
        now: OffsetDateTime,
    ) -> Result<(), DomainError> {
        self.require_active(
            definition,
            "terminal ceremony instances cannot approve guards",
        )?;
        let guard = definition
            .guards()
            .get(guard_name)
            .ok_or(DomainError::NotFound {
                what: "ceremony_guard",
            })?;
        if !matches!(guard.condition(), GuardCondition::HumanApproval) {
            return Err(DomainError::InvariantViolated {
                reason: "only human approval guards can be approved",
            });
        }
        self.require_declared_role(definition, &approved_by)?;
        self.context = self.context.clone().with_guard_approval(guard_name)?;
        self.guard_approvals.push(CeremonyGuardApproval::record(
            guard_name.clone(),
            approved_by,
            approved_by_kind,
            now,
        ));
        self.updated_at = now;
        Ok(())
    }

    /// Seat a role for this session.
    ///
    /// Rebinding is allowed and deliberate: a panel can become
    /// unavailable halfway through a working session, and a ceremony
    /// that could not be re-seated would have to be abandoned and
    /// started again, losing everything already decided. What was
    /// seated before stays in the journal; the instance carries who is
    /// seated now, which is what the next step needs.
    pub fn bind_participant(
        &mut self,
        definition: &CeremonyDefinition,
        role_id: RoleId,
        specialty: Specialty,
        now: OffsetDateTime,
    ) -> Result<(), DomainError> {
        self.require_active(
            definition,
            "terminal ceremony instances cannot be re-seated",
        )?;
        // A seat that the ceremony never declared is not a seat.
        if definition.role(&role_id).is_none() {
            return Err(DomainError::NotFound {
                what: "ceremony_role",
            });
        }
        self.participant_bindings.insert(
            role_id.clone(),
            CeremonyParticipantBinding::record(role_id, specialty, now),
        );
        self.updated_at = now;
        Ok(())
    }

    #[must_use]
    pub fn participant_bindings(&self) -> &BTreeMap<RoleId, CeremonyParticipantBinding> {
        &self.participant_bindings
    }

    /// The specialty a role's work should be put to, if this session
    /// seated one. `None` means the definition decides, as usual.
    #[must_use]
    pub fn bound_specialty(&self, role_id: &RoleId) -> Option<&Specialty> {
        self.participant_bindings
            .get(role_id)
            .map(CeremonyParticipantBinding::specialty)
    }

    pub fn defer_guard(
        &mut self,
        definition: &CeremonyDefinition,
        guard_name: GuardName,
        content: CeremonyGuardDeferralContent,
        deferred_by: RoleId,
        deferred_by_kind: AuditActorKind,
        now: OffsetDateTime,
    ) -> Result<(), DomainError> {
        self.require_active(
            definition,
            "terminal ceremony instances cannot defer guard decisions",
        )?;
        let guard = definition
            .guards()
            .get(&guard_name)
            .ok_or(DomainError::NotFound {
                what: "ceremony_guard",
            })?;
        if !matches!(guard.condition(), GuardCondition::HumanApproval) {
            return Err(DomainError::InvariantViolated {
                reason: "only human approval guards can be deferred",
            });
        }
        if self.context.is_guard_approved(&guard_name) {
            return Err(DomainError::InvariantViolated {
                reason: "approved human guards cannot be deferred",
            });
        }
        let is_currently_required = definition
            .available_transitions(&self.current_state)
            .any(|transition| transition.required_guards().contains(&guard_name));
        if !is_currently_required {
            return Err(DomainError::InvariantViolated {
                reason: "human guard is not required from the current state",
            });
        }

        self.require_declared_role(definition, &deferred_by)?;
        self.guard_deferrals.push(CeremonyGuardDeferral::record(
            guard_name,
            deferred_by,
            deferred_by_kind,
            content,
            now,
        ));
        self.updated_at = now;
        Ok(())
    }

    #[allow(clippy::too_many_arguments)]
    pub fn request_intervention_as(
        &mut self,
        definition: &CeremonyDefinition,
        intervention_id: CeremonyInterventionId,
        role_id: RoleId,
        kind: CeremonyInterventionKind,
        target: CeremonyInterventionTarget,
        content: CeremonyInterventionContent,
        now: OffsetDateTime,
    ) -> Result<(), DomainError> {
        self.request_intervention_with_provenance_as(
            definition,
            intervention_id,
            role_id,
            kind,
            target,
            content,
            None,
            now,
        )
    }

    #[allow(clippy::too_many_arguments)]
    pub fn request_intervention_with_provenance_as(
        &mut self,
        definition: &CeremonyDefinition,
        intervention_id: CeremonyInterventionId,
        role_id: RoleId,
        kind: CeremonyInterventionKind,
        target: CeremonyInterventionTarget,
        content: CeremonyInterventionContent,
        provenance: Option<CeremonyInterventionProvenance>,
        now: OffsetDateTime,
    ) -> Result<(), DomainError> {
        self.require_active(
            definition,
            "terminal ceremony instances cannot accept interventions",
        )?;
        self.require_role(definition, &role_id, &RoleAction::request_intervention())?;
        Self::require_intervention_target(definition, &target)?;
        if let Some(provenance) = provenance.as_ref() {
            self.require_intervention_provenance(definition, &role_id, &target, provenance)?;
        }
        if self
            .interventions
            .iter()
            .any(|intervention| intervention.id() == &intervention_id)
        {
            return Err(DomainError::AlreadyExists {
                what: "ceremony_intervention",
            });
        }
        let intervention = CeremonyIntervention::open_with_provenance(
            intervention_id,
            kind,
            role_id,
            target,
            content,
            provenance,
            now,
        );
        self.interventions.push(intervention);
        self.updated_at = now;
        Ok(())
    }

    pub fn respond_to_intervention_as(
        &mut self,
        definition: &CeremonyDefinition,
        intervention_id: &CeremonyInterventionId,
        role_id: RoleId,
        content: CeremonyInterventionContent,
        now: OffsetDateTime,
    ) -> Result<(), DomainError> {
        self.require_active(
            definition,
            "terminal ceremony instances cannot receive intervention responses",
        )?;
        self.require_role(definition, &role_id, &RoleAction::respond_to_intervention())?;
        self.interventions
            .iter_mut()
            .find(|intervention| intervention.id() == intervention_id)
            .ok_or(DomainError::NotFound {
                what: "ceremony_intervention",
            })?
            .respond(role_id, content, now)?;
        self.record_that_it_answers(intervention_id, now);
        self.updated_at = now;
        Ok(())
    }

    pub fn prepare_evidence_request_as(
        &self,
        definition: &CeremonyDefinition,
        intervention_id: CeremonyInterventionId,
        role_id: RoleId,
        source_id: CeremonyEvidenceSourceId,
        query: CeremonyInterventionContent,
    ) -> Result<CeremonyEvidenceRequest, DomainError> {
        self.require_active(
            definition,
            "terminal ceremony instances cannot collect intervention evidence",
        )?;
        self.require_role(definition, &role_id, &RoleAction::respond_to_intervention())?;
        self.intervention(&intervention_id)
            .ok_or(DomainError::NotFound {
                what: "ceremony_intervention",
            })?
            .ensure_can_respond(&role_id)?;
        Ok(CeremonyEvidenceRequest::new(
            self.id.clone(),
            intervention_id,
            role_id,
            source_id,
            query,
            self.context.clone(),
        ))
    }

    pub fn respond_to_intervention_with_evidence_as(
        &mut self,
        definition: &CeremonyDefinition,
        intervention_id: &CeremonyInterventionId,
        role_id: RoleId,
        evidence_pack: super::CeremonyEvidencePack,
        now: OffsetDateTime,
    ) -> Result<(), DomainError> {
        self.require_active(
            definition,
            "terminal ceremony instances cannot receive intervention evidence",
        )?;
        self.require_role(definition, &role_id, &RoleAction::respond_to_intervention())?;
        self.interventions
            .iter_mut()
            .find(|intervention| intervention.id() == intervention_id)
            .ok_or(DomainError::NotFound {
                what: "ceremony_intervention",
            })?
            .respond_with_evidence(role_id, evidence_pack, now)?;
        self.record_that_it_answers(intervention_id, now);
        self.updated_at = now;
        Ok(())
    }

    /// State why one thing here led to another.
    ///
    /// Its own act rather than a field on contributing, because a
    /// reason is often known later — "in fact I did that because…" is
    /// how people reason — and because a field gets filled in by
    /// inertia while an act is chosen.
    ///
    /// What it refuses is the point:
    ///
    /// - a kind only the engine may assert, because a participant able
    ///   to relabel the structure could rewrite the session's shape;
    /// - a kind only an author may assert, claimed by anyone else,
    ///   because nobody else has access to another's reasoning;
    /// - either end naming something this session never produced.
    #[allow(clippy::too_many_arguments)]
    pub fn assert_reason_as(
        &mut self,
        definition: &CeremonyDefinition,
        role_id: RoleId,
        from: CeremonyRecordRef,
        to: CeremonyRecordRef,
        kind: CeremonyReasonKind,
        why: impl Into<String>,
        confidence: MemoryConfidence,
        now: OffsetDateTime,
    ) -> Result<(), DomainError> {
        self.require_declared_role(definition, &role_id)?;
        self.require_record(&from)?;
        self.require_record(&to)?;

        match kind.asserter() {
            ReasonAsserter::TheEngine => {
                return Err(DomainError::InvariantViolated {
                    reason:
                        "this kind of reason states the shape of the session, not a judgement, \
                             and only the engine may assert it",
                });
            }
            ReasonAsserter::ItsAuthor => {
                if self.author_of(&from) != Some(&role_id) {
                    return Err(DomainError::InvariantViolated {
                        reason: "only whoever produced something may say why they decided it or \
                                 how they did it",
                    });
                }
            }
            ReasonAsserter::AnySeat => {}
        }

        self.reasons.push(CeremonyReason::new(
            from,
            to,
            kind,
            why,
            confidence,
            Some(role_id),
            now,
        )?);
        self.updated_at = now;
        Ok(())
    }

    /// Who produced a record, where anyone did.
    ///
    /// A step has none: the engine ran it. A transition the engine
    /// took has none either. Both are absences rather than gaps, and
    /// they are what stops a reason of testimony being made about
    /// something nobody can testify to.
    fn author_of(&self, record: &CeremonyRecordRef) -> Option<&RoleId> {
        match record {
            CeremonyRecordRef::Step { .. } => None,
            CeremonyRecordRef::AgendaItem { agenda_item } => self
                .intervention(agenda_item)
                .map(CeremonyIntervention::requested_by),
            CeremonyRecordRef::Contribution {
                agenda_item,
                ordinal,
            } => self
                .intervention(agenda_item)
                .and_then(|item| item.responses().get(*ordinal as usize))
                .map(CeremonyInterventionResponse::role_id),
            CeremonyRecordRef::GuardDecision { guard_name } => self
                .guard_approvals
                .iter()
                .find(|approval| approval.guard_name() == guard_name)
                .map(CeremonyGuardApproval::approved_by)
                .or_else(|| {
                    self.guard_deferrals
                        .iter()
                        .find(|deferral| deferral.guard_name() == guard_name)
                        .map(CeremonyGuardDeferral::deferred_by)
                }),
            CeremonyRecordRef::Transition { ordinal } => self
                .transitions
                .get(ordinal.saturating_sub(1) as usize)
                .and_then(CeremonyTransitionRecord::applied_by),
        }
    }

    /// A record this session actually produced.
    ///
    /// Memory cannot check this — an edge there may reach something
    /// written an hour ago — but a session knows everything it has
    /// done, and letting a reason cite what never happened would be
    /// declining to use the one advantage it has.
    fn require_record(&self, record: &CeremonyRecordRef) -> Result<(), DomainError> {
        let exists = match record {
            CeremonyRecordRef::Step { step_id } => self.step_records.contains_key(step_id),
            CeremonyRecordRef::AgendaItem { agenda_item } => {
                self.intervention(agenda_item).is_some()
            }
            CeremonyRecordRef::Contribution {
                agenda_item,
                ordinal,
            } => self
                .intervention(agenda_item)
                .is_some_and(|item| item.responses().len() > *ordinal as usize),
            CeremonyRecordRef::GuardDecision { guard_name } => {
                self.guard_approvals
                    .iter()
                    .any(|approval| approval.guard_name() == guard_name)
                    || self
                        .guard_deferrals
                        .iter()
                        .any(|deferral| deferral.guard_name() == guard_name)
            }
            CeremonyRecordRef::Transition { ordinal } => {
                *ordinal >= 1 && (*ordinal as usize) <= self.transitions.len()
            }
        };
        if exists {
            Ok(())
        } else {
            Err(DomainError::NotFound {
                what: "ceremony_record",
            })
        }
    }

    /// The reason the engine can see on its own: a contribution is the
    /// reply to the item it was made against.
    ///
    /// The only kind it asserts. Everything explanatory comes from
    /// whoever reasoned, because a session ending well after an action
    /// is not the action having worked.
    fn record_that_it_answers(
        &mut self,
        agenda_item: &CeremonyInterventionId,
        now: OffsetDateTime,
    ) {
        let Some(ordinal) = self
            .intervention(agenda_item)
            .map(|item| item.responses().len())
            .and_then(|count| u32::try_from(count.checked_sub(1)?).ok())
        else {
            return;
        };
        if let Ok(reason) = CeremonyReason::new(
            CeremonyRecordRef::contribution(agenda_item.clone(), ordinal),
            CeremonyRecordRef::agenda_item(agenda_item.clone()),
            CeremonyReasonKind::Answers,
            "a contribution made against this agenda item",
            MemoryConfidence::High,
            None,
            now,
        ) {
            self.reasons.push(reason);
        }
    }

    pub fn close_intervention_as(
        &mut self,
        definition: &CeremonyDefinition,
        intervention_id: &CeremonyInterventionId,
        role_id: &RoleId,
        now: OffsetDateTime,
    ) -> Result<(), DomainError> {
        self.require_active(
            definition,
            "terminal ceremony instances cannot close interventions",
        )?;
        self.require_role(definition, role_id, &RoleAction::request_intervention())?;
        self.interventions
            .iter_mut()
            .find(|intervention| intervention.id() == intervention_id)
            .ok_or(DomainError::NotFound {
                what: "ceremony_intervention",
            })?
            .close(role_id, now)?;
        self.updated_at = now;
        Ok(())
    }

    pub fn apply_transition_as(
        &mut self,
        definition: &CeremonyDefinition,
        role_id: &RoleId,
        trigger: &TransitionTrigger,
        now: OffsetDateTime,
    ) -> Result<StateId, DomainError> {
        self.require_role(
            definition,
            role_id,
            &RoleAction::transition(trigger.clone()),
        )?;
        self.move_on(definition, trigger, Some(role_id.clone()), now)
    }

    pub fn apply_transition(
        &mut self,
        definition: &CeremonyDefinition,
        trigger: &TransitionTrigger,
        now: OffsetDateTime,
    ) -> Result<StateId, DomainError> {
        self.move_on(definition, trigger, None, now)
    }

    /// The one place a session moves.
    ///
    /// `applied_by` is absent when the engine took the move itself,
    /// and naming somebody would be inventing them.
    fn move_on(
        &mut self,
        definition: &CeremonyDefinition,
        trigger: &TransitionTrigger,
        applied_by: Option<RoleId>,
        now: OffsetDateTime,
    ) -> Result<StateId, DomainError> {
        self.require_definition(definition)?;
        if self.is_terminal(definition) {
            return Err(DomainError::InvariantViolated {
                reason: "terminal ceremony instances cannot transition",
            });
        }

        let transition = definition
            .transition_for_trigger(&self.current_state, trigger)
            .ok_or(DomainError::InvalidTransition {
                from: "ceremony_instance.current_state",
                to: "transition_trigger",
            })?;
        if !definition.repeat_requirements_are_satisfied(&self.current_state, &self.step_records) {
            return Err(DomainError::InvariantViolated {
                reason: "ceremony step repeat condition is not satisfied",
            });
        }
        if !definition.guards_are_satisfied(transition, &self.step_records, &self.context) {
            return Err(DomainError::InvariantViolated {
                reason: "ceremony transition guards are not satisfied",
            });
        }

        let from_state = self.current_state.clone();
        self.current_state = transition.to().clone();
        self.transitions.push(CeremonyTransitionRecord::record(
            trigger.clone(),
            from_state,
            self.current_state.clone(),
            applied_by,
            now,
        ));
        self.updated_at = now;
        if definition.is_terminal_state(&self.current_state) {
            self.completed_at = Some(now);
        }
        Ok(self.current_state.clone())
    }

    fn matches_definition(&self, definition: &CeremonyDefinition) -> bool {
        self.definition_name == *definition.name()
            && self.definition_version == *definition.version()
    }

    fn require_definition(&self, definition: &CeremonyDefinition) -> Result<(), DomainError> {
        if self.matches_definition(definition) {
            Ok(())
        } else {
            Err(DomainError::InvariantViolated {
                reason: "ceremony instance definition mismatch",
            })
        }
    }

    fn require_active(
        &self,
        definition: &CeremonyDefinition,
        terminal_reason: &'static str,
    ) -> Result<(), DomainError> {
        self.require_definition(definition)?;
        if self.is_terminal(definition) {
            Err(DomainError::InvariantViolated {
                reason: terminal_reason,
            })
        } else {
            Ok(())
        }
    }

    fn require_intervention_target(
        definition: &CeremonyDefinition,
        target: &CeremonyInterventionTarget,
    ) -> Result<(), DomainError> {
        let Some(role_ids) = target.role_ids() else {
            return Ok(());
        };
        for role_id in role_ids {
            if definition.role(role_id).is_none() {
                return Err(DomainError::NotFound {
                    what: "ceremony_intervention.target_role",
                });
            }
            if !definition.role_allows(role_id, &RoleAction::respond_to_intervention()) {
                return Err(DomainError::InvariantViolated {
                    reason: "target role cannot respond to ceremony interventions",
                });
            }
        }
        Ok(())
    }

    fn require_intervention_provenance(
        &self,
        definition: &CeremonyDefinition,
        requested_by: &RoleId,
        target: &CeremonyInterventionTarget,
        provenance: &CeremonyInterventionProvenance,
    ) -> Result<(), DomainError> {
        let source = self
            .intervention(provenance.source_intervention_id())
            .ok_or(DomainError::NotFound {
                what: "ceremony_intervention.provenance_source",
            })?;
        if source.requested_by() != requested_by {
            return Err(DomainError::InvariantViolated {
                reason: "only the source requester can select an intervention response",
            });
        }
        if !source
            .responses()
            .iter()
            .any(|response| response.role_id() == provenance.source_response_role_id())
        {
            return Err(DomainError::NotFound {
                what: "ceremony_intervention.provenance_response",
            });
        }
        if definition.role(provenance.selected_role_id()).is_none() {
            return Err(DomainError::NotFound {
                what: "ceremony_intervention.provenance_selected_role",
            });
        }
        if !definition.role_allows(
            provenance.selected_role_id(),
            &RoleAction::respond_to_intervention(),
        ) {
            return Err(DomainError::InvariantViolated {
                reason: "selected intervention role cannot respond",
            });
        }
        if !target.accepts(provenance.selected_role_id()) {
            return Err(DomainError::InvariantViolated {
                reason: "intervention target does not include the selected role",
            });
        }
        Ok(())
    }

    /// A seat this session's definition declares.
    ///
    /// Weaker on purpose than [`Self::require_role`]: a definition says
    /// which roles may run a step or fire a transition, and says
    /// nothing about who may approve a human guard. Demanding a
    /// capability that no definition grants would leave every existing
    /// ceremony with no one able to approve anything. Which seats may
    /// decide a guard is a question for whenever guards grow an
    /// authority model; until then, being a seat at this table is the
    /// check, and it is enough to make the receipt name someone.
    fn require_declared_role(
        &self,
        definition: &CeremonyDefinition,
        role_id: &RoleId,
    ) -> Result<(), DomainError> {
        self.require_definition(definition)?;
        if definition.role(role_id).is_some() {
            Ok(())
        } else {
            Err(DomainError::NotFound {
                what: "ceremony_role",
            })
        }
    }

    fn require_role(
        &self,
        definition: &CeremonyDefinition,
        role_id: &RoleId,
        action: &RoleAction,
    ) -> Result<(), DomainError> {
        self.require_definition(definition)?;
        if definition.role_allows(role_id, action) {
            Ok(())
        } else {
            Err(DomainError::InvariantViolated {
                reason: "ceremony role is not allowed to perform action",
            })
        }
    }
}

fn next_attempt_for_start(record: &StepExecutionRecord) -> Result<StepAttempt, DomainError> {
    if matches!(record.status(), StepStatus::Failed | StepStatus::InProgress) {
        record.attempt().next()
    } else {
        Ok(record.attempt())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::value_objects::{
        Attributes, CeremonyGuard, CeremonyState, CeremonyStep, CeremonyTransition, GuardCondition,
        GuardName, LeaseOwnerId, RepeatUntilCondition, RetryPolicy, StepHandlerConfig,
        StepHandlerKind, StepIteration, StepOutput, StepOutputField, StepRepeatPolicy,
    };
    use serde_json::json;
    use time::macros::datetime;

    fn now() -> OffsetDateTime {
        datetime!(2026-06-06 12:00:00 UTC)
    }

    fn state_id(raw: &str) -> StateId {
        StateId::new(raw).unwrap()
    }

    fn step_id(raw: &str) -> StepId {
        StepId::new(raw).unwrap()
    }

    fn trigger(raw: &str) -> TransitionTrigger {
        TransitionTrigger::new(raw).unwrap()
    }

    fn role_id(raw: &str) -> RoleId {
        RoleId::new(raw).unwrap()
    }

    fn guard_name(raw: &str) -> GuardName {
        GuardName::new(raw).unwrap()
    }

    fn handler_kind() -> StepHandlerKind {
        StepHandlerKind::new("multiagent_round").unwrap()
    }

    fn retrying_step(raw_step_id: &str, raw_state_id: &str) -> CeremonyStep {
        CeremonyStep::new(
            step_id(raw_step_id),
            state_id(raw_state_id),
            handler_kind(),
            StepHandlerConfig::empty(),
            RetryPolicy::new(
                StepAttempt::new(3).unwrap(),
                crate::value_objects::DurationMs::ZERO,
            ),
            None,
        )
    }

    fn single_attempt_step(raw_step_id: &str, raw_state_id: &str) -> CeremonyStep {
        CeremonyStep::new(
            step_id(raw_step_id),
            state_id(raw_state_id),
            handler_kind(),
            StepHandlerConfig::empty(),
            RetryPolicy::single_attempt(),
            None,
        )
    }

    fn repeating_plan(max_iterations: u32) -> CeremonyStep {
        retrying_step("plan", "drafting").with_repeat_policy(StepRepeatPolicy::new(
            RepeatUntilCondition::output_field_equals(
                StepOutputField::new("ready").unwrap(),
                json!(true),
            ),
            StepIteration::new(max_iterations).unwrap(),
        ))
    }

    fn readiness_output(ready: bool) -> StepOutput {
        StepOutput::new(
            Attributes::new(std::collections::BTreeMap::from([(
                "ready".to_owned(),
                json!(ready),
            )]))
            .unwrap(),
        )
    }

    fn lease(
        raw_owner_id: &str,
        raw_key: &str,
        acquired_at: OffsetDateTime,
        expires_at: OffsetDateTime,
    ) -> StepLease {
        StepLease::new(
            LeaseOwnerId::new(raw_owner_id).unwrap(),
            IdempotencyKey::new(raw_key).unwrap(),
            acquired_at,
            expires_at,
        )
        .unwrap()
    }

    fn role(actions: Vec<RoleAction>) -> crate::value_objects::CeremonyRole {
        crate::value_objects::CeremonyRole::new(role_id("facilitator"), actions).unwrap()
    }

    fn definition_with_steps(steps: Vec<CeremonyStep>) -> CeremonyDefinition {
        let plan_done = CeremonyGuard::new(
            guard_name("plan_done"),
            GuardCondition::StepStatus {
                step_id: step_id("plan"),
                status: StepStatus::Completed,
            },
        );
        let finish = CeremonyTransition::new(
            state_id("drafting"),
            state_id("done"),
            trigger("finish"),
            vec![plan_done.name().clone()],
        )
        .unwrap();
        let role = role(vec![
            RoleAction::step(step_id("plan")),
            RoleAction::transition(finish.trigger().clone()),
            RoleAction::request_intervention(),
        ]);
        let observer = crate::value_objects::CeremonyRole::new(
            role_id("observer"),
            vec![RoleAction::respond_to_intervention()],
        )
        .unwrap();

        CeremonyDefinition::new(
            crate::value_objects::CeremonyName::new("planning_ceremony").unwrap(),
            CeremonyVersion::v1(),
            None,
            Vec::new(),
            Vec::new(),
            vec![
                CeremonyState::initial(state_id("drafting")),
                CeremonyState::intermediate(state_id("review")),
                CeremonyState::terminal(state_id("done")),
            ],
            vec![finish],
            steps,
            vec![plan_done],
            vec![role, observer],
        )
        .unwrap()
    }

    fn definition() -> CeremonyDefinition {
        definition_with_steps(vec![
            retrying_step("plan", "drafting"),
            single_attempt_step("review_step", "review"),
        ])
    }

    #[test]
    fn a_verified_digest_migration_rebinds_only_its_exact_definition() {
        let definition = definition();
        let published = PublishedCeremonyDefinition::seal(definition.clone()).unwrap();
        let migration = definition.choreographer_v1_digest_migration().unwrap();
        let mut value = serde_json::to_value(CeremonyInstance::start_bound(
            CeremonyId::new("legacy-bound").unwrap(),
            &published,
            CeremonyContext::empty(),
            now(),
        ))
        .unwrap();
        value["bound_definition"] = serde_json::to_value(migration.source()).unwrap();
        let mut instance: CeremonyInstance = serde_json::from_value(value).unwrap();

        assert!(instance.migrate_definition_binding(&migration).unwrap());
        assert_eq!(instance.bound_definition(), Some(migration.destination()));
        assert!(!instance.migrate_definition_binding(&migration).unwrap());
    }

    #[test]
    fn a_digest_migration_for_another_definition_is_rejected() {
        let definition = definition();
        let published = PublishedCeremonyDefinition::seal(definition.clone()).unwrap();
        let mut instance = CeremonyInstance::start_bound(
            CeremonyId::new("still-bound").unwrap(),
            &published,
            CeremonyContext::empty(),
            now(),
        );
        let other = CeremonyDefinition::new(
            CeremonyName::new("another_ceremony").unwrap(),
            CeremonyVersion::v1(),
            None,
            [],
            [],
            [CeremonyState::initial(state_id("OPEN"))],
            [],
            [],
            [],
            [],
        )
        .unwrap()
        .choreographer_v1_digest_migration()
        .unwrap();

        assert!(instance.migrate_definition_binding(&other).is_err());
        assert_eq!(instance.bound_definition(), Some(published.digest()));
    }

    /// The smallest ceremony that waits on a person: one guard, one
    /// transition it blocks, one seat allowed to fire it.
    fn definition_with_human_guard(approval: &CeremonyGuard) -> CeremonyDefinition {
        let finish = CeremonyTransition::new(
            state_id("drafting"),
            state_id("done"),
            trigger("approve"),
            vec![approval.name().clone()],
        )
        .unwrap();
        CeremonyDefinition::new(
            crate::value_objects::CeremonyName::new("approval_ceremony").unwrap(),
            CeremonyVersion::v1(),
            None,
            Vec::new(),
            Vec::new(),
            vec![
                CeremonyState::initial(state_id("drafting")),
                CeremonyState::terminal(state_id("done")),
            ],
            vec![finish.clone()],
            Vec::new(),
            vec![approval.clone()],
            vec![role(vec![RoleAction::transition(finish.trigger().clone())])],
        )
        .unwrap()
    }

    fn instance(definition: &CeremonyDefinition) -> CeremonyInstance {
        CeremonyInstance::start(
            CeremonyId::new("ceremony-1").unwrap(),
            definition,
            CeremonyContext::empty(),
            now(),
        )
    }

    #[test]
    fn starts_in_initial_state_with_pending_records() {
        let definition = definition();
        let instance = instance(&definition);

        assert_eq!(instance.current_state(), &state_id("drafting"));
        assert_eq!(
            instance.step_record(&step_id("plan")).unwrap().status(),
            StepStatus::Pending
        );
        assert_eq!(
            instance
                .step_record(&step_id("review_step"))
                .unwrap()
                .status(),
            StepStatus::Pending
        );
    }

    #[test]
    fn instances_without_iteration_fields_load_as_the_first_iteration() {
        let definition = definition();
        let mut value = serde_json::to_value(instance(&definition)).unwrap();
        value.as_object_mut().unwrap().remove("step_record_history");
        for record in value["step_records"].as_object_mut().unwrap().values_mut() {
            record.as_object_mut().unwrap().remove("iteration");
        }

        let restored: CeremonyInstance = serde_json::from_value(value).unwrap();

        assert!(restored.step_record_history(&step_id("plan")).is_empty());
        assert_eq!(
            restored.step_record(&step_id("plan")).unwrap().iteration(),
            StepIteration::FIRST
        );
    }

    #[test]
    fn dynamic_intervention_collects_role_scoped_response_and_requester_closes_it() {
        let definition = definition();
        let mut instance = instance(&definition);
        let intervention_id = CeremonyInterventionId::new("queue-check").unwrap();
        let facilitator = role_id("facilitator");
        let observer = role_id("observer");

        instance
            .request_intervention_as(
                &definition,
                intervention_id.clone(),
                facilitator.clone(),
                CeremonyInterventionKind::Investigation,
                CeremonyInterventionTarget::roles([observer.clone()]).unwrap(),
                CeremonyInterventionContent::new(
                    "Inspect the queue without consuming messages.",
                    Attributes::empty(),
                )
                .unwrap(),
                now(),
            )
            .unwrap();
        instance
            .respond_to_intervention_as(
                &definition,
                &intervention_id,
                observer.clone(),
                CeremonyInterventionContent::new("Queue depth is stable.", Attributes::empty())
                    .unwrap(),
                now(),
            )
            .unwrap();
        let selected_intervention_id = CeremonyInterventionId::new("selected-check").unwrap();
        instance
            .request_intervention_with_provenance_as(
                &definition,
                selected_intervention_id.clone(),
                facilitator.clone(),
                CeremonyInterventionKind::Investigation,
                CeremonyInterventionTarget::roles([observer.clone()]).unwrap(),
                CeremonyInterventionContent::new(
                    "Inspect the proposed signal.",
                    Attributes::empty(),
                )
                .unwrap(),
                Some(CeremonyInterventionProvenance::selected_from(
                    intervention_id.clone(),
                    observer.clone(),
                    observer.clone(),
                )),
                now(),
            )
            .unwrap();
        instance
            .close_intervention_as(&definition, &intervention_id, &facilitator, now())
            .unwrap();

        let intervention = instance.intervention(&intervention_id).unwrap();
        assert_eq!(intervention.responses().len(), 1);
        assert_eq!(
            intervention.status(),
            crate::value_objects::CeremonyInterventionStatus::Closed
        );
        let provenance = instance
            .intervention(&selected_intervention_id)
            .unwrap()
            .provenance()
            .unwrap();
        assert_eq!(provenance.source_intervention_id(), &intervention_id);
        assert_eq!(provenance.selected_role_id(), &observer);
    }

    #[test]
    fn intervention_rejects_roles_without_the_required_capability() {
        let definition = definition();
        let mut instance = instance(&definition);

        let error = instance
            .request_intervention_as(
                &definition,
                CeremonyInterventionId::new("not-allowed").unwrap(),
                role_id("observer"),
                CeremonyInterventionKind::Opinion,
                CeremonyInterventionTarget::table(),
                CeremonyInterventionContent::new("What do you think?", Attributes::empty())
                    .unwrap(),
                now(),
            )
            .unwrap_err();

        assert!(matches!(error, DomainError::InvariantViolated { .. }));
    }

    #[test]
    fn rejects_step_execution_outside_current_state() {
        let definition = definition();
        let mut instance = instance(&definition);

        let err = instance
            .start_step(
                &definition,
                &step_id("review_step"),
                lease(
                    "runner-1",
                    "key-1",
                    now(),
                    datetime!(2026-06-06 12:05:00 UTC),
                ),
                now(),
            )
            .unwrap_err();

        assert!(matches!(err, DomainError::InvalidTransition { .. }));
    }

    #[test]
    fn completed_step_unlocks_guarded_transition() {
        let definition = definition();
        let mut instance = instance(&definition);

        instance
            .start_step_as(
                &definition,
                &role_id("facilitator"),
                &step_id("plan"),
                lease(
                    "runner-1",
                    "key-1",
                    now(),
                    datetime!(2026-06-06 12:05:00 UTC),
                ),
                now(),
            )
            .unwrap();
        instance
            .apply_step_result(
                &definition,
                &step_id("plan"),
                StepResult::completed(StepOutput::empty()).unwrap(),
                datetime!(2026-06-06 12:01:00 UTC),
            )
            .unwrap();
        let state = instance
            .apply_transition_as(
                &definition,
                &role_id("facilitator"),
                &trigger("finish"),
                datetime!(2026-06-06 12:02:00 UTC),
            )
            .unwrap();

        assert_eq!(state, state_id("done"));
        assert!(instance.is_completed(&definition));
    }

    #[test]
    fn false_repeat_condition_archives_iteration_and_schedules_the_next() {
        let definition = definition_with_steps(vec![repeating_plan(3)]);
        let mut instance = instance(&definition);

        instance
            .start_step(
                &definition,
                &step_id("plan"),
                lease(
                    "runner-1",
                    "repeat-1",
                    now(),
                    datetime!(2026-06-06 12:05:00 UTC),
                ),
                now(),
            )
            .unwrap();
        instance
            .apply_step_result(
                &definition,
                &step_id("plan"),
                StepResult::completed(readiness_output(false)).unwrap(),
                datetime!(2026-06-06 12:01:00 UTC),
            )
            .unwrap();

        let current = instance.step_record(&step_id("plan")).unwrap();
        assert_eq!(current.status(), StepStatus::Pending);
        assert_eq!(current.iteration().get(), 2);
        assert_eq!(current.attempt(), StepAttempt::FIRST);
        let history = instance.step_record_history(&step_id("plan"));
        assert_eq!(history.len(), 1);
        assert_eq!(history[0].iteration(), StepIteration::FIRST);
        assert_eq!(history[0].output(), &readiness_output(false));
        assert!(instance
            .apply_transition(&definition, &trigger("finish"), now())
            .is_err());

        instance
            .start_step(
                &definition,
                &step_id("plan"),
                lease(
                    "runner-1",
                    "repeat-2",
                    datetime!(2026-06-06 12:02:00 UTC),
                    datetime!(2026-06-06 12:07:00 UTC),
                ),
                datetime!(2026-06-06 12:02:00 UTC),
            )
            .unwrap();
        instance
            .apply_step_result(
                &definition,
                &step_id("plan"),
                StepResult::completed(readiness_output(true)).unwrap(),
                datetime!(2026-06-06 12:03:00 UTC),
            )
            .unwrap();

        let current = instance.step_record(&step_id("plan")).unwrap();
        assert_eq!(current.status(), StepStatus::Completed);
        assert_eq!(current.iteration().get(), 2);
        assert!(!instance.step_repeat_limit_reached(&definition, &step_id("plan")));
        assert_eq!(
            instance
                .apply_transition(&definition, &trigger("finish"), now())
                .unwrap(),
            state_id("done")
        );
    }

    #[test]
    fn repeat_limit_is_terminal_for_the_step_and_blocks_transition() {
        let definition = definition_with_steps(vec![repeating_plan(2)]);
        let mut instance = instance(&definition);

        for iteration in 1..=2 {
            instance
                .start_step(
                    &definition,
                    &step_id("plan"),
                    lease(
                        "runner-1",
                        &format!("limit-{iteration}"),
                        now(),
                        datetime!(2026-06-06 12:05:00 UTC),
                    ),
                    now(),
                )
                .unwrap();
            instance
                .apply_step_result(
                    &definition,
                    &step_id("plan"),
                    StepResult::completed(readiness_output(false)).unwrap(),
                    now(),
                )
                .unwrap();
        }

        assert!(instance.step_repeat_limit_reached(&definition, &step_id("plan")));
        assert_eq!(
            instance
                .step_record(&step_id("plan"))
                .unwrap()
                .iteration()
                .get(),
            2
        );
        assert_eq!(instance.step_record_history(&step_id("plan")).len(), 1);
        assert!(instance
            .apply_transition(&definition, &trigger("finish"), now())
            .is_err());
        assert!(instance
            .start_step(
                &definition,
                &step_id("plan"),
                lease(
                    "runner-1",
                    "limit-3",
                    now(),
                    datetime!(2026-06-06 12:05:00 UTC),
                ),
                now(),
            )
            .is_err());
    }

    #[test]
    fn active_lease_blocks_failover_takeover() {
        let definition = definition();
        let mut instance = instance(&definition);

        instance
            .start_step(
                &definition,
                &step_id("plan"),
                lease(
                    "runner-1",
                    "key-1",
                    now(),
                    datetime!(2026-06-06 12:05:00 UTC),
                ),
                now(),
            )
            .unwrap();
        let err = instance
            .start_step(
                &definition,
                &step_id("plan"),
                lease(
                    "runner-2",
                    "key-2",
                    datetime!(2026-06-06 12:01:00 UTC),
                    datetime!(2026-06-06 12:06:00 UTC),
                ),
                datetime!(2026-06-06 12:01:00 UTC),
            )
            .unwrap_err();

        assert!(matches!(err, DomainError::InvariantViolated { .. }));
        assert_eq!(
            instance
                .step_record(&step_id("plan"))
                .unwrap()
                .lease()
                .unwrap()
                .owner_id()
                .as_str(),
            "runner-1"
        );
    }

    #[test]
    fn expired_lease_allows_failover_takeover_with_next_attempt() {
        let definition = definition();
        let mut instance = instance(&definition);

        instance
            .start_step(
                &definition,
                &step_id("plan"),
                lease(
                    "runner-1",
                    "key-1",
                    now(),
                    datetime!(2026-06-06 12:05:00 UTC),
                ),
                now(),
            )
            .unwrap();
        let attempt = instance
            .start_step(
                &definition,
                &step_id("plan"),
                lease(
                    "runner-2",
                    "key-2",
                    datetime!(2026-06-06 12:06:00 UTC),
                    datetime!(2026-06-06 12:11:00 UTC),
                ),
                datetime!(2026-06-06 12:06:00 UTC),
            )
            .unwrap();

        assert_eq!(attempt, StepAttempt::new(2).unwrap());
        let record = instance.step_record(&step_id("plan")).unwrap();
        assert_eq!(record.attempt(), StepAttempt::new(2).unwrap());
        assert_eq!(record.lease().unwrap().owner_id().as_str(), "runner-2");
    }

    #[test]
    fn approving_a_guard_the_ceremony_never_declared_is_refused() {
        let approval =
            CeremonyGuard::new(guard_name("human_approved"), GuardCondition::HumanApproval);
        let finish = CeremonyTransition::new(
            state_id("drafting"),
            state_id("done"),
            trigger("approve"),
            vec![approval.name().clone()],
        )
        .unwrap();
        let definition = CeremonyDefinition::new(
            crate::value_objects::CeremonyName::new("approval_ceremony").unwrap(),
            CeremonyVersion::v1(),
            None,
            Vec::new(),
            Vec::new(),
            vec![
                CeremonyState::initial(state_id("drafting")),
                CeremonyState::terminal(state_id("done")),
            ],
            vec![finish.clone()],
            Vec::new(),
            vec![approval],
            vec![role(vec![RoleAction::transition(finish.trigger().clone())])],
        )
        .unwrap();
        let mut instance = instance(&definition);

        // This used to succeed and write `not_a_guard: true` into the
        // session context: a caller could put any key at all there,
        // and a typo answered "approved" while leaving a session that
        // would never move.
        assert!(matches!(
            instance.approve_guard(
                &definition,
                &guard_name("not_a_guard"),
                role_id("facilitator"),
                AuditActorKind::Human,
                now()
            ),
            Err(DomainError::NotFound {
                what: "ceremony_guard"
            })
        ));
        assert!(!instance
            .context()
            .is_guard_approved(&guard_name("not_a_guard")));
    }

    #[test]
    fn human_approval_guard_uses_typed_context() {
        let approval =
            CeremonyGuard::new(guard_name("human_approved"), GuardCondition::HumanApproval);
        let finish = CeremonyTransition::new(
            state_id("drafting"),
            state_id("done"),
            trigger("approve"),
            vec![approval.name().clone()],
        )
        .unwrap();
        let definition = CeremonyDefinition::new(
            crate::value_objects::CeremonyName::new("approval_ceremony").unwrap(),
            CeremonyVersion::v1(),
            None,
            Vec::new(),
            Vec::new(),
            vec![
                CeremonyState::initial(state_id("drafting")),
                CeremonyState::terminal(state_id("done")),
            ],
            vec![finish.clone()],
            Vec::new(),
            vec![approval.clone()],
            vec![role(vec![RoleAction::transition(finish.trigger().clone())])],
        )
        .unwrap();
        let mut instance = instance(&definition);

        assert!(matches!(
            instance.apply_transition(&definition, &trigger("approve"), now()),
            Err(DomainError::InvariantViolated { .. })
        ));
        instance
            .approve_guard(
                &definition,
                approval.name(),
                role_id("facilitator"),
                AuditActorKind::Human,
                datetime!(2026-06-06 12:01:00 UTC),
            )
            .unwrap();
        instance
            .apply_transition(
                &definition,
                &trigger("approve"),
                datetime!(2026-06-06 12:02:00 UTC),
            )
            .unwrap();

        assert!(instance.is_completed(&definition));
    }

    #[test]
    fn human_guard_deferral_preserves_uncertainty_without_approving() {
        let approval =
            CeremonyGuard::new(guard_name("human_approved"), GuardCondition::HumanApproval);
        let finish = CeremonyTransition::new(
            state_id("drafting"),
            state_id("done"),
            trigger("approve"),
            vec![approval.name().clone()],
        )
        .unwrap();
        let definition = CeremonyDefinition::new(
            crate::value_objects::CeremonyName::new("deferral_ceremony").unwrap(),
            CeremonyVersion::v1(),
            None,
            Vec::new(),
            Vec::new(),
            vec![
                CeremonyState::initial(state_id("drafting")),
                CeremonyState::terminal(state_id("done")),
            ],
            vec![finish.clone()],
            Vec::new(),
            vec![approval.clone()],
            vec![role(vec![RoleAction::transition(finish.trigger().clone())])],
        )
        .unwrap();
        let mut instance = instance(&definition);

        instance
            .defer_guard(
                &definition,
                approval.name().clone(),
                CeremonyGuardDeferralContent::new(
                    "I do not know.",
                    "I cannot explain how the issue was resolved.",
                    vec!["New evidence explains the resolution.".to_owned()],
                )
                .unwrap(),
                role_id("facilitator"),
                AuditActorKind::Human,
                datetime!(2026-06-06 12:01:00 UTC),
            )
            .unwrap();

        assert!(!instance.context().is_guard_approved(approval.name()));
        assert!(instance
            .apply_transition(&definition, &trigger("approve"), now())
            .is_err());
        let deferral = &instance.guard_deferrals()[0];
        assert_eq!(deferral.guard_name(), approval.name());
        assert_eq!(deferral.content().statement(), "I do not know.");
    }
    /// An approval that names nobody is a receipt for a human decision
    /// nobody can be shown to have taken. This is that made checkable.
    #[test]
    fn approving_a_human_guard_records_the_seat_that_did_it() {
        let approval =
            CeremonyGuard::new(guard_name("human_approved"), GuardCondition::HumanApproval);
        let definition = definition_with_human_guard(&approval);
        let mut instance = instance(&definition);

        instance
            .approve_guard(
                &definition,
                approval.name(),
                role_id("facilitator"),
                AuditActorKind::Human,
                datetime!(2026-06-06 12:01:00 UTC),
            )
            .unwrap();

        let [recorded] = instance.guard_approvals() else {
            panic!(
                "expected one approval, got {:?}",
                instance.guard_approvals()
            );
        };
        assert_eq!(recorded.guard_name(), approval.name());
        assert_eq!(recorded.approved_by(), &role_id("facilitator"));
        assert_eq!(recorded.approved_at(), datetime!(2026-06-06 12:01:00 UTC));
        assert!(instance.context().is_guard_approved(approval.name()));
    }

    /// A seat this session does not have cannot approve anything on it.
    /// Weaker than the capability check the other verbs use, and
    /// deliberately so — but not so weak that any string will do.
    #[test]
    fn a_seat_the_definition_does_not_declare_cannot_approve() {
        let approval =
            CeremonyGuard::new(guard_name("human_approved"), GuardCondition::HumanApproval);
        let definition = definition_with_human_guard(&approval);
        let mut instance = instance(&definition);

        let outcome = instance.approve_guard(
            &definition,
            approval.name(),
            role_id("someone-who-is-not-here"),
            AuditActorKind::Human,
            now(),
        );

        assert!(matches!(
            outcome,
            Err(DomainError::NotFound {
                what: "ceremony_role"
            })
        ));
        assert!(instance.guard_approvals().is_empty());
        assert!(!instance.context().is_guard_approved(approval.name()));
    }
    /// A session with one agenda item and one contribution to it —
    /// the smallest thing that has something to explain.
    fn session_with_a_contribution(
        definition: &CeremonyDefinition,
    ) -> (CeremonyInstance, CeremonyInterventionId) {
        let mut instance = instance(definition);
        let agenda_item = CeremonyInterventionId::new("queue-check").unwrap();
        instance
            .request_intervention_as(
                definition,
                agenda_item.clone(),
                role_id("facilitator"),
                CeremonyInterventionKind::Investigation,
                CeremonyInterventionTarget::roles([role_id("observer")]).unwrap(),
                CeremonyInterventionContent::new("Inspect the queue.", Attributes::empty())
                    .unwrap(),
                now(),
            )
            .unwrap();
        instance
            .respond_to_intervention_as(
                definition,
                &agenda_item,
                role_id("observer"),
                CeremonyInterventionContent::new("Queue depth is stable.", Attributes::empty())
                    .unwrap(),
                now(),
            )
            .unwrap();
        (instance, agenda_item)
    }

    /// The one reason the engine sees on its own, and it records it
    /// without being asked.
    #[test]
    fn a_contribution_is_recorded_as_answering_its_agenda_item() {
        let definition = definition();
        let (instance, agenda_item) = session_with_a_contribution(&definition);

        let [answered] = instance.reasons() else {
            panic!("expected exactly one reason, got {:?}", instance.reasons());
        };
        assert_eq!(answered.kind(), CeremonyReasonKind::Answers);
        assert_eq!(
            answered.from(),
            &CeremonyRecordRef::contribution(agenda_item.clone(), 0)
        );
        assert_eq!(answered.to(), &CeremonyRecordRef::agenda_item(agenda_item));
        assert_eq!(
            answered.asserted_by(),
            None,
            "the engine observed it; naming a seat would be inventing one"
        );
    }

    /// Structure is not a judgement. A seat able to assert it could
    /// rewrite the shape of the session by relabelling it.
    #[test]
    fn a_seat_cannot_assert_what_only_the_engine_observes() {
        let definition = definition();
        let (mut instance, agenda_item) = session_with_a_contribution(&definition);

        let outcome = instance.assert_reason_as(
            &definition,
            role_id("observer"),
            CeremonyRecordRef::contribution(agenda_item.clone(), 0),
            CeremonyRecordRef::agenda_item(agenda_item),
            CeremonyReasonKind::Answers,
            "because I say it does",
            MemoryConfidence::High,
            now(),
        );

        assert!(matches!(
            outcome,
            Err(DomainError::InvariantViolated { .. })
        ));
    }

    /// Testimony about one's own reasoning. Nobody else has access to
    /// it, so nobody else may claim it.
    #[test]
    fn only_whoever_contributed_may_say_why_they_did() {
        let definition = definition();
        let (mut instance, agenda_item) = session_with_a_contribution(&definition);
        let contribution = CeremonyRecordRef::contribution(agenda_item.clone(), 0);
        let item = CeremonyRecordRef::agenda_item(agenda_item);

        let by_someone_else = instance.assert_reason_as(
            &definition,
            role_id("facilitator"),
            contribution.clone(),
            item.clone(),
            CeremonyReasonKind::ChosenBecause,
            "they must have thought the queue mattered",
            MemoryConfidence::Low,
            now(),
        );
        assert!(matches!(
            by_someone_else,
            Err(DomainError::InvariantViolated { .. })
        ));

        instance
            .assert_reason_as(
                &definition,
                role_id("observer"),
                contribution,
                item,
                CeremonyReasonKind::ChosenBecause,
                "the depth graph had been flat for an hour",
                MemoryConfidence::High,
                now(),
            )
            .expect("its author may say why");
        assert_eq!(instance.reasons().len(), 2);
    }

    /// A claim about the world, not about a mind. Anyone may make one
    /// and everyone may weigh it.
    #[test]
    fn any_seat_may_claim_that_one_thing_came_from_another() {
        let definition = definition();
        let (mut instance, agenda_item) = session_with_a_contribution(&definition);

        instance
            .assert_reason_as(
                &definition,
                role_id("facilitator"),
                CeremonyRecordRef::agenda_item(agenda_item.clone()),
                CeremonyRecordRef::contribution(agenda_item, 0),
                CeremonyReasonKind::FollowsFrom,
                "the item stayed open because the answer raised a new question",
                MemoryConfidence::Medium,
                now(),
            )
            .expect("a claim about the world is open to any seat");

        let asserted = instance.reasons().last().unwrap();
        assert_eq!(asserted.confidence(), MemoryConfidence::Medium);
        assert_eq!(asserted.asserted_by(), Some(&role_id("facilitator")));
    }

    /// A session knows everything it has done, so a reason may not
    /// cite something it never produced.
    #[test]
    fn a_reason_cannot_cite_something_that_never_happened() {
        let definition = definition();
        let (mut instance, agenda_item) = session_with_a_contribution(&definition);

        let outcome = instance.assert_reason_as(
            &definition,
            role_id("observer"),
            CeremonyRecordRef::contribution(agenda_item.clone(), 7),
            CeremonyRecordRef::agenda_item(agenda_item),
            CeremonyReasonKind::FollowsFrom,
            "a contribution nobody made",
            MemoryConfidence::Low,
            now(),
        );

        assert!(matches!(
            outcome,
            Err(DomainError::NotFound {
                what: "ceremony_record"
            })
        ));
    }

    /// A move is recorded with the seat that fired it, so "the session
    /// resolved because…" has something to point at.
    #[test]
    fn a_move_is_recorded_with_whoever_made_it() {
        let approval =
            CeremonyGuard::new(guard_name("human_approved"), GuardCondition::HumanApproval);
        let definition = definition_with_human_guard(&approval);
        let mut instance = instance(&definition);
        instance
            .approve_guard(
                &definition,
                approval.name(),
                role_id("facilitator"),
                AuditActorKind::Human,
                now(),
            )
            .unwrap();

        instance
            .apply_transition_as(
                &definition,
                &role_id("facilitator"),
                &trigger("approve"),
                datetime!(2026-06-06 12:05:00 UTC),
            )
            .unwrap();

        let [moved] = instance.transitions() else {
            panic!("expected one move, got {:?}", instance.transitions());
        };
        assert_eq!(moved.trigger(), &trigger("approve"));
        assert_eq!(moved.from_state(), &state_id("drafting"));
        assert_eq!(moved.to_state(), &state_id("done"));
        assert_eq!(moved.applied_by(), Some(&role_id("facilitator")));
    }

    /// And without one when the engine took the move itself. An
    /// absence, not a gap — and it is what stops testimony being
    /// claimed about something nobody can testify to.
    #[test]
    fn a_move_the_engine_took_names_nobody() {
        let approval =
            CeremonyGuard::new(guard_name("human_approved"), GuardCondition::HumanApproval);
        let definition = definition_with_human_guard(&approval);
        let mut instance = instance(&definition);
        instance
            .approve_guard(
                &definition,
                approval.name(),
                role_id("facilitator"),
                AuditActorKind::Human,
                now(),
            )
            .unwrap();

        instance
            .apply_transition(&definition, &trigger("approve"), now())
            .unwrap();

        assert_eq!(instance.transitions()[0].applied_by(), None);
    }
    /// What kind of party filled the seat is recorded as declared and
    /// never inferred.
    ///
    /// The engine knows this guard demands a human. That says one was
    /// required, not that one turned up — and a receipt that read
    /// compliance off its own requirement would assert exactly what
    /// nobody can demonstrate. So an agent approving a human-approval
    /// guard is recorded as an agent, and whether that is acceptable
    /// is a question for whoever reads it.
    #[test]
    fn an_approval_records_the_kind_it_was_told_not_the_one_the_guard_wanted() {
        let approval =
            CeremonyGuard::new(guard_name("human_approved"), GuardCondition::HumanApproval);
        let definition = definition_with_human_guard(&approval);
        let mut instance = instance(&definition);

        instance
            .approve_guard(
                &definition,
                approval.name(),
                role_id("facilitator"),
                AuditActorKind::Agent,
                now(),
            )
            .unwrap();

        let [recorded] = instance.guard_approvals() else {
            panic!("expected one approval");
        };
        assert_eq!(
            recorded.approved_by_kind(),
            AuditActorKind::Agent,
            "the guard asked for a human and an agent answered; saying otherwise \
             would be the engine vouching for something it cannot see"
        );
        assert!(instance.context().is_guard_approved(approval.name()));
    }
}