canic_host/fleet_ensure/workflow/

mod.rs

//! Module: fleet_ensure::workflow
//!
//! Responsibility: plan and advance one durable idempotent Fleet convergence operation.
//! Does not own: policy decisions, IC transport, or storage mechanics.
//! Boundary: persists exact intent before each ops effect and reconciles it before any retry.

pub mod attempt_recovery;
#[cfg(test)]
mod balance_observation;
pub mod capacity_import;
pub mod clean_reinstall;
mod continuation;
mod funding;
pub mod funding_observation;
#[cfg(test)]
mod funding_tests;
mod independent_effects;
pub mod infrastructure_bootstrap;
pub mod operator_mint;
pub mod readiness;
mod reinstall;
pub mod release;
mod retained_plan;
mod root_reinstall;

use crate::fleet_ensure::{
    dto::{FleetEnsurePhase, FleetEnsureProgress, FleetEnsureProgressState},
    inventory::project_current_fleet_inventory,
    model::{
        ActualCycleConservation, CanisterDisposition, CurrentFleetProtocolAction,
        CycleConservation, EffectRecord, EffectState, EnsureAction, EstateFundingDomainPlan,
        EstateFundingRequiredRecord, FLEET_ENSURE_SCHEMA_VERSION, FleetEnsureCompletion,
        FleetEnsureJournalRecord, FleetEnsurePlan, FleetEnsurePlanScope, FleetEnsureReport,
        FleetEnsureStateRecord, FleetObservation, RootManagementObservation,
        create_balance_is_terminal,
    },
    ops::{
        EffectRetry, EnsurePaths, EnsurePlatform, EnsureStateError, action_sha256,
        continuation::verify_release_transition, effect_preparation::prepare_effect,
        lock_operation, read_root_start_authority, read_state, reserve_fixture_publication_attempt,
        resolve_desired_artifacts, retain_configured_principal_bindings, write_journal, write_plan,
        write_state,
    },
    policy::{
        EnsurePolicyError, RootStartPlanInput, compile_plan, compile_root_start_prerequisite_plan,
        effect_observation_policy, expected_plan_sha256, operation_id,
        recompile_root_start_prerequisite_plan, validate_creation_fee_scope,
        validate_path_identity, validate_path_labels,
    },
};
use canic_contracts::cycles::Cycles;
use std::{
    collections::{BTreeMap, BTreeSet},
    path::Path,
    time::Instant,
};
use thiserror::Error as ThisError;

pub use reinstall::plan_reinstall;

/// Exact no-effect evidence that a fresh pool cannot retain its readiness floor.

#[derive(Debug, ThisError)]
#[error(
    "fresh pool canister {canister} requested {requested_creation_funding_cycles} creation cycles for readiness floor {readiness_floor_cycles}; its first live balance is {live_balance_cycles} cycles (readiness shortfall {readiness_shortfall_cycles}) and must retain {remaining_controller_burn_cycles} cycles for controller finalization (required pre-finalization balance {required_pre_finalization_balance_cycles}, pre-finalization shortfall {pre_finalization_shortfall_cycles}; maximum first-observation burn {maximum_observation_burn_cycles}); no controller or protocol action followed"
)]
pub struct FreshPoolCreationUnderfundedError {
    pub canister: String,
    pub live_balance_cycles: u128,
    pub maximum_observation_burn_cycles: u128,
    pub pre_finalization_shortfall_cycles: u128,
    pub readiness_floor_cycles: u128,
    pub readiness_shortfall_cycles: u128,
    pub remaining_controller_burn_cycles: u128,
    pub requested_creation_funding_cycles: u128,
    pub required_pre_finalization_balance_cycles: u128,
}

/// Exact reviewed Root-account funding prerequisite that blocks protocol mutation.
#[derive(Debug, ThisError)]
#[error(
    "Root {root} Cycles Ledger account {root_principal} on {cycles_ledger} has {available_cycles} cycles; {required_creation_count} autonomous creations require at most {maximum_creation_debit_cycles} cycles ({creation_amount_cycles} gross amount containing {readiness_floor_cycles} readiness floor, {management_creation_fee_cycles} management creation fee and {creation_execution_margin_cycles} execution margin, plus {ledger_fee_cycles} Ledger fee each; total reviewed fees {maximum_creation_fee_cycles}), shortfall {shortfall_cycles}; pending_creation={pending_creation_operation_id:?} attempts={attempt_count:?} last_attempt_at_ns={last_attempt_at_ns:?} retry_at_ns={retry_at_ns:?}; preserve this operation, repeat Fleet ensure without --apply and review funding_review, then apply its review_sha256 for the exact shortfall; do not fund the account out of band"
)]
pub struct EstateFundingRequiredError {
    pub attempt_count: Option<u32>,
    pub available_cycles: u128,
    pub creation_amount_cycles: u128,
    pub cycles_ledger: String,
    pub creation_execution_margin_cycles: u128,
    pub readiness_floor_cycles: u128,
    pub ledger_fee_cycles: u128,
    pub management_creation_fee_cycles: u128,
    pub maximum_creation_debit_cycles: u128,
    pub maximum_creation_fee_cycles: u128,
    pub pending_creation_operation_id: Option<String>,
    pub required_creation_count: u32,
    pub root: String,
    pub root_principal: String,
    pub last_attempt_at_ns: Option<u64>,
    pub retry_at_ns: Option<u64>,
    pub shortfall_cycles: u128,
}

/// Exact terminal-inventory merge or cycle-observation invariant failure.
#[derive(Debug, ThisError)]
pub enum TerminalInventoryError {
    #[error("one Principal is retained under more than one desired role")]
    DuplicateRetainedPrincipal,

    #[error("verified canister {canister} lost its Principal binding")]
    LostPrincipalBinding { canister: String },

    #[error("canister {canister} names unknown parent {parent}")]
    UnknownParent { canister: String, parent: String },

    #[error(
        "canister {canister} retained parent {retained_parent} differs from terminal parent {terminal_parent}"
    )]
    RetainedParentConflict {
        canister: String,
        retained_parent: String,
        terminal_parent: String,
    },

    #[error(
        "terminal Root-owned Idle canister {canister} cycle observations differ: expected {expected}, observed {observed}"
    )]
    IdleCycleObservationConflict {
        canister: String,
        expected: u128,
        observed: u128,
    },

    #[error(
        "terminal cycle observation duplicates canister {canister} outside its exact Root-owned Idle or Workload lifecycle"
    )]
    CycleObservationLifecycleConflict { canister: String },
}

/// Fleet ensure planning or convergence failure.

#[derive(Debug, ThisError)]
pub enum EnsureWorkflowError<E>
where
    E: std::error::Error + 'static,
{
    #[error(transparent)]
    InfrastructureBootstrap(
        #[from] crate::fleet_ensure::ops::infrastructure_bootstrap::InfrastructureBootstrapError,
    ),
    #[error(transparent)]
    RetainedContract(#[from] crate::fleet_ensure::ops::retained_contract::RetainedContractError),

    #[error(transparent)]
    FundingObservation(
        #[from] crate::fleet_ensure::model::funding_observation::FundingObservationError,
    ),

    #[error(
        "operator conversion must be resolved or its unapproved review cancelled before Fleet effects continue: {review_sha256}; inspect with fleet ensure --operator-mint"
    )]
    OperatorMintPending { review_sha256: String },

    #[error(
        "operator conversion required for retained funding review {review_sha256}; preserve the original withdrawal intent"
    )]
    OperatorFundingRequired { review_sha256: String },

    #[error("operator conversion review, approval digest or current authority differs")]
    OperatorMintReviewConflict,

    #[error("operator conversion reply conflicts with a retained transaction outcome")]
    OperatorMintReplyConflict,

    #[error("operator conversion wire data is invalid: {0}")]
    OperatorMintWire(#[from] crate::fleet_ensure::ops::operator_mint::OperatorMintWireError),

    #[error(
        "retained Fleet operation cannot be resumed: {source}; local source evidence identifies operation {operation_id}, journal plan reference {plan_sha256}, source document {source_document_sha256}; preserve plan, journal, state and artifacts; request an explicit --reinstall review without --apply using the selected corrected release; live authority, complete estate and cycle checks must pass before any reset; see docs/features/operations/fleet-ensure.md#unreadable-retained-plan"
    )]
    RetainedActivationReviewRequired {
        operation_id: String,
        plan_sha256: String,
        source_document_sha256: String,
        #[source]
        source: Box<EnsureStateError>,
    },

    #[error(
        "retained Fleet operation is unreadable: {source}; preserve plan, journal, state, artifacts and paid-effect receipts; local evidence does not establish a supported separate recovery review; resolve issued effects under their exact authority before a release transition; do not edit fields, replace digests or discard the operation; see docs/features/operations/fleet-ensure.md#unreadable-retained-plan"
    )]
    RetainedPlanUnreadable {
        #[source]
        source: Box<EnsureStateError>,
    },

    #[error(
        "activation {operation_id} has an Issued provisioning action after its Applied host prefix (source document {source_document_sha256}); preserve the plan and journal: required live evidence for the bounded partial-activation recovery is unavailable"
    )]
    PartialActivationResetUnavailable {
        operation_id: String,
        source_document_sha256: String,
    },

    #[error(
        "reinstall requires a converged Fleet with retained physical authority or an admitted single-Root partial-activation recovery source; the supplied source or target does not satisfy either contract; preserve the retained plan and journal"
    )]
    ReinstallConflict,

    #[error(
        "retained Fleet operation {operation_id} must recover before a new reinstall; preserve original plan {plan_sha256}, journal and receipts; run ensure without --reinstall to review retained funding, use --operator-mint if conversion is required, and resume the original reviewed operation before requesting a selected-release reinstall"
    )]
    RetainedOperationRecoveryRequired {
        operation_id: String,
        plan_sha256: String,
    },

    #[error("reviewed Fleet plan digest changed before its first effect")]
    DriftedBeforeApply,

    #[error(
        "replan-required Fleet operation owns completed reinstall evidence for desired input {expected}; refusing alternate desired input {actual}"
    )]
    RetainedReinstallDesiredConflict { actual: String, expected: String },

    #[error(
        "Root prerequisite retains activation authority for desired input {expected}; refreshed input {actual} is refused before Store effects; preserve the retained plan and journal for recovery review"
    )]
    RootActivationDesiredConflict { actual: String, expected: String },

    #[error("reviewed Fleet plan {expected} does not match retained plan {actual}")]
    PlanDigestMismatch { actual: String, expected: String },

    #[error("reviewed Fleet plan is missing; run `canic fleet ensure <fleet>` first")]
    PlanMissing,

    #[error("retained Fleet plan failed its content digest")]
    PlanIntegrity,

    #[error("retained Fleet journal does not match the reviewed Fleet operation")]
    JournalIntegrity,

    #[error(
        "Fleet ensure action {action} made no progress for {observations} consecutive observations (last progress: {progress_identity}); operation remains resumable"
    )]
    Stalled {
        action: String,
        observations: u32,
        progress_identity: String,
    },

    #[error("Root {root} provisioning has made no progress for {observations} observations: stage={stage:?}, target={target}, operation={operation}, diagnostic={diagnostic_code}, retry=Backoff, failed_at_ns={failed_at_ns}; the issued operation remains retained; resume the same reviewed apply after the dependency recovers", operation = canic_core::cdk::utils::hash::hex_bytes(.operation_id))]
    ProvisioningRetryPending {
        root: candid::Principal,
        stage: canic_contracts::dto::component_provisioning::ProvisioningFailureStage,
        target: candid::Principal,
        operation_id: [u8; 32],
        diagnostic_code: u16,
        observations: u32,
        failed_at_ns: u64,
    },

    #[error(
        "Root-owned canister {target} remained {last_lifecycle} for {observations} consecutive passive observations; no command was reissued and the operation remains resumable"
    )]
    RootOwnedObservationStalled {
        target: String,
        observations: u32,
        last_lifecycle: String,
    },

    #[error(
        "selected Cycles Ledger account has {actual} cycles, below reviewed maximum debit {required}; shortfall {} cycles; no new execution journal or funding intent was retained; fund the selected operator account before retrying this reviewed plan; repeat fleet ensure with --operator-mint and without --apply for a read-only ICP quote",
        required.saturating_sub(*actual)
    )]
    InsufficientOperatorCycles { actual: u128, required: u128 },

    #[error("terminal Fleet cycle conservation failed: {0}")]
    Conservation(String),

    #[error(
        "completed Fleet operation {operation_id} cannot replay accounting for plan {plan_sha256}: current operator balance {current_operator_cycles} is outside the reviewed range {minimum_operator_cycles}..={operator_source_cycles}; subsequent account activity cannot be attributed to this Fleet operation; preserve the completed plan, journal and receipts, run `canic fleet ensure {fleet}` without --apply to review a fresh plan, then apply only its newly reviewed digest and debit; do not repeat the completed deployment or edit its accounting"
    )]
    TerminalReplayBalanceChanged {
        fleet: String,
        operation_id: String,
        plan_sha256: String,
        current_operator_cycles: u128,
        minimum_operator_cycles: u128,
        operator_source_cycles: u128,
    },

    #[error("terminal Fleet inventory failed exact current-authority validation: {0}")]
    TerminalInventory(#[source] TerminalInventoryError),

    #[error(
        "reviewed effects completed, but live Fleet state still differs from desired state; review a new plan"
    )]
    ConvergenceDrift,

    #[error(
        "fixture publication action {action} exhausted its {maximum_attempts} reviewed paid attempts; preserve the operation and reconcile its exact source status; no additional update was issued"
    )]
    FixturePublicationBound {
        action: String,
        maximum_attempts: u32,
    },

    #[error(
        "Root pool maintenance {action} exhausted its reviewed limit of {maximum_updates} updates; retained receipts remain authoritative"
    )]
    PoolMaintenanceBound {
        action: String,
        maximum_updates: u32,
    },

    #[error(
        "created canister {canister} retained one exact Ledger creation receipt, but its first live balance is {actual_cycles} cycles instead of requested target {requested_cycles} (deficit {deficit_cycles}; configured management creation fee {configured_fee_cycles}); no controller or protocol action followed, and the immutable operation is now replan-required"
    )]
    ReplanRequiredAfterCreateBalanceDrift {
        actual_cycles: u128,
        canister: String,
        configured_fee_cycles: u128,
        deficit_cycles: u128,
        requested_cycles: u128,
    },

    #[error("Fleet successor requires a new review: {reason:?}; {}", .review.as_deref().map(ToString::to_string).unwrap_or_default())]
    SuccessorReviewRequired {
        reason: crate::fleet_ensure::model::FleetEnsureSuccessorReviewReason,
        review: Option<Box<crate::fleet_ensure::model::FleetSuccessorReview>>,
    },

    #[error(transparent)]
    FreshPoolCreationUnderfunded(Box<FreshPoolCreationUnderfundedError>),

    #[error(transparent)]
    EstateFundingRequired(Box<EstateFundingRequiredError>),

    #[error("native Root funding needs explicit approval of review {review_sha256}")]
    NativeFundingRequired { review_sha256: String },

    #[error(transparent)]
    Policy(#[from] EnsurePolicyError),

    #[error(transparent)]
    Inventory(#[from] crate::fleet_ensure::CurrentFleetInventoryError),

    #[error("Fleet platform operation failed: {0}")]
    Platform(#[source] E),

    #[error(transparent)]
    State(#[from] EnsureStateError),
}

/// Recover the selected target for an exact reinstall apply, including terminal replay.
pub fn retained_reinstall_apply_plan<E: std::error::Error + 'static>(
    root: &Path,
    environment: &str,
    requested_fleet: &str,
    reviewed_plan_sha256: &str,
) -> Result<Option<FleetEnsurePlan>, EnsureWorkflowError<E>> {
    validate_path_labels(environment, requested_fleet)?;
    let paths = EnsurePaths::under(root, environment, requested_fleet);
    let _lock = lock_operation(&paths)?;
    // The staged review owns apply; its source remains receipt-only evidence.
    let plan = if let Some(review) = crate::fleet_ensure::ops::reinstall::adoption::review(&paths)?
    {
        Some(review)
    } else {
        if crate::fleet_ensure::ops::operation_selection::completed(
            &paths,
            environment,
            requested_fleet,
        )?
        .is_some_and(|completed| completed.plan_sha256 != reviewed_plan_sha256)
        {
            return Ok(None);
        }
        retained_plan::read(&paths, environment, requested_fleet)?
    };
    let Some(plan) = plan else {
        return Ok(None);
    };
    let plan = verified_plan(plan)?;
    if plan.environment != environment || plan.fleet != requested_fleet {
        return Err(EnsureWorkflowError::PlanIntegrity);
    }
    if plan.reinstall.is_some() && plan.plan_sha256 == reviewed_plan_sha256 {
        Ok(Some(plan))
    } else {
        Ok(None)
    }
}

/// Load and verify the sole immutable in-progress operation for one exact Fleet path.
pub fn retained_in_progress_plan<E>(
    root: &Path,
    environment: &str,
    requested_fleet: &str,
) -> Result<Option<FleetEnsurePlan>, EnsureWorkflowError<E>>
where
    E: std::error::Error + 'static,
{
    validate_path_labels(environment, requested_fleet)?;
    let paths = EnsurePaths::under(root, environment, requested_fleet);
    let _lock = lock_operation(&paths)?;
    if let Some(review) = crate::fleet_ensure::ops::reinstall::adoption::review(&paths)? {
        if review.environment != environment || review.fleet != requested_fleet {
            return Err(EnsureWorkflowError::PlanIntegrity);
        }
        return Ok(Some(review));
    }
    if crate::fleet_ensure::ops::operation_selection::completed(
        &paths,
        environment,
        requested_fleet,
    )?
    .is_some()
    {
        return Ok(None);
    }
    let Some(journal) = retained_plan::journal(&paths, environment, requested_fleet)? else {
        return Ok(None);
    };
    if !matches!(
        journal.completion,
        FleetEnsureCompletion::InProgress | FleetEnsureCompletion::Prepared
    ) {
        return Ok(None);
    }
    let retained = verified_plan(
        retained_plan::read(&paths, environment, requested_fleet)?
            .ok_or(EnsureWorkflowError::PlanMissing)?,
    )?;
    if retained.environment != environment || retained.fleet != requested_fleet {
        return Err(EnsureWorkflowError::PlanIntegrity);
    }
    let state = read_state(&paths, requested_fleet)?;
    let prepared_full = journal.completion == FleetEnsureCompletion::Prepared
        && (retained.scope == FleetEnsurePlanScope::Full
            || (retained.scope == FleetEnsurePlanScope::RootReinstallPrerequisite
                && retained.reinstall.as_ref().is_some_and(|intent| {
                    intent
                        .activation_reset
                        .as_ref()
                        .is_some_and(|activation| activation.preparation.is_some())
                })))
        && retained.reinstall.as_ref().is_some_and(|intent| {
            intent.operation_id == journal.operation_id
                && intent.operation_id == retained.operation_id
        });
    if !prepared_full {
        verify_journal_integrity(&journal, &retained, requested_fleet, &state)?;
    }
    Ok(Some(retained))
}

/// Build and retain one read-only plan from current desired state plus live observation.
pub fn plan<P>(
    root: &Path,
    desired: &crate::fleet_ensure::model::DesiredFleet,
    desired_sha256: &str,
    requested_fleet: &str,
    created_at_time: u64,
    platform: &mut P,
) -> Result<FleetEnsureReport, EnsureWorkflowError<P::Error>>
where
    P: EnsurePlatform,
{
    platform.with_planning_observations(|platform| {
        plan_with_observations(
            root,
            desired,
            desired_sha256,
            requested_fleet,
            created_at_time,
            platform,
        )
    })
}

#[expect(
    clippy::too_many_lines,
    reason = "one locked read-only transaction reconciles retained authority, observation and the immutable plan"
)]
fn plan_with_observations<P>(
    root: &Path,
    desired: &crate::fleet_ensure::model::DesiredFleet,
    desired_sha256: &str,
    requested_fleet: &str,
    created_at_time: u64,
    platform: &mut P,
) -> Result<FleetEnsureReport, EnsureWorkflowError<P::Error>>
where
    P: EnsurePlatform,
{
    validate_path_identity(desired, requested_fleet)?;
    let paths = EnsurePaths::under(root, &desired.environment, requested_fleet);
    let _lock = lock_operation(&paths)?;
    if let Some(review) = crate::fleet_ensure::ops::reinstall::adoption::review(&paths)? {
        if review.desired_sha256 != desired_sha256 {
            return Err(EnsureWorkflowError::DriftedBeforeApply);
        }
        verify_release_transition(root, desired, review.scope)?;
        return Ok(FleetEnsureReport {
            funding_review: None,
            actual_conservation: None,
            effects_applied: 0,
            plan: review,
            terminal: false,
        });
    }
    if let Some(mut journal) =
        retained_plan::journal(&paths, &desired.environment, requested_fleet)?
        && journal.completion == FleetEnsureCompletion::InProgress
    {
        let retained = verified_plan(
            retained_plan::read(&paths, &desired.environment, requested_fleet)?
                .ok_or(EnsureWorkflowError::PlanMissing)?,
        )?;
        let state = read_state(&paths, requested_fleet)?;
        verify_journal(&journal, &retained, requested_fleet, &state)?;
        let operation_desired = retained
            .reviewed_desired
            .as_deref()
            .ok_or(EnsureWorkflowError::PlanIntegrity)?
            .desired();
        verify_release_transition(root, operation_desired, retained.scope)?;
        if journal.estate_funding_required.is_some()
            || funding::native_review_applicable(&retained, &journal)
        {
            platform
                .bind_reviewed_desired(operation_desired)
                .map_err(EnsureWorkflowError::Platform)?;
        }
        let funding_review = funding::prepare(
            &paths,
            &retained,
            &mut journal,
            &state,
            created_at_time,
            platform,
        )?;
        return Ok(FleetEnsureReport {
            funding_review,
            actual_conservation: None,
            effects_applied: applied_count(&journal),
            plan: retained,
            terminal: false,
        });
    }
    verify_release_transition(root, desired, FleetEnsurePlanScope::RootStartPrerequisite)?;
    let mut state = read_state(&paths, requested_fleet)?;
    let prior_plan = retained_plan::read(&paths, &desired.environment, requested_fleet)?
        .map(verified_plan)
        .transpose()?;
    let prior_journal = retained_plan::journal(&paths, &desired.environment, requested_fleet)?;
    if let Some(prior) = prior_plan
        .as_ref()
        .filter(|prior| prior.scope == FleetEnsurePlanScope::InfrastructureBootstrap)
    {
        crate::fleet_ensure::ops::infrastructure_bootstrap::convergence::prepare(
            &paths,
            prior,
            prior_journal
                .as_ref()
                .ok_or(EnsureWorkflowError::JournalIntegrity)?,
            &state,
            desired,
        )?;
    }
    if let Some(prior) = &prior_plan {
        let recovery_ready = prior_journal
            .as_ref()
            .map(|journal| reinstall_recovery_ready(prior, journal, &state))
            .transpose()?
            .unwrap_or(false);
        if recovery_ready && let Some(journal) = &prior_journal {
            if prior.desired_sha256 != desired_sha256 {
                return Err(EnsureWorkflowError::ReinstallConflict);
            }
            retain_completed_reinstalls(&mut state, prior, journal);
        }
        let activation_reset = prior
            .reinstall
            .as_ref()
            .is_some_and(|intent| intent.activation_reset.is_some());
        if activation_reset
            && prior.scope == FleetEnsurePlanScope::RootReinstallPrerequisite
            && let Some(journal) = prior_journal
                .as_ref()
                .filter(|journal| journal.completion == FleetEnsureCompletion::Converged)
        {
            verify_journal(journal, prior, requested_fleet, &state)?;
            retain_completed_reinstalls(&mut state, prior, journal);
            write_state(&paths, &state)?;
        }
        let unfinished_reset = prior.reinstall.is_some()
            && !recovery_ready
            && !prior_journal.as_ref().is_some_and(|journal| {
                journal.plan_sha256 == prior.plan_sha256
                    && journal.completion == FleetEnsureCompletion::Converged
            });
        if prior.scope == FleetEnsurePlanScope::ReinstallPreparation
            || unfinished_reset
            || (activation_reset && prior.scope == FleetEnsurePlanScope::RootReinstallPrerequisite)
        {
            if prior.desired_sha256 != desired_sha256 {
                return Err(EnsureWorkflowError::ReinstallConflict);
            }
            return reinstall::continue_plan(
                root,
                &paths,
                desired,
                prior,
                prior_journal.as_ref(),
                &state,
                platform,
            );
        }
    }
    let matching_prior = reconcile_retained_planning_state(
        &mut state,
        prior_plan.as_ref(),
        prior_journal.as_ref(),
        requested_fleet,
        &desired.environment,
        desired_sha256,
    )?;
    if let Some(prior) = matching_prior {
        retain_plan_cycles(&mut state, prior);
        if let Some(journal) = &prior_journal {
            retain_completed_reinstalls(&mut state, prior, journal);
        }
    }
    // Store and Component Registry preparation bind the setup operation identity.
    // Its verified import boundary continues that identity into workload provisioning.
    let operation_id = prior_plan
        .as_ref()
        .filter(|prior| prior.scope == FleetEnsurePlanScope::InfrastructureBootstrap)
        .or_else(|| matching_prior.filter(|prior| prior.scope == FleetEnsurePlanScope::Full))
        .map_or_else(
            || operation_id(desired_sha256, &desired.environment, requested_fleet),
            |prior| prior.operation_id.clone(),
        );
    if let Some(management) = platform
        .observe_root_management(&state, &BTreeSet::new())
        .map_err(EnsureWorkflowError::Platform)?
    {
        let root_start_authority = read_root_start_authority(&paths)?;
        if let Some(plan) = compile_root_start_prerequisite_plan(RootStartPlanInput {
            state: &state,
            authority: root_start_authority.as_ref(),
            created_at_time,
            desired,
            desired_sha256,
            observation: &management,
            requested_fleet,
        })? {
            write_plan(&paths, &plan)?;
            return Ok(FleetEnsureReport {
                funding_review: None,
                actual_conservation: None,
                effects_applied: 0,
                plan,
                terminal: false,
            });
        }
        verify_release_transition(root, desired, FleetEnsurePlanScope::Full)?;
        if let Some(plan) = crate::fleet_ensure::policy::root_reinstall::compile(
            RootStartPlanInput {
                state: &state,
                authority: None,
                created_at_time,
                desired,
                desired_sha256,
                observation: &management,
                requested_fleet,
            },
            &resolve_desired_artifacts(root, desired)?,
        )? {
            write_plan(&paths, &plan)?;
            return Ok(FleetEnsureReport {
                funding_review: None,
                actual_conservation: None,
                effects_applied: 0,
                plan,
                terminal: false,
            });
        }
    }
    verify_release_transition(root, desired, FleetEnsurePlanScope::Full)?;
    let artifacts = resolve_desired_artifacts(root, desired)?;
    let mut observation = platform
        .observe(&operation_id, &state)
        .map_err(EnsureWorkflowError::Platform)?;
    retain_observed_cycles(&mut state, &observation);
    // The prior Converged journal can remain visible until apply. Planning may refresh only
    // backup-inert observation and prior-operation evidence, never successor topology authority.
    write_state(&paths, &state)?;
    let terminal_inventory_operation_id = current_terminal_inventory_operation(
        &state,
        prior_plan.as_ref(),
        prior_journal.as_ref(),
        requested_fleet,
    )?;
    if let Some(operation_id) = terminal_inventory_operation_id.as_deref() {
        attach_terminal_inventory_cycles(operation_id, &state, platform, &mut observation)?;
    }
    let mut protocol_actions = platform
        .protocol_actions(
            terminal_inventory_operation_id
                .as_deref()
                .unwrap_or(&operation_id),
            &state,
        )
        .map_err(EnsureWorkflowError::Platform)?;
    let mut retained_observations = 0_u32;
    let mut plan = loop {
        match compile_plan(
            desired,
            &artifacts,
            &protocol_actions,
            desired_sha256,
            requested_fleet,
            &observation,
            created_at_time,
            &operation_id,
            None,
        ) {
            Ok(plan) => break plan,
            Err(EnsurePolicyError::PendingRootOwnedBalance { name }) => {
                retained_observations = retained_observations.saturating_add(1);
                if retained_observations >= desired.maximum_stalled_observations {
                    return Err(EnsureWorkflowError::RootOwnedObservationStalled {
                        last_lifecycle: root_owned_lifecycle_label(&observation, &name).to_string(),
                        observations: retained_observations,
                        target: name,
                    });
                }
                platform.pace_root_owned_observation(&name, retained_observations);
                observation = platform
                    .observe(&operation_id, &state)
                    .map_err(EnsureWorkflowError::Platform)?;
                retain_observed_cycles(&mut state, &observation);
                write_state(&paths, &state)?;
                if let Some(operation_id) = terminal_inventory_operation_id.as_deref() {
                    attach_terminal_inventory_cycles(
                        operation_id,
                        &state,
                        platform,
                        &mut observation,
                    )?;
                }
                protocol_actions = platform
                    .protocol_actions(
                        terminal_inventory_operation_id
                            .as_deref()
                            .unwrap_or(&operation_id),
                        &state,
                    )
                    .map_err(EnsureWorkflowError::Platform)?;
            }
            Err(error) => return Err(error.into()),
        }
    };
    for action in ordered_actions(&plan) {
        verify_store_activation_binding(&plan, action, &state, platform)?;
    }
    bind_terminal_inventory_operation(&mut plan, terminal_inventory_operation_id);
    write_plan(&paths, &plan)?;
    Ok(FleetEnsureReport {
        funding_review: None,
        actual_conservation: None,
        effects_applied: 0,
        plan,
        terminal: false,
    })
}

