type_bridge_schema_migration/

execution.rs

//! Provider-neutral fenced migration journal and recovery contracts.

use std::fmt;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::task::{Poll, Waker};
use std::time::Instant;

use type_bridge_contract::diagnostic::{Diagnostic, DiagnosticCategory, DiagnosticCode};
use type_bridge_contract::fingerprint::Fingerprint;
use type_bridge_contract::managed_scope::{ManagedScopeId, SemanticProfileFingerprint};
use type_bridge_contract::migration::{
    MigrationId, MigrationManifestDigest, MigrationPlanFingerprint,
};
use type_bridge_contract::schema_delta::ManagedSchemaState;
use type_bridge_contract::schema_fingerprint::{
    ManagedDeclaredIdentityFingerprint, ManagedSemanticSchemaFingerprint,
};
use type_bridge_contract::schema_lowering::SchemaLoweringProfileFingerprint;

use crate::{
    VerifiedMigrationApplyManifest, VerifiedMigrationApplyPlan, VerifiedMigrationRollbackManifest,
    VerifiedMigrationRollbackPlan, VerifiedMigrationTransactionGroup,
};

const MAX_LEASE_HOLDER_BYTES: usize = 128;

/// Shared maximum number of migration transaction groups in one invocation.
pub const MAX_MIGRATION_EXECUTION_GROUPS: usize = 65_536;
/// Shared maximum number of retained terminal backfill observations.
pub const MAX_MIGRATION_BACKFILL_OBSERVATIONS: usize = 65_536;

/// Cloneable, wakeable cancellation authority for migration operations.
#[derive(Clone, Debug, Default)]
pub struct MigrationCancellation {
    inner: Arc<MigrationCancellationInner>,
}

#[derive(Debug, Default)]
struct MigrationCancellationInner {
    cancelled: AtomicBool,
    waiters: std::sync::Mutex<Vec<Waker>>,
}

impl MigrationCancellation {
    /// Request cancellation and wake every currently registered provider wait.
    pub fn cancel(&self) {
        if !self.inner.cancelled.swap(true, Ordering::AcqRel) {
            let waiters = {
                let mut waiters = self.inner.waiters.lock().expect("cancellation waiters");
                std::mem::take(&mut *waiters)
            };
            for waiter in waiters {
                waiter.wake();
            }
        }
    }

    /// Return whether cancellation has been requested.
    #[must_use]
    pub fn is_cancelled(&self) -> bool {
        self.inner.cancelled.load(Ordering::Acquire)
    }

    /// Await cancellation without polling or spawning a worker thread.
    pub async fn cancelled(&self) {
        std::future::poll_fn(|context| {
            if self.is_cancelled() {
                return Poll::Ready(());
            }
            let mut waiters = self.inner.waiters.lock().expect("cancellation waiters");
            if self.is_cancelled() {
                return Poll::Ready(());
            }
            if !waiters
                .iter()
                .any(|waiter| waiter.will_wake(context.waker()))
            {
                waiters.push(context.waker().clone());
            }
            Poll::Pending
        })
        .await
    }
}

/// Tighten-only bounded resources for one migration execution invocation.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct MigrationExecutionResourceLimits {
    transaction_groups: usize,
    backfill_observations: usize,
}

impl MigrationExecutionResourceLimits {
    /// Construct limits clamped to the shared ceilings.
    #[must_use]
    pub const fn tightened(transaction_groups: usize, backfill_observations: usize) -> Self {
        Self {
            transaction_groups: if transaction_groups < MAX_MIGRATION_EXECUTION_GROUPS {
                transaction_groups
            } else {
                MAX_MIGRATION_EXECUTION_GROUPS
            },
            backfill_observations: if backfill_observations < MAX_MIGRATION_BACKFILL_OBSERVATIONS {
                backfill_observations
            } else {
                MAX_MIGRATION_BACKFILL_OBSERVATIONS
            },
        }
    }

    /// Return the maximum transaction groups admitted by this invocation.
    pub const fn transaction_groups(self) -> usize {
        self.transaction_groups
    }

    /// Return the maximum retained terminal backfill observations.
    pub const fn backfill_observations(self) -> usize {
        self.backfill_observations
    }
}

impl Default for MigrationExecutionResourceLimits {
    fn default() -> Self {
        Self::tightened(
            MAX_MIGRATION_EXECUTION_GROUPS,
            MAX_MIGRATION_BACKFILL_OBSERVATIONS,
        )
    }
}

/// Immutable controls shared by apply and rollback execution.
#[derive(Clone, Debug, Default)]
pub struct MigrationExecutionControl {
    cancellation: MigrationCancellation,
    deadline: Option<Instant>,
    resources: MigrationExecutionResourceLimits,
}

impl MigrationExecutionControl {
    /// Bind cancellation, one absolute monotonic deadline, and tightened limits.
    #[must_use]
    pub const fn new(
        cancellation: MigrationCancellation,
        deadline: Option<Instant>,
        resources: MigrationExecutionResourceLimits,
    ) -> Self {
        Self {
            cancellation,
            deadline,
            resources,
        }
    }

    /// Return the cancellation authority.
    pub const fn cancellation(&self) -> &MigrationCancellation {
        &self.cancellation
    }

    /// Return the absolute monotonic deadline, when bounded.
    pub const fn deadline(&self) -> Option<Instant> {
        self.deadline
    }

    /// Return caller-tightened resource limits.
    pub const fn resources(&self) -> MigrationExecutionResourceLimits {
        self.resources
    }

    /// Reject work at a safe coordinator boundary before another effect begins.
    pub fn check(&self) -> Result<(), Diagnostic> {
        if self.cancellation.is_cancelled() {
            return Err(failure(
                DiagnosticCategory::Cancelled,
                "migration_execution_cancelled",
                "migration execution was cancelled before the next effect",
            ));
        }
        if self
            .deadline
            .is_some_and(|deadline| Instant::now() >= deadline)
        {
            return Err(failure(
                DiagnosticCategory::ResourceLimit,
                "migration_execution_deadline_exceeded",
                "migration execution reached its absolute deadline before the next effect",
            ));
        }
        Ok(())
    }
}

/// Boxed future returned by provider-neutral execution stores.
pub type ExecutionFuture<'a, T> = Pin<Box<dyn Future<Output = Result<T, Diagnostic>> + Send + 'a>>;

/// A monotonically increasing store-issued migration fencing token.
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct ExecutionFence(u64);

impl ExecutionFence {
    /// Construct a non-zero fence.
    pub fn new(value: u64) -> Result<Self, Diagnostic> {
        if value == 0 {
            return Err(failure(
                DiagnosticCategory::InvalidContract,
                "migration_execution_zero_fence",
                "migration execution fences must be non-zero",
            ));
        }
        Ok(Self(value))
    }

    /// Return the numeric fence value.
    pub const fn get(self) -> u64 {
        self.0
    }

    /// Derive the next strictly greater fence without wrapping.
    pub fn checked_successor(self) -> Result<Self, Diagnostic> {
        let value = self.0.checked_add(1).ok_or_else(|| {
            failure(
                DiagnosticCategory::ResourceLimit,
                "migration_execution_fence_exhausted",
                "migration execution fence range is exhausted",
            )
        })?;
        Self::new(value)
    }
}

/// One durable managed migration scope.
#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct ExecutionScope(ManagedScopeId);

impl ExecutionScope {
    /// Bind execution to an existing managed-scope identity.
    pub const fn new(scope: ManagedScopeId) -> Self {
        Self(scope)
    }

    /// Return the managed-scope identity.
    pub const fn managed_scope_id(&self) -> &ManagedScopeId {
        &self.0
    }
}

/// Caller-supplied lease holder identity with no machine-derived garnish.
#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct LeaseHolderId(String);

impl LeaseHolderId {
    /// Validate a bounded canonical holder label.
    pub fn new(value: impl Into<String>) -> Result<Self, Diagnostic> {
        let value = value.into();
        if value.is_empty()
            || value.len() > MAX_LEASE_HOLDER_BYTES
            || !value
                .bytes()
                .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-'))
        {
            return Err(failure(
                DiagnosticCategory::InvalidContract,
                "migration_execution_invalid_holder",
                "lease holder must be bounded non-empty ASCII [A-Za-z0-9._-]",
            ));
        }
        Ok(Self(value))
    }

    /// Return the canonical holder label.
    pub fn as_str(&self) -> &str {
        &self.0
    }
}

/// Store-issued authority to mutate one migration scope.
#[derive(Clone, Debug)]
pub struct MigrationLease {
    scope: ExecutionScope,
    holder: LeaseHolderId,
    fence: ExecutionFence,
    local_binding: Option<ExecutionBindingToken>,
}

/// Opaque, process-local identity shared by one provider/store execution pair.
///
/// The token is deliberately absent from canonical and persisted migration
/// records. Provider-neutral stores continue to issue unbound leases through
/// [`MigrationLease::new`]; provider adapters that require local composition
/// integrity can mint a token and use [`MigrationLease::new_bound`].
#[doc(hidden)]
#[derive(Clone)]
pub struct ExecutionBindingToken(Arc<()>);

impl ExecutionBindingToken {
    /// Mint one fresh, unforgeable process-local identity.
    #[doc(hidden)]
    #[must_use]
    pub fn fresh() -> Self {
        Self(Arc::new(()))
    }

    fn matches(&self, other: &Self) -> bool {
        Arc::ptr_eq(&self.0, &other.0)
    }
}

impl fmt::Debug for ExecutionBindingToken {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str("ExecutionBindingToken([OPAQUE])")
    }
}

impl MigrationLease {
    /// Construct an unbound lease returned by a provider-neutral store.
    pub const fn new(scope: ExecutionScope, holder: LeaseHolderId, fence: ExecutionFence) -> Self {
        Self {
            scope,
            holder,
            fence,
            local_binding: None,
        }
    }

    /// Construct a lease carrying one process-local provider/store binding.
    #[doc(hidden)]
    #[must_use]
    pub fn new_bound(
        scope: ExecutionScope,
        holder: LeaseHolderId,
        fence: ExecutionFence,
        local_binding: ExecutionBindingToken,
    ) -> Self {
        Self {
            scope,
            holder,
            fence,
            local_binding: Some(local_binding),
        }
    }

    /// Return whether this lease carries the exact supplied local binding.
    #[doc(hidden)]
    #[must_use]
    pub fn is_bound_to(&self, expected: &ExecutionBindingToken) -> bool {
        self.local_binding
            .as_ref()
            .is_some_and(|actual| actual.matches(expected))
    }

    /// Return the leased scope.
    pub const fn scope(&self) -> &ExecutionScope {
        &self.scope
    }

    /// Return the holder identity.
    pub const fn holder(&self) -> &LeaseHolderId {
        &self.holder
    }

    /// Return the store-issued fence.
    pub const fn fence(&self) -> ExecutionFence {
        self.fence
    }
}

impl PartialEq for MigrationLease {
    fn eq(&self, other: &Self) -> bool {
        self.scope == other.scope
            && self.holder == other.holder
            && self.fence == other.fence
            && match (&self.local_binding, &other.local_binding) {
                (None, None) => true,
                (Some(left), Some(right)) => left.matches(right),
                (None, Some(_)) | (Some(_), None) => false,
            }
    }
}

impl Eq for MigrationLease {}

/// Commit certainty owned by the migration journal layer.
///
/// Provider adapters must map absent certainty information to [`Self::Unknown`].
/// Only an explicit provider proof that commit could not have occurred may map
/// to [`Self::DefinitelyAborted`].
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum GroupCommitCertainty {
    /// The provider proves that the group transaction did not commit.
    DefinitelyAborted,
    /// The provider cannot determine whether the group transaction committed.
    Unknown,
}

impl GroupCommitCertainty {
    /// Convert certainty into its durable journal event.
    pub const fn journal_event(self) -> GroupJournalEventKind {
        match self {
            Self::DefinitelyAborted => GroupJournalEventKind::DefinitelyAborted,
            Self::Unknown => GroupJournalEventKind::CommitOutcomeUnknown,
        }
    }
}

/// Durable commit-boundary event vocabulary for one transaction group.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum GroupJournalEventKind {
    /// The group statements completed and the commit call is about to begin.
    BeforeCommit,
    /// The provider commit succeeded and exact live target semantics were observed.
    Committed,
    /// The commit response cannot prove whether durability occurred.
    CommitOutcomeUnknown,
    /// The provider proves that the transaction did not commit.
    DefinitelyAborted,
    /// An empty formal-only group advanced without a provider transaction.
    FormalOnlyAdvanced,
}

/// Optional live managed-semantic evidence used during recovery.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum GroupRecoveryObservation {
    /// The provider cannot supply a trustworthy managed-semantic observation.
    Unavailable,
    /// Exact live managed semantics observed under the current fence.
    ManagedSemantics(ManagedSemanticSchemaFingerprint),
}

/// Fail-closed recovery decision for one positional transaction group.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum GroupRecoveryDecision {
    /// The group is proven absent and may execute under the current fence.
    ExecuteNormally,
    /// The target is proven reached and only the journal checkpoint needs repair.
    RepairCheckpoint,
    /// Evidence is ambiguous or contradictory and requires verified operator action.
    RequiresExplicitRecovery,
}

/// Direction of one closed backfill program at execution time.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum BackfillExecutionDirection {
    /// Execute the canonical forward data program.
    Forward,
    /// Execute its independently verified reverse program.
    Reverse,
}

