canwu_transport/

lib.rs

//! Domain-neutral transport execution records.
//!
//! The crate owns transport semantics, not the simulation scheduler or the
//! information lifecycle. Applications persist these values through their
//! domain-record schemas and drive transitions through canonical ingress.

#![allow(clippy::missing_errors_doc)]

use canwu_core::{
    DomainRecordRef, DomainRecordVersionRef, EntityRef, EvidenceRef, KnowledgeHolderRef,
};
use canwu_routing::{RoutePlan, RoutingNodeRef};
use canwu_time::SimTime;
use serde::{Deserialize, Serialize};
use std::cmp::Ordering;

/// Semantic identity of the transport record and transition contract.
///
/// `v5` adds capacity pools with deterministic booking allocation, booking
/// requests on an execution, cancellation, arrival settlement for executions
/// without a delivery attempt, and booking confirmation or cancellation before
/// the booking window opens.
pub const TRANSPORT_SEMANTIC_VERSION: &str = "canwu-transport.v5";

/// Hash domain of [`CapacityBookingAllocationEvidenceV1::semantic_digest`].
pub const CAPACITY_BOOKING_ALLOCATION_DIGEST_DOMAIN: &str =
    "canwu.transport.capacity-booking-allocation.v1";

#[must_use]
pub fn delivery_completion_operation_key(
    execution: TransportExecutionId,
    revision: ItineraryRevisionId,
    attempt_version: u64,
) -> String {
    format!(
        "transport/{}/revision/{}/delivery-completion/attempt-version/{}",
        execution.0, revision.0, attempt_version
    )
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct TransportExecutionId(pub u64);

/// Stable identity for an admitted transport-domain movement intent.
///
/// The simulation admits movement through a host- or integration-defined command.
/// This record adds route-plan and custody evidence when a transport domain needs
/// to persist a richer execution than its own projected transit state.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct MovementOrderId(pub u64);

/// Who initiated a movement intent. The runtime must derive this from the
/// admitted authority and never trust an unvalidated caller-supplied label.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum MovementInitiative {
    SelfDirected,
    Commanded,
    Delegated,
    Forced,
    Automatic,
}

/// The physical role of a subject in a movement manifest.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum MovementSubjectRole {
    MovablePrincipal,
    Cargo,
    Carrier,
    Passenger,
    Attached,
    /// An aggregate of people moving as one subject. The subject identity is
    /// normally an application domain-record reference, and the manifest
    /// quantity is its positive head count.
    PersonsGroup,
}

impl MovementSubjectRole {
    /// Whether this role is counted by a positive integer manifest quantity:
    /// units for [`Self::Cargo`], heads for [`Self::PersonsGroup`].
    #[must_use]
    pub const fn requires_quantity(self) -> bool {
        matches!(self, Self::Cargo | Self::PersonsGroup)
    }
}

/// One typed identity in a movement manifest.
#[derive(Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
pub struct MovementSubject {
    pub entity: EntityRef,
    pub role: MovementSubjectRole,
    /// Cargo quantities are integer units and persons-group quantities are
    /// head counts; both must be present and positive. Other roles leave this
    /// unset because their cardinality is one identity.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub quantity: Option<u64>,
    /// Expected carrier/custodian identity at admission, when applicable.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub expected_custody: Option<EntityRef>,
}

/// Immutable, admitted intent shared by transport movement domains.
///
/// `MovementOrder` is a contract for planning and authority evidence; it does
/// not directly mutate a world entity. Domain handlers still own location,
/// custody, quantity, arrival, and knowledge effects.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct MovementOrder {
    pub id: MovementOrderId,
    pub subjects: Vec<MovementSubject>,
    pub origin: RoutingNodeRef,
    pub destination: RoutingNodeRef,
    pub plan: RoutePlan,
    pub initiative: MovementInitiative,
    pub ordered_at: SimTime,
    pub expected_position_revision: u64,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub enum MovementOrderError {
    Invalid(String),
}