fn current_terminal_inventory_operation<E>(
    state: &FleetEnsureStateRecord,
    prior_plan: Option<&FleetEnsurePlan>,
    prior_journal: Option<&FleetEnsureJournalRecord>,
    requested_fleet: &str,
) -> Result<Option<String>, EnsureWorkflowError<E>>
where
    E: std::error::Error + 'static,
{
    if state.active_registry.is_none() {
        return Ok(None);
    }
    let prior_plan = prior_plan.ok_or(EnsureWorkflowError::PlanMissing)?;
    let prior_journal = prior_journal.ok_or(EnsureWorkflowError::JournalIntegrity)?;
    verify_journal(prior_journal, prior_plan, requested_fleet, state)?;
    // The separately receipted setup/import boundary has no workload inventory yet.
    // Planning admitted and archived its exact completion before refreshing observations.
    if prior_plan.scope == FleetEnsurePlanScope::InfrastructureBootstrap
        && prior_journal.completion == FleetEnsureCompletion::Converged
    {
        return Ok(None);
    }
    if prior_journal.completion != FleetEnsureCompletion::Converged
        || prior_plan.scope != FleetEnsurePlanScope::Full
    {
        return Err(EnsureWorkflowError::JournalIntegrity);
    }
    let operation_id = prior_plan
        .terminal_inventory_operation_id
        .as_ref()
        .unwrap_or(&prior_plan.operation_id);
    if operation_id.is_empty() {
        return Err(EnsureWorkflowError::JournalIntegrity);
    }
    Ok(Some(operation_id.clone()))
}

fn bind_terminal_inventory_operation(plan: &mut FleetEnsurePlan, operation_id: Option<String>) {
    plan.terminal_inventory_operation_id = operation_id;
    plan.plan_sha256 = expected_plan_sha256(plan);
}

fn reconcile_retained_planning_state<'a, E>(
    state: &mut FleetEnsureStateRecord,
    prior_plan: Option<&'a FleetEnsurePlan>,
    prior_journal: Option<&FleetEnsureJournalRecord>,
    requested_fleet: &str,
    desired_environment: &str,
    desired_sha256: &str,
) -> Result<Option<&'a FleetEnsurePlan>, EnsureWorkflowError<E>>
where
    E: std::error::Error + 'static,
{
    if let Some((prior, journal)) = prior_plan.zip(prior_journal)
        && prior.fleet == requested_fleet
        && prior.environment == desired_environment
        && retained_reinstall_desired_conflict(
            state,
            &prior.fleet,
            &prior.operation_id,
            &prior.desired_sha256,
            desired_sha256,
            journal,
        )
    {
        return Err(EnsureWorkflowError::RetainedReinstallDesiredConflict {
            actual: desired_sha256.to_string(),
            expected: prior.desired_sha256.clone(),
        });
    }
    if let Some((prior, journal)) = prior_plan.zip(prior_journal)
        && prior.scope == FleetEnsurePlanScope::RootReinstallPrerequisite
        && prior.desired_sha256 != desired_sha256
    {
        verify_journal(journal, prior, requested_fleet, state)?;
        return Err(EnsureWorkflowError::RootActivationDesiredConflict {
            actual: desired_sha256.to_string(),
            expected: prior.desired_sha256.clone(),
        });
    }
    let matching_prior = prior_plan.filter(|prior| {
        prior.fleet == requested_fleet
            && prior.environment == desired_environment
            && prior.desired_sha256 == desired_sha256
    });
    let retained_operation_matches =
        matching_prior
            .zip(prior_journal)
            .is_some_and(|(prior, journal)| {
                retained_reinstall_operation_matches(
                    state,
                    &prior.fleet,
                    &prior.operation_id,
                    journal,
                )
            });
    if !retained_operation_matches {
        state.completed_reinstall_action_sha256.clear();
        state.completed_reinstall_operation_id = None;
        state.completed_reinstalls.clear();
    }
    Ok(matching_prior)
}

fn retain_completed_reinstalls(
    state: &mut FleetEnsureStateRecord,
    plan: &FleetEnsurePlan,
    journal: &FleetEnsureJournalRecord,
) {
    if journal.fleet != plan.fleet || journal.operation_id != plan.operation_id {
        return;
    }
    for action in ordered_actions(plan) {
        let EnsureAction::Install {
            mode: crate::fleet_ensure::model::InstallMode::Reinstall,
            name,
            principal,
            wasm_sha256,
            ..
        } = action
        else {
            continue;
        };
        let action_sha256 = action_sha256(action);
        let Some(effect) = journal
            .effects
            .iter()
            .find(|effect| effect.action_sha256 == action_sha256)
        else {
            continue;
        };
        if effect.state == EffectState::Applied
            && let Some(pre_canister_version) = effect.pre_canister_version
            && state
                .principals
                .get(name)
                .is_some_and(|retained| retained == principal)
        {
            state.completed_reinstall_operation_id = Some(journal.operation_id.clone());
            state
                .completed_reinstall_action_sha256
                .insert(name.clone(), action_sha256);
            state
                .completed_reinstalls
                .insert(name.clone(), pre_canister_version);
            if let Some(topology) = state.topology.get_mut(name) {
                topology.module_hash = Some(wasm_sha256.clone());
            }
        }
    }
}

fn completed_reinstall_evidence_matches(
    state: &FleetEnsureStateRecord,
    journal: &FleetEnsureJournalRecord,
) -> bool {
    !state.completed_reinstalls.is_empty()
        && state
            .completed_reinstalls
            .iter()
            .all(|(name, pre_version)| {
                state
                    .completed_reinstall_action_sha256
                    .get(name)
                    .and_then(|action_sha256| {
                        journal
                            .effects
                            .iter()
                            .find(|effect| effect.action_sha256 == *action_sha256)
                    })
                    .is_some_and(|effect| {
                        effect.state == EffectState::Applied
                            && effect.pre_canister_version == Some(*pre_version)
                    })
            })
        && state.completed_reinstall_action_sha256.len() == state.completed_reinstalls.len()
}

/// Admit another review only after every effect of the selected reset is accounted for.
fn reinstall_recovery_ready<E: std::error::Error + 'static>(
    plan: &FleetEnsurePlan,
    journal: &FleetEnsureJournalRecord,
    state: &FleetEnsureStateRecord,
) -> Result<bool, EnsureWorkflowError<E>> {
    if plan.scope != FleetEnsurePlanScope::Full
        || plan.reinstall.is_none()
        || journal.completion != FleetEnsureCompletion::ReplanRequired
        || journal.plan_sha256 != plan.plan_sha256
    {
        return Ok(false);
    }
    verify_journal(journal, plan, &plan.fleet, state)?;
    let actions = continuation::actions(plan, journal);
    let complete = actions.len() == journal.effects.len()
        && journal
            .effects
            .iter()
            .all(|effect| effect.state == EffectState::Applied);
    let installs_recorded = actions
        .iter()
        .zip(&journal.effects)
        .all(|(action, effect)| {
            !matches!(action, EnsureAction::Install { .. }) || effect.pre_canister_version.is_some()
        });
    if !complete || !installs_recorded {
        return Err(EnsureWorkflowError::JournalIntegrity);
    }
    Ok(true)
}

fn retained_reinstall_operation_matches(
    state: &FleetEnsureStateRecord,
    prior_fleet: &str,
    prior_operation_id: &str,
    journal: &FleetEnsureJournalRecord,
) -> bool {
    journal.completion == FleetEnsureCompletion::ReplanRequired
        && journal.fleet == prior_fleet
        && journal.operation_id == prior_operation_id
        && state.completed_reinstall_operation_id.as_deref() == Some(journal.operation_id.as_str())
        && completed_reinstall_evidence_matches(state, journal)
}

fn retained_reinstall_desired_conflict(
    state: &FleetEnsureStateRecord,
    prior_fleet: &str,
    prior_operation_id: &str,
    prior_desired_sha256: &str,
    supplied_desired_sha256: &str,
    journal: &FleetEnsureJournalRecord,
) -> bool {
    prior_desired_sha256 != supplied_desired_sha256
        && retained_reinstall_operation_matches(state, prior_fleet, prior_operation_id, journal)
}

fn retain_plan_cycles(state: &mut FleetEnsureStateRecord, plan: &FleetEnsurePlan) {
    for canister in &plan.canisters {
        if let Some(principal) = &canister.principal {
            state
                .retained_cycles_by_principal
                .entry(principal.clone())
                .or_insert(canister.observed_cycles);
        }
    }
}

fn retain_observed_cycles(state: &mut FleetEnsureStateRecord, observation: &FleetObservation) {
    for live in observation.canisters.values().filter_map(Option::as_ref) {
        state
            .retained_cycles_by_principal
            .insert(live.principal.clone(), live.cycles);
    }
    state
        .retained_cycles_by_principal
        .extend(observation.additional_controlled_cycles.clone());
}

fn root_owned_lifecycle_label(observation: &FleetObservation, target: &str) -> &'static str {
    observation
        .canisters
        .get(target)
        .and_then(Option::as_ref)
        .and_then(|live| live.root_owned_lifecycle)
        .map_or("unavailable", |lifecycle| lifecycle.label())
}