/// Bounded aggregate counts from a completely verified backfill execution.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct BackfillExecutionCounts {
    matched: u64,
    changed: u64,
    skipped: u64,
    transaction_groups: u32,
}

impl BackfillExecutionCounts {
    /// Construct internally consistent terminal counts.
    pub fn new(
        matched: u64,
        changed: u64,
        skipped: u64,
        transaction_groups: u32,
    ) -> Result<Self, Diagnostic> {
        if transaction_groups == 0 {
            return Err(failure(
                DiagnosticCategory::InvalidContract,
                "migration_backfill_zero_transaction_groups",
                "terminal backfill evidence must include at least one transaction group",
            ));
        }
        if changed.checked_add(skipped) != Some(matched) {
            return Err(failure(
                DiagnosticCategory::InvalidContract,
                "migration_backfill_count_mismatch",
                "backfill matched count must equal changed plus skipped counts",
            ));
        }
        Ok(Self {
            matched,
            changed,
            skipped,
            transaction_groups,
        })
    }

    /// Return the number of source rows selected by the closed plan.
    pub const fn matched(self) -> u64 {
        self.matched
    }

    /// Return the number of destination rows changed.
    pub const fn changed(self) -> u64 {
        self.changed
    }

    /// Return the number of already-equal rows skipped idempotently.
    pub const fn skipped(self) -> u64 {
        self.skipped
    }

    /// Return the number of committed deterministic transaction groups.
    pub const fn transaction_groups(self) -> u32 {
        self.transaction_groups
    }
}

/// Exact terminal proof that one closed backfill program satisfies its postcondition.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct BackfillCompletionEvidence {
    plan_fingerprint: Fingerprint,
    direction: BackfillExecutionDirection,
    counts: BackfillExecutionCounts,
}

impl BackfillCompletionEvidence {
    /// Bind terminal counts to one exact canonical plan and execution direction.
    #[must_use]
    pub const fn new(
        plan_fingerprint: Fingerprint,
        direction: BackfillExecutionDirection,
        counts: BackfillExecutionCounts,
    ) -> Self {
        Self {
            plan_fingerprint,
            direction,
            counts,
        }
    }

    /// Return the exact canonical backfill-plan identity.
    pub const fn plan_fingerprint(&self) -> &Fingerprint {
        &self.plan_fingerprint
    }

    /// Return whether the forward or checked reverse program completed.
    pub const fn direction(&self) -> BackfillExecutionDirection {
        self.direction
    }

    /// Return internally consistent aggregate counts.
    pub const fn counts(&self) -> BackfillExecutionCounts {
        self.counts
    }
}

/// Fresh provider observation of one exact backfill postcondition.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum BackfillRecoveryObservation {
    /// No trustworthy data observation is available.
    Unavailable,
    /// The exact postcondition is not currently satisfied.
    Incomplete,
    /// The exact plan and direction have a verified terminal postcondition.
    Complete(BackfillCompletionEvidence),
}

/// Decide whether a complete backfill may execute or only repair its checkpoint.
///
/// Unlike schema groups, an incomplete data observation after a before-commit
/// or unknown-commit event cannot prove that no partition committed. Automatic
/// replay therefore remains forbidden until partition checkpoints are present.
/// With no prior event, both incomplete and already-complete postconditions are
/// safe to execute: the journal proves that execution has never begun, and an
/// already-complete closed backfill is a deterministic no-op.
pub fn decide_backfill_recovery(
    last_event: Option<GroupJournalEventKind>,
    observation: &BackfillRecoveryObservation,
    expected_plan: &Fingerprint,
    expected_direction: BackfillExecutionDirection,
) -> GroupRecoveryDecision {
    let complete = matches!(
        observation,
        BackfillRecoveryObservation::Complete(evidence)
            if evidence.plan_fingerprint() == expected_plan
                && evidence.direction() == expected_direction
    );
    match last_event {
        None if matches!(
            observation,
            BackfillRecoveryObservation::Incomplete | BackfillRecoveryObservation::Complete(_)
        ) =>
        {
            GroupRecoveryDecision::ExecuteNormally
        }
        Some(GroupJournalEventKind::DefinitelyAborted)
            if matches!(observation, BackfillRecoveryObservation::Incomplete) =>
        {
            GroupRecoveryDecision::ExecuteNormally
        }
        Some(
            GroupJournalEventKind::BeforeCommit
            | GroupJournalEventKind::CommitOutcomeUnknown
            | GroupJournalEventKind::Committed,
        ) if complete => GroupRecoveryDecision::RepairCheckpoint,
        _ => GroupRecoveryDecision::RequiresExplicitRecovery,
    }
}

/// Decide recovery from durable event and freshly observed managed semantics.
///
/// Equal source and target fingerprints are intentionally uninformative for
/// unknown commit outcomes. They never authorize replay or checkpoint repair.
pub fn decide_group_recovery(
    last_event: Option<GroupJournalEventKind>,
    observation: &GroupRecoveryObservation,
    source: &ManagedSemanticSchemaFingerprint,
    target: &ManagedSemanticSchemaFingerprint,
) -> GroupRecoveryDecision {
    let distinct = source != target;
    let observed = match observation {
        GroupRecoveryObservation::Unavailable => ObservedRelation::Unavailable,
        GroupRecoveryObservation::ManagedSemantics(value) if value == source && value == target => {
            ObservedRelation::Both
        }
        GroupRecoveryObservation::ManagedSemantics(value) if value == source => {
            ObservedRelation::Source
        }
        GroupRecoveryObservation::ManagedSemantics(value) if value == target => {
            ObservedRelation::Target
        }
        GroupRecoveryObservation::ManagedSemantics(_) => ObservedRelation::Neither,
    };
    match (last_event, distinct, observed) {
        (None, true, ObservedRelation::Source)
        | (Some(GroupJournalEventKind::DefinitelyAborted), true, ObservedRelation::Source)
        | (None, false, ObservedRelation::Both)
        | (Some(GroupJournalEventKind::DefinitelyAborted), false, ObservedRelation::Both) => {
            GroupRecoveryDecision::ExecuteNormally
        }
        (
            Some(GroupJournalEventKind::BeforeCommit | GroupJournalEventKind::CommitOutcomeUnknown),
            true,
            ObservedRelation::Source,
        ) => GroupRecoveryDecision::ExecuteNormally,
        (
            Some(GroupJournalEventKind::BeforeCommit | GroupJournalEventKind::CommitOutcomeUnknown),
            true,
            ObservedRelation::Target,
        )
        | (Some(GroupJournalEventKind::Committed), true, ObservedRelation::Target)
        | (
            Some(GroupJournalEventKind::Committed | GroupJournalEventKind::FormalOnlyAdvanced),
            false,
            ObservedRelation::Both,
        ) => GroupRecoveryDecision::RepairCheckpoint,
        _ => GroupRecoveryDecision::RequiresExplicitRecovery,
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum ObservedRelation {
    Source,
    Target,
    Both,
    Neither,
    Unavailable,
}

/// Store-assigned monotonic ordering identity for journal entries.
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct JournalSequence(u64);

impl JournalSequence {
    /// Construct a non-zero journal sequence.
    pub fn new(value: u64) -> Result<Self, Diagnostic> {
        if value == 0 {
            return Err(failure(
                DiagnosticCategory::InvalidContract,
                "migration_execution_zero_sequence",
                "journal sequence numbers must be non-zero",
            ));
        }
        Ok(Self(value))
    }

    /// Return the numeric sequence value.
    pub const fn get(self) -> u64 {
        self.0
    }
}

/// One record after the store atomically assigns its ordering sequence.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct JournalEntry<T> {
    sequence: JournalSequence,
    record: T,
}

impl<T> JournalEntry<T> {
    /// Attach a sequence allocated by the authoritative store.
    pub const fn from_store(sequence: JournalSequence, record: T) -> Self {
        Self { sequence, record }
    }

    /// Return the store ordering sequence.
    pub const fn sequence(&self) -> JournalSequence {
        self.sequence
    }

    /// Return the trusted record.
    pub const fn record(&self) -> &T {
        &self.record
    }

    /// Consume the envelope and return its record.
    pub fn into_record(self) -> T {
        self.record
    }
}

/// One trusted, still-open plan and its ordered group events.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct OpenPlanRecord {
    plan: JournalEntry<PlanRecord>,
    events: Vec<JournalEntry<GroupEventRecord>>,
    backfill_events: Vec<JournalEntry<BackfillEventRecord>>,
}

impl OpenPlanRecord {
    /// Rebuild store output while checking sequence, scope, fence, and manifest binding.
    ///
    /// The plan retains its original fence while recovery events may be written
    /// under later fences. Event fences therefore must be monotonic and no older
    /// than the plan fence; requiring equality would hide durable recovery work
    /// after the lease rolls forward.
    pub fn from_store(
        plan: JournalEntry<PlanRecord>,
        events: Vec<JournalEntry<GroupEventRecord>>,
    ) -> Result<Self, Diagnostic> {
        Self::from_store_with_backfills(plan, events, Vec::new())
    }

    /// Rebuild store output including independently journaled data-step events.
    pub fn from_store_with_backfills(
        plan: JournalEntry<PlanRecord>,
        events: Vec<JournalEntry<GroupEventRecord>>,
        backfill_events: Vec<JournalEntry<BackfillEventRecord>>,
    ) -> Result<Self, Diagnostic> {
        let mut previous = plan.sequence();
        let mut previous_fence = plan.record().fence();
        let mut ordered: Vec<(
            JournalSequence,
            ExecutionFence,
            &MigrationId,
            MigrationManifestDigest,
        )> = events
            .iter()
            .map(|event| {
                (
                    event.sequence(),
                    event.record().fence(),
                    event.record().migration_id(),
                    event.record().manifest_digest(),
                )
            })
            .chain(backfill_events.iter().map(|event| {
                (
                    event.sequence(),
                    event.record().fence(),
                    event.record().migration_id(),
                    event.record().manifest_digest(),
                )
            }))
            .collect();
        ordered.sort_by_key(|event| event.0);
        for (sequence, fence, migration_id, manifest_digest) in ordered {
            let manifest_index = plan
                .record()
                .manifest_digests()
                .iter()
                .position(|digest| digest == &manifest_digest);
            if sequence <= previous
                || fence < previous_fence
                || manifest_index.is_none_or(|index| {
                    plan.record().migration_ids().get(index) != Some(migration_id)
                })
            {
                return Err(failure(
                    DiagnosticCategory::Integrity,
                    "migration_execution_invalid_open_plan",
                    "loaded open-plan events are not ordered and bound to the plan",
                ));
            }
            previous = sequence;
            previous_fence = fence;
        }
        if events
            .iter()
            .any(|event| event.record().scope() != plan.record().scope())
            || backfill_events
                .iter()
                .any(|event| event.record().scope() != plan.record().scope())
        {
            return Err(failure(
                DiagnosticCategory::Integrity,
                "migration_execution_invalid_open_plan",
                "loaded open-plan events are not bound to the plan scope",
            ));
        }
        Ok(Self {
            plan,
            events,
            backfill_events,
        })
    }

    /// Return the sequenced plan record.
    pub const fn plan(&self) -> &JournalEntry<PlanRecord> {
        &self.plan
    }

    /// Return ordered sequenced group events.
    pub fn events(&self) -> &[JournalEntry<GroupEventRecord>] {
        &self.events
    }

    /// Return ordered sequenced backfill events.
    pub fn backfill_events(&self) -> &[JournalEntry<BackfillEventRecord>] {
        &self.backfill_events
    }
}

/// Identity-only journal record for one complete verified apply plan.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PlanRecord {
    scope: ExecutionScope,
    fence: ExecutionFence,
    source_applied: Vec<MigrationId>,
    source_frontier: Vec<MigrationId>,
    target_frontier: Vec<MigrationId>,
    migration_ids: Vec<MigrationId>,
    manifest_digests: Vec<MigrationManifestDigest>,
    manifest_plan_fingerprints: Vec<MigrationPlanFingerprint>,
    source_declared: ManagedDeclaredIdentityFingerprint,
    target_declared: ManagedDeclaredIdentityFingerprint,
    source_semantics: ManagedSemanticSchemaFingerprint,
    target_semantics: ManagedSemanticSchemaFingerprint,
    semantic_profile: SemanticProfileFingerprint,
    lowering_profile: SchemaLoweringProfileFingerprint,
    observed_live_source: ManagedSemanticSchemaFingerprint,
}