impl MovementOrder {
    /// Validates the structural invariants that are domain-neutral.
    ///
    /// Existence, authority, capability, and position/custody matching remain
    /// runtime or plugin responsibilities because this crate does not own the
    /// world or domain records.
    pub fn validate(&self) -> Result<(), MovementOrderError> {
        if self.id.0 == 0
            || self.expected_position_revision == 0
            || self.origin.as_str().trim().is_empty()
            || self.destination.as_str().trim().is_empty()
        {
            return Err(MovementOrderError::Invalid(
                "movement order identity, endpoints, and expected position revision must be valid"
                    .to_owned(),
            ));
        }
        if self.subjects.is_empty()
            || self
                .subjects
                .windows(2)
                .any(|pair| pair[0].entity >= pair[1].entity)
        {
            return Err(MovementOrderError::Invalid(
                "movement subjects must be non-empty, sorted, and unique by entity".to_owned(),
            ));
        }
        for subject in &self.subjects {
            if subject.quantity.is_some_and(|quantity| quantity == 0)
                || subject.role.requires_quantity() != subject.quantity.is_some()
            {
                return Err(MovementOrderError::Invalid(
                    "cargo and persons-group subjects require a positive quantity; other subjects cannot carry one"
                        .to_owned(),
                ));
            }
        }
        if self.plan.origin != self.origin
            || self.plan.destination != self.destination
            || self.plan.departure_at < self.ordered_at
            || self.plan.estimated_arrival_at < self.plan.departure_at
            || self.plan.digest.trim().is_empty()
        {
            return Err(MovementOrderError::Invalid(
                "movement plan does not match the order endpoints or time range".to_owned(),
            ));
        }
        Ok(())
    }
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct ItineraryRevisionId(pub u64);

#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct LegExecutionId(pub u64);

#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct HandoffId(pub u64);

#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct CapacityBookingId(pub u64);

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum TransportExecutionState {
    Prepared,
    Planning,
    Booking,
    Ready,
    Executing,
    ReplanPending,
    ArrivalPending,
    Settled,
    Failed,
    Cancelled,
}

impl TransportExecutionState {
    #[must_use]
    pub const fn is_terminal(self) -> bool {
        matches!(self, Self::Settled | Self::Failed | Self::Cancelled)
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ItineraryRevisionReason {
    Initial,
    Disaster {
        explanation: String,
    },
    CapacityUnavailable {
        explanation: String,
    },
    KnowledgeUpdate {
        explanation: String,
    },
    Recovery {
        explanation: String,
    },
    /// A revision caused by an application-owned condition record, cited at
    /// the exact positive record version the planner acted on. `kind` is an
    /// application label for the condition; transport does not interpret it.
    ExternalCondition {
        record: DomainRecordRef,
        version: u64,
        kind: String,
    },
}

impl ItineraryRevisionReason {
    fn validate(&self) -> Result<(), TransportError> {
        if let Self::ExternalCondition {
            record,
            version,
            kind,
        } = self
            && (*version == 0
                || kind.trim().is_empty()
                || !domain_record_ref_is_well_formed(record))
        {
            return Err(TransportError::InvalidRevision(
                "external-condition reason requires a well-formed record, a positive version, and a kind"
                    .to_owned(),
            ));
        }
        Ok(())
    }
}

fn domain_record_ref_is_well_formed(record: &DomainRecordRef) -> bool {
    !record.kind.namespace.trim().is_empty()
        && !record.kind.name.trim().is_empty()
        && !record.id.trim().is_empty()
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct ItineraryRevision {
    pub id: ItineraryRevisionId,
    pub predecessor: Option<ItineraryRevisionId>,
    pub plan: RoutePlan,
    pub planned_at: SimTime,
    pub valid_from: SimTime,
    pub reason: ItineraryRevisionReason,
    pub superseded_at: Option<SimTime>,
    pub evidence: Vec<EvidenceRef>,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum LegExecutionStatus {
    Planned,
    Booked,
    Loaded,
    Departed,
    Arrived,
    Waiting,
    Failed,
    Cancelled,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct LegExecution {
    pub id: LegExecutionId,
    pub itinerary_revision: ItineraryRevisionId,
    pub leg_index: usize,
    pub status: LegExecutionStatus,
    pub actual_departure_at: Option<SimTime>,
    pub actual_arrival_at: Option<SimTime>,
    pub failed_at: Option<SimTime>,
    pub failure_reason: Option<String>,
    pub evidence: Vec<EvidenceRef>,
}

/// How custody changed hands between two legs.
#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum HandoffKind {
    /// A custody transfer the itinerary planned for.
    #[default]
    Planned,
    /// Custody taken by an entity outside the itinerary. Transport records the
    /// seizing identity; the incident, hostility, and authority behind it stay
    /// application systems. Unless it is terminal (below), a seizure follows
    /// the same leg rules as a planned handoff: it is recorded after the
    /// source leg arrived or failed, into a startable leg (typically the first
    /// leg of a replacement revision), and `to_custodian` remains the
    /// application's custody label.
    ///
    /// A seizure may instead end the itinerary's custody. Such a *terminal*
    /// seizure names the failed current leg of a non-terminal execution's
    /// active itinerary as both `from_leg` and `to_leg`. It is recorded no
    /// earlier than that leg failed and only if no other handoff left it, and
    /// at most once per execution. Afterwards the execution records no
    /// handoff and cannot fail or reroute a leg, arrive, settle successfully,
    /// or book capacity (see [`TransportExecution::custody_left_itinerary`]); its
    /// owner closes it as failed or cancelled.
    Seizure { by: EntityRef },
}

impl HandoffKind {
    #[must_use]
    pub const fn is_planned(&self) -> bool {
        matches!(self, Self::Planned)
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct Handoff {
    pub id: HandoffId,
    pub from_leg: LegExecutionId,
    pub to_leg: LegExecutionId,
    pub from_custodian: String,
    pub to_custodian: String,
    pub at: SimTime,
    pub location: String,
    pub evidence: Vec<EvidenceRef>,
    /// Omitted from JSON when [`HandoffKind::Planned`], so handoffs recorded
    /// before seizure handoffs existed keep their exact shape.
    #[serde(default, skip_serializing_if = "HandoffKind::is_planned")]
    pub kind: HandoffKind,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct DeliveryCompletionRequest {
    pub operation_key: String,
    pub execution: TransportExecutionId,
    pub itinerary_revision: ItineraryRevisionId,
    pub delivery_attempt: DomainRecordVersionRef,
    pub completed_at: SimTime,
    pub evidence: Vec<EvidenceRef>,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum CapacityBookingStatus {
    Requested,
    Confirmed,
    Consumed,
    Released,
    Expired,
    Cancelled,
    Failed,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct CapacityBooking {
    pub id: CapacityBookingId,
    pub execution: TransportExecutionId,
    pub resource: String,
    pub valid_from: SimTime,
    pub valid_until: SimTime,
    pub quantity: u64,
    pub priority: i32,
    pub status: CapacityBookingStatus,
    pub allocation_evidence: Vec<EvidenceRef>,
}

impl CapacityBooking {
    pub fn new(
        id: CapacityBookingId,
        execution: TransportExecutionId,
        resource: String,
        valid_from: SimTime,
        valid_until: SimTime,
        quantity: u64,
        priority: i32,
    ) -> Result<Self, TransportError> {
        if valid_until < valid_from || quantity == 0 {
            return Err(TransportError::InvalidBooking(
                "capacity booking requires a positive quantity and non-inverted window".to_owned(),
            ));
        }
        Ok(Self {
            id,
            execution,
            resource,
            valid_from,
            valid_until,
            quantity,
            priority,
            status: CapacityBookingStatus::Requested,
            allocation_evidence: Vec::new(),
        })
    }

    /// Moves the booking through its windowed state machine.
    ///
    /// A request may be confirmed, failed, or cancelled, and a confirmed
    /// booking released or cancelled, before its window opens; capacity can
    /// only be consumed inside the window, confirmed only until it ends, and
    /// expired only after it ends.
    pub fn transition(
        &mut self,
        status: CapacityBookingStatus,
        at: SimTime,
    ) -> Result<(), TransportError> {
        if status == CapacityBookingStatus::Consumed
            && (at < self.valid_from || at > self.valid_until)
        {
            return Err(TransportError::InvalidBooking(
                "booking capacity can only be consumed inside its validity window".to_owned(),
            ));
        }
        let allowed = matches!(
            (self.status, status),
            (
                CapacityBookingStatus::Requested,
                CapacityBookingStatus::Confirmed
                    | CapacityBookingStatus::Failed
                    | CapacityBookingStatus::Cancelled
            ) | (
                CapacityBookingStatus::Confirmed,
                CapacityBookingStatus::Consumed
                    | CapacityBookingStatus::Released
                    | CapacityBookingStatus::Cancelled
                    | CapacityBookingStatus::Expired
            ) | (
                CapacityBookingStatus::Consumed,
                CapacityBookingStatus::Released
            )
        );
        if !allowed {
            return Err(TransportError::InvalidBooking(
                "capacity booking transition is not allowed".to_owned(),
            ));
        }
        if at > self.valid_until && status == CapacityBookingStatus::Confirmed {
            return Err(TransportError::InvalidBooking(
                "capacity booking cannot be confirmed after its validity window".to_owned(),
            ));
        }
        if status == CapacityBookingStatus::Expired && at <= self.valid_until {
            return Err(TransportError::InvalidBooking(
                "capacity booking cannot expire before its validity window ends".to_owned(),
            ));
        }
        self.status = status;
        Ok(())
    }
}

/// A windowed pool of interchangeable transport capacity, such as ferry
/// crossings, carriage places, or relay mounts for one period.
///
/// `quantity` is the capacity offered for the whole window. `booked` is held
/// by confirmed bookings and `consumed` by consumed ones; released, expired,
/// cancelled, and failed bookings hold nothing. `revision` advances whenever
/// the offer or either counter changes, so allocation evidence can cite the
/// exact pool state it read.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct TransportCapacityPoolV1 {
    pub id: String,
    pub resource: String,
    pub custodian: KnowledgeHolderRef,
    pub window_from: SimTime,
    pub window_until: SimTime,
    pub quantity: u64,
    pub booked: u64,
    pub consumed: u64,
    pub revision: u64,
}

impl TransportCapacityPoolV1 {
    /// Creates an empty pool at revision one.
    pub fn new(
        id: String,
        resource: String,
        custodian: KnowledgeHolderRef,
        window_from: SimTime,
        window_until: SimTime,
        quantity: u64,
    ) -> Result<Self, TransportError> {
        let pool = Self {
            id,
            resource,
            custodian,
            window_from,
            window_until,
            quantity,
            booked: 0,
            consumed: 0,
            revision: 1,
        };
        pool.validate()?;
        Ok(pool)
    }

    pub fn validate(&self) -> Result<(), TransportError> {
        if !is_canonical_label(&self.id)
            || !is_canonical_label(&self.resource)
            || self.window_until < self.window_from
            || self.revision == 0
        {
            return Err(TransportError::InvalidCapacityPool(
                "capacity pool requires a canonical identity and resource, a non-inverted window, and a positive revision"
                    .to_owned(),
            ));
        }
        if self
            .booked
            .checked_add(self.consumed)
            .is_none_or(|held| held > self.quantity)
        {
            return Err(TransportError::InvalidCapacityPool(
                "capacity pool holds more than it offers".to_owned(),
            ));
        }
        Ok(())
    }

    /// Capacity neither booked nor consumed.
    #[must_use]
    pub const fn available(&self) -> u64 {
        self.quantity
            .saturating_sub(self.booked)
            .saturating_sub(self.consumed)
    }

    /// Replaces the offered window and quantity. The new quantity must still
    /// cover everything booked or consumed.
    pub fn revise(
        &mut self,
        window_from: SimTime,
        window_until: SimTime,
        quantity: u64,
    ) -> Result<(), TransportError> {
        let mut revised = self.clone();
        revised.window_from = window_from;
        revised.window_until = window_until;
        revised.quantity = quantity;
        revised.revision = self
            .revision
            .checked_add(1)
            .ok_or(TransportError::Overflow)?;
        revised.validate()?;
        *self = revised;
        Ok(())
    }

    /// Commits one allocation pass computed by [`allocate_capacity_bookings`]
    /// against this exact pool revision, then advances the revision.
    pub fn apply_allocations(
        &mut self,
        allocations: &[BookingAllocationV1],
    ) -> Result<(), TransportError> {
        if allocations.is_empty() {
            return Ok(());
        }
        let mut bookings = allocations
            .iter()
            .map(|allocation| allocation.booking)
            .collect::<Vec<_>>();
        bookings.sort_unstable();
        if bookings.windows(2).any(|pair| pair[0] == pair[1]) {
            return Err(TransportError::InvalidCapacityPool(
                "an allocation pass names each booking once".to_owned(),
            ));
        }
        let mut confirmed = 0_u64;
        for allocation in allocations {
            let evidence = &allocation.evidence;
            if evidence.pool != self.id
                || evidence.pool_revision != self.revision
                || evidence.booking != allocation.booking
                || evidence.quantity != allocation.quantity
                || evidence.status != allocation.status
                || !evidence.digest_matches()
            {
                return Err(TransportError::InvalidCapacityPool(
                    "allocation evidence does not bind this pool revision".to_owned(),
                ));
            }
            match allocation.status {
                CapacityBookingStatus::Confirmed => {
                    confirmed = confirmed
                        .checked_add(allocation.quantity)
                        .ok_or(TransportError::Overflow)?;
                }
                CapacityBookingStatus::Failed if allocation.quantity == 0 => {}
                _ => {
                    return Err(TransportError::InvalidCapacityPool(
                        "an allocation either confirms its quantity or fails with none".to_owned(),
                    ));
                }
            }
        }
        let mut next = self.clone();
        next.booked = self
            .booked
            .checked_add(confirmed)
            .ok_or(TransportError::Overflow)?;
        next.revision = self
            .revision
            .checked_add(1)
            .ok_or(TransportError::Overflow)?;
        next.validate()?;
        *self = next;
        Ok(())
    }

    /// Applies the capacity effect of one booking transition made after
    /// allocation: consumption moves quantity from `booked` to `consumed`,
    /// while releasing, cancelling, or expiring a confirmed booking and
    /// releasing a consumed one return it. Cancelling or failing a request
    /// holds nothing and leaves the pool unchanged. Confirmation happens only
    /// through [`Self::apply_allocations`].
    pub fn apply_booking_transition(
        &mut self,
        quantity: u64,
        from: CapacityBookingStatus,
        to: CapacityBookingStatus,
    ) -> Result<(), TransportError> {
        use CapacityBookingStatus::{
            Cancelled, Confirmed, Consumed, Expired, Failed, Released, Requested,
        };
        let underflow =
            || TransportError::InvalidCapacityPool("capacity pool counter underflow".to_owned());
        let mut next = self.clone();
        match (from, to) {
            (Requested, Cancelled | Failed) => return Ok(()),
            (Confirmed, Consumed) => {
                next.booked = next.booked.checked_sub(quantity).ok_or_else(underflow)?;
                next.consumed = next
                    .consumed
                    .checked_add(quantity)
                    .ok_or(TransportError::Overflow)?;
            }
            (Confirmed, Released | Cancelled | Expired) => {
                next.booked = next.booked.checked_sub(quantity).ok_or_else(underflow)?;
            }
            (Consumed, Released) => {
                next.consumed = next.consumed.checked_sub(quantity).ok_or_else(underflow)?;
            }
            _ => {
                return Err(TransportError::InvalidCapacityPool(
                    "booking transition has no capacity-pool effect".to_owned(),
                ));
            }
        }
        next.revision = self
            .revision
            .checked_add(1)
            .ok_or(TransportError::Overflow)?;
        next.validate()?;
        *self = next;
        Ok(())
    }
}

/// One requested booking offered to a pool allocation pass, with the caller's
/// deterministic tie-break key and admission sequence.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct CapacityBookingRequestV1 {
    pub booking: CapacityBooking,
    pub tie_break: String,
    pub admitted_sequence: u64,
}

/// Why an allocation pass failed a booking request.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum CapacityAllocationFailureV1 {
    /// The request does not fit the capacity left after earlier grants.
    InsufficientCapacity,
    /// The booking window is not inside the pool window.
    OutsidePoolWindow,
    /// The booking window ended before the allocation time.
    WindowElapsed,
}

/// Replayable evidence of one allocation decision.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct CapacityBookingAllocationEvidenceV1 {
    pub pool: String,
    pub pool_revision: u64,
    pub booking: CapacityBookingId,
    pub execution: TransportExecutionId,
    pub requested: u64,
    /// Granted quantity: the full request when confirmed, zero when failed.
    pub quantity: u64,
    pub status: CapacityBookingStatus,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub failure: Option<CapacityAllocationFailureV1>,
    /// Pool capacity left after this decision, in allocation order.
    pub remaining_after: u64,
    pub allocated_at: SimTime,
    pub operation_key: String,
    /// Domain-separated BLAKE3 digest over every other field; see
    /// [`CAPACITY_BOOKING_ALLOCATION_DIGEST_DOMAIN`].
    pub semantic_digest: String,
}

impl CapacityBookingAllocationEvidenceV1 {
    /// Recomputes the digest over every field except `semantic_digest`.
    #[must_use]
    pub fn expected_digest(&self) -> String {
        let material = (
            &self.pool,
            self.pool_revision,
            self.booking,
            self.execution,
            self.requested,
            self.quantity,
            self.status,
            self.failure,
            self.remaining_after,
            self.allocated_at,
            &self.operation_key,
        );
        domain_digest(CAPACITY_BOOKING_ALLOCATION_DIGEST_DOMAIN, &material)
    }

    #[must_use]
    pub fn digest_matches(&self) -> bool {
        self.semantic_digest == self.expected_digest()
    }
}

/// One booking's allocation result, in allocation order.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct BookingAllocationV1 {
    pub booking: CapacityBookingId,
    /// Granted quantity: the full request when confirmed, zero when failed.
    pub quantity: u64,
    /// `Confirmed` or `Failed`.
    pub status: CapacityBookingStatus,
    pub evidence: CapacityBookingAllocationEvidenceV1,
}

/// Stable operation key of one booking's allocation at one pool revision.
#[must_use]
pub fn capacity_booking_allocation_operation_key(
    pool: &str,
    pool_revision: u64,
    booking: CapacityBookingId,
) -> String {
    format!(
        "transport/pool/{pool}/revision/{pool_revision}/booking/{}/allocation",
        booking.0
    )
}

/// Allocates requested bookings against one pool revision, all or nothing
/// per booking.
///
/// Requests are visited by descending priority, then ascending `valid_from`,
/// tie-break key, admission sequence, and booking identity, so the result
/// does not depend on input order. A request is confirmed when its window
/// lies inside the pool window, has not ended at `at`, and its full quantity
/// fits the capacity left by earlier grants; otherwise it fails with a
/// recorded reason, and a later, smaller request may still fit. The function
/// reads only its arguments; commit the result with
/// [`TransportCapacityPoolV1::apply_allocations`] and
/// [`CapacityBooking::transition`].
pub fn allocate_capacity_bookings(
    pool: &TransportCapacityPoolV1,
    requests: &[CapacityBookingRequestV1],
    at: SimTime,
) -> Result<Vec<BookingAllocationV1>, TransportError> {
    pool.validate()?;
    let mut ids = Vec::with_capacity(requests.len());
    for request in requests {
        let booking = &request.booking;
        if booking.status != CapacityBookingStatus::Requested
            || booking.resource != pool.resource
            || booking.quantity == 0
            || booking.valid_until < booking.valid_from
            || !is_canonical_label(&request.tie_break)
        {
            return Err(TransportError::InvalidCapacityPool(
                "allocation accepts only well-formed requested bookings for the pool resource"
                    .to_owned(),
            ));
        }
        ids.push(booking.id);
    }
    ids.sort_unstable();
    if ids.windows(2).any(|pair| pair[0] == pair[1]) {
        return Err(TransportError::InvalidCapacityPool(
            "allocation requests must have unique booking identities".to_owned(),
        ));
    }
    let mut ordered = requests.iter().collect::<Vec<_>>();
    ordered.sort_by(|left, right| allocation_order(left, right));
    let mut remaining = pool.available();
    let mut allocations = Vec::with_capacity(ordered.len());
    for request in ordered {
        let booking = &request.booking;
        let failure = if at > booking.valid_until {
            Some(CapacityAllocationFailureV1::WindowElapsed)
        } else if booking.valid_from < pool.window_from || booking.valid_until > pool.window_until {
            Some(CapacityAllocationFailureV1::OutsidePoolWindow)
        } else if booking.quantity > remaining {
            Some(CapacityAllocationFailureV1::InsufficientCapacity)
        } else {
            None
        };
        let (status, quantity) = if failure.is_none() {
            remaining -= booking.quantity;
            (CapacityBookingStatus::Confirmed, booking.quantity)
        } else {
            (CapacityBookingStatus::Failed, 0)
        };
        let mut evidence = CapacityBookingAllocationEvidenceV1 {
            pool: pool.id.clone(),
            pool_revision: pool.revision,
            booking: booking.id,
            execution: booking.execution,
            requested: booking.quantity,
            quantity,
            status,
            failure,
            remaining_after: remaining,
            allocated_at: at,
            operation_key: capacity_booking_allocation_operation_key(
                &pool.id,
                pool.revision,
                booking.id,
            ),
            semantic_digest: String::new(),
        };
        evidence.semantic_digest = evidence.expected_digest();
        allocations.push(BookingAllocationV1 {
            booking: booking.id,
            quantity,
            status,
            evidence,
        });
    }
    Ok(allocations)
}

fn allocation_order(left: &CapacityBookingRequestV1, right: &CapacityBookingRequestV1) -> Ordering {
    right
        .booking
        .priority
        .cmp(&left.booking.priority)
        .then_with(|| left.booking.valid_from.cmp(&right.booking.valid_from))
        .then_with(|| left.tie_break.cmp(&right.tie_break))
        .then_with(|| left.admitted_sequence.cmp(&right.admitted_sequence))
        .then_with(|| left.booking.id.cmp(&right.booking.id))
}

fn is_canonical_label(value: &str) -> bool {
    !value.is_empty() && value.trim() == value
}

fn domain_digest<T: Serialize>(domain: &str, value: &T) -> String {
    let encoded = serde_json::to_vec(value).expect("transport digest material must serialize");
    let mut hasher = blake3::Hasher::new();
    hasher.update(domain.as_bytes());
    hasher.update(&[0]);
    hasher.update(&encoded);
    hasher.finalize().to_hex().to_string()
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum SagaState {
    TransportIntent,
    WaitingForInformation,
    Executing,
    ArrivalPending,
    Settled,
    CompensationPending,
    Failed,
}

/// Result of reconciling the information-system delivery attempt.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub enum ReconciliationOutcome {
    Success,
    Failure { error: String },
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct DeliverySaga {
    pub operation_key: String,
    pub delivery_attempt: DomainRecordVersionRef,
    pub state: SagaState,
    pub step: u32,
    pub expected_attempt_version: u64,
    pub last_error: Option<String>,
    pub evidence: Vec<EvidenceRef>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct TransportExecution {
    pub id: TransportExecutionId,
    pub delivery_attempt: Option<DomainRecordVersionRef>,
    pub state: TransportExecutionState,
    pub active_itinerary_revision: Option<ItineraryRevisionId>,
    pub current_leg_index: usize,
    pub estimated_arrival_at: Option<SimTime>,
    pub current_endpoint: Option<String>,
    pub revisions: Vec<ItineraryRevision>,
    pub legs: Vec<LegExecution>,
    pub handoffs: Vec<Handoff>,
    pub bookings: Vec<CapacityBooking>,
    pub saga: Option<DeliverySaga>,
}

impl TransportExecution {
    #[must_use]
    pub fn new(id: TransportExecutionId, delivery_attempt: Option<DomainRecordVersionRef>) -> Self {
        Self {
            id,
            delivery_attempt,
            state: TransportExecutionState::Prepared,
            active_itinerary_revision: None,
            current_leg_index: 0,
            estimated_arrival_at: None,
            current_endpoint: None,
            revisions: Vec::new(),
            legs: Vec::new(),
            handoffs: Vec::new(),
            bookings: Vec::new(),
            saga: None,
        }
    }

    pub fn install_initial_itinerary(
        &mut self,
        revision: ItineraryRevision,
    ) -> Result<(), TransportError> {
        if !self.revisions.is_empty() || revision.predecessor.is_some() {
            return Err(TransportError::InvalidRevision(
                "initial itinerary must be the first revision".to_owned(),
            ));
        }
        revision.reason.validate()?;
        self.estimated_arrival_at = Some(revision.plan.estimated_arrival_at);
        self.active_itinerary_revision = Some(revision.id);
        self.current_endpoint = Some(revision.plan.origin.as_str().to_owned());
        self.legs = revision
            .plan
            .legs
            .iter()
            .enumerate()
            .map(|(index, _)| {
                let id = u64::try_from(index)
                    .ok()
                    .and_then(|index| index.checked_add(1))
                    .ok_or(TransportError::Overflow)?;
                Ok(LegExecution {
                    id: LegExecutionId(id),
                    itinerary_revision: revision.id,
                    leg_index: index,
                    status: LegExecutionStatus::Planned,
                    actual_departure_at: None,
                    actual_arrival_at: None,
                    failed_at: None,
                    failure_reason: None,
                    evidence: Vec::new(),
                })
            })
            .collect::<Result<Vec<_>, TransportError>>()?;
        self.revisions.push(revision);
        self.state = TransportExecutionState::Planning;
        Ok(())
    }

    pub fn reroute(
        &mut self,
        revision: ItineraryRevision,
        at: SimTime,
    ) -> Result<(), TransportError> {
        self.ensure_custody_in_itinerary()?;
        let active = self
            .active_itinerary_revision
            .ok_or(TransportError::MissingItinerary)?;
        if revision.predecessor != Some(active) || revision.valid_from < at {
            return Err(TransportError::InvalidRevision("reroute must reference the active revision and start no earlier than the reroute time".to_owned()));
        }
        revision.reason.validate()?;
        if self.revisions.iter().any(|item| item.id == revision.id) {
            return Err(TransportError::InvalidRevision(
                "itinerary revision identity must be unique within an execution".to_owned(),
            ));
        }
        if let Some(previous) = self.revisions.iter_mut().find(|item| item.id == active) {
            previous.superseded_at = Some(at);
        }
        self.estimated_arrival_at = Some(revision.plan.estimated_arrival_at);
        self.active_itinerary_revision = Some(revision.id);
        self.current_leg_index = 0;
        let next_leg_id = self
            .legs
            .iter()
            .map(|leg| leg.id.0)
            .max()
            .unwrap_or_default()
            .checked_add(1)
            .ok_or(TransportError::Overflow)?;
        let mut legs = Vec::with_capacity(revision.plan.legs.len());
        for (index, _) in revision.plan.legs.iter().enumerate() {
            let id = u64::try_from(index)
                .ok()
                .and_then(|index| next_leg_id.checked_add(index))
                .ok_or(TransportError::Overflow)?;
            legs.push(LegExecution {
                id: LegExecutionId(id),
                itinerary_revision: revision.id,
                leg_index: index,
                status: LegExecutionStatus::Planned,
                actual_departure_at: None,
                actual_arrival_at: None,
                failed_at: None,
                failure_reason: None,
                evidence: Vec::new(),
            });
        }
        self.legs.extend(legs);
        self.revisions.push(revision);
        if let Some(saga) = self.saga.as_mut() {
            saga.operation_key = delivery_completion_operation_key(
                self.id,
                self.active_itinerary_revision
                    .ok_or(TransportError::MissingItinerary)?,
                saga.expected_attempt_version,
            );
            saga.evidence.extend(
                self.revisions
                    .last()
                    .map(|current| current.evidence.clone())
                    .unwrap_or_default(),
            );
        }
        self.state = TransportExecutionState::Planning;
        Ok(())
    }

    pub fn begin_saga(
        &mut self,
        delivery_attempt: DomainRecordVersionRef,
        operation_key: String,
    ) -> Result<(), TransportError> {
        if self.saga.is_some() {
            return Err(TransportError::SagaAlreadyExists);
        }
        self.saga = Some(DeliverySaga {
            expected_attempt_version: delivery_attempt.version,
            operation_key,
            delivery_attempt,
            state: SagaState::TransportIntent,
            step: 0,
            last_error: None,
            evidence: Vec::new(),
        });
        self.state = TransportExecutionState::Executing;
        Ok(())
    }

    pub fn start_current_leg(&mut self, at: SimTime) -> Result<(), TransportError> {
        if self.state != TransportExecutionState::Ready
            && self.state != TransportExecutionState::Executing
            && self.state != TransportExecutionState::Planning
        {
            return Err(TransportError::InvalidState(
                "transport execution cannot start a leg in its current state".to_owned(),
            ));
        }
        let active = self
            .active_itinerary_revision
            .ok_or(TransportError::MissingItinerary)?;
        let leg = self
            .legs
            .iter_mut()
            .find(|leg| leg.itinerary_revision == active && leg.leg_index == self.current_leg_index)
            .ok_or(TransportError::MissingLeg)?;
        if !matches!(
            leg.status,
            LegExecutionStatus::Planned | LegExecutionStatus::Booked | LegExecutionStatus::Waiting
        ) {
            return Err(TransportError::InvalidState(
                "current leg is not startable".to_owned(),
            ));
        }
        leg.status = LegExecutionStatus::Departed;
        leg.actual_departure_at = Some(at);
        self.state = TransportExecutionState::Executing;
        Ok(())
    }

    pub fn complete_current_leg(
        &mut self,
        at: SimTime,
        endpoint: String,
    ) -> Result<bool, TransportError> {
        let active = self
            .active_itinerary_revision
            .ok_or(TransportError::MissingItinerary)?;
        let active_leg_count = self
            .legs
            .iter()
            .filter(|leg| leg.itinerary_revision == active)
            .count();
        let next_leg_index = self
            .current_leg_index
            .checked_add(1)
            .ok_or(TransportError::Overflow)?;
        let final_leg = next_leg_index >= active_leg_count;
        if final_leg && self.saga.is_none() {
            return Err(TransportError::MissingSaga);
        }
        let leg = self
            .legs
            .iter_mut()
            .find(|leg| leg.itinerary_revision == active && leg.leg_index == self.current_leg_index)
            .ok_or(TransportError::MissingLeg)?;
        if leg.status != LegExecutionStatus::Departed {
            return Err(TransportError::InvalidState(
                "current leg must be departed before arrival".to_owned(),
            ));
        }
        if leg
            .actual_departure_at
            .is_some_and(|departure| at < departure)
        {
            return Err(TransportError::InvalidState(
                "arrival precedes departure".to_owned(),
            ));
        }
        leg.status = LegExecutionStatus::Arrived;
        leg.actual_arrival_at = Some(at);
        self.current_endpoint = Some(endpoint);
        self.current_leg_index = next_leg_index;
        if self.current_leg_index >= active_leg_count {
            self.state = TransportExecutionState::ArrivalPending;
            self.mark_arrival_pending()?;
            Ok(true)
        } else {
            self.state = TransportExecutionState::Ready;
            Ok(false)
        }
    }

    pub fn fail_current_leg(&mut self, reason: String, at: SimTime) -> Result<(), TransportError> {
        self.ensure_custody_in_itinerary()?;
        let active = self
            .active_itinerary_revision
            .ok_or(TransportError::MissingItinerary)?;
        let leg = self
            .legs
            .iter_mut()
            .find(|leg| leg.itinerary_revision == active && leg.leg_index == self.current_leg_index)
            .ok_or(TransportError::MissingLeg)?;
        leg.status = LegExecutionStatus::Failed;
        leg.failed_at = Some(at);
        leg.failure_reason = Some(reason);
        self.state = TransportExecutionState::ReplanPending;
        Ok(())
    }

    pub fn mark_arrival_pending(&mut self) -> Result<(), TransportError> {
        self.ensure_custody_in_itinerary()?;
        let saga = self.saga.as_mut().ok_or(TransportError::MissingSaga)?;
        saga.step = saga.step.checked_add(1).ok_or(TransportError::Overflow)?;
        saga.state = SagaState::ArrivalPending;
        self.state = TransportExecutionState::ArrivalPending;
        Ok(())
    }

    pub fn record_handoff(&mut self, handoff: Handoff) -> Result<(), TransportError> {
        let terminal_seizure = !handoff.kind.is_planned() && handoff.from_leg == handoff.to_leg;
        if (handoff.from_leg == handoff.to_leg && !terminal_seizure)
            || handoff.from_custodian.trim().is_empty()
            || handoff.to_custodian.trim().is_empty()
            || handoff.location.trim().is_empty()
        {
            return Err(TransportError::InvalidHandoff(
                "handoff requires distinct legs, custodians, and location".to_owned(),
            ));
        }
        if let HandoffKind::Seizure {
            by: EntityRef::Domain(record),
        } = &handoff.kind
            && !domain_record_ref_is_well_formed(record)
        {
            return Err(TransportError::InvalidHandoff(
                "seizure handoff requires a well-formed seizing identity".to_owned(),
            ));
        }
        if self
            .handoffs
            .iter()
            .any(|existing| existing.id == handoff.id)
        {
            return Err(TransportError::InvalidHandoff(
                "handoff identity is already recorded".to_owned(),
            ));
        }
        if self.custody_left_itinerary() {
            return Err(TransportError::InvalidHandoff(
                "custody already left the itinerary by a terminal seizure".to_owned(),
            ));
        }
        let from_leg = self
            .legs
            .iter()
            .find(|leg| leg.id == handoff.from_leg)
            .ok_or(TransportError::MissingLeg)?;
        if terminal_seizure {
            if self.state.is_terminal()
                || Some(from_leg.itinerary_revision) != self.active_itinerary_revision
                || from_leg.leg_index != self.current_leg_index
                || from_leg.status != LegExecutionStatus::Failed
                || from_leg
                    .failed_at
                    .is_some_and(|failed_at| handoff.at < failed_at)
                || self
                    .handoffs
                    .iter()
                    .any(|existing| existing.from_leg == handoff.from_leg)
            {
                return Err(TransportError::InvalidHandoff(
                    "a terminal seizure must follow the failed current leg of a live execution, which no other handoff leaves"
                        .to_owned(),
                ));
            }
            self.handoffs.push(handoff);
            return Ok(());
        }
        let to_leg = self
            .legs
            .iter()
            .find(|leg| leg.id == handoff.to_leg)
            .ok_or(TransportError::MissingLeg)?;
        if !matches!(
            from_leg.status,
            LegExecutionStatus::Arrived | LegExecutionStatus::Failed
        ) || !matches!(
            to_leg.status,
            LegExecutionStatus::Planned | LegExecutionStatus::Booked | LegExecutionStatus::Waiting
        ) || from_leg
            .actual_arrival_at
            .is_some_and(|arrived_at| handoff.at < arrived_at)
            || from_leg
                .failed_at
                .is_some_and(|failed_at| handoff.at < failed_at)
        {
            return Err(TransportError::InvalidHandoff(
                "handoff must follow an arrived leg and precede the next leg".to_owned(),
            ));
        }
        self.handoffs.push(handoff);
        Ok(())
    }

    /// Whether a terminal seizure has taken custody out of the itinerary.
    ///
    /// Such an execution cannot fail or reroute a leg, arrive, settle
    /// successfully, book capacity, or record another handoff. Its owner
    /// closes it: an execution with a delivery saga through
    /// [`Self::reconcile_information`] with a failure, which leaves it
    /// `Failed`, and one without a saga through [`Self::cancel`].
    #[must_use]
    pub fn custody_left_itinerary(&self) -> bool {
        self.handoffs
            .iter()
            .any(|handoff| handoff.from_leg == handoff.to_leg && !handoff.kind.is_planned())
    }

    fn ensure_custody_in_itinerary(&self) -> Result<(), TransportError> {
        if self.custody_left_itinerary() {
            return Err(TransportError::InvalidState(
                "custody left the itinerary by a terminal seizure; the execution can only be closed"
                    .to_owned(),
            ));
        }
        Ok(())
    }

    /// Adds one requested capacity booking to this execution.
    ///
    /// The booking must name this execution, still be `Requested`, carry no
    /// allocation evidence, and have an identity unique within the execution.
    /// A terminal execution accepts no further bookings.
    pub fn request_booking(&mut self, booking: CapacityBooking) -> Result<(), TransportError> {
        self.ensure_custody_in_itinerary()?;
        if self.state.is_terminal() {
            return Err(TransportError::InvalidState(
                "a terminal transport execution cannot request capacity".to_owned(),
            ));
        }
        if booking.execution != self.id
            || booking.status != CapacityBookingStatus::Requested
            || !booking.allocation_evidence.is_empty()
            || booking.quantity == 0
            || booking.valid_until < booking.valid_from
        {
            return Err(TransportError::InvalidBooking(
                "a booking request must name this execution, be requested, and carry a positive quantity and window"
                    .to_owned(),
            ));
        }
        if self
            .bookings
            .iter()
            .any(|existing| existing.id == booking.id)
        {
            return Err(TransportError::InvalidBooking(
                "capacity booking identity is already recorded on this execution".to_owned(),
            ));
        }
        self.bookings.push(booking);
        Ok(())
    }

    /// Completes the final departed leg of an execution that carries no
    /// delivery attempt and settles it.
    ///
    /// A movement that does not complete an information delivery has nothing
    /// to reconcile, so arrival is its terminal fact. An execution with a
    /// delivery attempt or saga must instead use
    /// [`Self::complete_current_leg`], which enters `ArrivalPending`, and then
    /// [`Self::reconcile_information`].
    pub fn settle_arrival(&mut self, at: SimTime, endpoint: String) -> Result<(), TransportError> {
        if self.delivery_attempt.is_some() || self.saga.is_some() {
            return Err(TransportError::InvalidState(
                "an execution with a delivery attempt settles through reconciliation".to_owned(),
            ));
        }
        let active = self
            .active_itinerary_revision
            .ok_or(TransportError::MissingItinerary)?;
        let active_leg_count = self
            .legs
            .iter()
            .filter(|leg| leg.itinerary_revision == active)
            .count();
        let next_leg_index = self
            .current_leg_index
            .checked_add(1)
            .ok_or(TransportError::Overflow)?;
        if next_leg_index != active_leg_count {
            return Err(TransportError::InvalidState(
                "only the final leg of the active itinerary can settle an arrival".to_owned(),
            ));
        }
        let leg = self
            .legs
            .iter_mut()
            .find(|leg| leg.itinerary_revision == active && leg.leg_index == self.current_leg_index)
            .ok_or(TransportError::MissingLeg)?;
        if leg.status != LegExecutionStatus::Departed
            || leg
                .actual_departure_at
                .is_some_and(|departure| at < departure)
        {
            return Err(TransportError::InvalidState(
                "the final leg must have departed no later than its arrival".to_owned(),
            ));
        }
        leg.status = LegExecutionStatus::Arrived;
        leg.actual_arrival_at = Some(at);
        self.current_endpoint = Some(endpoint);
        self.current_leg_index = next_leg_index;
        self.state = TransportExecutionState::Settled;
        Ok(())
    }

    /// Cancels a non-terminal execution whose subject is not travelling.
    ///
    /// No leg of the active itinerary may be departed (a leg in progress must
    /// first arrive or fail) and an arrival-pending execution must reconcile
    /// instead. Unstarted legs of the active itinerary become `Cancelled`; a
    /// delivery saga enters `CompensationPending`. Bookings are not changed:
    /// the caller closes them through [`CapacityBooking::transition`].
    pub fn cancel(&mut self) -> Result<(), TransportError> {
        if self.state.is_terminal() || self.state == TransportExecutionState::ArrivalPending {
            return Err(TransportError::InvalidState(
                "only a non-terminal execution that has not arrived can be cancelled".to_owned(),
            ));
        }
        let active = self.active_itinerary_revision;
        if self.legs.iter().any(|leg| {
            Some(leg.itinerary_revision) == active && leg.status == LegExecutionStatus::Departed
        }) {
            return Err(TransportError::InvalidState(
                "a departed leg must arrive or fail before its execution is cancelled".to_owned(),
            ));
        }
        for leg in &mut self.legs {
            if Some(leg.itinerary_revision) == active
                && matches!(
                    leg.status,
                    LegExecutionStatus::Planned
                        | LegExecutionStatus::Booked
                        | LegExecutionStatus::Loaded
                        | LegExecutionStatus::Waiting
                )
            {
                leg.status = LegExecutionStatus::Cancelled;
            }
        }
        if let Some(saga) = self.saga.as_mut() {
            saga.step = saga.step.checked_add(1).ok_or(TransportError::Overflow)?;
            saga.state = SagaState::CompensationPending;
            saga.last_error = Some("transport execution cancelled".to_owned());
        }
        self.state = TransportExecutionState::Cancelled;
        Ok(())
    }

    pub fn completion_request(&self) -> Result<DeliveryCompletionRequest, TransportError> {
        let saga = self.saga.as_ref().ok_or(TransportError::MissingSaga)?;
        if self.state != TransportExecutionState::ArrivalPending
            || saga.state != SagaState::ArrivalPending
        {
            return Err(TransportError::InvalidState(
                "delivery completion requires an arrival-pending execution".to_owned(),
            ));
        }
        let revision = self
            .active_itinerary_revision
            .ok_or(TransportError::MissingItinerary)?;
        let completed_at = self
            .legs
            .iter()
            .rev()
            .find(|leg| leg.itinerary_revision == revision)
            .and_then(|leg| leg.actual_arrival_at)
            .ok_or(TransportError::MissingLeg)?;
        let attempt = self
            .delivery_attempt
            .clone()
            .ok_or(TransportError::MissingDeliveryAttempt)?;
        if attempt != saga.delivery_attempt {
            return Err(TransportError::InvalidState(
                "saga delivery attempt does not match execution".to_owned(),
            ));
        }
        Ok(DeliveryCompletionRequest {
            operation_key: saga.operation_key.clone(),
            execution: self.id,
            itinerary_revision: revision,
            delivery_attempt: attempt,
            completed_at,
            evidence: saga.evidence.clone(),
        })
    }

    pub fn reconcile_information(
        &mut self,
        outcome: ReconciliationOutcome,
    ) -> Result<(), TransportError> {
        if matches!(outcome, ReconciliationOutcome::Success) {
            self.ensure_custody_in_itinerary()?;
        }
        let saga = self.saga.as_mut().ok_or(TransportError::MissingSaga)?;
        saga.step = saga.step.checked_add(1).ok_or(TransportError::Overflow)?;
        match outcome {
            ReconciliationOutcome::Success => {
                saga.state = SagaState::Settled;
                saga.last_error = None;
                self.state = TransportExecutionState::Settled;
            }
            ReconciliationOutcome::Failure { error } => {
                saga.state = SagaState::CompensationPending;
                saga.last_error = Some(error);
                self.state = TransportExecutionState::Failed;
            }
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub enum TransportError {
    InvalidRevision(String),
    MissingItinerary,
    MissingSaga,
    SagaAlreadyExists,
    Overflow,
    MissingLeg,
    InvalidState(String),
    InvalidBooking(String),
    InvalidHandoff(String),
    MissingDeliveryAttempt,
    /// A capacity pool, a pool revision, or an allocation input is invalid.
    InvalidCapacityPool(String),
}

#[cfg(test)]
mod tests {
    use super::*;
    use canwu_core::{PersonId, TerritoryId};
    use canwu_routing::{
        ROUTING_ALGORITHM_VERSION, RouteCost, RouteLeg, RoutingConnectionRef, RoutingNodeRef,
        TransferMode,
    };

    fn plan() -> RoutePlan {
        RoutePlan {
            algorithm_version: ROUTING_ALGORITHM_VERSION.to_owned(),
            policy_version: "policy.v1".to_owned(),
            planning_snapshot_digest: "snapshot".to_owned(),
            origin: RoutingNodeRef::new("a"),
            destination: RoutingNodeRef::new("b"),
            departure_at: canwu_time::SimTime::EPOCH,
            estimated_arrival_at: canwu_time::SimTime::from_minutes(10),
            cost: RouteCost {
                estimated_arrival_at: canwu_time::SimTime::from_minutes(10),
                risk_per_mille: 0,
                resource_cost: 0,
                transfers: 1,
            },
            legs: vec![RouteLeg {
                connection: RoutingConnectionRef::new("ab"),
                from: RoutingNodeRef::new("a"),
                to: RoutingNodeRef::new("b"),
                mode: TransferMode::Horse,
                planned_departure_at: canwu_time::SimTime::EPOCH,
                planned_arrival_at: canwu_time::SimTime::from_minutes(10),
            }],
            digest: "route".to_owned(),
        }
    }

    #[test]
    fn movement_order_accepts_a_self_directed_person_subject() {
        let order = MovementOrder {
            id: MovementOrderId(1),
            subjects: vec![MovementSubject {
                entity: EntityRef::Person(PersonId::new(7)),
                role: MovementSubjectRole::MovablePrincipal,
                quantity: None,
                expected_custody: None,
            }],
            origin: RoutingNodeRef::new("a"),
            destination: RoutingNodeRef::new("b"),
            plan: plan(),
            initiative: MovementInitiative::SelfDirected,
            ordered_at: SimTime::EPOCH,
            expected_position_revision: 1,
        };
        order.validate().unwrap();
    }

    #[test]
    fn movement_order_rejects_duplicate_subjects_and_missing_cargo_quantity() {
        let mut order = MovementOrder {
            id: MovementOrderId(2),
            subjects: vec![
                MovementSubject {
                    entity: EntityRef::Person(PersonId::new(7)),
                    role: MovementSubjectRole::MovablePrincipal,
                    quantity: None,
                    expected_custody: None,
                },
                MovementSubject {
                    entity: EntityRef::Person(PersonId::new(7)),
                    role: MovementSubjectRole::Cargo,
                    quantity: None,
                    expected_custody: None,
                },
            ],
            origin: RoutingNodeRef::new("a"),
            destination: RoutingNodeRef::new("b"),
            plan: plan(),
            initiative: MovementInitiative::Delegated,
            ordered_at: SimTime::EPOCH,
            expected_position_revision: 1,
        };
        assert!(matches!(
            order.validate(),
            Err(MovementOrderError::Invalid(_))
        ));
        order.subjects[1].entity = EntityRef::Territory(TerritoryId::new(8));
        assert!(matches!(
            order.validate(),
            Err(MovementOrderError::Invalid(_))
        ));
    }

    #[test]
    fn movement_order_rejects_a_plan_that_does_not_match_the_order() {
        let mut order = MovementOrder {
            id: MovementOrderId(3),
            subjects: vec![MovementSubject {
                entity: EntityRef::Person(PersonId::new(9)),
                role: MovementSubjectRole::MovablePrincipal,
                quantity: None,
                expected_custody: None,
            }],
            origin: RoutingNodeRef::new("a"),
            destination: RoutingNodeRef::new("c"),
            plan: plan(),
            initiative: MovementInitiative::SelfDirected,
            ordered_at: SimTime::EPOCH,
            expected_position_revision: 1,
        };
        assert!(matches!(
            order.validate(),
            Err(MovementOrderError::Invalid(_))
        ));
        order.destination = RoutingNodeRef::new("b");
        order.plan.digest = String::new();
        assert!(matches!(
            order.validate(),
            Err(MovementOrderError::Invalid(_))
        ));
    }

    #[test]
    fn reroute_supersedes_without_creating_a_new_delivery_attempt() {
        let mut execution = TransportExecution::new(TransportExecutionId(1), None);
        execution
            .install_initial_itinerary(ItineraryRevision {
                id: ItineraryRevisionId(1),
                predecessor: None,
                plan: plan(),
                planned_at: canwu_time::SimTime::EPOCH,
                valid_from: canwu_time::SimTime::EPOCH,
                reason: ItineraryRevisionReason::Initial,
                superseded_at: None,
                evidence: Vec::new(),
            })
            .unwrap();
        let mut replacement = plan();
        replacement.destination = RoutingNodeRef::new("b");
        execution
            .reroute(
                ItineraryRevision {
                    id: ItineraryRevisionId(2),
                    predecessor: Some(ItineraryRevisionId(1)),
                    plan: replacement,
                    planned_at: canwu_time::SimTime::from_minutes(1),
                    valid_from: canwu_time::SimTime::from_minutes(1),
                    reason: ItineraryRevisionReason::Disaster {
                        explanation: "bridge closed".to_owned(),
                    },
                    superseded_at: None,
                    evidence: Vec::new(),
                },
                canwu_time::SimTime::from_minutes(1),
            )
            .unwrap();
        assert_eq!(execution.revisions.len(), 2);
        assert_eq!(
            execution.active_itinerary_revision,
            Some(ItineraryRevisionId(2))
        );
        assert_eq!(execution.state, TransportExecutionState::Planning);
    }

    #[test]
    fn capacity_booking_is_a_persisted_windowed_state_machine() {
        let mut booking = CapacityBooking::new(
            CapacityBookingId(1),
            TransportExecutionId(1),
            "relay-horse:wu-xi:01".to_owned(),
            canwu_time::SimTime::EPOCH,
            canwu_time::SimTime::from_minutes(60),
            1,
            10,
        )
        .unwrap();
        booking
            .transition(CapacityBookingStatus::Confirmed, canwu_time::SimTime::EPOCH)
            .unwrap();
        booking
            .transition(
                CapacityBookingStatus::Consumed,
                canwu_time::SimTime::from_minutes(10),
            )
            .unwrap();
        assert_eq!(booking.status, CapacityBookingStatus::Consumed);
    }

    #[test]
    fn completion_operation_key_is_stable_and_revision_scoped() {
        let first =
            delivery_completion_operation_key(TransportExecutionId(4), ItineraryRevisionId(2), 7);
        let second =
            delivery_completion_operation_key(TransportExecutionId(4), ItineraryRevisionId(3), 7);
        assert_ne!(first, second);
        assert_eq!(
            first,
            delivery_completion_operation_key(TransportExecutionId(4), ItineraryRevisionId(2), 7)
        );
    }

    #[test]
    fn completion_requires_saga_and_exposes_stable_bridge_request() {
        let attempt = DomainRecordVersionRef {
            record: canwu_core::DomainRecordRef::new(
                "fixture.information",
                "delivery_attempt",
                "delivery",
            ),
            version: 3,
            established_by: canwu_core::DomainRecordVersionSource::InitialScenario,
        };
        let mut execution = TransportExecution::new(TransportExecutionId(9), Some(attempt.clone()));
        let revision = ItineraryRevision {
            id: ItineraryRevisionId(1),
            predecessor: None,
            plan: plan(),
            planned_at: SimTime::EPOCH,
            valid_from: SimTime::EPOCH,
            reason: ItineraryRevisionReason::Initial,
            superseded_at: None,
            evidence: Vec::new(),
        };
        execution.install_initial_itinerary(revision).unwrap();
        assert_eq!(
            execution.complete_current_leg(SimTime::from_minutes(10), "b".to_owned()),
            Err(TransportError::MissingSaga)
        );
        execution
            .begin_saga(
                attempt,
                delivery_completion_operation_key(
                    TransportExecutionId(9),
                    ItineraryRevisionId(1),
                    3,
                ),
            )
            .unwrap();
        execution.start_current_leg(SimTime::EPOCH).unwrap();
        assert!(
            execution
                .complete_current_leg(SimTime::from_minutes(10), "b".to_owned())
                .unwrap()
        );
        let request = execution.completion_request().unwrap();
        assert_eq!(request.execution, TransportExecutionId(9));
        assert_eq!(request.itinerary_revision, ItineraryRevisionId(1));
        assert_eq!(request.delivery_attempt.version, 3);
    }

    #[test]
    fn handoff_requires_arrival_and_is_idempotency_safe() {
        let attempt = DomainRecordVersionRef {
            record: canwu_core::DomainRecordRef::new(
                "fixture.information",
                "delivery_attempt",
                "delivery",
            ),
            version: 1,
            established_by: canwu_core::DomainRecordVersionSource::InitialScenario,
        };
        let mut execution =
            TransportExecution::new(TransportExecutionId(10), Some(attempt.clone()));
        let mut route = plan();
        route.legs.push(RouteLeg {
            connection: RoutingConnectionRef::new("bc"),
            from: RoutingNodeRef::new("b"),
            to: RoutingNodeRef::new("c"),
            mode: TransferMode::Rail,
            planned_departure_at: SimTime::from_minutes(10),
            planned_arrival_at: SimTime::from_minutes(20),
        });
        route.destination = RoutingNodeRef::new("c");
        route.estimated_arrival_at = SimTime::from_minutes(20);
        let revision = ItineraryRevision {
            id: ItineraryRevisionId(1),
            predecessor: None,
            plan: route,
            planned_at: SimTime::EPOCH,
            valid_from: SimTime::EPOCH,
            reason: ItineraryRevisionReason::Initial,
            superseded_at: None,
            evidence: Vec::new(),
        };
        execution.install_initial_itinerary(revision).unwrap();
        execution.start_current_leg(SimTime::EPOCH).unwrap();
        execution
            .complete_current_leg(SimTime::from_minutes(10), "b".to_owned())
            .unwrap();
        let handoff = Handoff {
            id: HandoffId(1),
            from_leg: LegExecutionId(1),
            to_leg: LegExecutionId(2),
            from_custodian: "courier/wuxi".to_owned(),
            to_custodian: "rail/beijing".to_owned(),
            at: SimTime::from_minutes(10),
            location: "b".to_owned(),
            evidence: Vec::new(),
            kind: HandoffKind::Planned,
        };
        execution.record_handoff(handoff.clone()).unwrap();
        assert_eq!(
            execution.record_handoff(handoff),
            Err(TransportError::InvalidHandoff(
                "handoff identity is already recorded".to_owned()
            ))
        );
    }
}