/// Apply or resume exactly one reviewed plan until it converges or returns a typed blocker.
#[expect(
    clippy::too_many_lines,
    reason = "the sole effect driver keeps intent, reconciliation and terminal publication visibly ordered"
)]
pub fn apply<P>(
    root: &Path,
    desired: &crate::fleet_ensure::model::DesiredFleet,
    desired_sha256: &str,
    requested_fleet: &str,
    reviewed_plan_sha256: &str,
    platform: &mut P,
) -> Result<FleetEnsureReport, EnsureWorkflowError<P::Error>>
where
    P: EnsurePlatform,
{
    validate_path_identity(desired, requested_fleet)?;
    let paths = EnsurePaths::under(root, &desired.environment, requested_fleet);
    let _lock = lock_operation(&paths)?;
    let pending = crate::fleet_ensure::ops::reinstall::adoption::review(&paths)?;
    let adopting = pending.is_some();
    let retained_plan = verified_plan(match pending {
        Some(plan) => plan,
        None => retained_plan::read(&paths, &desired.environment, requested_fleet)?
            .ok_or(EnsureWorkflowError::PlanMissing)?,
    })?;
    let retained_journal = if adopting {
        None
    } else {
        retained_plan::journal(&paths, &desired.environment, requested_fleet)?
    };
    if retained_plan.plan_sha256 != reviewed_plan_sha256
        && !retained_journal
            .as_ref()
            .is_some_and(|journal| funding::accepts(&retained_plan, journal, reviewed_plan_sha256))
    {
        return Err(EnsureWorkflowError::PlanDigestMismatch {
            actual: retained_plan.plan_sha256,
            expected: reviewed_plan_sha256.to_string(),
        });
    }
    let mut state = read_state(&paths, requested_fleet)?;
    let in_progress = retained_journal
        .as_ref()
        .is_some_and(|journal| journal.completion == FleetEnsureCompletion::InProgress);
    if let Some(journal) = retained_journal.as_ref().filter(|_| in_progress) {
        verify_journal(journal, &retained_plan, requested_fleet, &state)?;
    }
    if !in_progress
        && retained_plan.reinstall.is_none()
        && retained_plan.desired_sha256 != desired_sha256
    {
        return Err(EnsureWorkflowError::DriftedBeforeApply);
    }
    let operation_desired = if in_progress || retained_plan.reinstall.is_some() {
        let reviewed = retained_plan
            .reviewed_desired
            .as_deref()
            .ok_or(EnsureWorkflowError::PlanIntegrity)?
            .desired();
        validate_path_identity(reviewed, requested_fleet)?;
        if reviewed.environment != retained_plan.environment {
            return Err(EnsureWorkflowError::PlanIntegrity);
        }
        reviewed
    } else {
        if retained_plan.desired_sha256 != desired_sha256 {
            return Err(EnsureWorkflowError::DriftedBeforeApply);
        }
        desired
    };
    if let Some(journal) = &retained_journal
        && let Some(actual) = crate::fleet_ensure::ops::terminal::read(
            &paths,
            &retained_plan,
            journal,
            &state,
            &funding_plan::<P::Error>(&retained_plan, journal)?.conservation,
        )?
    {
        verify_journal(journal, &retained_plan, requested_fleet, &state)?;
        if journal.completion != FleetEnsureCompletion::Converged {
            let mut completed = journal.clone();
            completed.completion = FleetEnsureCompletion::Converged;
            completed.stalled_observations = 0;
            write_journal(&paths, &completed)?;
        }
        return Ok(FleetEnsureReport {
            funding_review: None,
            actual_conservation: Some(actual),
            effects_applied: 0,
            plan: retained_plan,
            terminal: true,
        });
    }
    if let Some(journal) = &retained_journal
        && let Some(actual) = crate::fleet_ensure::ops::infrastructure_bootstrap::terminal::read(
            &paths,
            &retained_plan,
            journal,
            &state,
        )?
    {
        verify_journal(journal, &retained_plan, requested_fleet, &state)?;
        if journal.completion != FleetEnsureCompletion::Converged {
            let mut completed = journal.clone();
            completed.completion = FleetEnsureCompletion::Converged;
            completed.stalled_observations = 0;
            write_journal(&paths, &completed)?;
        }
        return Ok(FleetEnsureReport {
            funding_review: None,
            actual_conservation: Some(actual),
            effects_applied: 0,
            plan: retained_plan,
            terminal: true,
        });
    }
    if retained_plan.scope == FleetEnsurePlanScope::InfrastructureBootstrap {
        crate::fleet_ensure::ops::infrastructure_bootstrap::verify_plan(root, &retained_plan)?;
    }
    verify_release_transition(root, operation_desired, retained_plan.scope)?;
    if !retained_journal.as_ref().is_some_and(|journal| {
        journal.completion == FleetEnsureCompletion::Converged
            && journal.plan_sha256 == retained_plan.plan_sha256
    }) {
        validate_creation_fee_scope(
            operation_desired,
            &retained_plan.canisters,
            &retained_plan.conservation.estate_funding_domains,
        )?;
    }
    platform
        .bind_reviewed_desired(operation_desired)
        .map_err(EnsureWorkflowError::Platform)?;
    if let Some(source) = &retained_plan.infrastructure_bootstrap {
        platform
            .bind_infrastructure_bootstrap(source)
            .map_err(EnsureWorkflowError::Platform)?;
    }
    let operation_desired_sha256 = retained_plan.desired_sha256.as_str();
    if let Some(review) = retained_journal.as_ref().and_then(
        crate::fleet_ensure::ops::infrastructure_bootstrap::registration_recovery::approved,
    ) && !platform
        .bind_bootstrap_registration_recovery(review)
        .map_err(EnsureWorkflowError::Platform)?
    {
        return Err(EnsureWorkflowError::PlanIntegrity);
    }
    continuation::verify_inputs(root, operation_desired, &retained_plan)?;
    if let Some(journal) = &retained_journal {
        continuation::verify_canonical(&retained_plan, journal, &state, platform)?;
        if reinstall_recovery_ready(&retained_plan, journal, &state)? {
            return Err(EnsureWorkflowError::SuccessorReviewRequired {
                reason:
                    crate::fleet_ensure::model::FleetEnsureSuccessorReviewReason::AdditionalEffect,
                review: None,
            });
        }
    }
    let completed_journal = retained_journal.as_ref().filter(|journal| {
        (journal.completion == FleetEnsureCompletion::Converged
            || (journal.completion == FleetEnsureCompletion::Prepared
                && retained_plan.scope == FleetEnsurePlanScope::ReinstallPreparation))
            && journal.plan_sha256 == retained_plan.plan_sha256
    });
    if retained_plan.scope == FleetEnsurePlanScope::ReinstallPreparation
        && let Some(journal) = completed_journal
    {
        let actual = reinstall::complete(&retained_plan, journal, &state, platform)?;
        return Ok(FleetEnsureReport {
            funding_review: None,
            actual_conservation: Some(actual),
            effects_applied: 0,
            plan: retained_plan,
            terminal: true,
        });
    }
    if retained_plan.scope == FleetEnsurePlanScope::RootReinstallPrerequisite
        && let Some(journal) = completed_journal
    {
        verify_journal(journal, &retained_plan, requested_fleet, &state)?;
        let actual = root_reinstall::complete(&retained_plan, journal, &state, platform)?;
        report_progress_state(
            platform,
            &retained_plan,
            journal,
            FleetEnsurePhase::Infrastructure,
            FleetEnsureProgressState::PrerequisiteComplete,
        );
        return Ok(FleetEnsureReport {
            funding_review: None,
            actual_conservation: Some(actual),
            effects_applied: 0,
            plan: retained_plan,
            terminal: true,
        });
    }
    if retained_plan.scope == FleetEnsurePlanScope::RootStartPrerequisite
        && let Some(journal) = completed_journal
    {
        verify_terminal_root_start_replay(
            operation_desired,
            &retained_plan,
            journal,
            requested_fleet,
            platform,
            &state,
        )?;
        report_progress_state(
            platform,
            &retained_plan,
            journal,
            FleetEnsurePhase::Infrastructure,
            FleetEnsureProgressState::PrerequisiteComplete,
        );
        return Ok(FleetEnsureReport {
            funding_review: None,
            actual_conservation: None,
            effects_applied: applied_count(journal),
            plan: retained_plan,
            terminal: true,
        });
    }
    let mut journal = match retained_journal {
        Some(journal) if journal.completion == FleetEnsureCompletion::InProgress => {
            verify_journal(&journal, &retained_plan, requested_fleet, &state)?;
            journal
        }
        _ => {
            let (observation, initial_controlled_cycles) = verify_fresh_plan(
                root,
                operation_desired,
                operation_desired_sha256,
                requested_fleet,
                &retained_plan,
                platform,
                &state,
            )?;
            // Every fresh scope crosses this boundary, including reinstall verification.
            // Retained intents must instead reconcile their exact, possibly paid request.
            let required = retained_plan.conservation.maximum_operator_debit_cycles;
            if observation.operator_cycles < required {
                return Err(EnsureWorkflowError::InsufficientOperatorCycles {
                    actual: observation.operator_cycles,
                    required,
                });
            }
            root_reinstall::verify_initial_funding(
                &retained_plan,
                observation.operator_cycles,
                platform,
            )?;
            let journal = FleetEnsureJournalRecord {
                release: None,
                bootstrap_registration_recovery: None,
                funding_observations: BTreeMap::new(),
                funding_reviews: Vec::new(),
                successor_phases: Vec::new(),
                completion: FleetEnsureCompletion::InProgress,
                estate_funding_required: None,
                effects: Vec::new(),
                fleet: requested_fleet.to_string(),
                initial_controlled_cycles,
                initial_estate_funding_cycles_by_root: retained_plan
                    .conservation
                    .estate_funding_domains
                    .iter()
                    .map(|domain| {
                        (
                            domain.root.clone(),
                            domain.available_cycles.unwrap_or_default(),
                        )
                    })
                    .collect(),
                initial_operator_cycles: observation.operator_cycles,
                operation_id: retained_plan.operation_id.clone(),
                plan_sha256: retained_plan.plan_sha256.clone(),
                schema_version: FLEET_ENSURE_SCHEMA_VERSION,
                stalled_observations: 0,
            };
            // Cross the nonterminal boundary before any effect-owned state can be retained.
            if adopting {
                crate::fleet_ensure::ops::reinstall::adoption::adopt(
                    &paths,
                    &retained_plan,
                    &journal,
                )?;
            } else {
                write_journal(&paths, &journal)?;
            }
            journal
        }
    };

    funding::resume(
        &paths,
        &retained_plan,
        &mut journal,
        &state,
        reviewed_plan_sha256,
        platform,
    )?;
    if journal
        .estate_funding_required
        .as_ref()
        .is_some_and(|pause| funding::covered(&journal, pause))
    {
        retain_estate_funding_pause(&paths, &mut journal, None)?;
    }
    let mut terminal_state;
    let mut bootstrap_coordinator_verified = false;
    // Only the immediately following protocol phase may consume this evidence.
    // It is invocation-local and never survives interruption or a preceding effect.
    let mut continuation_observation: Option<ContinuationObservation> = None;
    loop {
        let execution_actions = continuation::actions(&retained_plan, &journal)
            .into_iter()
            .cloned()
            .collect::<Vec<_>>();
        let actions = execution_actions.iter().collect::<Vec<_>>();
        let mut replayed_issued_commands = BTreeSet::new();
        let mut reported_phase = None;
        loop {
            let mut deferred_controller_observation = false;
            for (index, action) in actions.iter().enumerate() {
                let observation_policy = effect_observation_policy(operation_desired, action)?;
                let action_hash = action_sha256(action);
                let retained_effect = journal.effects.get(index);
                let continued_observation = retained_effect
                    .is_none_or(|effect| effect.state != EffectState::Applied)
                    .then(|| continuation_observation.take())
                    .flatten()
                    .filter(|evidence| evidence.first_action_sha256 == action_hash);
                if retained_effect.is_none_or(|effect| effect.state != EffectState::Applied) {
                    if retained_plan.infrastructure_bootstrap.is_some()
                        && index < ordered_actions(&retained_plan).len()
                        && !bootstrap_coordinator_verified
                        && operation_desired
                            .bootstrap
                            .as_ref()
                            .is_some_and(|bootstrap| action.name() != bootstrap.coordinator)
                    {
                        if !platform
                            .verify_bootstrap_coordinator(&state)
                            .map_err(EnsureWorkflowError::Platform)?
                        {
                            return Err(EnsureWorkflowError::ConvergenceDrift);
                        }
                        bootstrap_coordinator_verified = true;
                    }
                    let phase = action_progress_phase(action);
                    if reported_phase != Some(phase) {
                        report_progress(platform, &retained_plan, &journal, phase);
                        reported_phase = Some(phase);
                    }
                }
                if retained_plan.infrastructure_bootstrap.is_none()
                    && matches!(action, EnsureAction::FleetProtocol { .. })
                    && retained_effect.is_none_or(|effect| effect.state == EffectState::Intent)
                    && !prior_fleet_protocol_effect_started(&actions, &journal, index)
                {
                    let funding_observation = match continued_observation {
                        Some(evidence) => evidence.observation,
                        None => platform
                            .observe(&journal.operation_id, &state)
                            .map_err(EnsureWorkflowError::Platform)?,
                    };
                    let funded_plan = funding_plan(&retained_plan, &journal)?;
                    let required =
                        estate_funding_requirement(&funded_plan, &state, &funding_observation)?;
                    retain_estate_funding_pause(&paths, &mut journal, required.as_ref())?;
                    if let Some(required) = required {
                        report_progress_state(
                            platform,
                            &retained_plan,
                            &journal,
                            action_progress_phase(action),
                            FleetEnsureProgressState::FundingRequired,
                        );
                        return Err(EnsureWorkflowError::EstateFundingRequired(Box::new(
                            estate_funding_error(&required),
                        )));
                    }
                }
                independent_effects::apply_batch(
                    &paths,
                    &retained_plan,
                    &actions,
                    index,
                    &mut journal,
                    &state,
                    platform,
                    observation_policy.maximum_stalled_observations,
                )?;
                let mut initial_observation = if journal.effects.len() <= index {
                    crate::fleet_ensure::ops::infrastructure_bootstrap::inspection::reserve_effect(
                        &paths,
                        &retained_plan,
                        action,
                    )?;
                    let prepared = prepare_effect(platform, &journal.operation_id, action, &state)
                        .map_err(EnsureWorkflowError::Platform)?;
                    journal.effects.push(prepared.record);
                    write_journal(&paths, &journal)?;
                    prepared.observation
                } else {
                    None
                };
                let observation_started = Instant::now();
                loop {
                    let record = journal
                        .effects
                        .get(index)
                        .ok_or(EnsureWorkflowError::JournalIntegrity)?;
                    if record.action_sha256 != action_hash {
                        return Err(EnsureWorkflowError::DriftedBeforeApply);
                    }
                    if !create_record_is_exact(action, record, &state) {
                        return Err(EnsureWorkflowError::JournalIntegrity);
                    }
                    if matches!(record.state, EffectState::Applied) {
                        let retained_create = journal
                            .effects
                            .get(index)
                            .ok_or(EnsureWorkflowError::JournalIntegrity)
                            .and_then(|record| {
                                retain_applied_create_authority(
                                    operation_desired,
                                    action,
                                    record,
                                    &mut state,
                                )
                            })?;
                        let retained_funding = journal.effects.get(index).is_some_and(|record| {
                            retain_applied_funding_cycles(&mut state, action, record)
                        });
                        if retained_create || retained_funding {
                            write_state(&paths, &state)?;
                        }
                        break;
                    }

                    let observed = if let Some(observed) = initial_observation.take() {
                        observed
                    } else {
                        crate::fleet_ensure::ops::infrastructure_bootstrap::inspection::reserve_effect(
                            &paths, &retained_plan, action,
                        )?;
                        platform
                            .observe_effect(&journal.operation_id, action, record, &state)
                            .map_err(EnsureWorkflowError::Platform)?
                    };
                    let source_cycles = if observed.post_cycles.is_some() {
                        observed.post_cycles
                    } else {
                        platform
                            .action_cycles(action, &state)
                            .map_err(EnsureWorkflowError::Platform)?
                    };
                    let destination_cycles = platform
                        .action_destination_cycles(action, &state)
                        .map_err(EnsureWorkflowError::Platform)?;
                    let replan_after_create_balance =
                        observed.retry == EffectRetry::ReplanRequiredAfterCreateBalanceDrift;
                    let retained_create_balance_is_exact = !replan_after_create_balance
                        || (journal.effects.len() == index + 1
                            && journal.effects[..index]
                                .iter()
                                .all(|effect| effect.state == EffectState::Applied)
                            && record.state == EffectState::Issued
                            && create_identity_is_exact(action.name(), record, &state)
                            && observed.post_cycles.is_some()
                            && observed.post_cycles == source_cycles);
                    let protocol_funding_required = if record.state == EffectState::Issued {
                        observed
                            .estate_funding_required
                            .as_ref()
                            .map(|funding| {
                                protocol_estate_funding_requirement(&retained_plan, &state, funding)
                            })
                            .transpose()?
                    } else {
                        None
                    };
                    if let Some(required) = protocol_funding_required.as_ref()
                        && !funding::covered(&journal, required)
                    {
                        retain_estate_funding_pause(&paths, &mut journal, Some(required))?;
                        report_progress_state(
                            platform,
                            &retained_plan,
                            &journal,
                            action_progress_phase(action),
                            FleetEnsureProgressState::FundingRequired,
                        );
                        return Err(EnsureWorkflowError::EstateFundingRequired(Box::new(
                            estate_funding_error(required),
                        )));
                    }
                    if journal
                        .estate_funding_required
                        .as_ref()
                        .is_some_and(|required| required.pending_creation_operation_id.is_some())
                    {
                        retain_estate_funding_pause(&paths, &mut journal, None)?;
                    }
                    let record = journal
                        .effects
                        .get_mut(index)
                        .ok_or(EnsureWorkflowError::JournalIntegrity)?;
                    merge_observed_cycles(record, source_cycles, destination_cycles);
                    if replan_after_create_balance {
                        if !retained_create_balance_is_exact {
                            return Err(EnsureWorkflowError::JournalIntegrity);
                        }
                        let EnsureAction::Create {
                            name,
                            requested_initial_cycles,
                            ..
                        } = action
                        else {
                            return Err(EnsureWorkflowError::JournalIntegrity);
                        };
                        let actual_cycles = record
                            .post_cycles
                            .ok_or(EnsureWorkflowError::JournalIntegrity)?;
                        record.progress_identity = Some(observed.progress_identity);
                        record.state = EffectState::Applied;
                        retain_created_canister_for_replan(
                            operation_desired,
                            name,
                            record,
                            &mut state,
                        )?;
                        journal.completion = FleetEnsureCompletion::ReplanRequired;
                        journal.stalled_observations = 0;
                        write_journal(&paths, &journal)?;
                        write_state(&paths, &state)?;
                        let configured_fee_cycles = operation_desired
                            .management_creation_fee_cycles
                            .parse::<Cycles>()
                            .map(|cycles| cycles.to_u128())
                            .map_err(|_| EnsureWorkflowError::JournalIntegrity)?;
                        return Err(EnsureWorkflowError::ReplanRequiredAfterCreateBalanceDrift {
                            actual_cycles,
                            canister: name.clone(),
                            configured_fee_cycles,
                            deficit_cycles: requested_initial_cycles.saturating_sub(actual_cycles),
                            requested_cycles: *requested_initial_cycles,
                        });
                    }
                    if observed.applied {
                        record.progress_identity = Some(observed.progress_identity);
                        record.state = EffectState::Applied;
                        journal.stalled_observations = 0;
                        write_journal(&paths, &journal)?;
                        let record = journal
                            .effects
                            .get(index)
                            .ok_or(EnsureWorkflowError::JournalIntegrity)?;
                        let retained_create = retain_applied_create_authority(
                            operation_desired,
                            action,
                            record,
                            &mut state,
                        )?;
                        let retained_funding =
                            retain_applied_funding_cycles(&mut state, action, record);
                        if retained_create || retained_funding {
                            write_state(&paths, &state)?;
                        }
                        // Emit once per durable transition, including a reconciled lost reply.
                        report_progress(
                            platform,
                            &retained_plan,
                            &journal,
                            action_progress_phase(action),
                        );
                        break;
                    }

                    if observed.retry == EffectRetry::DeferUntilControllerObservation {
                        if !deferred_create_observation_is_exact(
                            operation_desired,
                            &actions,
                            index,
                            action,
                            record,
                            &state,
                        ) {
                            return Err(EnsureWorkflowError::JournalIntegrity);
                        }
                        let progress_identity = observed.progress_identity.clone();
                        if record.progress_identity.as_deref() == Some(&observed.progress_identity)
                        {
                            journal.stalled_observations =
                                journal.stalled_observations.saturating_add(1);
                        } else {
                            record.progress_identity = Some(observed.progress_identity);
                            journal.stalled_observations = 0;
                        }
                        write_journal(&paths, &journal)?;
                        if journal.stalled_observations
                            >= observation_policy.maximum_stalled_observations
                        {
                            return Err(EnsureWorkflowError::Stalled {
                                action: action.name().to_string(),
                                observations: journal.stalled_observations,
                                progress_identity,
                            });
                        }
                        deferred_controller_observation = true;
                        break;
                    }

                    let maintenance_continuation =
                        observed.retry == EffectRetry::ContinuePoolMaintenance;
                    if matches!(record.state, EffectState::Intent) || maintenance_continuation {
                        crate::fleet_ensure::policy::infrastructure_bootstrap::validate_effect_headroom(
                            &retained_plan,
                            action,
                            source_cycles,
                        )?;
                        if let Some(maximum_attempts) = action.fixture_publication_attempt_limit() {
                            if !reserve_fixture_publication_attempt(record, maximum_attempts) {
                                return Err(EnsureWorkflowError::FixturePublicationBound {
                                    action: action.name().to_string(),
                                    maximum_attempts,
                                });
                            }
                            // Reserve before the external call, even if its response never returns.
                            write_journal(&paths, &journal)?;
                        }
                        let record = journal
                            .effects
                            .get_mut(index)
                            .ok_or(EnsureWorkflowError::JournalIntegrity)?;
                        if let EnsureAction::FleetProtocol {
                            action: current, ..
                        } = action
                            && let CurrentFleetProtocolAction::MaintainPoolReadiness {
                                maximum_updates,
                                ..
                            } = current.as_ref()
                        {
                            if record.maintenance_attempts >= *maximum_updates {
                                return Err(EnsureWorkflowError::PoolMaintenanceBound {
                                    action: action.name().to_string(),
                                    maximum_updates: *maximum_updates,
                                });
                            }
                            if record.progress_identity.as_deref()
                                == Some(&observed.progress_identity)
                            {
                                journal.stalled_observations =
                                    journal.stalled_observations.saturating_add(1);
                            } else {
                                journal.stalled_observations = 0;
                            }
                            record.maintenance_attempts += 1;
                            let attempts = record.maintenance_attempts;
                            // A failed or lost call consumes its reviewed attempt across process restart.
                            write_journal(&paths, &journal)?;
                            if attempts > 1 {
                                platform
                                    .pace_effect_observation(action, journal.stalled_observations);
                            }
                        } else if maintenance_continuation {
                            return Err(EnsureWorkflowError::JournalIntegrity);
                        }
                        let record = journal
                            .effects
                            .get(index)
                            .ok_or(EnsureWorkflowError::JournalIntegrity)?;
                        if retained_plan.scope == FleetEnsurePlanScope::RootReinstallPrerequisite {
                            root_reinstall::verify_effect_authority(
                                &retained_plan,
                                action,
                                record.pre_cycles,
                                &state,
                                platform,
                            )?;
                        }
                        verify_store_activation_binding(&retained_plan, action, &state, platform)?;
                        reinstall::verify_effect_authority(
                            &retained_plan,
                            action,
                            &state,
                            platform,
                        )?;
                        let outcome =
                            match platform.apply(&journal.operation_id, action, record, &state) {
                                Ok(outcome) => outcome,
                                Err(source) => {
                                    let progress_identity = record
                                        .progress_identity
                                        .clone()
                                        .unwrap_or_else(|| "effect-call-failed".to_string());
                                    journal.stalled_observations =
                                        journal.stalled_observations.saturating_add(1);
                                    write_journal(&paths, &journal)?;
                                    if journal.stalled_observations
                                        >= observation_policy.maximum_stalled_observations
                                    {
                                        return Err(EnsureWorkflowError::Stalled {
                                            action: action.name().to_string(),
                                            observations: journal.stalled_observations,
                                            progress_identity,
                                        });
                                    }
                                    return Err(EnsureWorkflowError::Platform(source));
                                }
                            };
                        if !create_outcome_is_exact(action, record, &state, &outcome) {
                            return Err(EnsureWorkflowError::JournalIntegrity);
                        }
                        if let Some(created) = &outcome.created_principal {
                            state
                                .pending_principals
                                .insert(action.name().to_string(), created.clone());
                            write_state(&paths, &state)?;
                        }
                        let record = journal
                            .effects
                            .get_mut(index)
                            .ok_or(EnsureWorkflowError::JournalIntegrity)?;
                        record.created_principal = outcome.created_principal;
                        record.receipt = outcome.receipt;
                        record.post_cycles = outcome.post_cycles;
                        record.progress_identity = Some(observed.progress_identity);
                        record.state = EffectState::Issued;
                        // A successful maintenance call can report no progress while Root work is in flight.
                        if !maintenance_continuation {
                            journal.stalled_observations = 0;
                        }
                        write_journal(&paths, &journal)?;
                        continue;
                    }

                    if observed.retry == EffectRetry::ReplayExactIssuedCommand
                        && !replayed_issued_commands.contains(&index)
                    {
                        if !matches!(
                            action,
                            EnsureAction::FleetProtocol { action, .. }
                                if matches!(
                                    action.as_ref(),
                                    crate::fleet_ensure::model::CurrentFleetProtocolAction::ProvisionComponents { .. }
                                )
                        ) {
                            return Err(EnsureWorkflowError::JournalIntegrity);
                        }
                        let retained_receipt = record.receipt.clone();
                        let outcome = platform
                            .apply(&journal.operation_id, action, record, &state)
                            .map_err(EnsureWorkflowError::Platform)?;
                        if outcome.created_principal.is_some()
                            || outcome.receipt != retained_receipt
                            || outcome.post_cycles.is_some()
                        {
                            return Err(EnsureWorkflowError::JournalIntegrity);
                        }
                        let record = journal
                            .effects
                            .get_mut(index)
                            .ok_or(EnsureWorkflowError::JournalIntegrity)?;
                        record.progress_identity = Some(observed.progress_identity);
                        journal.stalled_observations = 0;
                        write_journal(&paths, &journal)?;
                        replayed_issued_commands.insert(index);
                        continue;
                    }

                    if record.progress_identity.as_deref() == Some(&observed.progress_identity) {
                        journal.stalled_observations =
                            journal.stalled_observations.saturating_add(1);
                    } else {
                        record.progress_identity = Some(observed.progress_identity);
                        journal.stalled_observations = 0;
                    }
                    let progress_identity = record
                        .progress_identity
                        .clone()
                        .ok_or(EnsureWorkflowError::JournalIntegrity)?;
                    write_journal(&paths, &journal)?;
                    if journal.stalled_observations
                        >= observation_policy.maximum_stalled_observations
                    {
                        if let Some((failure, origin)) = observed
                            .provisioning_failure
                            .and_then(|failure| failure.origin.map(|origin| (failure, origin)))
                        {
                            return Err(EnsureWorkflowError::ProvisioningRetryPending {
                                root: failure.fleet_subnet_root,
                                stage: origin.stage,
                                target: origin.target,
                                operation_id: origin.operation_id,
                                diagnostic_code: origin.diagnostic_code,
                                observations: journal.stalled_observations,
                                failed_at_ns: origin.failed_at_ns,
                            });
                        }
                        return Err(EnsureWorkflowError::Stalled {
                            action: action.name().to_string(),
                            observations: journal.stalled_observations,
                            progress_identity,
                        });
                    }
                    if observation_policy.paced {
                        report_progress_state(
                            platform,
                            &retained_plan,
                            &journal,
                            action_progress_phase(action),
                            FleetEnsureProgressState::AwaitingProgress {
                                elapsed_seconds: observation_started.elapsed().as_secs(),
                                provisioning: observed.provisioning_progress,
                            },
                        );
                        platform.pace_effect_observation(action, journal.stalled_observations);
                    }
                }
            }
            if journal.effects.len() == actions.len()
                && journal
                    .effects
                    .iter()
                    .all(|record| record.state == EffectState::Applied)
            {
                break;
            }
            if !deferred_controller_observation {
                return Err(EnsureWorkflowError::JournalIntegrity);
            }
        }

        report_progress(
            platform,
            &retained_plan,
            &journal,
            FleetEnsurePhase::TerminalVerification,
        );
        if retained_plan.scope == FleetEnsurePlanScope::InfrastructureBootstrap {
            if infrastructure_bootstrap::advance(
                root,
                &retained_plan,
                &mut journal,
                &mut state,
                platform,
            )? {
                continue;
            }
            let phase = journal
                .successor_phases
                .first()
                .and_then(|phase| phase.plan.as_ref())
                .ok_or(EnsureWorkflowError::JournalIntegrity)?;
            let expected =
                crate::fleet_ensure::ops::infrastructure_bootstrap::registration::registry(phase)?;
            if !platform
                .verify_bootstrap_registry(&expected)
                .map_err(EnsureWorkflowError::Platform)?
            {
                return Err(EnsureWorkflowError::ConvergenceDrift);
            }
            let (actual, observed) = infrastructure_bootstrap::complete(
                root,
                &retained_plan,
                &journal,
                &state,
                platform,
            )?;
            if !platform
                .verify_bootstrap_registry(&expected)
                .map_err(EnsureWorkflowError::Platform)?
            {
                return Err(EnsureWorkflowError::ConvergenceDrift);
            }
            publish_terminal_state(operation_desired, &retained_plan, &journal, &mut state)
                .map_err(|_| EnsureWorkflowError::JournalIntegrity)?;
            crate::fleet_ensure::ops::infrastructure_bootstrap::registration::bind_state(
                root, &mut state, phase,
            )?;
            write_state(&paths, &state)?;
            crate::fleet_ensure::ops::infrastructure_bootstrap::terminal::retain(
                &paths,
                &retained_plan,
                &journal,
                &state,
                actual.clone(),
                &observed,
            )?;
            journal.completion = FleetEnsureCompletion::Converged;
            journal.stalled_observations = 0;
            write_journal(&paths, &journal)?;
            report_progress_state(
                platform,
                &retained_plan,
                &journal,
                FleetEnsurePhase::Infrastructure,
                FleetEnsureProgressState::PrerequisiteComplete,
            );
            return Ok(FleetEnsureReport {
                funding_review: None,
                actual_conservation: Some(actual),
                effects_applied: applied_count(&journal),
                plan: retained_plan,
                terminal: true,
            });
        }
        if retained_plan.scope == FleetEnsurePlanScope::ReinstallPreparation {
            let actual = reinstall::complete(&retained_plan, &journal, &state, platform)?;
            journal.completion = FleetEnsureCompletion::Prepared;
            journal.stalled_observations = 0;
            write_journal(&paths, &journal)?;
            report_progress_state(
                platform,
                &retained_plan,
                &journal,
                FleetEnsurePhase::Infrastructure,
                FleetEnsureProgressState::PrerequisiteComplete,
            );
            return Ok(FleetEnsureReport {
                funding_review: None,
                actual_conservation: Some(actual),
                effects_applied: applied_count(&journal),
                plan: retained_plan,
                terminal: true,
            });
        }
        if retained_plan.scope == FleetEnsurePlanScope::RootReinstallPrerequisite {
            let actual = root_reinstall::complete(&retained_plan, &journal, &state, platform)?;
            publish_terminal_state(operation_desired, &retained_plan, &journal, &mut state)
                .map_err(|_| EnsureWorkflowError::JournalIntegrity)?;
            state.active_registry = None;
            write_state(&paths, &state)?;
            journal.completion = FleetEnsureCompletion::Converged;
            journal.stalled_observations = 0;
            write_journal(&paths, &journal)?;
            report_progress_state(
                platform,
                &retained_plan,
                &journal,
                FleetEnsurePhase::Infrastructure,
                FleetEnsureProgressState::PrerequisiteComplete,
            );
            return Ok(FleetEnsureReport {
                funding_review: None,
                actual_conservation: Some(actual),
                effects_applied: applied_count(&journal),
                plan: retained_plan,
                terminal: true,
            });
        }
        if retained_plan.scope == FleetEnsurePlanScope::RootStartPrerequisite {
            let actual_conservation = complete_root_start_prerequisite(
                operation_desired,
                &retained_plan,
                platform,
                &state,
                &journal,
            )?;
            journal.completion = FleetEnsureCompletion::Converged;
            journal.stalled_observations = 0;
            write_journal(&paths, &journal)?;
            report_progress_state(
                platform,
                &retained_plan,
                &journal,
                FleetEnsurePhase::Infrastructure,
                FleetEnsureProgressState::PrerequisiteComplete,
            );
            return Ok(FleetEnsureReport {
                funding_review: None,
                actual_conservation: Some(actual_conservation),
                effects_applied: applied_count(&journal),
                plan: retained_plan,
                terminal: true,
            });
        }

        terminal_state = state.clone();
        publish_terminal_state(
            operation_desired,
            &retained_plan,
            &journal,
            &mut terminal_state,
        )
        .map_err(|TerminalStatePublicationError::JournalIntegrity| {
            EnsureWorkflowError::JournalIntegrity
        })?;
        if completed_infrastructure_reinstall(operation_desired, &retained_plan, &journal) {
            terminal_state.active_registry = None;
        }
        project_current_fleet_inventory(&terminal_state)?;
        {
            // Replanning after a completed phase is one read-only decision.
            // Pacing expires its evidence; appending or issuing work happens only
            // after this scope has closed.
            let (terminal_observation, converged) =
                platform.with_planning_observations(|platform| {
                    let mut terminal_observation = platform
                        .observe(&retained_plan.operation_id, &terminal_state)
                        .map_err(EnsureWorkflowError::Platform)?;
                    let artifacts = resolve_desired_artifacts(root, operation_desired)?;
                    let observation_started = Instant::now();
                    let converged = loop {
                        let protocol_actions = platform
                            .protocol_actions(protocol_operation(&retained_plan), &terminal_state)
                            .map_err(EnsureWorkflowError::Platform)?;
                        match compile_plan(
                            operation_desired,
                            &artifacts,
                            &protocol_actions,
                            operation_desired_sha256,
                            requested_fleet,
                            &terminal_observation,
                            retained_plan.planned_at_time,
                            &retained_plan.operation_id,
                            None,
                        ) {
                            Ok(plan) => {
                                if journal.stalled_observations != 0 {
                                    journal.stalled_observations = 0;
                                    write_journal(&paths, &journal)?;
                                }
                                break plan;
                            }
                            Err(EnsurePolicyError::PendingRootOwnedBalance { name }) => {
                                journal.stalled_observations =
                                    journal.stalled_observations.saturating_add(1);
                                retain_observed_cycles(&mut state, &terminal_observation);
                                write_state(&paths, &state)?;
                                write_journal(&paths, &journal)?;
                                if journal.stalled_observations
                                    >= operation_desired.maximum_stalled_observations
                                {
                                    return Err(EnsureWorkflowError::RootOwnedObservationStalled {
                                        last_lifecycle: root_owned_lifecycle_label(
                                            &terminal_observation,
                                            &name,
                                        )
                                        .to_string(),
                                        observations: journal.stalled_observations,
                                        target: name,
                                    });
                                }
                                report_progress_state(
                                    platform,
                                    &retained_plan,
                                    &journal,
                                    FleetEnsurePhase::TerminalVerification,
                                    FleetEnsureProgressState::AwaitingProgress {
                                        elapsed_seconds: observation_started.elapsed().as_secs(),
                                        provisioning: None,
                                    },
                                );
                                platform.pace_root_owned_observation(
                                    &name,
                                    journal.stalled_observations,
                                );
                                terminal_observation = platform
                                    .observe(&retained_plan.operation_id, &terminal_state)
                                    .map_err(EnsureWorkflowError::Platform)?;
                            }
                            Err(error @ EnsurePolicyError::EstatePoolCapacity { .. }) => {
                                retain_observed_cycles(&mut terminal_state, &terminal_observation);
                                retain_completed_reinstalls(
                                    &mut terminal_state,
                                    &retained_plan,
                                    &journal,
                                );
                                write_state(&paths, &terminal_state)?;
                                journal.completion = FleetEnsureCompletion::ReplanRequired;
                                journal.stalled_observations = 0;
                                write_journal(&paths, &journal)?;
                                return Err(error.into());
                            }
                            Err(error) => return Err(error.into()),
                        }
                    };
                    Ok::<_, EnsureWorkflowError<P::Error>>((terminal_observation, converged))
                })?;
            if converged
                .canisters
                .iter()
                .any(|canister| !canister.actions.is_empty())
                || !converged.protocol_actions.is_empty()
            {
                if retained_plan.continuation.is_some() {
                    let appended = continuation::append(
                        &paths,
                        operation_desired,
                        &retained_plan,
                        &mut journal,
                        &terminal_state,
                        &terminal_observation,
                        converged,
                        platform,
                    );
                    match appended {
                        Ok(()) => {
                            write_state(&paths, &terminal_state)?;
                            state = terminal_state.clone();
                            continuation_observation = journal
                                .successor_phases
                                .last()
                                .and_then(|phase| phase.plan.as_ref())
                                .and_then(|phase| phase.protocol_actions.first())
                                .map(|first| ContinuationObservation {
                                    first_action_sha256: action_sha256(first),
                                    observation: terminal_observation,
                                });
                            report_progress_state(
                                platform,
                                &retained_plan,
                                &journal,
                                FleetEnsurePhase::TerminalVerification,
                                FleetEnsureProgressState::PrerequisiteComplete,
                            );
                            continue;
                        }
                        Err(EnsureWorkflowError::SuccessorReviewRequired { reason, review }) => {
                            retain_observed_cycles(&mut terminal_state, &terminal_observation);
                            retain_completed_reinstalls(
                                &mut terminal_state,
                                &retained_plan,
                                &journal,
                            );
                            write_state(&paths, &terminal_state)?;
                            journal.completion = FleetEnsureCompletion::ReplanRequired;
                            write_journal(&paths, &journal)?;
                            report_progress_state(
                                platform,
                                &retained_plan,
                                &journal,
                                FleetEnsurePhase::TerminalVerification,
                                FleetEnsureProgressState::ReviewRequired {
                                    reason,
                                    review: review.clone(),
                                },
                            );
                            return Err(EnsureWorkflowError::SuccessorReviewRequired {
                                reason,
                                review,
                            });
                        }
                        Err(error) => return Err(error),
                    }
                }
                retain_observed_cycles(&mut terminal_state, &terminal_observation);
                retain_completed_reinstalls(&mut terminal_state, &retained_plan, &journal);
                write_state(&paths, &terminal_state)?;
                journal.completion = FleetEnsureCompletion::ReplanRequired;
                journal.stalled_observations = 0;
                write_journal(&paths, &journal)?;
                let reason =
                    crate::fleet_ensure::model::FleetEnsureSuccessorReviewReason::AdditionalEffect;
                let review = Some(Box::new(
                    crate::fleet_ensure::ops::recovery::review_details(&converged),
                ));
                report_progress_state(
                    platform,
                    &retained_plan,
                    &journal,
                    FleetEnsurePhase::TerminalVerification,
                    FleetEnsureProgressState::ReviewRequired {
                        reason,
                        review: review.clone(),
                    },
                );
                return Err(EnsureWorkflowError::SuccessorReviewRequired { reason, review });
            }
        }
        break;
    }
    let terminal_inventory_operation_id =
        completed_inventory_operation(&retained_plan, &terminal_state)?;
    let terminal_inventory = platform
        .terminal_inventory(terminal_inventory_operation_id, &terminal_state)
        .map_err(EnsureWorkflowError::Platform)?;
    let terminal_cycles = terminal_inventory.controlled_cycles_by_principal.clone();
    merge_terminal_inventory(&mut terminal_state, terminal_inventory)?;
    project_current_fleet_inventory(&terminal_state)?;
    let mut final_observation = platform
        .observe(&retained_plan.operation_id, &terminal_state)
        .map_err(EnsureWorkflowError::Platform)?;
    attach_terminal_cycles(&mut final_observation, terminal_cycles)?;
    reinstall::verify_terminal_estate(&retained_plan, &final_observation)?;
    reinstall::verify_terminal_authority(&retained_plan, &terminal_state, platform)?;
    let actual_conservation = verify_terminal_conservation(
        &retained_plan,
        &journal,
        &terminal_state,
        &final_observation,
    )?;
    terminal_state.completed_reinstall_action_sha256.clear();
    terminal_state.completed_reinstall_operation_id = None;
    terminal_state.completed_reinstalls.clear();
    // Publish the fully validated topology before making its terminal journal visible to backup.
    write_state(&paths, &terminal_state)?;
    journal.completion = FleetEnsureCompletion::Converged;
    journal.stalled_observations = 0;
    crate::fleet_ensure::ops::terminal::retain(
        &paths,
        &retained_plan,
        &journal,
        &terminal_state,
        &actual_conservation,
        &funding_plan::<P::Error>(&retained_plan, &journal)?.conservation,
    )?;
    write_journal(&paths, &journal)?;
    report_progress(
        platform,
        &retained_plan,
        &journal,
        FleetEnsurePhase::Complete,
    );
    Ok(FleetEnsureReport {
        funding_review: None,
        actual_conservation: Some(actual_conservation),
        effects_applied: applied_count(&journal),
        plan: retained_plan,
        terminal: true,
    })
}