impl PlanRecord {
    /// Bind a fresh-lease ledger and live-state precondition to verified plan identities.
    pub fn from_verified_plan(
        lease: &MigrationLease,
        plan: &VerifiedMigrationApplyPlan,
        observed_applied_migrations: &[MigrationId],
        observed_live_source: &ManagedSchemaState,
    ) -> Result<Self, Diagnostic> {
        let source = plan.source_state().ok_or_else(|| {
            failure(
                DiagnosticCategory::InvalidContract,
                "migration_execution_empty_plan",
                "an executable migration plan requires a source state",
            )
        })?;
        let target = plan.target_state().ok_or_else(|| {
            failure(
                DiagnosticCategory::InvalidContract,
                "migration_execution_empty_plan",
                "an executable migration plan requires a target state",
            )
        })?;
        let first = plan.migrations().first().ok_or_else(|| {
            failure(
                DiagnosticCategory::InvalidContract,
                "migration_execution_empty_plan",
                "an executable migration plan requires at least one manifest",
            )
        })?;
        if observed_applied_migrations != plan.applied_migrations() {
            return Err(failure(
                DiagnosticCategory::Integrity,
                "migration_execution_stale_applied_set",
                "applied ledger changed after migration planning; rebuild the plan",
            ));
        }
        if observed_live_source != source {
            return Err(failure(
                DiagnosticCategory::Integrity,
                "migration_execution_stale_source_state",
                "live managed state differs from the planned source; rebuild the plan",
            ));
        }
        let scope = ExecutionScope::new(source.scope().id().clone());
        if lease.scope() != &scope || target.scope() != source.scope() {
            return Err(failure(
                DiagnosticCategory::Integrity,
                "migration_execution_scope_mismatch",
                "lease, source, and target must bind the same managed scope",
            ));
        }
        let semantic_profile = first.manifest().semantic_profile().fingerprint().clone();
        let lowering_profile = first.manifest().lowering_profile().fingerprint().clone();
        for migration in plan.migrations() {
            if migration.manifest().managed_scope().id() != scope.managed_scope_id()
                || migration.manifest().semantic_profile().fingerprint() != &semantic_profile
                || migration.manifest().lowering_profile().fingerprint() != &lowering_profile
            {
                return Err(failure(
                    DiagnosticCategory::Integrity,
                    "migration_execution_plan_binding_mismatch",
                    "planned manifests do not share exact scope and profile bindings",
                ));
            }
        }
        Ok(Self {
            scope,
            fence: lease.fence(),
            source_applied: plan.applied_migrations().to_vec(),
            source_frontier: plan.applied_frontier().to_vec(),
            target_frontier: plan.target_frontier().to_vec(),
            migration_ids: plan
                .migrations()
                .iter()
                .map(|migration| migration.manifest().id().clone())
                .collect(),
            manifest_digests: plan
                .migrations()
                .iter()
                .map(VerifiedMigrationApplyManifest::digest)
                .collect(),
            manifest_plan_fingerprints: plan
                .migrations()
                .iter()
                .map(|migration| migration.manifest().plan_fingerprint().clone())
                .collect(),
            source_declared: source.managed_declared_identity().clone(),
            target_declared: target.managed_declared_identity().clone(),
            source_semantics: source.managed_semantic_schema().clone(),
            target_semantics: target.managed_semantic_schema().clone(),
            semantic_profile,
            lowering_profile,
            observed_live_source: observed_live_source.managed_semantic_schema().clone(),
        })
    }

    /// Return the execution scope.
    pub const fn scope(&self) -> &ExecutionScope {
        &self.scope
    }
    /// Return the fence bound into this record.
    pub const fn fence(&self) -> ExecutionFence {
        self.fence
    }
    /// Return the planned source frontier.
    pub fn source_frontier(&self) -> &[MigrationId] {
        &self.source_frontier
    }
    /// Return the complete canonically ordered source applied set.
    pub fn source_applied(&self) -> &[MigrationId] {
        &self.source_applied
    }
    /// Return the planned target frontier.
    pub fn target_frontier(&self) -> &[MigrationId] {
        &self.target_frontier
    }
    /// Return ordered migration identities.
    pub fn migration_ids(&self) -> &[MigrationId] {
        &self.migration_ids
    }
    /// Return ordered canonical manifest digests.
    pub fn manifest_digests(&self) -> &[MigrationManifestDigest] {
        &self.manifest_digests
    }
    /// Return ordered manifest plan fingerprints.
    pub fn manifest_plan_fingerprints(&self) -> &[MigrationPlanFingerprint] {
        &self.manifest_plan_fingerprints
    }
    /// Return planned source managed-declared identity.
    pub const fn source_declared(&self) -> &ManagedDeclaredIdentityFingerprint {
        &self.source_declared
    }
    /// Return planned target managed-declared identity.
    pub const fn target_declared(&self) -> &ManagedDeclaredIdentityFingerprint {
        &self.target_declared
    }
    /// Return planned source managed semantics.
    pub const fn source_semantics(&self) -> &ManagedSemanticSchemaFingerprint {
        &self.source_semantics
    }
    /// Return planned target managed semantics.
    pub const fn target_semantics(&self) -> &ManagedSemanticSchemaFingerprint {
        &self.target_semantics
    }
    /// Return semantic-profile content identity.
    pub const fn semantic_profile(&self) -> &SemanticProfileFingerprint {
        &self.semantic_profile
    }
    /// Return lowering-registry content identity.
    pub const fn lowering_profile(&self) -> &SchemaLoweringProfileFingerprint {
        &self.lowering_profile
    }
    /// Return the pre-mutation observed source semantics.
    pub const fn observed_live_source(&self) -> &ManagedSemanticSchemaFingerprint {
        &self.observed_live_source
    }
}

/// Identity-only journal event for one verified transaction group.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct GroupEventRecord {
    scope: ExecutionScope,
    fence: ExecutionFence,
    manifest_digest: MigrationManifestDigest,
    migration_id: MigrationId,
    group_ordinal: u32,
    first_step_index: u32,
    schema_delta_step_index: u32,
    end_step_index: u32,
    kind: GroupJournalEventKind,
    observed_target: Option<ManagedSemanticSchemaFingerprint>,
}

impl GroupEventRecord {
    /// Derive a positional event from exact verified apply evidence.
    pub fn new(
        lease: &MigrationLease,
        migration: &VerifiedMigrationApplyManifest,
        group: &VerifiedMigrationTransactionGroup,
        kind: GroupJournalEventKind,
        observed_target: Option<ManagedSemanticSchemaFingerprint>,
    ) -> Result<Self, Diagnostic> {
        if migration.transaction_groups().get(group.ordinal()) != Some(group) {
            return Err(failure(
                DiagnosticCategory::Integrity,
                "migration_execution_foreign_group",
                "transaction group does not belong to the supplied verified manifest",
            ));
        }
        let step = migration
            .steps()
            .get(group.schema_delta_step_index())
            .ok_or_else(|| {
                failure(
                    DiagnosticCategory::Integrity,
                    "migration_execution_group_position_mismatch",
                    "transaction group delta position is outside the verified manifest",
                )
            })?;
        let delta = step
            .step()
            .as_schema_delta()
            .ok_or_else(|| {
                failure(
                    DiagnosticCategory::Integrity,
                    "migration_execution_group_position_mismatch",
                    "transaction group does not terminate in a schema delta",
                )
            })?
            .delta();
        let lowering = step.lowering().ok_or_else(|| {
            failure(
                DiagnosticCategory::Integrity,
                "migration_execution_group_lowering_missing",
                "transaction group delta has no verified lowering",
            )
        })?;
        let scope = ExecutionScope::new(migration.manifest().managed_scope().id().clone());
        if lease.scope() != &scope {
            return Err(failure(
                DiagnosticCategory::Integrity,
                "migration_execution_scope_mismatch",
                "lease scope differs from the verified migration scope",
            ));
        }
        match kind {
            GroupJournalEventKind::Committed
                if observed_target.as_ref() == Some(delta.target().managed_semantic_schema()) => {}
            GroupJournalEventKind::Committed => {
                return Err(failure(
                    DiagnosticCategory::Integrity,
                    "migration_execution_commit_evidence_mismatch",
                    "committed event requires the exact observed target semantics",
                ));
            }
            GroupJournalEventKind::FormalOnlyAdvanced
                if observed_target.is_none()
                    && group.assertion_count() == 0
                    && lowering.units().is_empty()
                    && delta.source().managed_semantic_schema()
                        == delta.target().managed_semantic_schema() => {}
            GroupJournalEventKind::FormalOnlyAdvanced => {
                return Err(failure(
                    DiagnosticCategory::InvalidContract,
                    "migration_execution_invalid_formal_advance",
                    "formal-only advancement requires an assertion-free empty equal-semantic group",
                ));
            }
            _ if observed_target.is_none() => {}
            _ => {
                return Err(failure(
                    DiagnosticCategory::InvalidContract,
                    "migration_execution_unexpected_observation",
                    "only committed events may carry observed target semantics",
                ));
            }
        }
        Ok(Self {
            scope,
            fence: lease.fence(),
            manifest_digest: migration.digest(),
            migration_id: migration.manifest().id().clone(),
            group_ordinal: position(group.ordinal())?,
            first_step_index: position(group.first_step_index())?,
            schema_delta_step_index: position(group.schema_delta_step_index())?,
            end_step_index: position(group.end_step_index())?,
            kind,
            observed_target,
        })
    }

    /// Return the execution scope.
    pub const fn scope(&self) -> &ExecutionScope {
        &self.scope
    }
    /// Return the event fence.
    pub const fn fence(&self) -> ExecutionFence {
        self.fence
    }
    /// Return the canonical manifest digest.
    pub const fn manifest_digest(&self) -> MigrationManifestDigest {
        self.manifest_digest
    }
    /// Return the migration identity.
    pub const fn migration_id(&self) -> &MigrationId {
        &self.migration_id
    }
    /// Return the group ordinal.
    pub const fn group_ordinal(&self) -> u32 {
        self.group_ordinal
    }
    /// Return the first group step index.
    pub const fn first_step_index(&self) -> u32 {
        self.first_step_index
    }
    /// Return the terminal delta step index.
    pub const fn schema_delta_step_index(&self) -> u32 {
        self.schema_delta_step_index
    }
    /// Return the exclusive group step end.
    pub const fn end_step_index(&self) -> u32 {
        self.end_step_index
    }
    /// Return the event kind.
    pub const fn kind(&self) -> GroupJournalEventKind {
        self.kind
    }
    /// Return exact target evidence carried only by committed events.
    pub const fn observed_target(&self) -> Option<&ManagedSemanticSchemaFingerprint> {
        self.observed_target.as_ref()
    }
}

/// Durable positional event for one exact forward or reverse backfill step.
///
/// Data effects cannot be inferred from unchanged schema semantics. This
/// record therefore carries the canonical plan identity on every boundary and
/// admits terminal counts only when the provider has verified the complete
/// postcondition.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct BackfillEventRecord {
    scope: ExecutionScope,
    fence: ExecutionFence,
    manifest_digest: MigrationManifestDigest,
    migration_id: MigrationId,
    operation_ordinal: u32,
    manifest_step_index: u32,
    plan_fingerprint: Fingerprint,
    direction: BackfillExecutionDirection,
    kind: GroupJournalEventKind,
    completion: Option<BackfillCompletionEvidence>,
}

impl BackfillEventRecord {
    /// Derive a forward event from one exact verified apply step position.
    pub fn new_apply(
        lease: &MigrationLease,
        migration: &VerifiedMigrationApplyManifest,
        step_index: usize,
        kind: GroupJournalEventKind,
        completion: Option<BackfillCompletionEvidence>,
    ) -> Result<Self, Diagnostic> {
        if lease.scope().managed_scope_id() != migration.manifest().managed_scope().id() {
            return Err(failure(
                DiagnosticCategory::Integrity,
                "migration_execution_scope_mismatch",
                "lease scope differs from the verified migration scope",
            ));
        }
        if !migration.backfill_step_indices().contains(&step_index) {
            return Err(failure(
                DiagnosticCategory::Integrity,
                "migration_execution_backfill_step_position",
                "backfill event position is outside the verified apply manifest",
            ));
        }
        let step = migration.steps().get(step_index).ok_or_else(|| {
            failure(
                DiagnosticCategory::Integrity,
                "migration_execution_backfill_step_position",
                "backfill event position is outside the verified apply manifest",
            )
        })?;
        let (contract, _) = step.step().as_backfill().ok_or_else(|| {
            failure(
                DiagnosticCategory::Integrity,
                "migration_execution_backfill_step_position",
                "verified backfill position does not contain a backfill step",
            )
        })?;
        Self::new_checked(
            lease,
            migration.digest(),
            migration.manifest().id().clone(),
            step_index,
            step_index,
            contract.plan_fingerprint().clone(),
            BackfillExecutionDirection::Forward,
            kind,
            completion,
        )
    }

    /// Derive a reverse event from one exact mixed rollback operation position.
    pub fn new_rollback(
        lease: &MigrationLease,
        rollback: &VerifiedMigrationRollbackManifest,
        operation_index: usize,
        kind: GroupJournalEventKind,
        completion: Option<BackfillCompletionEvidence>,
    ) -> Result<Self, Diagnostic> {
        if lease.scope().managed_scope_id() != rollback.manifest().managed_scope().id() {
            return Err(failure(
                DiagnosticCategory::Integrity,
                "migration_execution_scope_mismatch",
                "lease scope differs from the verified rollback scope",
            ));
        }
        let backfill_index = match rollback.operations().get(operation_index) {
            Some(crate::VerifiedMigrationRollbackOperation::Backfill(index)) => *index,
            _ => {
                return Err(failure(
                    DiagnosticCategory::Integrity,
                    "migration_execution_backfill_step_position",
                    "backfill event position is outside the verified rollback operations",
                ));
            }
        };
        let step = rollback
            .backfills()
            .get(backfill_index)
            .ok_or_else(|| {
                failure(
                    DiagnosticCategory::Integrity,
                    "migration_execution_backfill_step_position",
                    "rollback backfill index is outside the verified reverse program",
                )
            })?
            .forward_step();
        let (contract, _) = step.as_backfill().ok_or_else(|| {
            failure(
                DiagnosticCategory::Integrity,
                "migration_execution_backfill_step_position",
                "verified rollback backfill does not contain a backfill step",
            )
        })?;
        let manifest_step_index = rollback
            .manifest()
            .steps()
            .iter()
            .position(|candidate| candidate == step)
            .ok_or_else(|| {
                failure(
                    DiagnosticCategory::Integrity,
                    "migration_execution_backfill_step_position",
                    "rollback backfill is absent from its verified manifest",
                )
            })?;
        Self::new_checked(
            lease,
            *rollback.digest(),
            rollback.manifest().id().clone(),
            operation_index,
            manifest_step_index,
            contract.plan_fingerprint().clone(),
            BackfillExecutionDirection::Reverse,
            kind,
            completion,
        )
    }