fn action_progress_phase(action: &EnsureAction) -> FleetEnsurePhase {
    match action {
        EnsureAction::FleetProtocol { action, .. } => match action.as_ref() {
            CurrentFleetProtocolAction::ReconcilePoolAsset { .. } => {
                FleetEnsurePhase::ImportReconciliation
            }
            CurrentFleetProtocolAction::ProvisionComponents { .. } => {
                FleetEnsurePhase::WorkloadProvisioning
            }
            CurrentFleetProtocolAction::MaintainPoolReadiness { .. }
            | CurrentFleetProtocolAction::ObservePoolReadiness { .. } => {
                FleetEnsurePhase::PoolReadiness
            }
            _ => FleetEnsurePhase::ControlPlane,
        },
        _ => FleetEnsurePhase::Infrastructure,
    }
}

fn report_progress<P: EnsurePlatform>(
    platform: &mut P,
    plan: &FleetEnsurePlan,
    journal: &FleetEnsureJournalRecord,
    phase: FleetEnsurePhase,
) {
    let state = if phase == FleetEnsurePhase::Complete {
        FleetEnsureProgressState::Complete
    } else {
        FleetEnsureProgressState::Advancing
    };
    report_progress_state(platform, plan, journal, phase, state);
}

fn report_progress_state<P: EnsurePlatform>(
    platform: &mut P,
    plan: &FleetEnsurePlan,
    journal: &FleetEnsureJournalRecord,
    phase: FleetEnsurePhase,
    state: FleetEnsureProgressState,
) {
    platform.report_progress(FleetEnsureProgress {
        next_action: if matches!(
            state,
            FleetEnsureProgressState::Advancing
                | FleetEnsureProgressState::AwaitingProgress { .. }
                | FleetEnsureProgressState::FundingRequired
        ) && phase != FleetEnsurePhase::TerminalVerification
        {
            crate::fleet_ensure::ops::progress::next_action(
                &continuation::actions(plan, journal),
                journal,
            )
        } else {
            None
        },
        state,
        operation_id: plan.operation_id.clone(),
        plan_sha256: plan.plan_sha256.clone(),
        phase,
        applied_effects: applied_count(journal),
        reviewed_effects: continuation::actions(plan, journal).len()
            + journal
                .funding_reviews
                .iter()
                .filter(|review| review.effect.is_some())
                .count(),
    });
}

const fn merge_observed_cycles(
    record: &mut EffectRecord,
    source_cycles: Option<u128>,
    destination_cycles: Option<u128>,
) {
    if let Some(cycles) = source_cycles {
        record.post_cycles = Some(cycles);
    }
    if let Some(cycles) = destination_cycles {
        record.destination_post_cycles = Some(cycles);
    }
}

pub(super) fn retain_applied_create_authority<E>(
    desired: &crate::fleet_ensure::model::DesiredFleet,
    action: &EnsureAction,
    record: &EffectRecord,
    state: &mut FleetEnsureStateRecord,
) -> Result<bool, EnsureWorkflowError<E>>
where
    E: std::error::Error + 'static,
{
    let EnsureAction::Create {
        name,
        requested_initial_cycles,
        ..
    } = action
    else {
        return Ok(false);
    };
    if record.state != EffectState::Applied {
        return Err(EnsureWorkflowError::JournalIntegrity);
    }
    let configured = desired
        .canisters
        .iter()
        .find(|canister| {
            canister.name == *name
                && canister.presence == crate::fleet_ensure::model::DesiredPresence::Present
        })
        .ok_or(EnsureWorkflowError::JournalIntegrity)?;
    let principal = record
        .created_principal
        .as_deref()
        .ok_or(EnsureWorkflowError::JournalIntegrity)?;
    let cycles = record
        .post_cycles
        .ok_or(EnsureWorkflowError::JournalIntegrity)?;
    let maximum_observation_burn_cycles =
        crate::fleet_ensure::ops::maximum_creation_observation_burn(desired)
            .ok_or(EnsureWorkflowError::JournalIntegrity)?;
    if !create_identity_is_exact(name, record, state)
        || !create_balance_is_terminal(
            Some(cycles),
            *requested_initial_cycles,
            maximum_observation_burn_cycles,
        )
        || state
            .retained_cycles_by_principal
            .get(principal)
            .is_some_and(|retained| *retained != cycles)
    {
        return Err(EnsureWorkflowError::JournalIntegrity);
    }
    validate_fresh_pool_creation_balance(
        desired,
        configured,
        name,
        cycles,
        maximum_observation_burn_cycles,
        *requested_initial_cycles,
    )?;

    let prior_topology = state.topology.get(name);
    let topology_identity_matches = prior_topology.is_some_and(|topology| {
        topology.kind == configured.kind
            && topology.parent == configured.parent
            && topology.protocol_binding == configured.protocol_binding
            && topology.role
                == configured
                    .protocol_binding
                    .as_ref()
                    .map(|binding| binding.role.to_string())
    });
    let topology = crate::fleet_ensure::model::FleetEnsureTopologyRecord {
        kind: configured.kind,
        module_hash: if topology_identity_matches {
            prior_topology.and_then(|topology| topology.module_hash.clone())
        } else {
            None
        },
        parent: configured.parent.clone(),
        protocol_binding: configured.protocol_binding.clone(),
        role: configured
            .protocol_binding
            .as_ref()
            .map(|binding| binding.role.to_string()),
    };
    let changed = state.retained_cycles_by_principal.get(principal) != Some(&cycles)
        || state.topology.get(name) != Some(&topology);
    state
        .retained_cycles_by_principal
        .insert(principal.to_string(), cycles);
    state.topology.insert(name.clone(), topology);
    Ok(changed)
}

fn validate_fresh_pool_creation_balance<E>(
    desired: &crate::fleet_ensure::model::DesiredFleet,
    configured: &crate::fleet_ensure::model::DesiredCanister,
    name: &str,
    cycles: u128,
    maximum_observation_burn_cycles: u128,
    requested_initial_cycles: u128,
) -> Result<(), EnsureWorkflowError<E>>
where
    E: std::error::Error + 'static,
{
    if configured.kind == crate::fleet_ensure::model::DesiredCanisterKind::Pool
        && desired
            .bootstrap
            .as_ref()
            .is_some_and(|bootstrap| bootstrap.fresh_estate)
        && configured.principal.is_none()
        && configured.controllers.is_empty()
        && configured.parent.as_ref().is_some_and(|parent| {
            configured.controller_canisters.as_slice() == std::slice::from_ref(parent)
        })
    {
        let readiness_floor_cycles = configured
            .minimum_cycles
            .parse::<Cycles>()
            .map(|cycles| cycles.to_u128())
            .map_err(|_| EnsureWorkflowError::JournalIntegrity)?;
        let remaining_controller_burn_cycles = desired
            .maximum_update_burn_cycles
            .parse::<Cycles>()
            .map(|cycles| cycles.to_u128())
            .map_err(|_| EnsureWorkflowError::JournalIntegrity)?;
        let required_pre_finalization_balance_cycles = readiness_floor_cycles
            .checked_add(remaining_controller_burn_cycles)
            .ok_or(EnsureWorkflowError::JournalIntegrity)?;
        if cycles < required_pre_finalization_balance_cycles {
            return Err(EnsureWorkflowError::FreshPoolCreationUnderfunded(Box::new(
                FreshPoolCreationUnderfundedError {
                    canister: name.to_string(),
                    live_balance_cycles: cycles,
                    maximum_observation_burn_cycles,
                    pre_finalization_shortfall_cycles: required_pre_finalization_balance_cycles
                        - cycles,
                    readiness_floor_cycles,
                    readiness_shortfall_cycles: readiness_floor_cycles.saturating_sub(cycles),
                    remaining_controller_burn_cycles,
                    requested_creation_funding_cycles: requested_initial_cycles,
                    required_pre_finalization_balance_cycles,
                },
            )));
        }
    }
    Ok(())
}

fn retain_applied_funding_cycles(
    state: &mut FleetEnsureStateRecord,
    action: &EnsureAction,
    record: &EffectRecord,
) -> bool {
    let EnsureAction::Fund { principal, .. } = action else {
        return false;
    };
    if record.state != EffectState::Applied || record.receipt.is_none() {
        return false;
    }
    let Some(cycles) = record.post_cycles else {
        return false;
    };
    let principal = principal.strip_prefix("created:").map_or_else(
        || Some(principal.as_str()),
        |name| state.pending_principals.get(name).map(String::as_str),
    );
    let Some(principal) = principal else {
        return false;
    };
    if state.retained_cycles_by_principal.get(principal) == Some(&cycles) {
        return false;
    }
    state
        .retained_cycles_by_principal
        .insert(principal.to_string(), cycles);
    true
}

fn retain_created_canister_for_replan<E>(
    desired: &crate::fleet_ensure::model::DesiredFleet,
    name: &str,
    record: &EffectRecord,
    state: &mut FleetEnsureStateRecord,
) -> Result<(), EnsureWorkflowError<E>>
where
    E: std::error::Error + 'static,
{
    let configured = desired
        .canisters
        .iter()
        .find(|canister| {
            canister.name == name
                && canister.presence == crate::fleet_ensure::model::DesiredPresence::Present
        })
        .ok_or(EnsureWorkflowError::JournalIntegrity)?;
    let principal = record
        .created_principal
        .as_deref()
        .ok_or(EnsureWorkflowError::JournalIntegrity)?;
    let cycles = record
        .post_cycles
        .ok_or(EnsureWorkflowError::JournalIntegrity)?;
    if state.pending_principals.get(name).map(String::as_str) != Some(principal) {
        return Err(EnsureWorkflowError::JournalIntegrity);
    }
    state.pending_principals.remove(name);
    state
        .principals
        .insert(name.to_string(), principal.to_string());
    state
        .retained_cycles_by_principal
        .insert(principal.to_string(), cycles);
    state.topology.insert(
        name.to_string(),
        crate::fleet_ensure::model::FleetEnsureTopologyRecord {
            kind: configured.kind,
            module_hash: None,
            parent: configured.parent.clone(),
            protocol_binding: configured.protocol_binding.clone(),
            role: configured
                .protocol_binding
                .as_ref()
                .map(|binding| binding.role.to_string()),
        },
    );
    Ok(())
}

fn create_record_is_exact(
    action: &EnsureAction,
    record: &EffectRecord,
    state: &FleetEnsureStateRecord,
) -> bool {
    let EnsureAction::Create {
        name,
        requested_initial_cycles,
        ..
    } = action
    else {
        return true;
    };
    match record.state {
        EffectState::Intent => {
            record.created_principal.is_none()
                && record.receipt.is_none()
                && record.post_cycles.is_none()
        }
        EffectState::Issued => {
            create_identity_is_exact(name, record, state)
                && record.post_cycles == Some(*requested_initial_cycles)
        }
        EffectState::Applied => create_identity_is_exact(name, record, state),
    }
}

fn create_identity_is_exact(
    name: &str,
    record: &EffectRecord,
    state: &FleetEnsureStateRecord,
) -> bool {
    let Some(created_principal) = record
        .created_principal
        .as_deref()
        .filter(|value| !value.is_empty())
    else {
        return false;
    };
    if record.receipt.as_deref().is_none_or(str::is_empty) {
        return false;
    }
    state_create_principal_matches(name, created_principal, state, true)
}

fn state_create_principal_matches(
    name: &str,
    created_principal: &str,
    state: &FleetEnsureStateRecord,
    require_retained: bool,
) -> bool {
    let pending = state.pending_principals.get(name).map(String::as_str);
    let terminal = state.principals.get(name).map(String::as_str);
    let retained = pending.is_some() || terminal.is_some();
    let pending_matches = pending.is_none_or(|principal| principal == created_principal);
    let terminal_matches = terminal.is_none_or(|principal| principal == created_principal);
    (!require_retained || retained) && pending_matches && terminal_matches
}

fn create_outcome_is_exact(
    action: &EnsureAction,
    record: &EffectRecord,
    state: &FleetEnsureStateRecord,
    outcome: &crate::fleet_ensure::ops::EffectOutcome,
) -> bool {
    if matches!(action, EnsureAction::FundEstate { .. }) {
        return outcome.created_principal.is_none()
            && outcome.post_cycles.is_none()
            && outcome
                .receipt
                .as_deref()
                .is_some_and(|receipt| !receipt.is_empty());
    }
    let EnsureAction::Create {
        name,
        requested_initial_cycles,
        ..
    } = action
    else {
        return true;
    };
    let Some(created_principal) = outcome
        .created_principal
        .as_deref()
        .filter(|value| !value.is_empty())
    else {
        return false;
    };
    let Some(receipt) = outcome.receipt.as_deref().filter(|value| !value.is_empty()) else {
        return false;
    };
    if outcome.post_cycles != Some(*requested_initial_cycles) {
        return false;
    }
    let principal_matches = record
        .created_principal
        .as_deref()
        .is_none_or(|retained| retained == created_principal);
    let receipt_matches = record
        .receipt
        .as_deref()
        .is_none_or(|retained| retained == receipt);
    let state_matches = state_create_principal_matches(name, created_principal, state, false);
    principal_matches && receipt_matches && state_matches
}

fn verify_fresh_plan<P>(
    root: &Path,
    desired: &crate::fleet_ensure::model::DesiredFleet,
    desired_sha256: &str,
    requested_fleet: &str,
    retained_plan: &FleetEnsurePlan,
    platform: &mut P,
    state: &FleetEnsureStateRecord,
) -> Result<(FleetObservation, u128), EnsureWorkflowError<P::Error>>
where
    P: EnsurePlatform,
{
    if retained_plan.scope == FleetEnsurePlanScope::InfrastructureBootstrap {
        return infrastructure_bootstrap::verify_before_apply(root, retained_plan, state, platform);
    }
    if retained_plan.reinstall.is_some() {
        return reinstall::verify_before_apply(root, desired, retained_plan, platform, state);
    }
    if retained_plan.scope == FleetEnsurePlanScope::RootReinstallPrerequisite {
        return root_reinstall::verify_before_apply(root, desired, retained_plan, platform, state);
    }
    if retained_plan.scope == FleetEnsurePlanScope::RootStartPrerequisite {
        let targets = reviewed_root_start_targets(retained_plan)?;
        let management = platform
            .observe_root_management(state, &targets)
            .map_err(EnsureWorkflowError::Platform)?
            .ok_or(EnsureWorkflowError::PlanIntegrity)?;
        let current = recompile_root_start_prerequisite_plan(
            RootStartPlanInput {
                state,
                authority: retained_plan.root_start_authority.as_deref(),
                created_at_time: retained_plan.planned_at_time,
                desired,
                desired_sha256,
                observation: &management,
                requested_fleet,
            },
            &targets,
        )?;
        if !compatible_root_start_prerequisite(retained_plan, &current, desired) {
            return Err(EnsureWorkflowError::DriftedBeforeApply);
        }
        let observation = root_management_fleet_observation(&management, &targets)?;
        return Ok((observation, current.conservation.observed_controlled_cycles));
    }
    let mut observation = platform
        .observe(&retained_plan.operation_id, state)
        .map_err(EnsureWorkflowError::Platform)?;
    let terminal_inventory_operation_id =
        fresh_plan_inventory_operation(root, retained_plan, state)?;
    if let Some(operation_id) = terminal_inventory_operation_id {
        attach_terminal_inventory_cycles(operation_id, state, platform, &mut observation)?;
    }
    let artifacts = resolve_desired_artifacts(root, desired)?;
    let mut retained_observations = 0_u32;
    let mut current = loop {
        let protocol_actions = platform
            .protocol_actions(protocol_operation(retained_plan), state)
            .map_err(EnsureWorkflowError::Platform)?;
        match compile_plan(
            desired,
            &artifacts,
            &protocol_actions,
            desired_sha256,
            requested_fleet,
            &observation,
            retained_plan.planned_at_time,
            &retained_plan.operation_id,
            None,
        ) {
            Ok(plan) => break plan,
            Err(EnsurePolicyError::PendingRootOwnedBalance { name }) => {
                retained_observations = retained_observations.saturating_add(1);
                if retained_observations >= desired.maximum_stalled_observations {
                    return Err(EnsureWorkflowError::RootOwnedObservationStalled {
                        last_lifecycle: root_owned_lifecycle_label(&observation, &name).to_string(),
                        observations: retained_observations,
                        target: name,
                    });
                }
                platform.pace_root_owned_observation(&name, retained_observations);
                observation = platform
                    .observe(&retained_plan.operation_id, state)
                    .map_err(EnsureWorkflowError::Platform)?;
                if let Some(operation_id) = terminal_inventory_operation_id {
                    attach_terminal_inventory_cycles(
                        operation_id,
                        state,
                        platform,
                        &mut observation,
                    )?;
                }
            }
            Err(error) => return Err(error.into()),
        }
    };
    bind_terminal_inventory_operation(
        &mut current,
        retained_plan.terminal_inventory_operation_id.clone(),
    );
    if current.plan_sha256 != retained_plan.plan_sha256
        && !compatible_after_bounded_observation(retained_plan, &current, desired, &observation)
    {
        return Err(EnsureWorkflowError::DriftedBeforeApply);
    }
    Ok((observation, current.conservation.observed_controlled_cycles))
}

fn estate_funding_requirement<E>(
    plan: &FleetEnsurePlan,
    state: &FleetEnsureStateRecord,
    observation: &FleetObservation,
) -> Result<Option<EstateFundingRequiredRecord>, EnsureWorkflowError<E>>
where
    E: std::error::Error + 'static,
{
    for reviewed in &plan.conservation.estate_funding_domains {
        let observed = observation
            .estate_funding_domains
            .get(&reviewed.root)
            .ok_or(EnsureWorkflowError::PlanIntegrity)?;
        if observed.cycles_ledger != reviewed.cycles_ledger {
            return Err(EnsureWorkflowError::PlanIntegrity);
        }
        let root_principal = observed
            .root_principal
            .as_deref()
            .ok_or(EnsureWorkflowError::PlanIntegrity)?;
        if reviewed_estate_root_principal(reviewed, state) != Some(root_principal) {
            return Err(EnsureWorkflowError::PlanIntegrity);
        }
        let available_cycles = observed
            .balance_cycles
            .ok_or(EnsureWorkflowError::PlanIntegrity)?;
        let reviewed_available = reviewed.available_cycles.unwrap_or_default();
        // A surplus stays controlled and grants no additional spending authority.
        // An unexplained Ledger debit still requires exact reconciliation.
        if available_cycles < reviewed_available {
            return Err(EnsureWorkflowError::DriftedBeforeApply);
        }
        if reviewed.required_creation_count == 0 {
            continue;
        }
        if available_cycles < reviewed.maximum_creation_debit_cycles {
            return Ok(Some(EstateFundingRequiredRecord {
                attempt_count: None,
                available_cycles,
                creation_amount_cycles: reviewed.creation_amount_cycles,
                creation_execution_margin_cycles: reviewed.creation_execution_margin_cycles,
                readiness_floor_cycles: reviewed.readiness_floor_cycles,
                cycles_ledger: reviewed.cycles_ledger.clone(),
                ledger_fee_cycles: reviewed.ledger_fee_cycles,
                management_creation_fee_cycles: reviewed.management_creation_fee_cycles,
                maximum_creation_debit_cycles: reviewed.maximum_creation_debit_cycles,
                maximum_creation_fee_cycles: reviewed.maximum_creation_fee_cycles,
                operation_id: plan.operation_id.clone(),
                pending_creation_operation_id: None,
                plan_sha256: plan.plan_sha256.clone(),
                required_creation_count: reviewed.required_creation_count,
                root: reviewed.root.clone(),
                root_principal: root_principal.to_string(),
                last_attempt_at_ns: None,
                retry_at_ns: None,
                shortfall_cycles: reviewed
                    .maximum_creation_debit_cycles
                    .saturating_sub(available_cycles),
            }));
        }
    }
    Ok(None)
}

fn protocol_estate_funding_requirement<E>(
    plan: &FleetEnsurePlan,
    state: &FleetEnsureStateRecord,
    funding: &canic_contracts::dto::component_provisioning::RootEstateFundingRequired,
) -> Result<EstateFundingRequiredRecord, EnsureWorkflowError<E>>
where
    E: std::error::Error + 'static,
{
    let root_principal = funding.root.to_text();
    let reviewed = plan
        .conservation
        .estate_funding_domains
        .iter()
        .find(|domain| reviewed_estate_root_principal(domain, state) == Some(&root_principal))
        .ok_or(EnsureWorkflowError::PlanIntegrity)?;
    let creation_amount_cycles = funding.creation_amount.to_u128();
    let ledger_fee_cycles = funding.ledger_fee.to_u128();
    let management_creation_fee_cycles = funding.management_creation_fee.to_u128();
    let required_cycles = creation_amount_cycles
        .checked_add(ledger_fee_cycles)
        .ok_or(EnsureWorkflowError::PlanIntegrity)?;
    let maximum_creation_fee_cycles = management_creation_fee_cycles
        .checked_add(ledger_fee_cycles)
        .ok_or(EnsureWorkflowError::PlanIntegrity)?;
    let exact_reviewed_authority = reviewed.required_creation_count > 0
        && reviewed.cycles_ledger == funding.cycles_ledger.to_text()
        && reviewed.creation_amount_cycles == creation_amount_cycles
        && reviewed.creation_execution_margin_cycles == funding.execution_margin.to_u128()
        && reviewed.readiness_floor_cycles == funding.readiness_floor.to_u128()
        && reviewed.ledger_fee_cycles == ledger_fee_cycles
        && reviewed.management_creation_fee_cycles == management_creation_fee_cycles;
    let exact_runtime_arithmetic = funding.required.to_u128() == required_cycles
        && funding.available < funding.required
        && funding.shortfall.to_u128()
            == required_cycles.saturating_sub(funding.available.to_u128());
    if !exact_reviewed_authority || !exact_runtime_arithmetic {
        return Err(EnsureWorkflowError::PlanIntegrity);
    }
    Ok(EstateFundingRequiredRecord {
        attempt_count: Some(funding.attempt_count),
        available_cycles: funding.available.to_u128(),
        creation_amount_cycles,
        cycles_ledger: funding.cycles_ledger.to_text(),
        creation_execution_margin_cycles: funding.execution_margin.to_u128(),
        readiness_floor_cycles: funding.readiness_floor.to_u128(),
        ledger_fee_cycles,
        management_creation_fee_cycles,
        maximum_creation_debit_cycles: required_cycles,
        maximum_creation_fee_cycles,
        operation_id: plan.operation_id.clone(),
        pending_creation_operation_id: Some(canic_core::cdk::utils::hash::hex_bytes(
            funding.operation_id,
        )),
        plan_sha256: plan.plan_sha256.clone(),
        required_creation_count: 1,
        root: reviewed.root.clone(),
        root_principal,
        last_attempt_at_ns: funding.last_attempt_at_ns,
        retry_at_ns: Some(funding.retry_at_ns),
        shortfall_cycles: funding.shortfall.to_u128(),
    })
}

fn estate_funding_error(required: &EstateFundingRequiredRecord) -> EstateFundingRequiredError {
    EstateFundingRequiredError {
        attempt_count: required.attempt_count,
        available_cycles: required.available_cycles,
        creation_amount_cycles: required.creation_amount_cycles,
        creation_execution_margin_cycles: required.creation_execution_margin_cycles,
        readiness_floor_cycles: required.readiness_floor_cycles,
        cycles_ledger: required.cycles_ledger.clone(),
        ledger_fee_cycles: required.ledger_fee_cycles,
        management_creation_fee_cycles: required.management_creation_fee_cycles,
        maximum_creation_debit_cycles: required.maximum_creation_debit_cycles,
        maximum_creation_fee_cycles: required.maximum_creation_fee_cycles,
        pending_creation_operation_id: required.pending_creation_operation_id.clone(),
        required_creation_count: required.required_creation_count,
        root: required.root.clone(),
        root_principal: required.root_principal.clone(),
        last_attempt_at_ns: required.last_attempt_at_ns,
        retry_at_ns: required.retry_at_ns,
        shortfall_cycles: required.shortfall_cycles,
    }
}

fn retain_estate_funding_pause<E>(
    paths: &EnsurePaths,
    journal: &mut FleetEnsureJournalRecord,
    required: Option<&EstateFundingRequiredRecord>,
) -> Result<bool, EnsureWorkflowError<E>>
where
    E: std::error::Error + 'static,
{
    if journal.estate_funding_required.as_ref() == required {
        return Ok(false);
    }
    journal.estate_funding_required = required.cloned();
    write_journal(paths, journal)?;
    Ok(true)
}

fn reviewed_estate_root_principal<'a>(
    domain: &'a crate::fleet_ensure::model::EstateFundingDomainPlan,
    state: &'a FleetEnsureStateRecord,
) -> Option<&'a str> {
    domain
        .root_principal
        .as_deref()
        .or_else(|| {
            state
                .pending_principals
                .get(&domain.root)
                .map(String::as_str)
        })
        .or_else(|| state.principals.get(&domain.root).map(String::as_str))
}

/// Evidence captured after all preceding effects, bound to one successor's first action.
struct ContinuationObservation {
    first_action_sha256: String,
    observation: FleetObservation,
}

fn prior_fleet_protocol_effect_started(
    actions: &[&EnsureAction],
    journal: &FleetEnsureJournalRecord,
    current_index: usize,
) -> bool {
    actions
        .iter()
        .take(current_index)
        .zip(&journal.effects)
        .any(|(action, effect)| {
            matches!(action, EnsureAction::FleetProtocol { .. })
                && matches!(effect.state, EffectState::Issued | EffectState::Applied)
        })
}

// A later Host operation observes the completed Component owner's exact receipts.
// Changing Host retry settings must not allocate a new infrastructure protocol identity.
fn protocol_operation(plan: &FleetEnsurePlan) -> &str {
    plan.terminal_inventory_operation_id
        .as_deref()
        .unwrap_or(&plan.operation_id)
}

fn completed_inventory_operation<'a, E>(
    plan: &'a FleetEnsurePlan,
    state: &FleetEnsureStateRecord,
) -> Result<&'a str, EnsureWorkflowError<E>>
where
    E: std::error::Error + 'static,
{
    if plan.terminal_inventory_operation_id.is_some() {
        reviewed_terminal_inventory_operation(plan, state)?
            .ok_or(EnsureWorkflowError::PlanIntegrity)
    } else {
        Ok(&plan.operation_id)
    }
}

fn fresh_plan_inventory_operation<'a, E>(
    root: &Path,
    plan: &'a FleetEnsurePlan,
    state: &FleetEnsureStateRecord,
) -> Result<Option<&'a str>, EnsureWorkflowError<E>>
where
    E: std::error::Error + 'static,
{
    if state.active_registry.is_some() && plan.terminal_inventory_operation_id.is_none() {
        crate::fleet_ensure::ops::infrastructure_bootstrap::convergence::verify_origin(
            &EnsurePaths::under(root, &plan.environment, &plan.fleet),
            plan,
            state,
        )?;
        Ok(None)
    } else {
        reviewed_terminal_inventory_operation(plan, state)
    }
}

fn reviewed_terminal_inventory_operation<'a, E>(
    plan: &'a FleetEnsurePlan,
    state: &FleetEnsureStateRecord,
) -> Result<Option<&'a str>, EnsureWorkflowError<E>>
where
    E: std::error::Error + 'static,
{
    match (
        state.active_registry.is_some(),
        plan.terminal_inventory_operation_id.as_deref(),
    ) {
        (false, None) => Ok(None),
        (true, Some(operation_id)) if !operation_id.is_empty() => Ok(Some(operation_id)),
        _ => Err(EnsureWorkflowError::PlanIntegrity),
    }
}

fn complete_root_start_prerequisite<P>(
    desired: &crate::fleet_ensure::model::DesiredFleet,
    retained_plan: &FleetEnsurePlan,
    platform: &mut P,
    state: &FleetEnsureStateRecord,
    journal: &FleetEnsureJournalRecord,
) -> Result<ActualCycleConservation, EnsureWorkflowError<P::Error>>
where
    P: EnsurePlatform,
{
    let targets = reviewed_root_start_targets(retained_plan)?;
    let management = platform
        .observe_root_management(state, &targets)
        .map_err(EnsureWorkflowError::Platform)?
        .ok_or(EnsureWorkflowError::PlanIntegrity)?;
    let current = recompile_root_start_prerequisite_plan(
        RootStartPlanInput {
            state,
            authority: retained_plan.root_start_authority.as_deref(),
            created_at_time: retained_plan.planned_at_time,
            desired,
            desired_sha256: &retained_plan.desired_sha256,
            observation: &management,
            requested_fleet: &retained_plan.fleet,
        },
        &targets,
    )?;
    if !compatible_root_start_prerequisite(retained_plan, &current, desired) {
        return Err(EnsureWorkflowError::ConvergenceDrift);
    }
    if management.roots.iter().any(|(name, observed)| {
        targets.contains(name)
            && observed.live.status != crate::fleet_ensure::model::CanisterRuntimeStatus::Running
    }) {
        return Err(EnsureWorkflowError::ConvergenceDrift);
    }
    let terminal = root_management_fleet_observation(&management, &targets)?;
    verify_terminal_conservation(retained_plan, journal, state, &terminal)
}

fn verify_terminal_root_start_replay<P>(
    desired: &crate::fleet_ensure::model::DesiredFleet,
    plan: &FleetEnsurePlan,
    journal: &FleetEnsureJournalRecord,
    requested_fleet: &str,
    platform: &mut P,
    state: &FleetEnsureStateRecord,
) -> Result<(), EnsureWorkflowError<P::Error>>
where
    P: EnsurePlatform,
{
    verify_journal(journal, plan, requested_fleet, state)?;
    let actions = ordered_actions(plan);
    if actions.len() != journal.effects.len()
        || journal
            .effects
            .iter()
            .any(|effect| effect.state != EffectState::Applied)
    {
        return Err(EnsureWorkflowError::JournalIntegrity);
    }
    for (action, record) in actions.into_iter().zip(&journal.effects) {
        if action_sha256(action) != record.action_sha256
            || !platform
                .observe_effect(&journal.operation_id, action, record, state)
                .map_err(EnsureWorkflowError::Platform)?
                .applied
        {
            return Err(EnsureWorkflowError::ConvergenceDrift);
        }
    }
    let targets = reviewed_root_start_targets(plan)?;
    let management = platform
        .observe_root_management(state, &targets)
        .map_err(EnsureWorkflowError::Platform)?
        .ok_or(EnsureWorkflowError::PlanIntegrity)?;
    let current = recompile_root_start_prerequisite_plan(
        RootStartPlanInput {
            state,
            authority: plan.root_start_authority.as_deref(),
            created_at_time: plan.planned_at_time,
            desired,
            desired_sha256: &plan.desired_sha256,
            observation: &management,
            requested_fleet: &plan.fleet,
        },
        &targets,
    )?;
    if normalized_plan(plan) != normalized_plan(&current)
        || management.roots.iter().any(|(name, observed)| {
            targets.contains(name)
                && observed.live.status
                    != crate::fleet_ensure::model::CanisterRuntimeStatus::Running
        })
    {
        return Err(EnsureWorkflowError::ConvergenceDrift);
    }
    Ok(())
}

fn reviewed_root_start_targets<E>(
    plan: &FleetEnsurePlan,
) -> Result<BTreeSet<String>, EnsureWorkflowError<E>>
where
    E: std::error::Error + 'static,
{
    if plan.scope != FleetEnsurePlanScope::RootStartPrerequisite
        || !plan.protocol_actions.is_empty()
        || plan.canisters.is_empty()
        || plan.conservation.maximum_new_funding_cycles != 0
        || plan.conservation.maximum_operator_debit_cycles != 0
        || plan.conservation.maximum_unavoidable_fee_cycles != 0
        || !plan.conservation.estate_funding_domains.is_empty()
        || plan.conservation.scheduled_transfer_cycles != 0
        || plan.conservation.observed_controlled_cycles
            != plan.conservation.retained_in_reused_canisters_cycles
        || plan
            .conservation
            .observed_controlled_cycles
            .checked_sub(plan.conservation.maximum_execution_burn_cycles)
            != Some(plan.conservation.expected_post_operation_cycles)
    {
        return Err(EnsureWorkflowError::PlanIntegrity);
    }
    let mut targets = BTreeSet::new();
    for canister in &plan.canisters {
        if canister.disposition != CanisterDisposition::Reuse
            || canister.principal.is_none()
            || !matches!(
                canister.actions.as_slice(),
                [EnsureAction::Start { name, principal }]
                    if name == &canister.name
                        && Some(principal) == canister.principal.as_ref()
            )
            || !targets.insert(canister.name.clone())
        {
            return Err(EnsureWorkflowError::PlanIntegrity);
        }
    }
    Ok(targets)
}

fn root_management_fleet_observation<E>(
    management: &RootManagementObservation,
    targets: &BTreeSet<String>,
) -> Result<FleetObservation, EnsureWorkflowError<E>>
where
    E: std::error::Error + 'static,
{
    let canisters = targets
        .iter()
        .map(|name| {
            management
                .roots
                .get(name)
                .map(|observed| (name.clone(), Some(observed.live.clone())))
                .ok_or(EnsureWorkflowError::PlanIntegrity)
        })
        .collect::<Result<BTreeMap<_, _>, _>>()?;
    Ok(FleetObservation {
        additional_controlled_cycles: BTreeMap::new(),
        canisters,
        estate_funding_domains: BTreeMap::new(),
        ledger_fee_cycles: 0,
        operator_cycles: management.operator_cycles,
        protocol_ready: BTreeMap::new(),
    })
}

fn compatible_root_start_prerequisite(
    retained: &FleetEnsurePlan,
    current: &FleetEnsurePlan,
    desired: &crate::fleet_ensure::model::DesiredFleet,
) -> bool {
    let Ok(maximum_movement) = desired
        .maximum_update_burn_cycles
        .parse::<Cycles>()
        .map(|cycles| cycles.to_u128())
    else {
        return false;
    };
    retained.canisters.len() == current.canisters.len()
        && retained
            .canisters
            .iter()
            .zip(&current.canisters)
            .all(|(retained, current)| {
                retained
                    .observed_cycles
                    .saturating_sub(current.observed_cycles)
                    <= maximum_movement
            })
        && normalized_plan(retained) == normalized_plan(current)
}

fn attach_terminal_inventory_cycles<P>(
    operation_id: &str,
    state: &FleetEnsureStateRecord,
    platform: &mut P,
    observation: &mut FleetObservation,
) -> Result<(), EnsureWorkflowError<P::Error>>
where
    P: EnsurePlatform,
{
    let inventory = platform
        .terminal_inventory(operation_id, state)
        .map_err(EnsureWorkflowError::Platform)?;
    attach_terminal_cycles(observation, inventory.controlled_cycles_by_principal)
}

pub(super) fn completed_infrastructure_reinstall(
    desired: &crate::fleet_ensure::model::DesiredFleet,
    plan: &FleetEnsurePlan,
    journal: &FleetEnsureJournalRecord,
) -> bool {
    if desired.protocol.is_none() {
        return false;
    }
    let infrastructure = desired.canisters.iter().filter(|canister| {
        canister.presence == crate::fleet_ensure::model::DesiredPresence::Present
            && matches!(
                canister.kind,
                crate::fleet_ensure::model::DesiredCanisterKind::Coordinator
                    | crate::fleet_ensure::model::DesiredCanisterKind::Root
                    | crate::fleet_ensure::model::DesiredCanisterKind::Store
            )
    });
    let mut count = 0_usize;
    for configured in infrastructure {
        count += 1;
        let Some(action) = plan
            .canisters
            .iter()
            .find(|canister| {
                canister.name == configured.name
                    && canister.disposition == CanisterDisposition::Reinstall
            })
            .and_then(|canister| {
                canister.actions.iter().find(|action| {
                    matches!(
                        action,
                        EnsureAction::Install {
                            mode: crate::fleet_ensure::model::InstallMode::Reinstall,
                            name,
                            ..
                        } if name == &configured.name
                    )
                })
            })
        else {
            return false;
        };
        let action_sha256 = action_sha256(action);
        if !journal.effects.iter().any(|effect| {
            effect.action_sha256 == action_sha256 && effect.state == EffectState::Applied
        }) {
            return false;
        }
    }
    count > 0
}

fn compatible_after_bounded_observation(
    retained: &FleetEnsurePlan,
    current: &FleetEnsurePlan,
    desired: &crate::fleet_ensure::model::DesiredFleet,
    observation: &FleetObservation,
) -> bool {
    let Ok(maximum_observation_movement) = desired
        .maximum_observation_burn_cycles
        .parse::<Cycles>()
        .map(|cycles| cycles.to_u128())
    else {
        return false;
    };
    if retained.canisters.len() != current.canisters.len() {
        return false;
    }
    for (retained_canister, current_canister) in retained.canisters.iter().zip(&current.canisters) {
        let movement = retained_canister
            .observed_cycles
            .saturating_sub(current_canister.observed_cycles);
        if movement > maximum_observation_movement {
            return false;
        }
    }
    if !retained_funding_remains_sufficient(retained, desired, observation) {
        return false;
    }
    normalized_plan(retained) == normalized_plan(current)
}

fn retained_funding_remains_sufficient(
    retained: &FleetEnsurePlan,
    desired: &crate::fleet_ensure::model::DesiredFleet,
    observation: &FleetObservation,
) -> bool {
    retained
        .canisters
        .iter()
        .flat_map(|canister| &canister.actions)
        .filter_map(|action| match action {
            EnsureAction::Fund {
                amount,
                name,
                pool_funding,
                principal,
                ..
            } => Some((amount, name, pool_funding, principal)),
            _ => None,
        })
        .all(|(amount, name, pool_funding, principal)| {
            if let Some(root) = pool_funding {
                return observation
                    .estate_funding_domains
                    .values()
                    .find(|domain| domain.root_principal.as_ref() == Some(&root.root))
                    .and_then(|domain| domain.pool.as_ref())
                    .is_some_and(|pool| {
                        pool.assets.iter().any(|asset| {
                            asset.principal == *principal
                                && asset.lifecycle == root.lifecycle
                                && asset
                                    .cycles
                                    .checked_add(*amount)
                                    .is_some_and(|funded| funded >= pool.readiness_floor_cycles)
                        })
                    });
            }
            let Some(live) = observation.canisters.get(name).and_then(Option::as_ref) else {
                return false;
            };
            let Some(configured) = desired
                .canisters
                .iter()
                .find(|configured| configured.name == *name)
            else {
                return false;
            };
            let Ok(minimum) = configured
                .minimum_cycles
                .parse::<Cycles>()
                .map(|cycles| cycles.to_u128())
            else {
                return false;
            };
            live.cycles
                .checked_add(*amount)
                .is_some_and(|funded| funded >= minimum)
        })
}

fn normalized_plan(plan: &FleetEnsurePlan) -> FleetEnsurePlan {
    let mut normalized = plan.clone();
    normalized.plan_sha256.clear();
    normalized.reviewed_desired = None;
    if let Some(review) = &mut normalized.recovery_review {
        // These estimates move with bounded observations; they grant no funding authority.
        review.continuation_reserve_cycles = 0;
        for funding in &mut review.known_pool_funding {
            funding.amount_cycles = 0;
            funding.funding_deficit_cycles = 0;
        }
    }
    normalized.conservation = CycleConservation {
        estate_funding_domains: plan
            .conservation
            .estate_funding_domains
            .iter()
            .cloned()
            .map(|mut domain| {
                domain.available_cycles = None;
                domain.maximum_funding_cycles = 0;
                domain.root_principal = None;
                domain.shortfall_cycles = 0;
                domain
            })
            .collect(),
        expected_post_operation_cycles: 0,
        maximum_execution_burn_cycles: 0,
        maximum_new_funding_cycles: 0,
        maximum_operator_debit_cycles: 0,
        maximum_unavoidable_fee_cycles: 0,
        observed_controlled_cycles: 0,
        retained_in_reused_canisters_cycles: 0,
        scheduled_transfer_cycles: 0,
    };
    for canister in &mut normalized.canisters {
        canister.observed_cycles = 0;
        for action in &mut canister.actions {
            if let EnsureAction::Fund {
                amount,
                funding_deficit_cycles,
                ..
            } = action
            {
                *amount = 0;
                *funding_deficit_cycles = 0;
            }
        }
    }
    normalized
}

fn attach_terminal_cycles<E>(
    observation: &mut FleetObservation,
    additional: BTreeMap<String, u128>,
) -> Result<(), EnsureWorkflowError<E>>
where
    E: std::error::Error + 'static,
{
    let mut retained = BTreeMap::new();
    for (principal, cycles) in additional {
        let Some(configured) = observation
            .canisters
            .values_mut()
            .filter_map(Option::as_mut)
            .find(|canister| canister.principal == principal)
        else {
            retained.insert(principal, cycles);
            continue;
        };
        match configured.root_owned_lifecycle {
            Some(crate::fleet_ensure::model::RootOwnedCanisterLifecycle::Workload) => {
                // Both protected observations identify the same running workload, but
                // execution may burn cycles between them. Retain the conservative lower value.
                configured.cycles = configured.cycles.min(cycles);
            }
            Some(crate::fleet_ensure::model::RootOwnedCanisterLifecycle::Idle)
                if cycles >= configured.cycles =>
            {
                // A stopped asset can receive donations between protected reads.
                // Count the latest balance once, without attributing its origin.
                configured.cycles = cycles;
            }
            Some(crate::fleet_ensure::model::RootOwnedCanisterLifecycle::Idle) => {
                return Err(EnsureWorkflowError::TerminalInventory(
                    TerminalInventoryError::IdleCycleObservationConflict {
                        canister: principal,
                        expected: configured.cycles,
                        observed: cycles,
                    },
                ));
            }
            _ => {
                return Err(EnsureWorkflowError::TerminalInventory(
                    TerminalInventoryError::CycleObservationLifecycleConflict {
                        canister: principal,
                    },
                ));
            }
        }
    }
    observation.additional_controlled_cycles = retained;
    Ok(())
}

fn funding_plan<'a, E: std::error::Error + 'static>(
    plan: &'a FleetEnsurePlan,
    journal: &FleetEnsureJournalRecord,
) -> Result<std::borrow::Cow<'a, FleetEnsurePlan>, EnsureWorkflowError<E>> {
    if journal.funding_reviews.is_empty()
        && journal.funding_observations.is_empty()
        && journal.bootstrap_registration_recovery.is_none()
    {
        return Ok(std::borrow::Cow::Borrowed(plan));
    }
    funding_observation::verify(plan, journal)?;
    let mut funded = plan.clone();
    funded.conservation =
        crate::fleet_ensure::ops::infrastructure_bootstrap::registration_recovery::conservation(
            plan, journal,
        )?;
    let (amount, fee) = funding::totals::<E>(journal, None)?;
    let conservation = &mut funded.conservation;
    conservation.maximum_execution_burn_cycles = conservation
        .maximum_execution_burn_cycles
        .checked_add(funding_observation::total(journal)?)
        .ok_or(EnsureWorkflowError::JournalIntegrity)?;
    conservation.maximum_operator_debit_cycles = conservation
        .maximum_operator_debit_cycles
        .checked_add(amount)
        .and_then(|n| n.checked_add(fee))
        .ok_or(EnsureWorkflowError::JournalIntegrity)?;
    conservation.maximum_new_funding_cycles = conservation
        .maximum_new_funding_cycles
        .checked_add(amount)
        .ok_or(EnsureWorkflowError::JournalIntegrity)?;
    conservation.maximum_unavoidable_fee_cycles = conservation
        .maximum_unavoidable_fee_cycles
        .checked_add(fee)
        .ok_or(EnsureWorkflowError::JournalIntegrity)?;
    for domain in &mut conservation.estate_funding_domains {
        domain.maximum_funding_cycles = domain
            .maximum_funding_cycles
            .checked_add(funding::totals::<E>(journal, Some(&domain.root))?.0)
            .ok_or(EnsureWorkflowError::JournalIntegrity)?;
    }
    Ok(std::borrow::Cow::Owned(funded))
}

fn verify_terminal_conservation<E>(
    plan: &FleetEnsurePlan,
    journal: &FleetEnsureJournalRecord,
    state: &FleetEnsureStateRecord,
    terminal: &FleetObservation,
) -> Result<ActualCycleConservation, EnsureWorkflowError<E>>
where
    E: std::error::Error + 'static,
{
    verify_terminal_conservation_with_total(
        plan,
        journal,
        state,
        terminal,
        controlled_cycles(terminal)?,
    )
}

fn verify_terminal_conservation_with_total<E: std::error::Error + 'static>(
    plan: &FleetEnsurePlan,
    journal: &FleetEnsureJournalRecord,
    state: &FleetEnsureStateRecord,
    terminal: &FleetObservation,
    final_controlled_cycles: u128,
) -> Result<ActualCycleConservation, EnsureWorkflowError<E>> {
    let funded = funding_plan(plan, journal)?;
    let conservation = &funded.conservation;
    let operator_source_cycles =
        crate::fleet_ensure::policy::operator_mint::operator_source(journal)
            .ok_or(EnsureWorkflowError::JournalIntegrity)?;
    let minimum_operator_cycles =
        operator_source_cycles.saturating_sub(conservation.maximum_operator_debit_cycles);
    if journal.completion == FleetEnsureCompletion::Converged
        && !(minimum_operator_cycles..=operator_source_cycles).contains(&terminal.operator_cycles)
    {
        return Err(EnsureWorkflowError::TerminalReplayBalanceChanged {
            fleet: plan.fleet.clone(),
            operation_id: plan.operation_id.clone(),
            plan_sha256: plan.plan_sha256.clone(),
            current_operator_cycles: terminal.operator_cycles,
            minimum_operator_cycles,
            operator_source_cycles,
        });
    }
    let operator_debit_cycles = operator_source_cycles
        .checked_sub(terminal.operator_cycles)
        .ok_or_else(|| {
            EnsureWorkflowError::Conservation(
                "operator balance increased during apply; review a new plan".to_string(),
            )
        })?;
    if operator_debit_cycles > conservation.maximum_operator_debit_cycles {
        return Err(EnsureWorkflowError::Conservation(format!(
            "operator debit {operator_debit_cycles} exceeded reviewed maximum {}",
            conservation.maximum_operator_debit_cycles
        )));
    }
    let received_new_funding_cycles = operator_debit_cycles
        .checked_sub(conservation.maximum_unavoidable_fee_cycles)
        .ok_or_else(|| {
            EnsureWorkflowError::Conservation(
                "operator debit was below the exact reviewed fee total".to_string(),
            )
        })?;
    if received_new_funding_cycles > conservation.maximum_new_funding_cycles {
        return Err(EnsureWorkflowError::Conservation(format!(
            "received funding {received_new_funding_cycles} exceeded reviewed maximum {}",
            conservation.maximum_new_funding_cycles
        )));
    }
    let (estate_funding_cycles, exact_estate_creation_fee_cycles) =
        reconcile_estate_funding(plan, journal, state, terminal)?;
    let available = journal
        .initial_controlled_cycles
        .checked_add(received_new_funding_cycles)
        .ok_or_else(|| {
            EnsureWorkflowError::Conservation(
                "terminal controlled-cycle arithmetic overflowed".to_string(),
            )
        })?;
    let after_estate_fees = available
        .checked_sub(exact_estate_creation_fee_cycles)
        .ok_or_else(|| {
            EnsureWorkflowError::Conservation(
                "terminal controlled cycles cannot cover reviewed estate creation fees".to_string(),
            )
        })?;
    // Native balances are open to third-party deposits. These are net observations,
    // not authenticated funding or a measurement of gross execution consumption.
    let observed_net_cycle_debit_cycles = after_estate_fees.saturating_sub(final_controlled_cycles);
    let observed_net_cycle_credit_cycles =
        final_controlled_cycles.saturating_sub(after_estate_fees);
    if observed_net_cycle_debit_cycles > conservation.maximum_execution_burn_cycles {
        return Err(EnsureWorkflowError::Conservation(format!(
            "observed net cycle debit {observed_net_cycle_debit_cycles} exceeded reviewed maximum {}",
            conservation.maximum_execution_burn_cycles
        )));
    }
    Ok(ActualCycleConservation {
        estate_funding_cycles,
        exact_estate_creation_fee_cycles,
        exact_unavoidable_fee_cycles: conservation.maximum_unavoidable_fee_cycles,
        final_controlled_cycles,
        observed_net_cycle_debit_cycles,
        observed_starting_cycles: journal.initial_controlled_cycles,
        observed_net_cycle_credit_cycles,
        operator_debit_cycles,
        received_new_funding_cycles,
    })
}

fn resolved_initial_pool_assets<E>(
    domain: &crate::fleet_ensure::model::EstateFundingDomainPlan,
    state: &FleetEnsureStateRecord,
) -> Result<BTreeSet<String>, EnsureWorkflowError<E>>
where
    E: std::error::Error + 'static,
{
    domain
        .initial_pool_assets
        .iter()
        .map(|identity| {
            identity.strip_prefix("created:").map_or_else(
                || Ok(identity.clone()),
                |name| {
                    state
                        .pending_principals
                        .get(name)
                        .or_else(|| state.principals.get(name))
                        .cloned()
                        .ok_or(EnsureWorkflowError::JournalIntegrity)
                },
            )
        })
        .collect()
}

fn exact_estate_creation_costs<E>(
    state: &FleetEnsureStateRecord,
    terminal: &FleetObservation,
    domain: &crate::fleet_ensure::model::EstateFundingDomainPlan,
) -> Result<(u128, u128), EnsureWorkflowError<E>>
where
    E: std::error::Error + 'static,
{
    let observed = terminal
        .estate_funding_domains
        .get(&domain.root)
        .ok_or(EnsureWorkflowError::JournalIntegrity)?;
    let Some(pool) = observed.pool.as_ref() else {
        return if domain.required_creation_count == 0 {
            Ok((0, 0))
        } else {
            Err(EnsureWorkflowError::JournalIntegrity)
        };
    };
    if pool.pending_creation.is_some() {
        return Err(EnsureWorkflowError::JournalIntegrity);
    }

    estate_creation_costs(state, terminal, domain)
}

fn estate_creation_costs<E: std::error::Error + 'static>(
    state: &FleetEnsureStateRecord,
    observation: &FleetObservation,
    domain: &EstateFundingDomainPlan,
) -> Result<(u128, u128), EnsureWorkflowError<E>> {
    let pool = observation
        .estate_funding_domains
        .get(&domain.root)
        .and_then(|domain| domain.pool.as_ref())
        .ok_or(EnsureWorkflowError::JournalIntegrity)?;
    let initial = resolved_initial_pool_assets(domain, state)?;
    let terminal_principals = pool
        .assets
        .iter()
        .map(|asset| asset.principal.clone())
        .collect::<BTreeSet<_>>();
    if terminal_principals.len() != pool.assets.len() || !initial.is_subset(&terminal_principals) {
        return Err(EnsureWorkflowError::JournalIntegrity);
    }
    let created = pool
        .assets
        .iter()
        .filter(|asset| !initial.contains(&asset.principal))
        .collect::<Vec<_>>();
    let creation_count_is_bounded = created.len()
        <= usize::try_from(domain.required_creation_count)
            .map_err(|_| EnsureWorkflowError::JournalIntegrity)?;
    // Root maintenance can spend its reviewed reserve without provisioning a new Workload.
    // The per-Root budget and exact receipts bind that autonomous creation authority.
    let pool_policy_is_exact = created.is_empty() || terminal_pool_policy_is_exact(pool, domain);
    if !creation_count_is_bounded || !pool_policy_is_exact {
        return Err(EnsureWorkflowError::JournalIntegrity);
    }

    let mut operation_ids = BTreeSet::new();
    let mut block_indexes = BTreeSet::new();
    let mut creation_debit = 0_u128;
    let mut creation_fee = 0_u128;
    for asset in created {
        let (asset_debit, asset_fee) = exact_estate_creation_receipt_costs(
            asset,
            domain,
            &mut operation_ids,
            &mut block_indexes,
        )?;
        creation_debit = creation_debit.checked_add(asset_debit).ok_or_else(|| {
            EnsureWorkflowError::Conservation(
                "estate creation-debit arithmetic overflowed".to_string(),
            )
        })?;
        creation_fee = creation_fee.checked_add(asset_fee).ok_or_else(|| {
            EnsureWorkflowError::Conservation(
                "estate creation-fee arithmetic overflowed".to_string(),
            )
        })?;
    }
    if creation_debit > domain.maximum_creation_debit_cycles
        || creation_fee > domain.maximum_creation_fee_cycles
    {
        return Err(EnsureWorkflowError::JournalIntegrity);
    }
    Ok((creation_debit, creation_fee))
}