    #[allow(clippy::too_many_arguments)]
    fn new_checked(
        lease: &MigrationLease,
        manifest_digest: MigrationManifestDigest,
        migration_id: MigrationId,
        operation_ordinal: usize,
        manifest_step_index: usize,
        plan_fingerprint: Fingerprint,
        direction: BackfillExecutionDirection,
        kind: GroupJournalEventKind,
        completion: Option<BackfillCompletionEvidence>,
    ) -> Result<Self, Diagnostic> {
        match kind {
            GroupJournalEventKind::Committed
                if completion.as_ref().is_some_and(|evidence| {
                    evidence.plan_fingerprint() == &plan_fingerprint
                        && evidence.direction() == direction
                }) => {}
            GroupJournalEventKind::Committed => {
                return Err(failure(
                    DiagnosticCategory::Integrity,
                    "migration_execution_backfill_completion_mismatch",
                    "committed backfill event requires exact terminal plan evidence",
                ));
            }
            GroupJournalEventKind::FormalOnlyAdvanced => {
                return Err(failure(
                    DiagnosticCategory::InvalidContract,
                    "migration_execution_backfill_formal_advance",
                    "a data backfill cannot advance as a formal-only operation",
                ));
            }
            _ if completion.is_none() => {}
            _ => {
                return Err(failure(
                    DiagnosticCategory::InvalidContract,
                    "migration_execution_unexpected_backfill_completion",
                    "only a committed backfill event may carry terminal evidence",
                ));
            }
        }
        Ok(Self {
            scope: lease.scope().clone(),
            fence: lease.fence(),
            manifest_digest,
            migration_id,
            operation_ordinal: position(operation_ordinal)?,
            manifest_step_index: position(manifest_step_index)?,
            plan_fingerprint,
            direction,
            kind,
            completion,
        })
    }

    /// Return the execution scope.
    pub const fn scope(&self) -> &ExecutionScope {
        &self.scope
    }

    /// Return the event fence.
    pub const fn fence(&self) -> ExecutionFence {
        self.fence
    }

    /// Return the exact manifest digest.
    pub const fn manifest_digest(&self) -> MigrationManifestDigest {
        self.manifest_digest
    }

    /// Return the migration identity.
    pub const fn migration_id(&self) -> &MigrationId {
        &self.migration_id
    }

    /// Return the position in forward steps or mixed reverse operations.
    pub const fn operation_ordinal(&self) -> u32 {
        self.operation_ordinal
    }

    /// Return the original canonical manifest step position.
    pub const fn manifest_step_index(&self) -> u32 {
        self.manifest_step_index
    }

    /// Return the exact canonical backfill plan identity.
    pub const fn plan_fingerprint(&self) -> &Fingerprint {
        &self.plan_fingerprint
    }

    /// Return the execution direction.
    pub const fn direction(&self) -> BackfillExecutionDirection {
        self.direction
    }

    /// Return the commit-boundary event kind.
    pub const fn kind(&self) -> GroupJournalEventKind {
        self.kind
    }

    /// Return terminal evidence carried only by committed events.
    pub const fn completion(&self) -> Option<&BackfillCompletionEvidence> {
        self.completion.as_ref()
    }
}

/// Identity-only applied-ledger record for one verified manifest.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct AppliedRecord {
    scope: ExecutionScope,
    fence: ExecutionFence,
    migration_id: MigrationId,
    manifest_digest: MigrationManifestDigest,
    source_declared: ManagedDeclaredIdentityFingerprint,
    target_declared: ManagedDeclaredIdentityFingerprint,
    source_semantics: ManagedSemanticSchemaFingerprint,
    target_semantics: ManagedSemanticSchemaFingerprint,
}

impl AppliedRecord {
    /// Derive an applied-ledger record from one exact verified manifest.
    pub fn from_verified_manifest(
        lease: &MigrationLease,
        migration: &VerifiedMigrationApplyManifest,
    ) -> Result<Self, Diagnostic> {
        Self::from_verified_manifest_contract(lease, migration.manifest())
    }

    /// Derive an applied-ledger record from one verified manifest contract.
    ///
    /// This is the reconstruction seam for persistent stores. The manifest
    /// remains the trust anchor: its digest is recomputed from verified bytes,
    /// and no persisted record claim enters the trusted value.
    pub fn from_verified_manifest_contract(
        lease: &MigrationLease,
        manifest: &crate::VerifiedSchemaMigrationManifest,
    ) -> Result<Self, Diagnostic> {
        let source = manifest.source_state();
        let target = manifest.target_state();
        let scope = ExecutionScope::new(source.scope().id().clone());
        if lease.scope() != &scope || target.scope() != source.scope() {
            return Err(failure(
                DiagnosticCategory::Integrity,
                "migration_execution_scope_mismatch",
                "lease and manifest endpoints must bind the same managed scope",
            ));
        }
        Ok(Self {
            scope,
            fence: lease.fence(),
            migration_id: manifest.id().clone(),
            manifest_digest: crate::verified_manifest_digest(manifest)?,
            source_declared: source.managed_declared_identity().clone(),
            target_declared: target.managed_declared_identity().clone(),
            source_semantics: source.managed_semantic_schema().clone(),
            target_semantics: target.managed_semantic_schema().clone(),
        })
    }

    /// Return the execution scope.
    pub const fn scope(&self) -> &ExecutionScope {
        &self.scope
    }
    /// Return the record fence.
    pub const fn fence(&self) -> ExecutionFence {
        self.fence
    }
    /// Return the migration identity.
    pub const fn migration_id(&self) -> &MigrationId {
        &self.migration_id
    }
    /// Return the canonical manifest digest.
    pub const fn manifest_digest(&self) -> MigrationManifestDigest {
        self.manifest_digest
    }
    /// Return source managed-declared identity.
    pub const fn source_declared(&self) -> &ManagedDeclaredIdentityFingerprint {
        &self.source_declared
    }
    /// Return target managed-declared identity.
    pub const fn target_declared(&self) -> &ManagedDeclaredIdentityFingerprint {
        &self.target_declared
    }
    /// Return source managed semantics.
    pub const fn source_semantics(&self) -> &ManagedSemanticSchemaFingerprint {
        &self.source_semantics
    }
    /// Return target managed semantics.
    pub const fn target_semantics(&self) -> &ManagedSemanticSchemaFingerprint {
        &self.target_semantics
    }
}

/// Identity-only journal record for one complete verified rollback plan.
///
/// The record binds the exact reverse-topological order, manifest digests,
/// surviving applied set, and endpoint managed states the rollback executes
/// under. It is the rollback analogue of [`PlanRecord`].
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct RollbackPlanRecord {
    scope: ExecutionScope,
    fence: ExecutionFence,
    source_applied: Vec<MigrationId>,
    rollback_ids: Vec<MigrationId>,
    manifest_digests: Vec<MigrationManifestDigest>,
    manifest_plan_fingerprints: Vec<MigrationPlanFingerprint>,
    remaining_applied: Vec<MigrationId>,
    source_declared: ManagedDeclaredIdentityFingerprint,
    target_declared: ManagedDeclaredIdentityFingerprint,
    source_semantics: ManagedSemanticSchemaFingerprint,
    target_semantics: ManagedSemanticSchemaFingerprint,
    semantic_profile: SemanticProfileFingerprint,
    lowering_profile: SchemaLoweringProfileFingerprint,
    observed_live_source: ManagedSemanticSchemaFingerprint,
}

impl RollbackPlanRecord {
    /// Bind a fresh-lease rollback ledger and live-state precondition to
    /// verified rollback plan identities.
    pub fn from_verified_rollback_plan(
        lease: &MigrationLease,
        plan: &VerifiedMigrationRollbackPlan,
        observed_applied_migrations: &[MigrationId],
        observed_live_source: &ManagedSchemaState,
    ) -> Result<Self, Diagnostic> {
        let first = plan.rollbacks().first().ok_or_else(|| {
            failure(
                DiagnosticCategory::InvalidContract,
                "migration_execution_empty_plan",
                "an executable rollback plan requires at least one manifest",
            )
        })?;
        let basis: Vec<MigrationId> = plan.applied_basis().into_iter().collect();
        if observed_applied_migrations != basis {
            return Err(failure(
                DiagnosticCategory::Integrity,
                "migration_execution_stale_applied_set",
                "applied ledger changed after rollback planning; rebuild the plan",
            ));
        }
        if observed_live_source != plan.source_state() {
            return Err(failure(
                DiagnosticCategory::Integrity,
                "migration_execution_stale_source_state",
                "live managed state differs from the planned source; rebuild the plan",
            ));
        }
        let scope = ExecutionScope::new(plan.source_state().scope().id().clone());
        if lease.scope() != &scope || plan.target_state().scope() != plan.source_state().scope() {
            return Err(failure(
                DiagnosticCategory::Integrity,
                "migration_execution_scope_mismatch",
                "lease, source, and target must bind the same managed scope",
            ));
        }
        let semantic_profile = first.manifest().semantic_profile().fingerprint().clone();
        let lowering_profile = first.manifest().lowering_profile().fingerprint().clone();
        for rollback in plan.rollbacks() {
            if rollback.manifest().managed_scope().id() != scope.managed_scope_id()
                || rollback.manifest().semantic_profile().fingerprint() != &semantic_profile
                || rollback.manifest().lowering_profile().fingerprint() != &lowering_profile
            {
                return Err(failure(
                    DiagnosticCategory::Integrity,
                    "migration_execution_plan_binding_mismatch",
                    "planned rollbacks do not share exact scope and profile bindings",
                ));
            }
        }
        Ok(Self {
            scope,
            fence: lease.fence(),
            source_applied: basis,
            rollback_ids: plan
                .rollbacks()
                .iter()
                .map(|rollback| rollback.manifest().id().clone())
                .collect(),
            manifest_digests: plan
                .rollbacks()
                .iter()
                .map(|rollback| *rollback.digest())
                .collect(),
            manifest_plan_fingerprints: plan
                .rollbacks()
                .iter()
                .map(|rollback| rollback.manifest().plan_fingerprint().clone())
                .collect(),
            remaining_applied: plan.remaining_applied().to_vec(),
            source_declared: plan.source_state().managed_declared_identity().clone(),
            target_declared: plan.target_state().managed_declared_identity().clone(),
            source_semantics: plan.source_state().managed_semantic_schema().clone(),
            target_semantics: plan.target_state().managed_semantic_schema().clone(),
            semantic_profile,
            lowering_profile,
            observed_live_source: observed_live_source.managed_semantic_schema().clone(),
        })
    }

    /// Return the execution scope.
    pub const fn scope(&self) -> &ExecutionScope {
        &self.scope
    }
    /// Return the fence bound into this record.
    pub const fn fence(&self) -> ExecutionFence {
        self.fence
    }
    /// Return the complete canonically ordered pre-rollback applied set.
    pub fn source_applied(&self) -> &[MigrationId] {
        &self.source_applied
    }
    /// Return rolled-back identities in reverse-topological execution order.
    pub fn rollback_ids(&self) -> &[MigrationId] {
        &self.rollback_ids
    }
    /// Return ordered canonical manifest digests.
    pub fn manifest_digests(&self) -> &[MigrationManifestDigest] {
        &self.manifest_digests
    }
    /// Return ordered manifest plan fingerprints.
    pub fn manifest_plan_fingerprints(&self) -> &[MigrationPlanFingerprint] {
        &self.manifest_plan_fingerprints
    }
    /// Return the applied identities that survive the rollback, in order.
    pub fn remaining_applied(&self) -> &[MigrationId] {
        &self.remaining_applied
    }
    /// Return planned pre-rollback managed-declared identity.
    pub const fn source_declared(&self) -> &ManagedDeclaredIdentityFingerprint {
        &self.source_declared
    }
    /// Return planned restored managed-declared identity.
    pub const fn target_declared(&self) -> &ManagedDeclaredIdentityFingerprint {
        &self.target_declared
    }
    /// Return planned pre-rollback managed semantics.
    pub const fn source_semantics(&self) -> &ManagedSemanticSchemaFingerprint {
        &self.source_semantics
    }
    /// Return planned restored managed semantics.
    pub const fn target_semantics(&self) -> &ManagedSemanticSchemaFingerprint {
        &self.target_semantics
    }
    /// Return semantic-profile content identity.
    pub const fn semantic_profile(&self) -> &SemanticProfileFingerprint {
        &self.semantic_profile
    }
    /// Return lowering-registry content identity.
    pub const fn lowering_profile(&self) -> &SchemaLoweringProfileFingerprint {
        &self.lowering_profile
    }
    /// Return the pre-mutation observed source semantics.
    pub const fn observed_live_source(&self) -> &ManagedSemanticSchemaFingerprint {
        &self.observed_live_source
    }
}

/// Identity-only journal event for one verified rollback step transaction.
///
/// Each rollback step executes one recorded reverse program in its own
/// provider transaction, so the commit-boundary vocabulary and recovery
/// decision table are shared with forward [`GroupEventRecord`] events.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct RollbackStepEventRecord {
    scope: ExecutionScope,
    fence: ExecutionFence,
    manifest_digest: MigrationManifestDigest,
    migration_id: MigrationId,
    step_ordinal: u32,
    kind: GroupJournalEventKind,
    observed_target: Option<ManagedSemanticSchemaFingerprint>,
}