const fn terminal_pool_policy_is_exact(
    pool: &crate::fleet_ensure::model::EstatePoolInventoryObservation,
    domain: &crate::fleet_ensure::model::EstateFundingDomainPlan,
) -> bool {
    let expected_maximum_size = domain.pool_maximum_size;
    let expected_readiness_floor = domain.readiness_floor_cycles;
    let expected_execution_margin = domain.creation_execution_margin_cycles;
    pool.maximum_size == expected_maximum_size
        && pool.readiness_floor_cycles == expected_readiness_floor
        && pool.creation_execution_margin_cycles == expected_execution_margin
}

#[derive(Eq, PartialEq)]
struct EstateCreationReceiptAuthority<'a> {
    creation_execution_margin_cycles: u128,
    cycles_ledger: &'a str,
    ledger_amount_cycles: u128,
    ledger_fee_cycles: u128,
    management_creation_fee_cycles: u128,
    readiness_floor_cycles: u128,
}

fn observed_estate_creation_authority(
    receipt: &crate::fleet_ensure::model::EstatePoolCreationReceiptObservation,
) -> EstateCreationReceiptAuthority<'_> {
    EstateCreationReceiptAuthority {
        creation_execution_margin_cycles: receipt.creation_execution_margin_cycles,
        cycles_ledger: &receipt.cycles_ledger,
        ledger_amount_cycles: receipt.ledger_amount_cycles,
        ledger_fee_cycles: receipt.ledger_fee_cycles,
        management_creation_fee_cycles: receipt.management_creation_fee_cycles,
        readiness_floor_cycles: receipt.readiness_floor_cycles,
    }
}

fn reviewed_estate_creation_authority(
    domain: &crate::fleet_ensure::model::EstateFundingDomainPlan,
) -> EstateCreationReceiptAuthority<'_> {
    EstateCreationReceiptAuthority {
        creation_execution_margin_cycles: domain.creation_execution_margin_cycles,
        cycles_ledger: &domain.cycles_ledger,
        ledger_amount_cycles: domain.creation_amount_cycles,
        ledger_fee_cycles: domain.ledger_fee_cycles,
        management_creation_fee_cycles: domain.management_creation_fee_cycles,
        readiness_floor_cycles: domain.readiness_floor_cycles,
    }
}

fn first_estate_creation_observation_is_exact(
    receipt: &crate::fleet_ensure::model::EstatePoolCreationReceiptObservation,
) -> bool {
    receipt
        .ledger_amount_cycles
        .checked_sub(receipt.management_creation_fee_cycles)
        .zip(receipt.first_observed_cycles)
        .is_some_and(|(funded_native_cycles, observed_cycles)| {
            observed_cycles >= receipt.readiness_floor_cycles
                && funded_native_cycles.saturating_sub(observed_cycles)
                    <= receipt.creation_execution_margin_cycles
        })
}

fn exact_estate_creation_receipt_costs<E>(
    asset: &crate::fleet_ensure::model::EstatePoolAssetObservation,
    domain: &crate::fleet_ensure::model::EstateFundingDomainPlan,
    operation_ids: &mut BTreeSet<String>,
    block_indexes: &mut BTreeSet<u64>,
) -> Result<(u128, u128), EnsureWorkflowError<E>>
where
    E: std::error::Error + 'static,
{
    let receipt = asset
        .creation_receipt
        .as_ref()
        .filter(|_| asset.origin == crate::fleet_ensure::model::EstatePoolAssetOrigin::Created)
        .ok_or(EnsureWorkflowError::JournalIntegrity)?;
    let authority_is_exact =
        observed_estate_creation_authority(receipt) == reviewed_estate_creation_authority(domain);
    if !is_sha256_hex(&receipt.operation_id)
        || !authority_is_exact
        || !first_estate_creation_observation_is_exact(receipt)
        || !operation_ids.insert(receipt.operation_id.clone())
        || !block_indexes.insert(receipt.block_index)
    {
        return Err(EnsureWorkflowError::JournalIntegrity);
    }
    let debit = receipt
        .ledger_amount_cycles
        .checked_add(receipt.ledger_fee_cycles)
        .ok_or_else(|| {
            EnsureWorkflowError::Conservation(
                "estate creation-debit arithmetic overflowed".to_string(),
            )
        })?;
    let fee = receipt
        .management_creation_fee_cycles
        .checked_add(receipt.ledger_fee_cycles)
        .ok_or_else(|| {
            EnsureWorkflowError::Conservation(
                "estate creation-fee arithmetic overflowed".to_string(),
            )
        })?;
    Ok((debit, fee))
}

fn reconcile_estate_funding<E>(
    plan: &FleetEnsurePlan,
    journal: &FleetEnsureJournalRecord,
    state: &FleetEnsureStateRecord,
    terminal: &FleetObservation,
) -> Result<(u128, u128), EnsureWorkflowError<E>>
where
    E: std::error::Error + 'static,
{
    plan.conservation
        .estate_funding_domains
        .iter()
        .try_fold((0_u128, 0_u128), |(funding_total, fee_total), domain| {
            let initial = journal
                .initial_estate_funding_cycles_by_root
                .get(&domain.root)
                .copied()
                .ok_or(EnsureWorkflowError::JournalIntegrity)?;
            let observed = terminal
                .estate_funding_domains
                .get(&domain.root)
                .ok_or(EnsureWorkflowError::JournalIntegrity)?;
            if observed.cycles_ledger != domain.cycles_ledger
                || reviewed_estate_root_principal(domain, state)
                    != observed.root_principal.as_deref()
            {
                return Err(EnsureWorkflowError::JournalIntegrity);
            }
            let terminal_balance = observed
                .balance_cycles
                .ok_or(EnsureWorkflowError::JournalIntegrity)?;
            let (creation_debit, creation_fee) = exact_estate_creation_costs(
                state,
                terminal,
                domain,
            )?;
            let funding = applied_estate_funding_for_domain(plan, journal, state, domain)?
                .checked_add(funding::totals::<E>(journal, Some(&domain.root))?.0)
                .ok_or(EnsureWorkflowError::JournalIntegrity)?;
            let expected_terminal_balance = initial
                .checked_add(funding)
                .and_then(|available| available.checked_sub(creation_debit))
                .ok_or_else(|| {
                    EnsureWorkflowError::Conservation(format!(
                        "Root {} funding-account balance cannot reconcile its reviewed funding and creation debit",
                        domain.root
                    ))
                })?;
            // Surplus Ledger cycles join the terminal net-credit observation;
            // authenticated funding and exact creation costs remain unchanged.
            if terminal_balance < expected_terminal_balance {
                return Err(EnsureWorkflowError::Conservation(format!(
                    "Root {} funding-account balance {terminal_balance} differs from reviewed terminal balance {expected_terminal_balance}",
                    domain.root
                )));
            }
            Ok((
                funding_total.checked_add(funding).ok_or_else(|| {
                    EnsureWorkflowError::Conservation(
                        "estate funding arithmetic overflowed".to_string(),
                    )
                })?,
                fee_total
                    .checked_add(creation_fee)
                    .ok_or_else(|| {
                        EnsureWorkflowError::Conservation(
                            "estate creation-fee arithmetic overflowed".to_string(),
                        )
                    })?,
            ))
        })
}

fn applied_estate_funding_for_domain<E>(
    plan: &FleetEnsurePlan,
    journal: &FleetEnsureJournalRecord,
    state: &FleetEnsureStateRecord,
    domain: &crate::fleet_ensure::model::EstateFundingDomainPlan,
) -> Result<u128, EnsureWorkflowError<E>>
where
    E: std::error::Error + 'static,
{
    let action = plan
        .canisters
        .iter()
        .find(|canister| canister.name == domain.root)
        .into_iter()
        .flat_map(|canister| &canister.actions)
        .find(|action| matches!(action, EnsureAction::FundEstate { .. }));
    let Some(action) = action else {
        return if domain.maximum_funding_cycles == 0 {
            Ok(0)
        } else {
            Err(EnsureWorkflowError::JournalIntegrity)
        };
    };
    let EnsureAction::FundEstate {
        amount,
        expected_post_cycles,
        ledger,
        ledger_fee_cycles,
        principal,
        ..
    } = action
    else {
        unreachable!("filtered estate-funding action")
    };
    let root_principal = principal.strip_prefix("created:").map_or_else(
        || Some(principal.as_str()),
        |name| {
            state
                .pending_principals
                .get(name)
                .or_else(|| state.principals.get(name))
                .map(String::as_str)
        },
    );
    let expected_destination = domain
        .available_cycles
        .unwrap_or_default()
        .checked_add(domain.shortfall_cycles)
        .ok_or(EnsureWorkflowError::JournalIntegrity)?;
    let action_authority_is_exact = *amount == domain.shortfall_cycles
        && *expected_post_cycles == expected_destination
        && ledger == &domain.cycles_ledger
        && *ledger_fee_cycles == domain.ledger_fee_cycles
        && root_principal == reviewed_estate_root_principal(domain, state);
    let effect = journal
        .effects
        .iter()
        .find(|effect| effect.action_sha256 == action_sha256(action));
    let Some(effect) = effect.filter(|effect| {
        effect.state == EffectState::Applied
            && effect
                .receipt
                .as_deref()
                .is_some_and(|receipt| !receipt.is_empty())
    }) else {
        return Err(EnsureWorkflowError::JournalIntegrity);
    };
    let balance_evidence_is_exact = crate::fleet_ensure::ops::estate_funding_applied(
        crate::fleet_ensure::ops::EstateFundingObservation {
            amount: *amount,
            destination_after: effect.destination_post_cycles.unwrap_or_default(),
            destination_before: effect.destination_pre_cycles,
            expected_destination_after: *expected_post_cycles,
            ledger_fee_cycles: *ledger_fee_cycles,
            source_after: effect.post_cycles.unwrap_or_default(),
            source_before: effect.pre_cycles,
        },
    );
    if !action_authority_is_exact || !balance_evidence_is_exact {
        return Err(EnsureWorkflowError::JournalIntegrity);
    }
    Ok(*amount)
}

fn controlled_cycles<E>(observation: &FleetObservation) -> Result<u128, EnsureWorkflowError<E>>
where
    E: std::error::Error + 'static,
{
    observation
        .canisters
        .values()
        .filter_map(Option::as_ref)
        .try_fold(0_u128, |total, live| total.checked_add(live.cycles))
        .and_then(|total| {
            observation
                .additional_controlled_cycles
                .values()
                .try_fold(total, |subtotal, cycles| subtotal.checked_add(*cycles))
        })
        .and_then(|total| {
            observation
                .estate_funding_domains
                .values()
                .filter_map(|domain| domain.balance_cycles)
                .try_fold(total, u128::checked_add)
        })
        .ok_or_else(|| {
            EnsureWorkflowError::Conservation("controlled-cycle total overflowed".to_string())
        })
}

fn verified_plan<E>(plan: FleetEnsurePlan) -> Result<FleetEnsurePlan, EnsureWorkflowError<E>>
where
    E: std::error::Error + 'static,
{
    if plan.reviewed_desired.is_none()
        || expected_plan_sha256(&plan) != plan.plan_sha256
        || (plan.scope == FleetEnsurePlanScope::InfrastructureBootstrap)
            != plan.infrastructure_bootstrap.is_some()
    {
        return Err(EnsureWorkflowError::PlanIntegrity);
    }
    Ok(plan)
}

fn verify_journal<E>(
    journal: &FleetEnsureJournalRecord,
    plan: &FleetEnsurePlan,
    requested_fleet: &str,
    state: &FleetEnsureStateRecord,
) -> Result<(), EnsureWorkflowError<E>>
where
    E: std::error::Error + 'static,
{
    verify_journal_integrity(journal, plan, requested_fleet, state)?;
    if let Some(review) = journal
        .funding_reviews
        .iter()
        .filter_map(|r| r.operator_mint.as_ref())
        .find(|review| review.receipt.is_none())
    {
        return Err(EnsureWorkflowError::OperatorMintPending {
            review_sha256: review.review_sha256.clone(),
        });
    }
    Ok(())
}

fn verify_journal_integrity<E>(
    journal: &FleetEnsureJournalRecord,
    plan: &FleetEnsurePlan,
    requested_fleet: &str,
    state: &FleetEnsureStateRecord,
) -> Result<(), EnsureWorkflowError<E>>
where
    E: std::error::Error + 'static,
{
    if journal.fleet != requested_fleet
        || journal.operation_id != plan.operation_id
        || journal.plan_sha256 != plan.plan_sha256
        || !initial_estate_funding_is_exact(journal, plan)
        || !estate_funding_pause_is_exact(journal, plan, state)
    {
        return Err(EnsureWorkflowError::JournalIntegrity);
    }
    crate::fleet_ensure::ops::infrastructure_bootstrap::registration_recovery::verify(
        plan, journal, state,
    )?;
    funding::verify(plan, journal, state)?;
    funding_observation::verify(plan, journal)?;
    operator_mint::verify(plan, journal)?;
    continuation::verify_records(plan, journal)?;
    let actions = continuation::actions(plan, journal);
    let effect_count_matches = journal.effects.len() <= actions.len();
    let action_hashes_match = journal.effects.iter().zip(actions).all(|(effect, action)| {
        effect.action_sha256 == action_sha256(action) && effect.attempts_match(action)
    });
    if !(effect_count_matches && action_hashes_match) {
        return Err(EnsureWorkflowError::JournalIntegrity);
    }
    Ok(())
}

fn initial_estate_funding_is_exact(
    journal: &FleetEnsureJournalRecord,
    plan: &FleetEnsurePlan,
) -> bool {
    journal.initial_estate_funding_cycles_by_root.len()
        == plan.conservation.estate_funding_domains.len()
        && plan
            .conservation
            .estate_funding_domains
            .iter()
            .all(|domain| {
                journal
                    .initial_estate_funding_cycles_by_root
                    .get(&domain.root)
                    .copied()
                    == Some(domain.available_cycles.unwrap_or_default())
            })
}

fn estate_funding_pause_is_exact(
    journal: &FleetEnsureJournalRecord,
    plan: &FleetEnsurePlan,
    state: &FleetEnsureStateRecord,
) -> bool {
    let Some(required) = journal.estate_funding_required.as_ref() else {
        return true;
    };
    journal.completion == FleetEnsureCompletion::InProgress
        && estate_funding_record_is_exact(required, plan, state)
}

fn estate_funding_record_is_exact(
    required: &EstateFundingRequiredRecord,
    plan: &FleetEnsurePlan,
    state: &FleetEnsureStateRecord,
) -> bool {
    let Some(reviewed) = plan
        .conservation
        .estate_funding_domains
        .iter()
        .find(|domain| domain.root == required.root)
    else {
        return false;
    };
    let operation_authority_is_exact =
        required.operation_id == plan.operation_id && required.plan_sha256 == plan.plan_sha256;
    let funding_authority_is_exact =
        EstateFundingPauseAuthority::from(required) == EstateFundingPauseAuthority::from(reviewed);
    let root_authority_is_exact =
        reviewed_estate_root_principal(reviewed, state) == Some(required.root_principal.as_str());
    let shortfall_is_exact = required.available_cycles < required.maximum_creation_debit_cycles
        && required.shortfall_cycles
            == required
                .maximum_creation_debit_cycles
                .saturating_sub(required.available_cycles);
    if !(operation_authority_is_exact
        && funding_authority_is_exact
        && root_authority_is_exact
        && shortfall_is_exact)
    {
        return false;
    }
    if required.pending_creation_operation_id.is_some() {
        current_creation_funding_pause_is_exact(required, reviewed)
    } else {
        planned_estate_funding_pause_is_exact(required, reviewed)
    }
}

#[derive(Eq, PartialEq)]
struct EstateFundingPauseAuthority<'a> {
    creation_amount_cycles: u128,
    creation_execution_margin_cycles: u128,
    cycles_ledger: &'a str,
    ledger_fee_cycles: u128,
    management_creation_fee_cycles: u128,
    readiness_floor_cycles: u128,
    root: &'a str,
}

impl<'a> From<&'a EstateFundingRequiredRecord> for EstateFundingPauseAuthority<'a> {
    fn from(required: &'a EstateFundingRequiredRecord) -> Self {
        Self {
            creation_amount_cycles: required.creation_amount_cycles,
            creation_execution_margin_cycles: required.creation_execution_margin_cycles,
            cycles_ledger: &required.cycles_ledger,
            ledger_fee_cycles: required.ledger_fee_cycles,
            management_creation_fee_cycles: required.management_creation_fee_cycles,
            readiness_floor_cycles: required.readiness_floor_cycles,
            root: &required.root,
        }
    }
}

impl<'a> From<&'a EstateFundingDomainPlan> for EstateFundingPauseAuthority<'a> {
    fn from(reviewed: &'a EstateFundingDomainPlan) -> Self {
        Self {
            creation_amount_cycles: reviewed.creation_amount_cycles,
            creation_execution_margin_cycles: reviewed.creation_execution_margin_cycles,
            cycles_ledger: &reviewed.cycles_ledger,
            ledger_fee_cycles: reviewed.ledger_fee_cycles,
            management_creation_fee_cycles: reviewed.management_creation_fee_cycles,
            readiness_floor_cycles: reviewed.readiness_floor_cycles,
            root: &reviewed.root,
        }
    }
}

fn current_creation_funding_pause_is_exact(
    required: &EstateFundingRequiredRecord,
    reviewed: &EstateFundingDomainPlan,
) -> bool {
    let operation_id_is_exact = required
        .pending_creation_operation_id
        .as_deref()
        .is_some_and(is_sha256_hex);
    let attempt_evidence_is_exact = required.attempt_count.is_some()
        && required
            .retry_at_ns
            .is_some_and(|retry_at_ns| retry_at_ns > 0)
        && required
            .last_attempt_at_ns
            .is_none_or(|attempted_at_ns| attempted_at_ns > 0);
    let single_creation_debit = required
        .creation_amount_cycles
        .checked_add(required.ledger_fee_cycles);
    let single_creation_fees = required
        .management_creation_fee_cycles
        .checked_add(required.ledger_fee_cycles);
    let debit_authority_is_exact = Some(required.maximum_creation_debit_cycles)
        == single_creation_debit
        && Some(required.maximum_creation_fee_cycles) == single_creation_fees;
    reviewed.required_creation_count > 0
        && required.required_creation_count == 1
        && operation_id_is_exact
        && attempt_evidence_is_exact
        && debit_authority_is_exact
}

const fn planned_estate_funding_pause_is_exact(
    required: &EstateFundingRequiredRecord,
    reviewed: &EstateFundingDomainPlan,
) -> bool {
    let runtime_attempt_is_absent = required.attempt_count.is_none()
        && required.last_attempt_at_ns.is_none()
        && required.retry_at_ns.is_none();
    let reviewed_debit_is_exact = required.maximum_creation_debit_cycles
        == reviewed.maximum_creation_debit_cycles
        && required.maximum_creation_fee_cycles == reviewed.maximum_creation_fee_cycles
        && required.required_creation_count == reviewed.required_creation_count;
    runtime_attempt_is_absent && reviewed_debit_is_exact
}

fn is_sha256_hex(value: &str) -> bool {
    value.len() == 64
        && value
            .bytes()
            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum TerminalStatePublicationError {
    JournalIntegrity,
}

fn publish_terminal_state(
    desired: &crate::fleet_ensure::model::DesiredFleet,
    plan: &FleetEnsurePlan,
    journal: &FleetEnsureJournalRecord,
    state: &mut FleetEnsureStateRecord,
) -> Result<(), TerminalStatePublicationError> {
    let maximum_observation_burn_cycles =
        crate::fleet_ensure::ops::maximum_creation_observation_burn(desired)
            .ok_or(TerminalStatePublicationError::JournalIntegrity)?;
    let actions = continuation::actions(plan, journal);
    if actions.len() != journal.effects.len() {
        return Err(TerminalStatePublicationError::JournalIntegrity);
    }
    for (action, record) in actions.into_iter().zip(&journal.effects) {
        if action_sha256(action) != record.action_sha256 || record.state != EffectState::Applied {
            return Err(TerminalStatePublicationError::JournalIntegrity);
        }
        let EnsureAction::Create {
            name,
            requested_initial_cycles,
            ..
        } = action
        else {
            continue;
        };
        let Some(created_principal) = record.created_principal.as_deref() else {
            return Err(TerminalStatePublicationError::JournalIntegrity);
        };
        let Some(post_cycles) = record.post_cycles else {
            return Err(TerminalStatePublicationError::JournalIntegrity);
        };
        if !create_balance_is_terminal(
            Some(post_cycles),
            *requested_initial_cycles,
            maximum_observation_burn_cycles,
        ) || !create_identity_is_exact(name, record, state)
            || state
                .retained_cycles_by_principal
                .get(created_principal)
                .is_some_and(|retained| *retained != post_cycles)
        {
            return Err(TerminalStatePublicationError::JournalIntegrity);
        }
        state
            .retained_cycles_by_principal
            .insert(created_principal.to_string(), post_cycles);
    }

    let prior_topology = state.topology.clone();
    for canister in &plan.canisters {
        if !desired
            .canisters
            .iter()
            .any(|configured| configured.name == canister.name)
        {
            // Dynamic pool assets remain in protected Root inventory. A funding-only
            // projection must not become a configured application identity.
            if !is_reviewed_dynamic_pool_funding(plan, canister) {
                return Err(TerminalStatePublicationError::JournalIntegrity);
            }
            continue;
        }

        match canister.disposition {
            CanisterDisposition::Create | CanisterDisposition::Replace => {
                if let Some(principal) = state.pending_principals.remove(&canister.name) {
                    state.principals.insert(canister.name.clone(), principal);
                }
            }
            CanisterDisposition::Reinstall | CanisterDisposition::Reuse => {
                if let Some(principal) = state
                    .pending_principals
                    .remove(&canister.name)
                    .or_else(|| canister.principal.clone())
                {
                    state.principals.insert(canister.name.clone(), principal);
                }
            }
            CanisterDisposition::Delete => {
                state.principals.remove(&canister.name);
                state.pending_principals.remove(&canister.name);
            }
        }
    }
    publish_topology(desired, plan, &prior_topology, state);
    retain_configured_principal_bindings(state);
    Ok(())
}

fn publish_topology(
    desired: &crate::fleet_ensure::model::DesiredFleet,
    plan: &FleetEnsurePlan,
    prior_topology: &BTreeMap<String, crate::fleet_ensure::model::FleetEnsureTopologyRecord>,
    state: &mut FleetEnsureStateRecord,
) {
    state.topology = desired
        .canisters
        .iter()
        .filter(|canister| {
            canister.presence == crate::fleet_ensure::model::DesiredPresence::Present
                && (plan.scope != FleetEnsurePlanScope::InfrastructureBootstrap
                    || plan
                        .canisters
                        .iter()
                        .any(|target| target.name == canister.name))
        })
        .map(|canister| {
            (
                canister.name.clone(),
                crate::fleet_ensure::model::FleetEnsureTopologyRecord {
                    kind: canister.kind,
                    module_hash: projected_module_hash(plan, prior_topology, canister),
                    parent: canister.parent.clone(),
                    protocol_binding: canister.protocol_binding.clone(),
                    role: canister
                        .protocol_binding
                        .as_ref()
                        .map(|binding| binding.role.to_string()),
                },
            )
        })
        .collect();
}

fn is_reviewed_dynamic_pool_funding(
    plan: &FleetEnsurePlan,
    canister: &crate::fleet_ensure::model::CanisterPlan,
) -> bool {
    canister.disposition == CanisterDisposition::Reuse
        && !canister.actions.is_empty()
        && canister
            .actions
            .iter()
            .all(|action| is_reviewed_pool_funding(action, plan, canister.principal.as_deref()))
}

fn is_reviewed_pool_funding(
    action: &EnsureAction,
    plan: &FleetEnsurePlan,
    expected_principal: Option<&str>,
) -> bool {
    let EnsureAction::Fund {
        pool_funding: Some(root),
        principal,
        ..
    } = action
    else {
        return false;
    };
    if Some(principal.as_str()) != expected_principal {
        return false;
    }
    plan.protocol_actions.iter().any(|candidate| {
        let EnsureAction::FleetProtocol {
            principal: target,
            action,
            ..
        } = candidate
        else {
            return false;
        };
        let CurrentFleetProtocolAction::ReconcilePoolAsset { request, .. } = action.as_ref() else {
            return false;
        };
        target == &root.root && request.canister_id.to_text() == *principal
    })
}

fn projected_module_hash(
    plan: &FleetEnsurePlan,
    prior_topology: &BTreeMap<String, crate::fleet_ensure::model::FleetEnsureTopologyRecord>,
    canister: &crate::fleet_ensure::model::DesiredCanister,
) -> Option<String> {
    let installed = plan
        .canisters
        .iter()
        .find(|planned| planned.name == canister.name)
        .and_then(|planned| {
            planned
                .actions
                .iter()
                .rev()
                .find_map(|action| match action {
                    EnsureAction::Install { wasm_sha256, .. } => Some(wasm_sha256.clone()),
                    _ => None,
                })
        });
    installed.or_else(|| {
        prior_topology
            .get(&canister.name)
            .filter(|retained| retained.kind == canister.kind && retained.parent == canister.parent)
            .and_then(|retained| retained.module_hash.clone())
    })
}

fn merge_terminal_inventory<E>(
    state: &mut FleetEnsureStateRecord,
    inventory: crate::fleet_ensure::ops::TerminalFleetInventory,
) -> Result<(), EnsureWorkflowError<E>>
where
    E: std::error::Error + 'static,
{
    let entries = inventory.entries;
    let mut names_by_principal = state
        .principals
        .iter()
        .map(|(name, principal)| (principal.clone(), name.clone()))
        .collect::<BTreeMap<_, _>>();
    if names_by_principal.len() != state.principals.len() {
        return Err(EnsureWorkflowError::TerminalInventory(
            TerminalInventoryError::DuplicateRetainedPrincipal,
        ));
    }
    for entry in &entries {
        names_by_principal
            .entry(entry.pid.clone())
            .or_insert_with(|| format!("observed:{}", entry.pid));
    }
    for entry in entries {
        let name = names_by_principal.get(&entry.pid).cloned().ok_or_else(|| {
            EnsureWorkflowError::TerminalInventory(TerminalInventoryError::LostPrincipalBinding {
                canister: entry.pid.clone(),
            })
        })?;
        let parent = entry
            .parent_pid
            .as_ref()
            .map(|principal| {
                names_by_principal.get(principal).cloned().ok_or_else(|| {
                    EnsureWorkflowError::TerminalInventory(TerminalInventoryError::UnknownParent {
                        canister: entry.pid.clone(),
                        parent: principal.clone(),
                    })
                })
            })
            .transpose()?;
        let existing = state.topology.get(&name);
        let kind = terminal_inventory_kind(existing, &entry);
        let parent_conflict = existing
            .and_then(|topology| topology.parent.as_ref())
            .zip(parent.as_ref())
            .filter(|(expected, actual)| expected != actual);
        if let Some((retained_parent, terminal_parent)) = parent_conflict
            && !verified_pool_to_component_parent_transition(existing, kind, &entry)
        {
            return Err(EnsureWorkflowError::TerminalInventory(
                TerminalInventoryError::RetainedParentConflict {
                    canister: entry.pid,
                    retained_parent: retained_parent.clone(),
                    terminal_parent: terminal_parent.clone(),
                },
            ));
        }
        state.principals.insert(name.clone(), entry.pid);
        state.topology.insert(
            name,
            crate::fleet_ensure::model::FleetEnsureTopologyRecord {
                kind,
                module_hash: entry.module_hash,
                parent,
                protocol_binding: entry.protocol_binding,
                role: entry.role,
            },
        );
    }
    state.active_registry = inventory.active_registry;
    Ok(())
}

fn terminal_inventory_kind(
    existing: Option<&crate::fleet_ensure::model::FleetEnsureTopologyRecord>,
    entry: &crate::registry::RegistryEntry,
) -> crate::fleet_ensure::model::DesiredCanisterKind {
    existing.map_or_else(
        || {
            if entry.module_hash.is_some() {
                crate::fleet_ensure::model::DesiredCanisterKind::Component
            } else {
                crate::fleet_ensure::model::DesiredCanisterKind::Auxiliary
            }
        },
        |topology| {
            if topology.kind == crate::fleet_ensure::model::DesiredCanisterKind::Pool
                && entry.module_hash.is_some()
                && entry.protocol_binding.is_some()
            {
                crate::fleet_ensure::model::DesiredCanisterKind::Component
            } else {
                topology.kind
            }
        },
    )
}

fn verified_pool_to_component_parent_transition(
    existing: Option<&crate::fleet_ensure::model::FleetEnsureTopologyRecord>,
    terminal_kind: crate::fleet_ensure::model::DesiredCanisterKind,
    entry: &crate::registry::RegistryEntry,
) -> bool {
    existing.is_some_and(|topology| {
        topology.kind == crate::fleet_ensure::model::DesiredCanisterKind::Pool
            && terminal_kind == crate::fleet_ensure::model::DesiredCanisterKind::Component
            && entry.module_hash.is_some()
            && entry.protocol_binding.is_some()
    })
}

fn applied_count(journal: &FleetEnsureJournalRecord) -> u32 {
    u32::try_from(
        journal
            .effects
            .iter()
            .chain(
                journal
                    .funding_reviews
                    .iter()
                    .filter_map(|review| review.effect.as_ref()),
            )
            .filter(|effect| matches!(effect.state, EffectState::Applied))
            .count(),
    )
    .unwrap_or(u32::MAX)
}

fn verify_store_activation_binding<P: EnsurePlatform>(
    plan: &FleetEnsurePlan,
    action: &EnsureAction,
    state: &FleetEnsureStateRecord,
    platform: &mut P,
) -> Result<(), EnsureWorkflowError<P::Error>> {
    use crate::fleet_ensure::model::DesiredCanisterInit;
    let EnsureAction::Install {
        canic_init: Some(DesiredCanisterInit::Store { root }),
        ..
    } = action
    else {
        return Ok(());
    };
    let root_will_be_installed = plan.canisters.iter().flat_map(|canister| &canister.actions)
        .any(|action| matches!(action,
            EnsureAction::Install { name, canic_init: Some(DesiredCanisterInit::Root { root: installed_root }), .. }
            if name == root && installed_root == root));
    if !root_will_be_installed {
        platform
            .require_retained_root_activation(&plan.operation_id, root, state)
            .map_err(EnsureWorkflowError::Platform)?;
    }
    Ok(())
}

pub(super) fn ordered_actions(plan: &FleetEnsurePlan) -> Vec<&EnsureAction> {
    let mut actions = plan
        .canisters
        .iter()
        .flat_map(|canister| canister.actions.iter())
        .chain(plan.protocol_actions.iter())
        .collect::<Vec<_>>();
    if plan.scope == FleetEnsurePlanScope::ReinstallPreparation
        && plan
            .reinstall
            .as_ref()
            .is_some_and(|intent| intent.activation_reset.is_some())
    {
        // Preserve the compiler's Coordinator-first order, then settle every Root before restart.
        actions.sort_by_key(|action| matches!(action, EnsureAction::Start { .. }));
        return actions;
    }
    if matches!(
        plan.scope,
        FleetEnsurePlanScope::RootReinstallPrerequisite
            | FleetEnsurePlanScope::InfrastructureBootstrap
    ) {
        return actions;
    }
    let temporary_pool_observation_finalizations = plan
        .reviewed_desired
        .as_deref()
        .map(|reviewed| {
            let desired = reviewed.desired();
            plan.canisters
                .iter()
                .filter_map(|canister_plan| {
                    let configured = desired.canisters.iter().find(|configured| {
                        configured.name == canister_plan.name
                            && configured.kind
                                == crate::fleet_ensure::model::DesiredCanisterKind::Pool
                    })?;
                    let temporary_create = desired
                        .bootstrap
                        .as_ref()
                        .is_some_and(|bootstrap| bootstrap.fresh_estate)
                        && configured.principal.is_none()
                        && desired.bootstrap.as_ref().is_some_and(|bootstrap| {
                            configured.controllers
                                == bootstrap
                                    .recovery_controllers
                                    .iter()
                                    .map(candid::Principal::to_text)
                                    .collect::<Vec<_>>()
                        })
                        && configured.controller_canisters.len() == 1
                        && canister_plan.actions.iter().any(|action| {
                            matches!(
                                action,
                                EnsureAction::Create {
                                    controller_canisters,
                                    controllers,
                                    ..
                                } if controller_canisters == &configured.controller_canisters
                                    && controllers == &{
                                        let mut expected = configured.controllers.clone();
                                        expected.push(desired.operator.clone());
                                        expected.sort();
                                        expected
                                    }
                            )
                        });
                    temporary_create.then_some(canister_plan.name.as_str())
                })
                .collect::<BTreeSet<_>>()
        })
        .unwrap_or_default();
    actions.sort_by_key(|action| {
        if matches!(action, EnsureAction::SetControllers { name, .. } if temporary_pool_observation_finalizations.contains(name.as_str()))
        {
            6
        } else {
            action_order(action)
        }
    });
    actions
}

fn deferred_create_observation_is_exact(
    desired: &crate::fleet_ensure::model::DesiredFleet,
    actions: &[&EnsureAction],
    index: usize,
    action: &EnsureAction,
    record: &EffectRecord,
    state: &FleetEnsureStateRecord,
) -> bool {
    let EnsureAction::Create {
        controller_canisters,
        controllers,
        name,
        ..
    } = action
    else {
        return false;
    };
    let Some(configured) = desired.canisters.iter().find(|canister| {
        canister.name == *name
            && canister.kind == crate::fleet_ensure::model::DesiredCanisterKind::Pool
    }) else {
        return false;
    };
    let Some(parent) = configured.parent.as_deref() else {
        return false;
    };
    let fresh_root_only_create = desired
        .bootstrap
        .as_ref()
        .is_some_and(|bootstrap| bootstrap.fresh_estate)
        && configured.principal.is_none()
        && configured.controllers.is_empty()
        && configured.controller_canisters.as_slice() == [parent]
        && controller_canisters == &configured.controller_canisters
        && controllers.is_empty();
    let exact_issued_create = record.state == EffectState::Issued
        && create_identity_is_exact(name, record, state)
        && record.post_cycles.is_some();
    let later_root_install = actions[index.saturating_add(1)..].iter().any(|candidate| {
        matches!(
            candidate,
            EnsureAction::Install {
                canic_init: Some(crate::fleet_ensure::model::DesiredCanisterInit::Root { root }),
                name: install_name,
                ..
            } if root == parent && install_name == parent
        )
    });
    let infrastructure_only = actions.iter().all(|candidate| {
        matches!(
            candidate,
            EnsureAction::Create { .. }
                | EnsureAction::FundEstate { .. }
                | EnsureAction::Install { .. }
                | EnsureAction::SetControllers { .. }
        )
    });
    fresh_root_only_create && exact_issued_create && later_root_install && infrastructure_only
}

pub(super) const fn action_order(action: &EnsureAction) -> u8 {
    match action {
        EnsureAction::Create { .. } => 0,
        EnsureAction::Fund { .. } | EnsureAction::FundEstate { .. } => 1,
        EnsureAction::Install {
            canic_init: Some(crate::fleet_ensure::model::DesiredCanisterInit::Coordinator),
            ..
        } => 2,
        EnsureAction::SetControllers { .. } => 3,
        EnsureAction::Install {
            canic_init: Some(crate::fleet_ensure::model::DesiredCanisterInit::Store { .. }),
            ..
        }
        | EnsureAction::DeleteSnapshot { .. } => 4,
        EnsureAction::Uninstall { .. } | EnsureAction::Install { .. } => 5,
        EnsureAction::Start { .. } => 6,
        EnsureAction::FleetProtocol { .. } | EnsureAction::Protocol { .. } => 7,
        EnsureAction::Transfer { .. } => 8,
        EnsureAction::Stop { .. } => 9,
        EnsureAction::Delete { .. } => 10,
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn store_binding_requires_exact_retained_root_or_reviewed_root_install() {
        use crate::fleet_ensure::{
            model::{CanisterDisposition, CanisterPlan, DesiredCanisterInit, InstallMode},
            tests::MockPlatform,
        };
        let (state, _) = retained_evidence();
        let mut plan = estate_funding_plan();
        let desired = serde_json::from_value(serde_json::json!({
            "canisters": [], "cycles_ledger": "um5iw-rqaaa-aaaaq-qaaba-cai",
            "environment": "local", "fleet": "fleet", "ledger_fee_cycles": "0",
            "management_creation_fee_cycles": "0", "material_cycle_threshold": "1",
            "maximum_observation_burn_cycles": "1", "maximum_stalled_observations": 2,
            "maximum_update_burn_cycles": "2", "operator": "rrkah-fqaaa-aaaaa-aaaaq-cai",
            "schema_version": FLEET_ENSURE_SCHEMA_VERSION, "treasury": "root"
        }))
        .unwrap();
        let mut platform = MockPlatform::new(desired, []);
        platform.rejected_activation_root = Some("root".into());
        let store = EnsureAction::Install {
            canic_init: Some(DesiredCanisterInit::Store {
                root: "root".into(),
            }),
            reinstall_witness: None,
            init_arg: None,
            init_arg_sha256: None,
            init_candid: None,
            init_candid_sha256: None,
            mode: InstallMode::Install,
            name: "store".into(),
            principal: "store-principal".into(),
            wasm: "store.wasm".into(),
            wasm_sha256: "store-hash".into(),
        };
        assert!(matches!(
            verify_store_activation_binding(&plan, &store, &state, &mut platform),
            Err(EnsureWorkflowError::Platform(_))
        ));
        assert_eq!(
            platform.retained_activation_checks,
            [(plan.operation_id.clone(), "root".into())]
        );
        let mut root_install = store.clone();
        if let EnsureAction::Install {
            name, canic_init, ..
        } = &mut root_install
        {
            *name = "other-root".into();
            *canic_init = Some(DesiredCanisterInit::Root {
                root: "other-root".into(),
            });
        }
        plan.canisters.push(CanisterPlan {
            actions: vec![root_install],
            disposition: CanisterDisposition::Create,
            name: "root".into(),
            observed_cycles: 0,
            principal: None,
        });
        assert!(verify_store_activation_binding(&plan, &store, &state, &mut platform).is_err());
        if let EnsureAction::Install {
            name, canic_init, ..
        } = &mut plan.canisters.last_mut().unwrap().actions[0]
        {
            *name = "root".into();
            *canic_init = Some(DesiredCanisterInit::Root {
                root: "root".into(),
            });
        }
        let reads = platform.retained_activation_checks.len();
        verify_store_activation_binding(&plan, &store, &state, &mut platform).unwrap();
        assert_eq!(platform.retained_activation_checks.len(), reads);
        plan.canisters.pop();
        platform.rejected_activation_root = None;
        verify_store_activation_binding(&plan, &store, &state, &mut platform).unwrap();
        assert_eq!(platform.retained_activation_checks.len(), reads + 1);
    }

    #[test]
    fn root_prerequisite_retains_activation_authority_across_desired_refresh() {
        let (mut state, mut journal) = retained_evidence();
        let mut plan = estate_funding_plan();
        plan.scope = FleetEnsurePlanScope::RootReinstallPrerequisite;
        plan.conservation.estate_funding_domains.clear();
        journal.effects.clear();
        journal.completion = FleetEnsureCompletion::Converged;
        let before = state.clone();
        assert!(matches!(
            reconcile_retained_planning_state::<std::io::Error>(
                &mut state,
                Some(&plan),
                Some(&journal),
                "fleet",
                "local",
                "refreshed-imports"
            ),
            Err(EnsureWorkflowError::RootActivationDesiredConflict { .. })
        ));
        assert_eq!(state, before);
        assert!(
            reconcile_retained_planning_state::<std::io::Error>(
                &mut state,
                Some(&plan),
                Some(&journal),
                "fleet",
                "local",
                "desired"
            )
            .unwrap()
            .is_some()
        );
    }

    fn terminal_component_entry(canister: &str, parent: &str) -> crate::registry::RegistryEntry {
        crate::registry::RegistryEntry {
            pid: canister.to_string(),
            role: Some("managed_component".to_string()),
            parent_pid: Some(parent.to_string()),
            module_hash: Some("11".repeat(32)),
            protocol_binding: Some(crate::protocol_binding::RegistryProtocolBinding {
                release_identity: "0.110.test".to_string(),
                role: canic_contracts::ids::CanisterRole::from("managed_component"),
                capabilities: BTreeSet::new(),
                candid_sha256: [1; 32],
                protocol_profile_digest:
                    canic_core::role_contract::ProtocolProfileDigest::from_bytes([2; 32]),
            }),
        }
    }

    pub(super) fn retained_evidence() -> (FleetEnsureStateRecord, FleetEnsureJournalRecord) {
        let state = FleetEnsureStateRecord {
            active_registry: None,
            completed_reinstall_action_sha256: BTreeMap::from([(
                "root".to_string(),
                "action".to_string(),
            )]),
            completed_reinstall_operation_id: Some("operation".to_string()),
            completed_reinstalls: BTreeMap::from([("root".to_string(), 7)]),
            fleet: "fleet".to_string(),
            pending_principals: BTreeMap::new(),
            principals: BTreeMap::new(),
            retained_cycles_by_principal: BTreeMap::new(),
            schema_version: crate::fleet_ensure::model::FLEET_ENSURE_SCHEMA_VERSION,
            topology: BTreeMap::new(),
        };
        let journal = FleetEnsureJournalRecord {
            release: None,
            bootstrap_registration_recovery: None,
            funding_observations: BTreeMap::new(),
            funding_reviews: Vec::new(),
            successor_phases: Vec::new(),
            completion: FleetEnsureCompletion::ReplanRequired,
            estate_funding_required: None,
            effects: vec![crate::fleet_ensure::model::EffectRecord {
                publication_attempts: 0,
                maintenance_attempts: 0,
                action_sha256: "action".to_string(),
                created_principal: None,
                destination_post_cycles: None,
                destination_pre_cycles: None,
                post_cycles: None,
                pre_cycles: None,
                pre_canister_version: Some(7),
                progress_identity: None,
                receipt: None,
                state: EffectState::Applied,
            }],
            fleet: "fleet".to_string(),
            initial_controlled_cycles: 0,
            initial_estate_funding_cycles_by_root: BTreeMap::new(),
            initial_operator_cycles: 0,
            operation_id: "operation".to_string(),
            plan_sha256: "plan".to_string(),
            schema_version: crate::fleet_ensure::model::FLEET_ENSURE_SCHEMA_VERSION,
            stalled_observations: 0,
        };
        (state, journal)
    }

    pub(super) fn estate_funding_plan() -> FleetEnsurePlan {
        FleetEnsurePlan {
            continuation: None,
            canisters: Vec::new(),
            conservation: CycleConservation {
                estate_funding_domains: vec![crate::fleet_ensure::model::EstateFundingDomainPlan {
                    allocated_workloads: 0,
                    available_cycles: Some(40),
                    available_pool_slots: 2,
                    creation_amount_cycles: 45,
                    creation_execution_margin_cycles: 5,
                    readiness_floor_cycles: 35,
                    cycles_ledger: "ledger".to_string(),
                    eligible_ready_pool_assets: 0,
                    initial_pool_assets: Vec::new(),
                    ledger_fee_cycles: 5,
                    management_creation_fee_cycles: 5,
                    maximum_creation_debit_cycles: 100,
                    maximum_creation_fee_cycles: 20,
                    maximum_funding_cycles: 60,
                    occupied_pool_assets: 0,
                    pending_creation_count: 0,
                    pending_creation: None,
                    planned_initial_workloads: 2,
                    pool_maximum_size: 2,
                    required_creation_count: 2,
                    root: "root".to_string(),
                    root_principal: Some("rrkah-fqaaa-aaaaa-aaaaq-cai".to_string()),
                    shortfall_cycles: 60,
                }],
                expected_post_operation_cycles: 80,
                maximum_execution_burn_cycles: 0,
                maximum_new_funding_cycles: 0,
                maximum_operator_debit_cycles: 0,
                maximum_unavoidable_fee_cycles: 0,
                observed_controlled_cycles: 40,
                retained_in_reused_canisters_cycles: 0,
                scheduled_transfer_cycles: 0,
            },
            desired_sha256: "desired".to_string(),
            environment: "local".to_string(),
            fleet: "fleet".to_string(),
            operation_id: "operation".to_string(),
            plan_sha256: "plan".to_string(),
            planned_at_time: 1,
            protocol_actions: Vec::new(),
            recovery_review: None,
            reinstall: None,
            infrastructure_bootstrap: None,
            root_reinstall_bindings: Vec::new(),
            root_start_authority: None,
            reviewed_desired: None,
            schema_version: FLEET_ENSURE_SCHEMA_VERSION,
            scope: FleetEnsurePlanScope::Full,
            terminal_inventory_operation_id: None,
        }
    }

    pub(super) fn estate_funding_observation(balance_cycles: Option<u128>) -> FleetObservation {
        FleetObservation {
            additional_controlled_cycles: BTreeMap::new(),
            canisters: BTreeMap::new(),
            estate_funding_domains: BTreeMap::from([(
                "root".to_string(),
                crate::fleet_ensure::model::EstateFundingDomainObservation {
                    balance_cycles,
                    cycles_ledger: "ledger".to_string(),
                    pool: None,
                    root_principal: Some("rrkah-fqaaa-aaaaa-aaaaq-cai".to_string()),
                },
            )]),
            ledger_fee_cycles: 0,
            operator_cycles: 0,
            protocol_ready: BTreeMap::new(),
        }
    }

    fn terminal_estate_creation_observation() -> FleetObservation {
        let mut observation = estate_funding_observation(Some(0));
        let receipt =
            |byte, block_index| crate::fleet_ensure::model::EstatePoolCreationReceiptObservation {
                block_index,
                operation_id: format!("{byte:02x}").repeat(32),
                cycles_ledger: "ledger".to_string(),
                ledger_amount_cycles: 45,
                ledger_fee_cycles: 5,
                readiness_floor_cycles: 35,
                creation_execution_margin_cycles: 5,
                management_creation_fee_cycles: 5,
                first_observed_cycles: Some(39),
            };
        observation
            .estate_funding_domains
            .get_mut("root")
            .expect("Root funding domain")
            .pool = Some(crate::fleet_ensure::model::EstatePoolInventoryObservation {
            assets: vec![
                crate::fleet_ensure::model::EstatePoolAssetObservation {
                    creation_receipt: Some(receipt(1, 10)),
                    cycles: 35,
                    lifecycle: crate::fleet_ensure::model::EstatePoolAssetLifecycle::Ready,
                    origin: crate::fleet_ensure::model::EstatePoolAssetOrigin::Created,
                    principal: "created-one".to_string(),
                },
                crate::fleet_ensure::model::EstatePoolAssetObservation {
                    creation_receipt: Some(receipt(2, 11)),
                    cycles: 35,
                    lifecycle: crate::fleet_ensure::model::EstatePoolAssetLifecycle::Workload,
                    origin: crate::fleet_ensure::model::EstatePoolAssetOrigin::Created,
                    principal: "created-two".to_string(),
                },
            ],
            maximum_size: 2,
            minimum_size: 0,
            pending_creation: None,
            readiness_floor_cycles: 35,
            creation_execution_margin_cycles: 5,
        });
        observation
    }

    fn estate_creation_receipt(
        observation: &mut FleetObservation,
        asset_index: usize,
    ) -> &mut crate::fleet_ensure::model::EstatePoolCreationReceiptObservation {
        observation
            .estate_funding_domains
            .get_mut("root")
            .expect("Root domain")
            .pool
            .as_mut()
            .expect("pool")
            .assets[asset_index]
            .creation_receipt
            .as_mut()
            .expect("receipt")
    }

    #[test]
    fn autonomous_reserve_creation_reconciles_without_host_provisioning() {
        let mut plan = estate_funding_plan();
        let domain = &mut plan.conservation.estate_funding_domains[0];
        domain.available_cycles = Some(100);
        domain.maximum_funding_cycles = 0;
        domain.shortfall_cycles = 0;
        let (state, mut journal) = retained_evidence();
        journal.effects.clear();
        journal
            .initial_estate_funding_cycles_by_root
            .insert("root".to_string(), 100);
        let terminal = terminal_estate_creation_observation();
        assert_eq!(
            reconcile_estate_funding::<std::io::Error>(&plan, &journal, &state, &terminal)
                .expect("reconcile exact Root-owned creation against the reviewed reserve"),
            (0, 20)
        );
    }

    #[test]
    fn terminal_estate_creation_uses_exact_protected_receipts() {
        let plan = estate_funding_plan();
        let domain = &plan.conservation.estate_funding_domains[0];
        let (state, _) = retained_evidence();
        let exact = terminal_estate_creation_observation();
        assert_eq!(
            exact_estate_creation_costs::<std::io::Error>(&state, &exact, domain)
                .expect("reconcile exact creation receipts"),
            (100, 20)
        );
        for required_creation_count in [0, 1] {
            let mut insufficient = domain.clone();
            insufficient.required_creation_count = required_creation_count;
            assert!(matches!(
                exact_estate_creation_costs::<std::io::Error>(&state, &exact, &insufficient),
                Err(EnsureWorkflowError::JournalIntegrity)
            ));
        }

        let mut drifted = Vec::new();
        let mut changed = exact.clone();
        estate_creation_receipt(&mut changed, 0).ledger_amount_cycles = 44;
        drifted.push(changed);
        let mut changed = exact.clone();
        estate_creation_receipt(&mut changed, 1).operation_id = "01".repeat(32);
        drifted.push(changed);
        let mut changed = exact.clone();
        estate_creation_receipt(&mut changed, 1).block_index = 10;
        drifted.push(changed);
        let mut changed = exact.clone();
        estate_creation_receipt(&mut changed, 0).first_observed_cycles = None;
        drifted.push(changed);
        let mut changed = exact.clone();
        estate_creation_receipt(&mut changed, 0).first_observed_cycles = Some(34);
        drifted.push(changed);
        let mut donated = exact.clone();
        estate_creation_receipt(&mut donated, 0).first_observed_cycles = Some(41);
        assert_eq!(
            exact_estate_creation_costs::<std::io::Error>(&state, &donated, domain)
                .expect("external donation leaves exact creation cost unchanged"),
            (100, 20)
        );
        let mut changed = exact;
        changed
            .estate_funding_domains
            .get_mut("root")
            .expect("Root domain")
            .pool
            .as_mut()
            .expect("pool")
            .assets[0]
            .origin = crate::fleet_ensure::model::EstatePoolAssetOrigin::Imported;
        drifted.push(changed);
        for observation in drifted {
            assert!(matches!(
                exact_estate_creation_costs::<std::io::Error>(&state, &observation, domain,),
                Err(EnsureWorkflowError::JournalIntegrity)
            ));
        }
    }

    #[test]
    fn estate_funding_pause_is_exact_and_clears_only_after_sufficient_balance() {
        let plan = estate_funding_plan();
        let (state, mut journal) = retained_evidence();
        let required = estate_funding_requirement::<std::io::Error>(
            &plan,
            &state,
            &estate_funding_observation(Some(40)),
        )
        .expect("validate funding observation")
        .expect("underfunded estate pauses");
        assert_eq!(required.available_cycles, 40);
        assert_eq!(required.maximum_creation_debit_cycles, 100);
        assert_eq!(required.shortfall_cycles, 60);
        assert!(
            estate_funding_requirement::<std::io::Error>(
                &plan,
                &state,
                &estate_funding_observation(Some(100)),
            )
            .expect("validate sufficient funding observation")
            .is_none()
        );

        journal.completion = FleetEnsureCompletion::InProgress;
        journal.effects.clear();
        journal.estate_funding_required = Some(required);
        journal.initial_estate_funding_cycles_by_root = BTreeMap::from([("root".to_string(), 40)]);
        assert!(initial_estate_funding_is_exact(&journal, &plan));
        assert!(estate_funding_pause_is_exact(&journal, &plan, &state));

        journal
            .estate_funding_required
            .as_mut()
            .expect("funding pause")
            .shortfall_cycles = 59;
        assert!(!estate_funding_pause_is_exact(&journal, &plan, &state));

        *journal
            .initial_estate_funding_cycles_by_root
            .get_mut("root")
            .expect("initial Root funding balance") = 41;
        assert!(!initial_estate_funding_is_exact(&journal, &plan));
    }

    #[test]
    fn current_creation_funding_pause_is_bound_to_exact_runtime_evidence() {
        let mut plan = estate_funding_plan();
        plan.conservation.estate_funding_domains[0].cycles_ledger = "aaaaa-aa".to_string();
        let (state, mut journal) = retained_evidence();
        let funding = canic_contracts::dto::component_provisioning::RootEstateFundingRequired {
            available: Cycles::new(40),
            attempt_count: 2,
            creation_amount: Cycles::new(45),
            cycles_ledger: candid::Principal::from_text("aaaaa-aa").expect("Ledger Principal"),
            execution_margin: Cycles::new(5),
            last_attempt_at_ns: Some(10),
            ledger_fee: Cycles::new(5),
            management_creation_fee: Cycles::new(5),
            operation_id: [17; 32],
            readiness_floor: Cycles::new(35),
            required: Cycles::new(50),
            retry_at_ns: 20,
            root: candid::Principal::from_text("rrkah-fqaaa-aaaaa-aaaaq-cai")
                .expect("Root Principal"),
            shortfall: Cycles::new(10),
        };
        let required =
            protocol_estate_funding_requirement::<std::io::Error>(&plan, &state, &funding)
                .expect("bind runtime funding evidence");
        journal.completion = FleetEnsureCompletion::InProgress;
        journal.effects.clear();
        journal.estate_funding_required = Some(required);
        assert!(estate_funding_pause_is_exact(&journal, &plan, &state));

        journal
            .estate_funding_required
            .as_mut()
            .expect("funding pause")
            .pending_creation_operation_id = Some("AA".repeat(32));
        assert!(!estate_funding_pause_is_exact(&journal, &plan, &state));

        let mut drifted = funding;
        drifted.execution_margin = Cycles::new(6);
        assert!(matches!(
            protocol_estate_funding_requirement::<std::io::Error>(&plan, &state, &drifted),
            Err(EnsureWorkflowError::PlanIntegrity)
        ));
    }

    #[test]
    fn estate_funding_observation_rejects_missing_or_mismatched_authority() {
        let plan = estate_funding_plan();
        let (state, _) = retained_evidence();
        let mut missing_balance = estate_funding_observation(None);
        assert!(matches!(
            estate_funding_requirement::<std::io::Error>(&plan, &state, &missing_balance),
            Err(EnsureWorkflowError::PlanIntegrity)
        ));

        let observed = missing_balance
            .estate_funding_domains
            .get_mut("root")
            .expect("Root funding domain");
        observed.balance_cycles = Some(40);
        observed.root_principal = Some("aaaaa-aa".to_string());
        assert!(matches!(
            estate_funding_requirement::<std::io::Error>(&plan, &state, &missing_balance),
            Err(EnsureWorkflowError::PlanIntegrity)
        ));

        for changed_balance in [0, 39] {
            assert!(matches!(
                estate_funding_requirement::<std::io::Error>(
                    &plan,
                    &state,
                    &estate_funding_observation(Some(changed_balance)),
                ),
                Err(EnsureWorkflowError::DriftedBeforeApply)
            ));
        }
    }

    #[test]
    fn estate_funding_surplus_preserves_reviewed_creation_limits() {
        let plan = estate_funding_plan();
        let original = plan.clone();
        let (state, _) = retained_evidence();
        assert!(
            estate_funding_requirement::<std::io::Error>(
                &plan,
                &state,
                &estate_funding_observation(Some(150)),
            )
            .unwrap()
            .is_none()
        );
        assert_eq!(plan, original);
        let mut extra_creation = terminal_estate_creation_observation();
        let pool = extra_creation
            .estate_funding_domains
            .get_mut("root")
            .unwrap()
            .pool
            .as_mut()
            .unwrap();
        let mut extra = pool.assets[0].clone();
        extra.principal = "unreviewed-third-creation".into();
        pool.assets.push(extra);
        assert!(matches!(
            exact_estate_creation_costs::<std::io::Error>(
                &state,
                &extra_creation,
                &plan.conservation.estate_funding_domains[0]
            ),
            Err(EnsureWorkflowError::JournalIntegrity)
        ));
    }

    #[test]
    fn non_creation_protocol_completion_does_not_consume_estate_creation_authority() {
        let mut plan = estate_funding_plan();
        plan.conservation.estate_funding_domains[0].maximum_creation_debit_cycles = 0;
        plan.conservation.estate_funding_domains[0].maximum_creation_fee_cycles = 0;
        plan.conservation.estate_funding_domains[0].maximum_funding_cycles = 0;
        plan.conservation.estate_funding_domains[0].required_creation_count = 0;
        plan.conservation.estate_funding_domains[0].shortfall_cycles = 0;
        let action = EnsureAction::FleetProtocol {
            action: Box::new(
                crate::fleet_ensure::model::CurrentFleetProtocolAction::PublishStoreChunk {
                    request: canic_contracts::dto::template::TemplateChunkInput {
                        preparation: None,
                        template_id: canic_contracts::ids::TemplateId::from("root"),
                        version: canic_contracts::ids::TemplateVersion::from("current"),
                        chunk_index: 0,
                        bytes: vec![2],
                    },
                },
            ),
            candid: "store.did".to_string(),
            candid_sha256: "11".repeat(32),
            maximum_execution_burn_cycles: 0,
            name: "publish-store-chunk".to_string(),
            principal: "rrkah-fqaaa-aaaaa-aaaaq-cai".to_string(),
        };
        let action_hash = action_sha256(&action);
        plan.protocol_actions.push(action);
        let (state, mut journal) = retained_evidence();
        journal.initial_estate_funding_cycles_by_root = BTreeMap::from([("root".to_string(), 40)]);
        journal.effects = vec![EffectRecord {
            publication_attempts: 0,
            maintenance_attempts: 0,
            action_sha256: action_hash,
            created_principal: None,
            destination_post_cycles: None,
            destination_pre_cycles: None,
            post_cycles: None,
            pre_cycles: None,
            pre_canister_version: None,
            progress_identity: Some("store chunks prepared".to_string()),
            receipt: Some("receipt".to_string()),
            state: EffectState::Applied,
        }];

        assert_eq!(
            reconcile_estate_funding::<std::io::Error>(
                &plan,
                &journal,
                &state,
                &estate_funding_observation(Some(40)),
            )
            .expect("non-creation protocol preserves the Root Ledger balance"),
            (0, 0),
        );
        journal.initial_controlled_cycles = 40;
        journal.initial_operator_cycles = 0;
        let original = plan.clone();
        for balance in [40, 41, 150] {
            let observed = estate_funding_observation(Some(balance));
            assert!(
                estate_funding_requirement::<std::io::Error>(&plan, &state, &observed)
                    .unwrap()
                    .is_none()
            );
            let actual =
                verify_terminal_conservation::<std::io::Error>(&plan, &journal, &state, &observed)
                    .unwrap();
            assert_eq!(actual.final_controlled_cycles, balance);
            assert_eq!(actual.observed_net_cycle_credit_cycles, balance - 40);
            assert_eq!(actual.estate_funding_cycles, 0);
            assert_eq!(actual.operator_debit_cycles, 0);
            assert_eq!(actual.exact_estate_creation_fee_cycles, 0);
        }
        assert_eq!(plan, original);
        assert!(matches!(
            reconcile_estate_funding::<std::io::Error>(
                &plan,
                &journal,
                &state,
                &estate_funding_observation(Some(39))
            ),
            Err(EnsureWorkflowError::Conservation(_))
        ));
    }

    #[test]
    fn fresh_estate_funding_binds_the_created_root_from_durable_state() {
        let mut plan = estate_funding_plan();
        plan.conservation.estate_funding_domains[0].available_cycles = None;
        plan.conservation.estate_funding_domains[0].maximum_funding_cycles = 100;
        plan.conservation.estate_funding_domains[0].root_principal = None;
        let (mut state, _) = retained_evidence();
        state.pending_principals.insert(
            "root".to_string(),
            "rrkah-fqaaa-aaaaa-aaaaq-cai".to_string(),
        );

        let required = estate_funding_requirement::<std::io::Error>(
            &plan,
            &state,
            &estate_funding_observation(Some(0)),
        )
        .expect("validate created Root funding authority")
        .expect("fresh underfunded estate pauses");
        assert_eq!(required.shortfall_cycles, 100);
        assert_eq!(required.root_principal, "rrkah-fqaaa-aaaaa-aaaaq-cai");

        state
            .pending_principals
            .insert("root".to_string(), "aaaaa-aa".to_string());
        assert!(matches!(
            estate_funding_requirement::<std::io::Error>(
                &plan,
                &state,
                &estate_funding_observation(Some(0)),
            ),
            Err(EnsureWorkflowError::PlanIntegrity)
        ));
    }

    #[test]
    fn unchanged_estate_funding_pause_does_not_rewrite_its_journal() {
        let root = crate::test_support::temp_dir("estate-funding-pause");
        let paths = EnsurePaths::under(&root, "local", "fleet");
        let plan = estate_funding_plan();
        let (state, mut journal) = retained_evidence();
        journal.completion = FleetEnsureCompletion::InProgress;
        journal.effects.clear();
        journal.initial_estate_funding_cycles_by_root = BTreeMap::from([("root".to_string(), 40)]);
        let required = estate_funding_requirement::<std::io::Error>(
            &plan,
            &state,
            &estate_funding_observation(Some(40)),
        )
        .expect("validate funding observation")
        .expect("underfunded estate pauses");

        assert!(
            retain_estate_funding_pause::<std::io::Error>(&paths, &mut journal, Some(&required),)
                .expect("retain first funding pause")
        );
        let retained = std::fs::read(&paths.journal).expect("read first retained pause");
        assert!(
            !retain_estate_funding_pause::<std::io::Error>(&paths, &mut journal, Some(&required),)
                .expect("replay unchanged funding pause")
        );
        assert_eq!(
            std::fs::read(&paths.journal).expect("reread unchanged retained pause"),
            retained
        );
        assert!(
            retain_estate_funding_pause::<std::io::Error>(&paths, &mut journal, None)
                .expect("clear funded pause")
        );
        assert!(journal.estate_funding_required.is_none());

        std::fs::remove_dir_all(root).expect("remove funding-pause fixture");
    }

    #[test]
    fn estate_funding_is_rechecked_until_the_first_protocol_effect_starts() {
        let protocol_action = || EnsureAction::FleetProtocol {
            action: Box::new(
                crate::fleet_ensure::model::CurrentFleetProtocolAction::PublishStoreChunk {
                    request: canic_contracts::dto::template::TemplateChunkInput {
                        preparation: None,
                        template_id: canic_contracts::ids::TemplateId::from("root"),
                        version: canic_contracts::ids::TemplateVersion::from("current"),
                        chunk_index: 0,
                        bytes: vec![2],
                    },
                },
            ),
            candid: "store.did".to_string(),
            candid_sha256: "11".repeat(32),
            maximum_execution_burn_cycles: 1,
            name: "prepare".to_string(),
            principal: "rrkah-fqaaa-aaaaa-aaaaq-cai".to_string(),
        };
        let first = protocol_action();
        let second = protocol_action();
        let actions = [&first, &second];
        let (_, mut journal) = retained_evidence();
        journal.effects = vec![EffectRecord {
            publication_attempts: 0,
            maintenance_attempts: 0,
            action_sha256: action_sha256(&first),
            created_principal: None,
            destination_post_cycles: None,
            destination_pre_cycles: None,
            post_cycles: None,
            pre_cycles: None,
            pre_canister_version: None,
            progress_identity: None,
            receipt: None,
            state: EffectState::Intent,
        }];
        assert!(!prior_fleet_protocol_effect_started(&actions, &journal, 1));

        journal.effects[0].state = EffectState::Issued;
        assert!(prior_fleet_protocol_effect_started(&actions, &journal, 1));
    }

    #[test]
    fn configured_projection_publishes_unique_bindings_without_losing_dynamic_cycles() {
        let asset = "rrkah-fqaaa-aaaaa-aaaaq-cai";
        let dynamic = "ryjl3-tyaaa-aaaaa-aaaba-cai";
        for kind in ["pool", "component"] {
            let desired = serde_json::from_value(serde_json::json!({
                "canisters": [{
                    "controllers": [], "initial_cycles": "0", "kind": kind,
                    "minimum_cycles": "0", "name": "configured", "presence": "present",
                    "principal": asset, "replace": false, "subnet": "subnet"
                }],
                "cycles_ledger": "um5iw-rqaaa-aaaaq-qaaba-cai",
                "environment": "local", "fleet": "fleet", "ledger_fee_cycles": "0",
                "management_creation_fee_cycles": "0", "material_cycle_threshold": "1",
                "maximum_observation_burn_cycles": "1", "maximum_stalled_observations": 2,
                "maximum_update_burn_cycles": "2", "operator": asset,
                "schema_version": FLEET_ENSURE_SCHEMA_VERSION, "treasury": "configured"
            }))
            .expect("current desired projection");
            let (mut state, mut journal) = retained_evidence();
            journal.effects.clear();
            state.principals = BTreeMap::from([
                ("configured".to_string(), asset.to_string()),
                (format!("observed:{asset}"), asset.to_string()),
                (format!("observed:{dynamic}"), dynamic.to_string()),
            ]);
            state.retained_cycles_by_principal =
                BTreeMap::from([(asset.to_string(), 100), (dynamic.to_string(), 60)]);
            let cycles = state.retained_cycles_by_principal.clone();
            let journal_before = serde_json::to_vec(&journal).expect("journal before projection");
            let plan = estate_funding_plan();

            publish_terminal_state(&desired, &plan, &journal, &mut state)
                .expect("publish current configured topology");
            assert_eq!(
                state.principals,
                BTreeMap::from([
                    ("configured".to_string(), asset.to_string()),
                    (format!("observed:{dynamic}"), dynamic.to_string()),
                ])
            );
            assert_eq!(state.retained_cycles_by_principal, cycles);
            let published = state.clone();
            publish_terminal_state(&desired, &plan, &journal, &mut state)
                .expect("repeat projection");
            assert_eq!(state, published);
            assert_eq!(
                serde_json::to_vec(&journal).expect("journal after projection"),
                journal_before
            );
        }
    }

    #[test]
    fn configured_projection_preserves_duplicate_authority_rejection() {
        let asset = "rrkah-fqaaa-aaaaa-aaaaq-cai";
        for configured in [false, true] {
            let (mut state, _) = retained_evidence();
            state.principals = BTreeMap::from([
                ("first".to_string(), asset.to_string()),
                ("second".to_string(), asset.to_string()),
            ]);
            if configured {
                let topology = crate::fleet_ensure::model::FleetEnsureTopologyRecord {
                    kind: crate::fleet_ensure::model::DesiredCanisterKind::Pool,
                    module_hash: None,
                    parent: None,
                    protocol_binding: None,
                    role: None,
                };
                state.topology = BTreeMap::from([
                    ("first".to_string(), topology.clone()),
                    ("second".to_string(), topology),
                ]);
            }
            let before = state.clone();
            retain_configured_principal_bindings(&mut state);
            assert_eq!(state, before);
            for entries in [Vec::new(), vec![terminal_component_entry(asset, "parent")]] {
                assert!(matches!(
                    merge_terminal_inventory::<std::io::Error>(
                        &mut state,
                        crate::fleet_ensure::ops::TerminalFleetInventory {
                            entries,
                            ..Default::default()
                        }
                    ),
                    Err(EnsureWorkflowError::TerminalInventory(
                        TerminalInventoryError::DuplicateRetainedPrincipal
                    ))
                ));
                assert_eq!(state, before);
            }
        }
    }

    #[test]
    fn terminal_inventory_allows_only_verified_pool_to_component_parent_transition() {
        let asset = "rrkah-fqaaa-aaaaa-aaaaq-cai";
        let old_root = "r7inp-6aaaa-aaaaa-aaabq-cai";
        let hub = "ryjl3-tyaaa-aaaaa-aaaba-cai";
        let state = || {
            let (mut state, _) = retained_evidence();
            state.principals = BTreeMap::from([
                ("asset".to_string(), asset.to_string()),
                ("hub".to_string(), hub.to_string()),
                ("root".to_string(), old_root.to_string()),
            ]);
            state.topology.insert(
                "asset".to_string(),
                crate::fleet_ensure::model::FleetEnsureTopologyRecord {
                    kind: crate::fleet_ensure::model::DesiredCanisterKind::Pool,
                    module_hash: None,
                    parent: Some("root".to_string()),
                    protocol_binding: None,
                    role: Some("canister_pool_asset".to_string()),
                },
            );
            state
        };

        let mut promoted = state();
        merge_terminal_inventory::<std::io::Error>(
            &mut promoted,
            crate::fleet_ensure::ops::TerminalFleetInventory {
                active_registry: None,
                controlled_cycles_by_principal: BTreeMap::new(),
                entries: vec![terminal_component_entry(asset, hub)],
            },
        )
        .expect("verified pool workload assumes its terminal Component parent");
        let promoted = promoted.topology.get("asset").expect("promoted asset");
        assert_eq!(
            promoted.kind,
            crate::fleet_ensure::model::DesiredCanisterKind::Component
        );
        assert_eq!(promoted.parent.as_deref(), Some("hub"));

        let mut already_component = state();
        already_component
            .topology
            .get_mut("asset")
            .expect("retained asset")
            .kind = crate::fleet_ensure::model::DesiredCanisterKind::Component;
        assert!(matches!(
            merge_terminal_inventory::<std::io::Error>(
                &mut already_component,
                crate::fleet_ensure::ops::TerminalFleetInventory {
                    active_registry: None,
                    controlled_cycles_by_principal: BTreeMap::new(),
                    entries: vec![terminal_component_entry(asset, hub)],
                },
            ),
            Err(EnsureWorkflowError::TerminalInventory(
                TerminalInventoryError::RetainedParentConflict {
                    canister,
                    retained_parent,
                    terminal_parent,
                }
            )) if canister == asset
                && retained_parent == "root"
                && terminal_parent == "hub"
        ));

        for incomplete_entry in [
            crate::registry::RegistryEntry {
                module_hash: None,
                ..terminal_component_entry(asset, hub)
            },
            crate::registry::RegistryEntry {
                protocol_binding: None,
                ..terminal_component_entry(asset, hub)
            },
        ] {
            assert!(matches!(
                merge_terminal_inventory::<std::io::Error>(
                    &mut state(),
                    crate::fleet_ensure::ops::TerminalFleetInventory {
                        active_registry: None,
                        controlled_cycles_by_principal: BTreeMap::new(),
                        entries: vec![incomplete_entry],
                    },
                ),
                Err(EnsureWorkflowError::TerminalInventory(
                    TerminalInventoryError::RetainedParentConflict {
                        canister,
                        retained_parent,
                        terminal_parent,
                    }
                )) if canister == asset
                    && retained_parent == "root"
                    && terminal_parent == "hub"
            ));
        }
    }

    #[test]
    fn reinstall_recovery_review_requires_complete_exact_install_evidence() {
        use crate::fleet_ensure::model::{CanisterPlan, FleetReinstallRecord, InstallMode};
        let (state, mut journal) = retained_evidence();
        let mut plan = estate_funding_plan();
        plan.conservation.estate_funding_domains.clear();
        plan.reinstall = Some(Box::new(FleetReinstallRecord {
            activation_reset: None,
            operation_id: plan.operation_id.clone(),
            source_operation_id: "source-operation".into(),
            authorities: Vec::new(),
            assets: Vec::new(),
        }));
        let action = EnsureAction::Install {
            canic_init: None,
            reinstall_witness: None,
            init_arg: None,
            init_arg_sha256: None,
            init_candid: None,
            init_candid_sha256: None,
            mode: InstallMode::Reinstall,
            name: "root".into(),
            principal: "rrkah-fqaaa-aaaaa-aaaaq-cai".into(),
            wasm: "root.wasm".into(),
            wasm_sha256: "22".repeat(32),
        };
        journal.effects[0].action_sha256 = action_sha256(&action);
        plan.canisters.push(CanisterPlan {
            actions: vec![action],
            disposition: CanisterDisposition::Reuse,
            name: "root".into(),
            observed_cycles: 0,
            principal: Some("rrkah-fqaaa-aaaaa-aaaaq-cai".into()),
        });
        let admit = |journal: &FleetEnsureJournalRecord| {
            reinstall_recovery_ready::<std::io::Error>(&plan, journal, &state)
        };
        assert!(admit(&journal).unwrap());
        let mut issued = journal.clone();
        issued.effects[0].state = EffectState::Issued;
        let mut missing = journal.clone();
        missing.effects.clear();
        let mut wrong_action = journal.clone();
        wrong_action.effects[0].action_sha256 = "other".into();
        let mut no_version = journal.clone();
        no_version.effects[0].pre_canister_version = None;
        let mut wrong_operation = journal.clone();
        wrong_operation.operation_id = "other".into();
        for invalid in [issued, missing, wrong_action, no_version, wrong_operation] {
            assert!(matches!(
                admit(&invalid),
                Err(EnsureWorkflowError::JournalIntegrity)
            ));
        }
        for completion in [
            FleetEnsureCompletion::InProgress,
            FleetEnsureCompletion::Converged,
        ] {
            let mut other = journal.clone();
            other.completion = completion;
            assert!(!admit(&other).unwrap());
        }
    }

    #[test]
    fn completed_reinstall_evidence_requires_exact_operation_action_and_version() {
        let (state, journal) = retained_evidence();
        assert!(retained_reinstall_operation_matches(
            &state,
            "fleet",
            "operation",
            &journal,
        ));

        let mut wrong_action = journal.clone();
        wrong_action.effects[0].action_sha256 = "other".to_string();
        assert!(!retained_reinstall_operation_matches(
            &state,
            "fleet",
            "operation",
            &wrong_action,
        ));

        let mut wrong_version = journal.clone();
        wrong_version.effects[0].pre_canister_version = Some(8);
        assert!(!retained_reinstall_operation_matches(
            &state,
            "fleet",
            "operation",
            &wrong_version,
        ));

        for (fleet, operation) in [("other", "operation"), ("fleet", "other")] {
            assert!(!retained_reinstall_operation_matches(
                &state, fleet, operation, &journal,
            ));
        }

        let mut incomplete = journal;
        incomplete.effects[0].state = EffectState::Issued;
        assert!(!retained_reinstall_operation_matches(
            &state,
            "fleet",
            "operation",
            &incomplete,
        ));
    }

    #[test]
    fn fresh_fleet_alternate_desired_cannot_clear_completed_reinstall_evidence() {
        let (state, journal) = retained_evidence();
        assert!(retained_reinstall_desired_conflict(
            &state,
            "fleet",
            "operation",
            "reviewed-desired",
            "alternate-desired",
            &journal,
        ));
        assert!(!retained_reinstall_desired_conflict(
            &state,
            "fleet",
            "operation",
            "reviewed-desired",
            "reviewed-desired",
            &journal,
        ));

        let mut terminal = journal;
        terminal.completion = FleetEnsureCompletion::Converged;
        assert!(!retained_reinstall_desired_conflict(
            &state,
            "fleet",
            "operation",
            "reviewed-desired",
            "alternate-desired",
            &terminal,
        ));
    }

    fn cycle_observation(
        lifecycle: crate::fleet_ensure::model::RootOwnedCanisterLifecycle,
        cycles: u128,
    ) -> FleetObservation {
        FleetObservation {
            additional_controlled_cycles: BTreeMap::new(),
            canisters: BTreeMap::from([(
                "asset".to_string(),
                Some(crate::fleet_ensure::model::LiveCanister {
                    canister_version: Some(1),
                    controllers: Vec::new(),
                    cycles,
                    module_sha256: None,
                    principal: "controlled-principal".to_string(),
                    reinstall_required: false,
                    root_owned_lifecycle: Some(lifecycle),
                    status: crate::fleet_ensure::model::CanisterRuntimeStatus::Stopped,
                }),
            )]),
            estate_funding_domains: BTreeMap::new(),
            ledger_fee_cycles: 0,
            operator_cycles: 0,
            protocol_ready: BTreeMap::new(),
        }
    }

    #[test]
    fn fresh_fleet_idle_observation_accepts_donations_without_double_counting() {
        use crate::fleet_ensure::model::RootOwnedCanisterLifecycle;

        let additional = BTreeMap::from([("controlled-principal".to_string(), 100)]);
        let mut idle = cycle_observation(RootOwnedCanisterLifecycle::Idle, 100);
        attach_terminal_cycles::<std::io::Error>(&mut idle, additional.clone())
            .expect("merge exact Idle observation");
        assert!(idle.additional_controlled_cycles.is_empty());

        let mut donated_idle = cycle_observation(RootOwnedCanisterLifecycle::Idle, 99);
        attach_terminal_cycles::<std::io::Error>(&mut donated_idle, additional.clone()).unwrap();
        assert_eq!(
            controlled_cycles::<std::io::Error>(&donated_idle).unwrap(),
            100
        );
        assert!(donated_idle.additional_controlled_cycles.is_empty());

        let mut conflicting_idle = cycle_observation(RootOwnedCanisterLifecycle::Idle, 101);
        assert!(matches!(
            attach_terminal_cycles::<std::io::Error>(&mut conflicting_idle, additional.clone()),
            Err(EnsureWorkflowError::TerminalInventory(
                TerminalInventoryError::IdleCycleObservationConflict {
                    canister,
                    expected: 101,
                    observed: 100,
                }
            )) if canister == "controlled-principal"
        ));

        let mut workload = cycle_observation(RootOwnedCanisterLifecycle::Workload, 99);
        attach_terminal_cycles::<std::io::Error>(&mut workload, additional)
            .expect("merge conservative workload observation");
        assert_eq!(
            workload
                .canisters
                .get("asset")
                .and_then(Option::as_ref)
                .map(|live| live.cycles),
            Some(99),
        );
    }
}