impl RollbackStepEventRecord {
    /// Derive a positional event from exact verified rollback evidence.
    pub fn new(
        lease: &MigrationLease,
        rollback: &VerifiedMigrationRollbackManifest,
        step_index: usize,
        kind: GroupJournalEventKind,
        observed_target: Option<ManagedSemanticSchemaFingerprint>,
    ) -> Result<Self, Diagnostic> {
        let step = rollback.steps().get(step_index).ok_or_else(|| {
            failure(
                DiagnosticCategory::Integrity,
                "migration_execution_rollback_step_position",
                "rollback event position is outside the verified rollback manifest",
            )
        })?;
        let reverse = rollback.reverse_delta(step)?;
        let scope = ExecutionScope::new(rollback.manifest().managed_scope().id().clone());
        if lease.scope() != &scope {
            return Err(failure(
                DiagnosticCategory::Integrity,
                "migration_execution_scope_mismatch",
                "lease scope differs from the verified rollback scope",
            ));
        }
        match kind {
            GroupJournalEventKind::Committed
                if observed_target.as_ref() == Some(reverse.target().managed_semantic_schema()) => {
            }
            GroupJournalEventKind::Committed => {
                return Err(failure(
                    DiagnosticCategory::Integrity,
                    "migration_execution_commit_evidence_mismatch",
                    "committed event requires the exact observed target semantics",
                ));
            }
            GroupJournalEventKind::FormalOnlyAdvanced
                if observed_target.is_none()
                    && step.lowering().units().is_empty()
                    && reverse.source().managed_semantic_schema()
                        == reverse.target().managed_semantic_schema() => {}
            GroupJournalEventKind::FormalOnlyAdvanced => {
                return Err(failure(
                    DiagnosticCategory::InvalidContract,
                    "migration_execution_invalid_formal_advance",
                    "formal-only advancement requires an empty equal-semantic reverse program",
                ));
            }
            _ if observed_target.is_none() => {}
            _ => {
                return Err(failure(
                    DiagnosticCategory::InvalidContract,
                    "migration_execution_unexpected_observation",
                    "only committed events may carry observed target semantics",
                ));
            }
        }
        Ok(Self {
            scope,
            fence: lease.fence(),
            manifest_digest: *rollback.digest(),
            migration_id: rollback.manifest().id().clone(),
            step_ordinal: position(step_index)?,
            kind,
            observed_target,
        })
    }

    /// Return the execution scope.
    pub const fn scope(&self) -> &ExecutionScope {
        &self.scope
    }
    /// Return the event fence.
    pub const fn fence(&self) -> ExecutionFence {
        self.fence
    }
    /// Return the canonical digest of the manifest being rolled back.
    pub const fn manifest_digest(&self) -> MigrationManifestDigest {
        self.manifest_digest
    }
    /// Return the migration identity.
    pub const fn migration_id(&self) -> &MigrationId {
        &self.migration_id
    }
    /// Return the rollback step position in execution order.
    pub const fn step_ordinal(&self) -> u32 {
        self.step_ordinal
    }
    /// Return the event kind.
    pub const fn kind(&self) -> GroupJournalEventKind {
        self.kind
    }
    /// Return exact target evidence carried only by committed events.
    pub const fn observed_target(&self) -> Option<&ManagedSemanticSchemaFingerprint> {
        self.observed_target.as_ref()
    }
}

/// Identity-only retirement record for one rolled-back applied migration.
///
/// Retirement is append-only history: the applied record stays in the durable
/// journal, and this record marks it inactive. The states enter swapped
/// relative to [`AppliedRecord`] because rolling back moves the managed schema
/// from the manifest's target back to its source.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct RolledBackRecord {
    scope: ExecutionScope,
    fence: ExecutionFence,
    migration_id: MigrationId,
    manifest_digest: MigrationManifestDigest,
    source_declared: ManagedDeclaredIdentityFingerprint,
    target_declared: ManagedDeclaredIdentityFingerprint,
    source_semantics: ManagedSemanticSchemaFingerprint,
    target_semantics: ManagedSemanticSchemaFingerprint,
}

impl RolledBackRecord {
    /// Derive a retirement record from one exact verified rollback manifest.
    pub fn from_verified_rollback(
        lease: &MigrationLease,
        rollback: &VerifiedMigrationRollbackManifest,
    ) -> Result<Self, Diagnostic> {
        Self::from_verified_manifest_contract(lease, rollback.manifest())
    }

    /// Derive a retirement record from one verified manifest contract.
    ///
    /// This is the reconstruction seam for persistent stores, mirroring
    /// [`AppliedRecord::from_verified_manifest_contract`].
    pub fn from_verified_manifest_contract(
        lease: &MigrationLease,
        manifest: &crate::VerifiedSchemaMigrationManifest,
    ) -> Result<Self, Diagnostic> {
        let source = manifest.target_state();
        let target = manifest.source_state();
        let scope = ExecutionScope::new(source.scope().id().clone());
        if lease.scope() != &scope || target.scope() != source.scope() {
            return Err(failure(
                DiagnosticCategory::Integrity,
                "migration_execution_scope_mismatch",
                "lease and manifest endpoints must bind the same managed scope",
            ));
        }
        Ok(Self {
            scope,
            fence: lease.fence(),
            migration_id: manifest.id().clone(),
            manifest_digest: crate::verified_manifest_digest(manifest)?,
            source_declared: source.managed_declared_identity().clone(),
            target_declared: target.managed_declared_identity().clone(),
            source_semantics: source.managed_semantic_schema().clone(),
            target_semantics: target.managed_semantic_schema().clone(),
        })
    }

    /// Return the execution scope.
    pub const fn scope(&self) -> &ExecutionScope {
        &self.scope
    }
    /// Return the record fence.
    pub const fn fence(&self) -> ExecutionFence {
        self.fence
    }
    /// Return the retired migration identity.
    pub const fn migration_id(&self) -> &MigrationId {
        &self.migration_id
    }
    /// Return the canonical manifest digest.
    pub const fn manifest_digest(&self) -> MigrationManifestDigest {
        self.manifest_digest
    }
    /// Return pre-rollback managed-declared identity.
    pub const fn source_declared(&self) -> &ManagedDeclaredIdentityFingerprint {
        &self.source_declared
    }
    /// Return restored managed-declared identity.
    pub const fn target_declared(&self) -> &ManagedDeclaredIdentityFingerprint {
        &self.target_declared
    }
    /// Return pre-rollback managed semantics.
    pub const fn source_semantics(&self) -> &ManagedSemanticSchemaFingerprint {
        &self.source_semantics
    }
    /// Return restored managed semantics.
    pub const fn target_semantics(&self) -> &ManagedSemanticSchemaFingerprint {
        &self.target_semantics
    }
}

/// One trusted, still-open rollback plan and its ordered step events.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct OpenRollbackPlanRecord {
    plan: JournalEntry<RollbackPlanRecord>,
    events: Vec<JournalEntry<RollbackStepEventRecord>>,
    backfill_events: Vec<JournalEntry<BackfillEventRecord>>,
}

impl OpenRollbackPlanRecord {
    /// Rebuild store output while checking sequence, scope, fence, and manifest binding.
    ///
    /// The same monotonic-fence rule as [`OpenPlanRecord::from_store`] applies:
    /// recovery events may be written under later fences than the plan itself.
    pub fn from_store(
        plan: JournalEntry<RollbackPlanRecord>,
        events: Vec<JournalEntry<RollbackStepEventRecord>>,
    ) -> Result<Self, Diagnostic> {
        Self::from_store_with_backfills(plan, events, Vec::new())
    }

    /// Rebuild store output including independently journaled reverse data-step events.
    pub fn from_store_with_backfills(
        plan: JournalEntry<RollbackPlanRecord>,
        events: Vec<JournalEntry<RollbackStepEventRecord>>,
        backfill_events: Vec<JournalEntry<BackfillEventRecord>>,
    ) -> Result<Self, Diagnostic> {
        let mut previous = plan.sequence();
        let mut previous_fence = plan.record().fence();
        let mut ordered: Vec<(
            JournalSequence,
            ExecutionFence,
            &MigrationId,
            MigrationManifestDigest,
        )> = events
            .iter()
            .map(|event| {
                (
                    event.sequence(),
                    event.record().fence(),
                    event.record().migration_id(),
                    event.record().manifest_digest(),
                )
            })
            .chain(backfill_events.iter().map(|event| {
                (
                    event.sequence(),
                    event.record().fence(),
                    event.record().migration_id(),
                    event.record().manifest_digest(),
                )
            }))
            .collect();
        ordered.sort_by_key(|event| event.0);
        for (sequence, fence, migration_id, manifest_digest) in ordered {
            let manifest_index = plan
                .record()
                .manifest_digests()
                .iter()
                .position(|digest| digest == &manifest_digest);
            if sequence <= previous
                || fence < previous_fence
                || manifest_index.is_none_or(|index| {
                    plan.record().rollback_ids().get(index) != Some(migration_id)
                })
            {
                return Err(failure(
                    DiagnosticCategory::Integrity,
                    "migration_execution_invalid_open_plan",
                    "loaded open-rollback events are not ordered and bound to the plan",
                ));
            }
            previous = sequence;
            previous_fence = fence;
        }
        if events
            .iter()
            .any(|event| event.record().scope() != plan.record().scope())
            || backfill_events
                .iter()
                .any(|event| event.record().scope() != plan.record().scope())
        {
            return Err(failure(
                DiagnosticCategory::Integrity,
                "migration_execution_invalid_open_plan",
                "loaded open-rollback events are not bound to the plan scope",
            ));
        }
        Ok(Self {
            plan,
            events,
            backfill_events,
        })
    }

    /// Return the sequenced rollback plan record.
    pub const fn plan(&self) -> &JournalEntry<RollbackPlanRecord> {
        &self.plan
    }

    /// Return ordered sequenced rollback step events.
    pub fn events(&self) -> &[JournalEntry<RollbackStepEventRecord>] {
        &self.events
    }

    /// Return ordered sequenced reverse backfill events.
    pub fn backfill_events(&self) -> &[JournalEntry<BackfillEventRecord>] {
        &self.backfill_events
    }
}

/// Store-backed exclusive lease service.
pub trait MigrationLeaseStore: Send + Sync {
    /// Atomically acquire an available scope and issue a fence strictly greater
    /// than every fence previously issued for that scope.
    ///
    /// Lease expiry, liveness, and takeover policy are store-defined. Every
    /// takeover after release, expiry, or failure must issue a strictly greater
    /// fence so a surviving stale holder cannot mutate the journal.
    fn acquire<'a>(
        &'a self,
        scope: &'a ExecutionScope,
        holder: &'a LeaseHolderId,
    ) -> ExecutionFuture<'a, MigrationLease>;

    /// Release only the exact currently active holder and fence.
    fn release<'a>(&'a self, lease: &'a MigrationLease) -> ExecutionFuture<'a, ()>;
}

/// Durable applied ledger and open-plan journal.
///
/// Every write must atomically reject a lease that is not the current active
/// scope, holder, and fence. There is deliberately no advisory validate method.
pub trait MigrationExecutionJournal: Send + Sync {
    /// Begin one fully verified plan after stale-ledger and live-source checks.
    ///
    /// An open plan of either direction is exclusive: the store must reject a
    /// new apply plan while a rollback plan is open and vice versa.
    fn begin_plan<'a>(
        &'a self,
        lease: &'a MigrationLease,
        record: PlanRecord,
    ) -> ExecutionFuture<'a, JournalEntry<PlanRecord>>;

    /// Append one positional commit-boundary event under the active fence.
    fn record_group_event<'a>(
        &'a self,
        lease: &'a MigrationLease,
        record: GroupEventRecord,
    ) -> ExecutionFuture<'a, JournalEntry<GroupEventRecord>>;

    /// Append one exact forward or reverse backfill commit-boundary event.
    fn record_backfill_event<'a>(
        &'a self,
        lease: &'a MigrationLease,
        record: BackfillEventRecord,
    ) -> ExecutionFuture<'a, JournalEntry<BackfillEventRecord>>;

    /// Add one exact verified manifest from the open plan to the applied ledger.
    ///
    /// The store must reject a new record whose migration identity and digest do
    /// not occur at the same position in the open plan. Importing an existing
    /// ledger is a separate concern and must not use this execution write. An
    /// exact retry is idempotent only under the same fence; seeing the same
    /// migration under a newer fence requires reloading the ledger rather than
    /// treating differently fenced evidence as equal.
    fn record_applied<'a>(
        &'a self,
        lease: &'a MigrationLease,
        record: AppliedRecord,
    ) -> ExecutionFuture<'a, JournalEntry<AppliedRecord>>;

    /// Load ordered active applied records under the active fence.
    ///
    /// Retired records — applied records whose migration was rolled back by a
    /// later [`RolledBackRecord`] — are excluded. The full append-only history
    /// stays durable in the store; only the active ledger is the apply basis.
    fn load_applied<'a>(
        &'a self,
        lease: &'a MigrationLease,
    ) -> ExecutionFuture<'a, Vec<JournalEntry<AppliedRecord>>>;

    /// Load the open plan and all ordered events written at its fence or later.
    fn load_open_plan<'a>(
        &'a self,
        lease: &'a MigrationLease,
    ) -> ExecutionFuture<'a, Option<OpenPlanRecord>>;

    /// Begin one fully verified rollback plan after stale-ledger checks.
    ///
    /// Subject to the same open-plan exclusivity as [`Self::begin_plan`].
    fn begin_rollback_plan<'a>(
        &'a self,
        lease: &'a MigrationLease,
        record: RollbackPlanRecord,
    ) -> ExecutionFuture<'a, JournalEntry<RollbackPlanRecord>>;

    /// Append one positional rollback commit-boundary event under the active fence.
    fn record_rollback_step_event<'a>(
        &'a self,
        lease: &'a MigrationLease,
        record: RollbackStepEventRecord,
    ) -> ExecutionFuture<'a, JournalEntry<RollbackStepEventRecord>>;

    /// Retire one exact applied record from the open rollback plan.
    ///
    /// The store must reject a record whose migration identity and digest do
    /// not occur at the same position in the open rollback plan, or whose
    /// migration is not currently active in the applied ledger. Retirement is
    /// append-only: the applied record and this record both stay durable, and
    /// the migration merely leaves the active ledger. An exact retry is
    /// idempotent only under the same fence.
    fn record_rolled_back<'a>(
        &'a self,
        lease: &'a MigrationLease,
        record: RolledBackRecord,
    ) -> ExecutionFuture<'a, JournalEntry<RolledBackRecord>>;

    /// Load ordered retirement records under the active fence.
    fn load_rolled_back<'a>(
        &'a self,
        lease: &'a MigrationLease,
    ) -> ExecutionFuture<'a, Vec<JournalEntry<RolledBackRecord>>>;

    /// Load the open rollback plan and all ordered events at its fence or later.
    fn load_open_rollback_plan<'a>(
        &'a self,
        lease: &'a MigrationLease,
    ) -> ExecutionFuture<'a, Option<OpenRollbackPlanRecord>>;
}

/// Filter an applied ledger down to its active records.
///
/// Each retirement record consumes the latest not-yet-retired applied record
/// with the same migration identity and manifest digest written before it.
/// A retirement that matches no applied record is corrupt history and fails
/// closed. Stores share this exact matching rule so every journal
/// implementation reports the same active basis.
pub fn active_applied_entries(
    applied: Vec<JournalEntry<AppliedRecord>>,
    rolled_back: &[JournalEntry<RolledBackRecord>],
) -> Result<Vec<JournalEntry<AppliedRecord>>, Diagnostic> {
    let mut retired = vec![false; applied.len()];
    for retirement in rolled_back {
        let matched = applied
            .iter()
            .enumerate()
            .filter(|(index, entry)| {
                !retired[*index]
                    && entry.sequence() < retirement.sequence()
                    && entry.record().migration_id() == retirement.record().migration_id()
                    && entry.record().manifest_digest() == retirement.record().manifest_digest()
            })
            .max_by_key(|(_, entry)| entry.sequence())
            .map(|(index, _)| index);
        let Some(index) = matched else {
            return Err(failure(
                DiagnosticCategory::Integrity,
                "migration_execution_unmatched_retirement",
                "retirement record matches no active applied record before it",
            ));
        };
        retired[index] = true;
    }
    Ok(applied
        .into_iter()
        .zip(retired)
        .filter_map(|(entry, retired)| (!retired).then_some(entry))
        .collect())
}

fn position(value: usize) -> Result<u32, Diagnostic> {
    u32::try_from(value).map_err(|_| {
        failure(
            DiagnosticCategory::ResourceLimit,
            "migration_execution_position_limit",
            "migration transaction position exceeds the canonical u32 range",
        )
    })
}

fn failure(category: DiagnosticCategory, code: &'static str, message: &'static str) -> Diagnostic {
    Diagnostic::new(
        category,
        DiagnosticCode::new(code).expect("static migration execution diagnostic code"),
        message,
    )
}

#[cfg(test)]
mod tests {
    use std::collections::BTreeMap;
    use std::future::Future;
    use std::sync::Mutex;
    use std::task::{Context, Poll, Waker};

    use type_bridge_contract::fingerprint::SemanticProfileId;
    use type_bridge_contract::managed_scope::{ManagedScopeId, SemanticProfileBinding};
    use type_bridge_contract::migration::{MigrationAppLabel, MigrationName};

    use super::*;

    #[test]
    fn migration_controls_cancel_and_only_tighten_shared_ceilings() {
        let cancellation = MigrationCancellation::default();
        let limits = MigrationExecutionResourceLimits::tightened(7, usize::MAX);
        let control = MigrationExecutionControl::new(cancellation.clone(), None, limits);

        assert_eq!(control.resources().transaction_groups(), 7);
        assert_eq!(
            control.resources().backfill_observations(),
            MAX_MIGRATION_BACKFILL_OBSERVATIONS
        );
        assert!(control.check().is_ok());

        cancellation.cancel();
        cancellation.cancel();
        let diagnostic = control.check().expect_err("cancelled control");
        assert_eq!(diagnostic.category(), DiagnosticCategory::Cancelled);
        assert_eq!(diagnostic.code().as_str(), "migration_execution_cancelled");
    }

    #[test]
    fn migration_control_uses_one_absolute_deadline() {
        let control = MigrationExecutionControl::new(
            MigrationCancellation::default(),
            Some(Instant::now()),
            MigrationExecutionResourceLimits::default(),
        );
        let diagnostic = control.check().expect_err("expired deadline");
        assert_eq!(diagnostic.category(), DiagnosticCategory::ResourceLimit);
        assert_eq!(
            diagnostic.code().as_str(),
            "migration_execution_deadline_exceeded"
        );
    }
    use crate::schema_lowering_profile_binding;

    #[derive(Default)]
    struct ScopeState {
        highest_fence: u64,
        active_lease: Option<MigrationLease>,
        next_sequence: u64,
        applied: Vec<JournalEntry<AppliedRecord>>,
        rolled_back: Vec<JournalEntry<RolledBackRecord>>,
        open_plan: Option<JournalEntry<PlanRecord>>,
        events: Vec<JournalEntry<GroupEventRecord>>,
        backfill_events: Vec<JournalEntry<BackfillEventRecord>>,
        open_rollback_plan: Option<JournalEntry<RollbackPlanRecord>>,
        rollback_events: Vec<JournalEntry<RollbackStepEventRecord>>,
    }

    #[derive(Default)]
    struct InMemoryStore {
        scopes: Mutex<BTreeMap<ExecutionScope, ScopeState>>,
    }

    impl InMemoryStore {
        fn check_lease<'a>(
            state: &'a mut ScopeState,
            lease: &MigrationLease,
        ) -> Result<&'a mut ScopeState, Diagnostic> {
            if state.active_lease.as_ref() != Some(lease) {
                return Err(failure(
                    DiagnosticCategory::Integrity,
                    "migration_execution_stale_fence",
                    "journal write does not carry the current active lease and fence",
                ));
            }
            Ok(state)
        }

        fn sequence(state: &mut ScopeState) -> Result<JournalSequence, Diagnostic> {
            state.next_sequence = state.next_sequence.checked_add(1).ok_or_else(|| {
                failure(
                    DiagnosticCategory::ResourceLimit,
                    "migration_execution_sequence_exhausted",
                    "journal sequence range is exhausted",
                )
            })?;
            JournalSequence::new(state.next_sequence)
        }
    }

    impl MigrationLeaseStore for InMemoryStore {
        fn acquire<'a>(
            &'a self,
            scope: &'a ExecutionScope,
            holder: &'a LeaseHolderId,
        ) -> ExecutionFuture<'a, MigrationLease> {
            Box::pin(async move {
                let mut scopes = self.scopes.lock().expect("store mutex");
                let state = scopes.entry(scope.clone()).or_default();
                if state.active_lease.is_some() {
                    return Err(failure(
                        DiagnosticCategory::InvalidContract,
                        "migration_execution_lease_contended",
                        "migration scope already has an active lease",
                    ));
                }
                let fence = if state.highest_fence == 0 {
                    ExecutionFence::new(1)?
                } else {
                    ExecutionFence::new(state.highest_fence)?.checked_successor()?
                };
                state.highest_fence = fence.get();
                let lease = MigrationLease::new(scope.clone(), holder.clone(), fence);
                state.active_lease = Some(lease.clone());
                Ok(lease)
            })
        }

        fn release<'a>(&'a self, lease: &'a MigrationLease) -> ExecutionFuture<'a, ()> {
            Box::pin(async move {
                let mut scopes = self.scopes.lock().expect("store mutex");
                let state = scopes.get_mut(lease.scope()).ok_or_else(|| {
                    failure(
                        DiagnosticCategory::Integrity,
                        "migration_execution_stale_fence",
                        "lease scope is not active",
                    )
                })?;
                Self::check_lease(state, lease)?;
                state.active_lease = None;
                Ok(())
            })
        }
    }

    impl MigrationExecutionJournal for InMemoryStore {
        fn begin_plan<'a>(
            &'a self,
            lease: &'a MigrationLease,
            record: PlanRecord,
        ) -> ExecutionFuture<'a, JournalEntry<PlanRecord>> {
            Box::pin(async move {
                let mut scopes = self.scopes.lock().expect("store mutex");
                let state = scopes.get_mut(lease.scope()).ok_or_else(stale_fence)?;
                Self::check_lease(state, lease)?;
                if record.scope() != lease.scope() || record.fence() != lease.fence() {
                    return Err(stale_fence());
                }
                if state.open_plan.is_some() || state.open_rollback_plan.is_some() {
                    return Err(failure(
                        DiagnosticCategory::InvalidContract,
                        "migration_execution_plan_already_open",
                        "migration scope already has an open plan",
                    ));
                }
                let entry = JournalEntry::from_store(Self::sequence(state)?, record);
                state.open_plan = Some(entry.clone());
                Ok(entry)
            })
        }

        fn record_group_event<'a>(
            &'a self,
            lease: &'a MigrationLease,
            record: GroupEventRecord,
        ) -> ExecutionFuture<'a, JournalEntry<GroupEventRecord>> {
            Box::pin(async move {
                let mut scopes = self.scopes.lock().expect("store mutex");
                let state = scopes.get_mut(lease.scope()).ok_or_else(stale_fence)?;
                Self::check_lease(state, lease)?;
                if record.scope() != lease.scope() || record.fence() != lease.fence() {
                    return Err(stale_fence());
                }
                let plan = state.open_plan.as_ref().ok_or_else(|| {
                    failure(
                        DiagnosticCategory::InvalidContract,
                        "migration_execution_no_open_plan",
                        "group event requires an open migration plan",
                    )
                })?;
                if !plan
                    .record()
                    .manifest_digests()
                    .contains(&record.manifest_digest())
                {
                    return Err(failure(
                        DiagnosticCategory::Integrity,
                        "migration_execution_foreign_event",
                        "group event manifest is absent from the open plan",
                    ));
                }
                let entry = JournalEntry::from_store(Self::sequence(state)?, record);
                state.events.push(entry.clone());
                Ok(entry)
            })
        }

        fn record_backfill_event<'a>(
            &'a self,
            lease: &'a MigrationLease,
            record: BackfillEventRecord,
        ) -> ExecutionFuture<'a, JournalEntry<BackfillEventRecord>> {
            Box::pin(async move {
                let mut scopes = self.scopes.lock().expect("store mutex");
                let state = scopes.get_mut(lease.scope()).ok_or_else(stale_fence)?;
                Self::check_lease(state, lease)?;
                if record.scope() != lease.scope() || record.fence() != lease.fence() {
                    return Err(stale_fence());
                }
                let member = state.open_plan.as_ref().is_some_and(|plan| {
                    plan.record()
                        .manifest_digests()
                        .contains(&record.manifest_digest())
                }) || state.open_rollback_plan.as_ref().is_some_and(|plan| {
                    plan.record()
                        .manifest_digests()
                        .contains(&record.manifest_digest())
                });
                if !member {
                    return Err(failure(
                        DiagnosticCategory::Integrity,
                        "migration_execution_foreign_event",
                        "backfill event manifest is absent from the open plan",
                    ));
                }
                let entry = JournalEntry::from_store(Self::sequence(state)?, record);
                state.backfill_events.push(entry.clone());
                Ok(entry)
            })
        }

        fn record_applied<'a>(
            &'a self,
            lease: &'a MigrationLease,
            record: AppliedRecord,
        ) -> ExecutionFuture<'a, JournalEntry<AppliedRecord>> {
            Box::pin(async move {
                let mut scopes = self.scopes.lock().expect("store mutex");
                let state = scopes.get_mut(lease.scope()).ok_or_else(stale_fence)?;
                Self::check_lease(state, lease)?;
                if record.scope() != lease.scope() || record.fence() != lease.fence() {
                    return Err(stale_fence());
                }
                let active = active_applied_entries(state.applied.clone(), &state.rolled_back)?;
                if let Some(existing) = active
                    .iter()
                    .find(|entry| entry.record().migration_id() == record.migration_id())
                {
                    return if existing.record() == &record {
                        Ok(existing.clone())
                    } else {
                        Err(failure(
                            DiagnosticCategory::Integrity,
                            "migration_execution_applied_identity_conflict",
                            "applied migration identity has different evidence",
                        ))
                    };
                }
                let plan = state.open_plan.as_ref().ok_or_else(|| {
                    failure(
                        DiagnosticCategory::InvalidContract,
                        "migration_execution_no_open_plan",
                        "applied migration requires an open migration plan",
                    )
                })?;
                let manifest_index = plan
                    .record()
                    .migration_ids()
                    .iter()
                    .position(|id| id == record.migration_id());
                if manifest_index.is_none_or(|index| {
                    plan.record().manifest_digests().get(index) != Some(&record.manifest_digest())
                }) {
                    return Err(failure(
                        DiagnosticCategory::Integrity,
                        "migration_execution_foreign_applied_record",
                        "applied migration identity and digest are absent from the open plan",
                    ));
                }
                let entry = JournalEntry::from_store(Self::sequence(state)?, record);
                state.applied.push(entry.clone());
                let active = active_applied_entries(state.applied.clone(), &state.rolled_back)?;
                let complete = state.open_plan.as_ref().is_some_and(|plan| {
                    plan.record().migration_ids().iter().all(|id| {
                        active
                            .iter()
                            .any(|applied| applied.record().migration_id() == id)
                    })
                });
                if complete {
                    state.open_plan = None;
                    state.events.clear();
                    state.backfill_events.clear();
                }
                Ok(entry)
            })
        }

        fn load_applied<'a>(
            &'a self,
            lease: &'a MigrationLease,
        ) -> ExecutionFuture<'a, Vec<JournalEntry<AppliedRecord>>> {
            Box::pin(async move {
                let mut scopes = self.scopes.lock().expect("store mutex");
                let state = scopes.get_mut(lease.scope()).ok_or_else(stale_fence)?;
                Self::check_lease(state, lease)?;
                active_applied_entries(state.applied.clone(), &state.rolled_back)
            })
        }

        fn load_open_plan<'a>(
            &'a self,
            lease: &'a MigrationLease,
        ) -> ExecutionFuture<'a, Option<OpenPlanRecord>> {
            Box::pin(async move {
                let mut scopes = self.scopes.lock().expect("store mutex");
                let state = scopes.get_mut(lease.scope()).ok_or_else(stale_fence)?;
                Self::check_lease(state, lease)?;
                let Some(plan) = state.open_plan.clone() else {
                    return Ok(None);
                };
                Ok(Some(OpenPlanRecord::from_store_with_backfills(
                    plan,
                    state.events.clone(),
                    state
                        .backfill_events
                        .iter()
                        .filter(|event| {
                            event.record().direction() == BackfillExecutionDirection::Forward
                        })
                        .cloned()
                        .collect(),
                )?))
            })
        }

        fn begin_rollback_plan<'a>(
            &'a self,
            lease: &'a MigrationLease,
            record: RollbackPlanRecord,
        ) -> ExecutionFuture<'a, JournalEntry<RollbackPlanRecord>> {
            Box::pin(async move {
                let mut scopes = self.scopes.lock().expect("store mutex");
                let state = scopes.get_mut(lease.scope()).ok_or_else(stale_fence)?;
                Self::check_lease(state, lease)?;
                if record.scope() != lease.scope() || record.fence() != lease.fence() {
                    return Err(stale_fence());
                }
                if state.open_plan.is_some() || state.open_rollback_plan.is_some() {
                    return Err(failure(
                        DiagnosticCategory::InvalidContract,
                        "migration_execution_plan_already_open",
                        "migration scope already has an open plan",
                    ));
                }
                let entry = JournalEntry::from_store(Self::sequence(state)?, record);
                state.open_rollback_plan = Some(entry.clone());
                Ok(entry)
            })
        }

        fn record_rollback_step_event<'a>(
            &'a self,
            lease: &'a MigrationLease,
            record: RollbackStepEventRecord,
        ) -> ExecutionFuture<'a, JournalEntry<RollbackStepEventRecord>> {
            Box::pin(async move {
                let mut scopes = self.scopes.lock().expect("store mutex");
                let state = scopes.get_mut(lease.scope()).ok_or_else(stale_fence)?;
                Self::check_lease(state, lease)?;
                if record.scope() != lease.scope() || record.fence() != lease.fence() {
                    return Err(stale_fence());
                }
                let plan = state.open_rollback_plan.as_ref().ok_or_else(|| {
                    failure(
                        DiagnosticCategory::InvalidContract,
                        "migration_execution_no_open_plan",
                        "rollback event requires an open rollback plan",
                    )
                })?;
                if !plan
                    .record()
                    .manifest_digests()
                    .contains(&record.manifest_digest())
                {
                    return Err(failure(
                        DiagnosticCategory::Integrity,
                        "migration_execution_foreign_event",
                        "rollback event manifest is absent from the open plan",
                    ));
                }
                let entry = JournalEntry::from_store(Self::sequence(state)?, record);
                state.rollback_events.push(entry.clone());
                Ok(entry)
            })
        }

        fn record_rolled_back<'a>(
            &'a self,
            lease: &'a MigrationLease,
            record: RolledBackRecord,
        ) -> ExecutionFuture<'a, JournalEntry<RolledBackRecord>> {
            Box::pin(async move {
                let mut scopes = self.scopes.lock().expect("store mutex");
                let state = scopes.get_mut(lease.scope()).ok_or_else(stale_fence)?;
                Self::check_lease(state, lease)?;
                if record.scope() != lease.scope() || record.fence() != lease.fence() {
                    return Err(stale_fence());
                }
                let active = active_applied_entries(state.applied.clone(), &state.rolled_back)?;
                let is_active = active.iter().any(|entry| {
                    entry.record().migration_id() == record.migration_id()
                        && entry.record().manifest_digest() == record.manifest_digest()
                });
                if !is_active {
                    if let Some(existing) = state.rolled_back.iter().find(|entry| {
                        entry.record() == &record && entry.record().fence() == lease.fence()
                    }) {
                        return Ok(existing.clone());
                    }
                    return Err(failure(
                        DiagnosticCategory::Integrity,
                        "migration_execution_retirement_conflict",
                        "retirement target is not active in the applied ledger",
                    ));
                }
                let plan = state.open_rollback_plan.as_ref().ok_or_else(|| {
                    failure(
                        DiagnosticCategory::InvalidContract,
                        "migration_execution_no_open_plan",
                        "retirement requires an open rollback plan",
                    )
                })?;
                let manifest_index = plan
                    .record()
                    .rollback_ids()
                    .iter()
                    .position(|id| id == record.migration_id());
                if manifest_index.is_none_or(|index| {
                    plan.record().manifest_digests().get(index) != Some(&record.manifest_digest())
                }) {
                    return Err(failure(
                        DiagnosticCategory::Integrity,
                        "migration_execution_foreign_applied_record",
                        "retirement identity and digest are absent from the open plan",
                    ));
                }
                let entry = JournalEntry::from_store(Self::sequence(state)?, record);
                state.rolled_back.push(entry.clone());
                let active = active_applied_entries(state.applied.clone(), &state.rolled_back)?;
                let complete = state.open_rollback_plan.as_ref().is_some_and(|plan| {
                    plan.record().rollback_ids().iter().all(|id| {
                        !active
                            .iter()
                            .any(|applied| applied.record().migration_id() == id)
                    })
                });
                if complete {
                    state.open_rollback_plan = None;
                    state.rollback_events.clear();
                    state.backfill_events.clear();
                }
                Ok(entry)
            })
        }

        fn load_rolled_back<'a>(
            &'a self,
            lease: &'a MigrationLease,
        ) -> ExecutionFuture<'a, Vec<JournalEntry<RolledBackRecord>>> {
            Box::pin(async move {
                let mut scopes = self.scopes.lock().expect("store mutex");
                let state = scopes.get_mut(lease.scope()).ok_or_else(stale_fence)?;
                Self::check_lease(state, lease)?;
                Ok(state.rolled_back.clone())
            })
        }

        fn load_open_rollback_plan<'a>(
            &'a self,
            lease: &'a MigrationLease,
        ) -> ExecutionFuture<'a, Option<OpenRollbackPlanRecord>> {
            Box::pin(async move {
                let mut scopes = self.scopes.lock().expect("store mutex");
                let state = scopes.get_mut(lease.scope()).ok_or_else(stale_fence)?;
                Self::check_lease(state, lease)?;
                let Some(plan) = state.open_rollback_plan.clone() else {
                    return Ok(None);
                };
                Ok(Some(OpenRollbackPlanRecord::from_store_with_backfills(
                    plan,
                    state.rollback_events.clone(),
                    state
                        .backfill_events
                        .iter()
                        .filter(|event| {
                            event.record().direction() == BackfillExecutionDirection::Reverse
                        })
                        .cloned()
                        .collect(),
                )?))
            })
        }
    }

    fn stale_fence() -> Diagnostic {
        failure(
            DiagnosticCategory::Integrity,
            "migration_execution_stale_fence",
            "journal write does not carry the current active lease and fence",
        )
    }

    fn scope() -> ExecutionScope {
        ExecutionScope::new(ManagedScopeId::new("journal-test").expect("scope"))
    }

    fn migration_id() -> MigrationId {
        MigrationId::from_components(
            MigrationAppLabel::new("example").expect("app"),
            MigrationName::new("0001_initial").expect("name"),
        )
    }

    fn semantic_fingerprint(bytes: &[u8]) -> ManagedSemanticSchemaFingerprint {
        ManagedSemanticSchemaFingerprint::compute(
            SemanticProfileId::new("typedb-3.12.1/v1").expect("profile"),
            bytes,
        )
        .expect("semantic fingerprint")
    }

    fn fake_plan(lease: &MigrationLease) -> PlanRecord {
        let id = migration_id();
        let semantic = SemanticProfileBinding::resolve(
            SemanticProfileId::new("typedb-3.12.1/v1").expect("profile"),
        )
        .expect("semantic binding");
        PlanRecord {
            scope: lease.scope().clone(),
            fence: lease.fence(),
            source_applied: Vec::new(),
            source_frontier: Vec::new(),
            target_frontier: vec![id.clone()],
            migration_ids: vec![id],
            manifest_digests: vec![MigrationManifestDigest::compute(b"manifest")],
            manifest_plan_fingerprints: vec![
                MigrationPlanFingerprint::compute(&[]).expect("plan fingerprint"),
            ],
            source_declared: ManagedDeclaredIdentityFingerprint::compute(b"source")
                .expect("source declared"),
            target_declared: ManagedDeclaredIdentityFingerprint::compute(b"target")
                .expect("target declared"),
            source_semantics: semantic_fingerprint(b"source"),
            target_semantics: semantic_fingerprint(b"target"),
            semantic_profile: semantic.fingerprint().clone(),
            lowering_profile: schema_lowering_profile_binding()
                .expect("lowering binding")
                .fingerprint()
                .clone(),
            observed_live_source: semantic_fingerprint(b"source"),
        }
    }

    fn fake_event(lease: &MigrationLease, plan: &PlanRecord) -> GroupEventRecord {
        GroupEventRecord {
            scope: lease.scope().clone(),
            fence: lease.fence(),
            manifest_digest: plan.manifest_digests()[0],
            migration_id: plan.migration_ids()[0].clone(),
            group_ordinal: 0,
            first_step_index: 0,
            schema_delta_step_index: 0,
            end_step_index: 1,
            kind: GroupJournalEventKind::BeforeCommit,
            observed_target: None,
        }
    }

    fn fake_applied(lease: &MigrationLease, plan: &PlanRecord) -> AppliedRecord {
        AppliedRecord {
            scope: lease.scope().clone(),
            fence: lease.fence(),
            migration_id: plan.migration_ids()[0].clone(),
            manifest_digest: plan.manifest_digests()[0],
            source_declared: plan.source_declared().clone(),
            target_declared: plan.target_declared().clone(),
            source_semantics: plan.source_semantics().clone(),
            target_semantics: plan.target_semantics().clone(),
        }
    }

    #[test]
    fn bound_lease_equality_includes_process_local_binding_identity() {
        let scope = scope();
        let holder = LeaseHolderId::new("owner").expect("holder");
        let fence = ExecutionFence::new(1).expect("fence");
        let token = ExecutionBindingToken::fresh();
        let same = MigrationLease::new_bound(scope.clone(), holder.clone(), fence, token.clone());
        let clone = same.clone();
        let foreign = MigrationLease::new_bound(
            scope.clone(),
            holder.clone(),
            fence,
            ExecutionBindingToken::fresh(),
        );
        let unbound = MigrationLease::new(scope.clone(), holder.clone(), fence);
        let same_unbound = MigrationLease::new(scope, holder, fence);

        assert_eq!(same, clone);
        assert_ne!(same, foreign);
        assert_ne!(same, unbound);
        assert_eq!(unbound, same_unbound);
    }

    #[test]
    fn store_assigns_monotonic_sequences_and_rejects_every_stale_fence_write() {
        let store = InMemoryStore::default();
        let scope = scope();
        let holder_a = LeaseHolderId::new("owner-a").expect("holder");
        let holder_b = LeaseHolderId::new("owner-b").expect("holder");
        let lease_a = block_on(store.acquire(&scope, &holder_a)).expect("lease a");
        assert!(block_on(store.acquire(&scope, &holder_b)).is_err());
        let plan_a = fake_plan(&lease_a);
        let plan_entry = block_on(store.begin_plan(&lease_a, plan_a.clone())).expect("plan");
        let event_entry =
            block_on(store.record_group_event(&lease_a, fake_event(&lease_a, &plan_a)))
                .expect("event");
        assert_eq!(plan_entry.sequence().get(), 1);
        assert_eq!(event_entry.sequence().get(), 2);
        block_on(store.release(&lease_a)).expect("release a");
        let lease_b = block_on(store.acquire(&scope, &holder_b)).expect("lease b");
        assert!(lease_b.fence() > lease_a.fence());
        assert!(block_on(store.begin_plan(&lease_a, plan_a.clone())).is_err());
        assert!(
            block_on(store.record_group_event(&lease_a, fake_event(&lease_a, &plan_a),)).is_err()
        );
        assert!(
            block_on(store.record_applied(&lease_a, fake_applied(&lease_a, &plan_a),)).is_err()
        );
        assert!(block_on(store.release(&lease_a)).is_err());
        let recovery_event =
            block_on(store.record_group_event(&lease_b, fake_event(&lease_b, &plan_a)))
                .expect("recovery event");
        assert_eq!(recovery_event.sequence().get(), 3);
        block_on(store.release(&lease_b)).expect("release recovery lease");
        let holder_c = LeaseHolderId::new("owner-c").expect("holder");
        let lease_c = block_on(store.acquire(&scope, &holder_c)).expect("lease c");
        assert!(lease_c.fence() > lease_b.fence());
        let open = block_on(store.load_open_plan(&lease_c))
            .expect("open plan")
            .expect("plan remains open after recovery crash");
        assert_eq!(open.events().len(), 2);
        assert_eq!(open.events()[0].record().fence(), lease_a.fence());
        assert_eq!(open.events()[1].record().fence(), lease_b.fence());
    }

    #[test]
    fn applied_records_are_plan_bound_idempotent_and_close_the_completed_plan() {
        let store = InMemoryStore::default();
        let scope = scope();
        let holder = LeaseHolderId::new("applied-owner").expect("holder");
        let lease = block_on(store.acquire(&scope, &holder)).expect("lease");
        let plan = fake_plan(&lease);
        let applied = fake_applied(&lease, &plan);

        assert!(block_on(store.record_applied(&lease, applied.clone())).is_err());
        block_on(store.begin_plan(&lease, plan.clone())).expect("open plan");

        let mut foreign = applied.clone();
        foreign.manifest_digest = MigrationManifestDigest::compute(b"foreign");
        assert!(block_on(store.record_applied(&lease, foreign)).is_err());

        let first = block_on(store.record_applied(&lease, applied.clone()))
            .expect("apply planned manifest");
        let duplicate = block_on(store.record_applied(&lease, applied))
            .expect("same-fence duplicate is idempotent");
        assert_eq!(duplicate, first);
        assert!(
            block_on(store.load_open_plan(&lease))
                .expect("load completed plan")
                .is_none()
        );
        assert_eq!(
            block_on(store.load_applied(&lease)).expect("load applied ledger"),
            vec![first],
        );
    }

    fn fake_rollback_plan(lease: &MigrationLease, plan: &PlanRecord) -> RollbackPlanRecord {
        RollbackPlanRecord {
            scope: lease.scope().clone(),
            fence: lease.fence(),
            source_applied: plan.migration_ids().to_vec(),
            rollback_ids: plan.migration_ids().to_vec(),
            manifest_digests: plan.manifest_digests().to_vec(),
            manifest_plan_fingerprints: plan.manifest_plan_fingerprints().to_vec(),
            remaining_applied: Vec::new(),
            source_declared: plan.target_declared().clone(),
            target_declared: plan.source_declared().clone(),
            source_semantics: plan.target_semantics().clone(),
            target_semantics: plan.source_semantics().clone(),
            semantic_profile: plan.semantic_profile().clone(),
            lowering_profile: plan.lowering_profile().clone(),
            observed_live_source: plan.target_semantics().clone(),
        }
    }

    fn fake_rolled_back(lease: &MigrationLease, plan: &PlanRecord) -> RolledBackRecord {
        RolledBackRecord {
            scope: lease.scope().clone(),
            fence: lease.fence(),
            migration_id: plan.migration_ids()[0].clone(),
            manifest_digest: plan.manifest_digests()[0],
            source_declared: plan.target_declared().clone(),
            target_declared: plan.source_declared().clone(),
            source_semantics: plan.target_semantics().clone(),
            target_semantics: plan.source_semantics().clone(),
        }
    }

    #[test]
    fn retirement_is_append_only_exclusive_and_reopens_the_identity() {
        let store = InMemoryStore::default();
        let scope = scope();
        let holder = LeaseHolderId::new("retirement-owner").expect("holder");
        let lease = block_on(store.acquire(&scope, &holder)).expect("lease");
        let plan = fake_plan(&lease);
        let rollback_plan = fake_rollback_plan(&lease, &plan);
        let retirement = fake_rolled_back(&lease, &plan);

        // Apply the migration through an ordinary forward plan.
        block_on(store.begin_plan(&lease, plan.clone())).expect("open plan");
        assert!(
            block_on(store.begin_rollback_plan(&lease, rollback_plan.clone())).is_err(),
            "open plans of either direction must be exclusive"
        );
        block_on(store.record_applied(&lease, fake_applied(&lease, &plan)))
            .expect("apply planned manifest");
        assert_eq!(
            block_on(store.load_applied(&lease)).expect("active").len(),
            1
        );

        // Retiring outside an open rollback plan fails closed.
        assert!(block_on(store.record_rolled_back(&lease, retirement.clone())).is_err());
        block_on(store.begin_rollback_plan(&lease, rollback_plan)).expect("open rollback plan");
        assert!(
            block_on(store.begin_plan(&lease, plan.clone())).is_err(),
            "an open rollback plan must block a new apply plan"
        );
        let first = block_on(store.record_rolled_back(&lease, retirement.clone()))
            .expect("retire applied migration");
        let duplicate = block_on(store.record_rolled_back(&lease, retirement))
            .expect("same-fence duplicate retirement is idempotent");
        assert_eq!(duplicate, first);
        assert!(
            block_on(store.load_open_rollback_plan(&lease))
                .expect("closed rollback plan")
                .is_none()
        );
        assert!(
            block_on(store.load_applied(&lease))
                .expect("active")
                .is_empty()
        );
        assert_eq!(
            block_on(store.load_rolled_back(&lease))
                .expect("retired")
                .len(),
            1,
        );

        // The identity is free again: a fresh forward plan re-applies it.
        block_on(store.begin_plan(&lease, plan.clone())).expect("reopen plan");
        block_on(store.record_applied(&lease, fake_applied(&lease, &plan)))
            .expect("re-apply retired migration");
        assert_eq!(
            block_on(store.load_applied(&lease)).expect("active").len(),
            1
        );
    }

    #[test]
    fn unmatched_retirements_are_corrupt_history() {
        let store = InMemoryStore::default();
        let scope = scope();
        let holder = LeaseHolderId::new("orphan-owner").expect("holder");
        let lease = block_on(store.acquire(&scope, &holder)).expect("lease");
        let plan = fake_plan(&lease);
        let orphan = JournalEntry::from_store(
            JournalSequence::new(7).expect("sequence"),
            fake_rolled_back(&lease, &plan),
        );
        let error = active_applied_entries(Vec::new(), &[orphan])
            .expect_err("a retirement without its applied record is corrupt");
        assert_eq!(
            error.code().as_str(),
            "migration_execution_unmatched_retirement"
        );
    }

    #[test]
    fn recovery_decision_table_is_exhaustive_for_distinct_and_equal_semantics() {
        let source = semantic_fingerprint(b"source");
        let target = semantic_fingerprint(b"target");
        let neither = semantic_fingerprint(b"neither");
        let events = [
            None,
            Some(GroupJournalEventKind::BeforeCommit),
            Some(GroupJournalEventKind::Committed),
            Some(GroupJournalEventKind::CommitOutcomeUnknown),
            Some(GroupJournalEventKind::DefinitelyAborted),
            Some(GroupJournalEventKind::FormalOnlyAdvanced),
        ];
        let observations = [
            GroupRecoveryObservation::ManagedSemantics(source.clone()),
            GroupRecoveryObservation::ManagedSemantics(target.clone()),
            GroupRecoveryObservation::ManagedSemantics(neither.clone()),
            GroupRecoveryObservation::Unavailable,
        ];
        for event in events {
            for observation in &observations {
                let expected = expected_distinct(event, observation, &source, &target);
                assert_eq!(
                    decide_group_recovery(event, observation, &source, &target),
                    expected,
                    "distinct case {event:?} {observation:?}",
                );
            }
        }
        let equal_observations = [
            GroupRecoveryObservation::ManagedSemantics(source.clone()),
            GroupRecoveryObservation::ManagedSemantics(neither),
            GroupRecoveryObservation::Unavailable,
        ];
        for event in events {
            for observation in &equal_observations {
                let expected = expected_equal(event, observation, &source);
                assert_eq!(
                    decide_group_recovery(event, observation, &source, &source),
                    expected,
                    "equal case {event:?} {observation:?}",
                );
            }
        }
    }

    #[test]
    fn backfill_counts_and_recovery_fail_closed_without_exact_completion() {
        assert_eq!(
            BackfillExecutionCounts::new(5, 3, 1, 1)
                .expect_err("inconsistent counts reject")
                .code()
                .as_str(),
            "migration_backfill_count_mismatch"
        );
        assert_eq!(
            BackfillExecutionCounts::new(0, 0, 0, 0)
                .expect_err("zero transaction groups reject")
                .code()
                .as_str(),
            "migration_backfill_zero_transaction_groups"
        );

        let plan = semantic_fingerprint(b"backfill-plan")
            .as_fingerprint()
            .clone();
        let foreign = semantic_fingerprint(b"foreign-backfill-plan")
            .as_fingerprint()
            .clone();
        let counts = BackfillExecutionCounts::new(5, 3, 2, 2).expect("valid counts");
        let complete = BackfillRecoveryObservation::Complete(BackfillCompletionEvidence::new(
            plan.clone(),
            BackfillExecutionDirection::Forward,
            counts,
        ));

        assert_eq!(
            decide_backfill_recovery(
                None,
                &BackfillRecoveryObservation::Incomplete,
                &plan,
                BackfillExecutionDirection::Forward,
            ),
            GroupRecoveryDecision::ExecuteNormally
        );
        assert_eq!(
            decide_backfill_recovery(None, &complete, &plan, BackfillExecutionDirection::Forward,),
            GroupRecoveryDecision::ExecuteNormally
        );
        assert_eq!(
            decide_backfill_recovery(
                Some(GroupJournalEventKind::BeforeCommit),
                &BackfillRecoveryObservation::Incomplete,
                &plan,
                BackfillExecutionDirection::Forward,
            ),
            GroupRecoveryDecision::RequiresExplicitRecovery
        );
        assert_eq!(
            decide_backfill_recovery(
                Some(GroupJournalEventKind::CommitOutcomeUnknown),
                &complete,
                &plan,
                BackfillExecutionDirection::Forward,
            ),
            GroupRecoveryDecision::RepairCheckpoint
        );
        assert_eq!(
            decide_backfill_recovery(
                Some(GroupJournalEventKind::Committed),
                &complete,
                &foreign,
                BackfillExecutionDirection::Forward,
            ),
            GroupRecoveryDecision::RequiresExplicitRecovery
        );
        assert_eq!(
            decide_backfill_recovery(
                Some(GroupJournalEventKind::Committed),
                &complete,
                &plan,
                BackfillExecutionDirection::Reverse,
            ),
            GroupRecoveryDecision::RequiresExplicitRecovery
        );
    }

    fn expected_distinct(
        event: Option<GroupJournalEventKind>,
        observation: &GroupRecoveryObservation,
        source: &ManagedSemanticSchemaFingerprint,
        target: &ManagedSemanticSchemaFingerprint,
    ) -> GroupRecoveryDecision {
        let source_seen = matches!(observation, GroupRecoveryObservation::ManagedSemantics(value) if value == source);
        let target_seen = matches!(observation, GroupRecoveryObservation::ManagedSemantics(value) if value == target);
        match event {
            None | Some(GroupJournalEventKind::DefinitelyAborted) if source_seen => {
                GroupRecoveryDecision::ExecuteNormally
            }
            Some(
                GroupJournalEventKind::BeforeCommit | GroupJournalEventKind::CommitOutcomeUnknown,
            ) if source_seen => GroupRecoveryDecision::ExecuteNormally,
            Some(
                GroupJournalEventKind::BeforeCommit
                | GroupJournalEventKind::CommitOutcomeUnknown
                | GroupJournalEventKind::Committed,
            ) if target_seen => GroupRecoveryDecision::RepairCheckpoint,
            _ => GroupRecoveryDecision::RequiresExplicitRecovery,
        }
    }

    fn expected_equal(
        event: Option<GroupJournalEventKind>,
        observation: &GroupRecoveryObservation,
        both: &ManagedSemanticSchemaFingerprint,
    ) -> GroupRecoveryDecision {
        let both_seen = matches!(observation, GroupRecoveryObservation::ManagedSemantics(value) if value == both);
        if !both_seen {
            return GroupRecoveryDecision::RequiresExplicitRecovery;
        }
        match event {
            None | Some(GroupJournalEventKind::DefinitelyAborted) => {
                GroupRecoveryDecision::ExecuteNormally
            }
            Some(GroupJournalEventKind::Committed | GroupJournalEventKind::FormalOnlyAdvanced) => {
                GroupRecoveryDecision::RepairCheckpoint
            }
            _ => GroupRecoveryDecision::RequiresExplicitRecovery,
        }
    }

    fn block_on<F: Future>(future: F) -> F::Output {
        let mut context = Context::from_waker(Waker::noop());
        let mut future = Box::pin(future);
        loop {
            match future.as_mut().poll(&mut context) {
                Poll::Ready(output) => return output,
                Poll::Pending => std::thread::yield_now(),
            }
        }
    }
}