canic-host 0.110.37

use crate::{
    fleet_ensure::{
        dto::{
            FleetEnsurePhase, FleetEnsureProgress, FleetEnsureProgressState,
            FleetProvisioningProgress,
        },
        model::{
            CanisterDisposition, CanisterRuntimeStatus, CurrentFleetProtocolAction,
            CycleConservation, DesiredCanister, DesiredCanisterInit, DesiredCanisterKind,
            DesiredFleet, DesiredFleetProtocol, DesiredPresence, DesiredProtocolStep,
            DrainAuthority, EffectRecord, EffectState, EnsureAction,
            EstateFundingDomainObservation, FLEET_ENSURE_SCHEMA_VERSION, FleetEnsureCompletion,
            FleetEnsureJournalRecord, FleetEnsurePlan, FleetEnsureStateRecord, FleetObservation,
            FleetReinstallRecord, FleetReinstallSourceRecord, FleetTerminalRetirementRecord,
            FleetTerminalSourceRecord, LiveCanister, RootOwnedCanisterLifecycle,
            create_balance_is_terminal,
            operator_mint::{
                OperatorMintAuthority, OperatorMintNotificationOutcomeRecord,
                OperatorMintReceiptRecord, OperatorMintTransferOutcomeRecord,
            },
        },
        ops::{
            EffectObservation, EffectOutcome, EffectRetry, EnsurePlatform, TerminalFleetInventory,
            action_sha256,
            operator_mint::{journal as records, prepare_intent},
        },
        workflow,
    },
    registry::RegistryEntry,
    test_support::{start_pocket_ic, temp_dir},
};
use candid::Principal;
use canic_control_plane::{
    dto::template::{TemplateChunkInput, TemplateChunkSetPrepareInput},
    ids::{TemplateId, TemplateVersion},
};
use canic_core::{
    cdk::{types::Cycles, utils::hash::sha256_hex},
    dto::{
        component_provisioning::{
            FleetComponentProvisioningOperation, FleetComponentProvisioningPlan,
            FleetComponentProvisioningPrepareRequest,
        },
        fleet_registry::{
            FleetRegistry, FleetRegistryActivationRequest, FleetRegistryVersion,
            FleetSubnetRootEntry, FleetSubnetRootJoinRequest, FleetSubnetRootStatus,
        },
        fleet_subnet_root::FleetSubnetWasmStoreAdoptionRequest,
    },
    ids::{
        AppId, CanonicalNetworkId, ComponentDeploymentConfigurationDigest, ComponentTopologyDigest,
        CyclesFundingBudget, FleetAdmissionPolicy, FleetBinding, FleetCoordinatorBinding, FleetId,
        FleetKey, FleetRegistryAuthority, FleetSubnetCanisterPoolConfig, FleetSubnetRootLimits,
        FleetSubnetRootReleaseSet, FleetSubnetWasmStoreAuthority, ReleaseBuildId,
        ReleaseBuildNonce, ReleaseSetDigest, SubnetId,
    },
};
use std::{
    collections::{BTreeMap, BTreeSet},
    env, fs,
    path::{Path, PathBuf},
};
use thiserror::Error as ThisError;

const TREASURY: &str = "rrkah-fqaaa-aaaaa-aaaaq-cai";
const OLD_APP: &str = "ryjl3-tyaaa-aaaaa-aaaba-cai";
const RETIRED: &str = "r7inp-6aaaa-aaaaa-aaabq-cai";
const OLD_REPLACED: &str = "renrk-eyaaa-aaaaa-aaada-cai";
const SUBNET: &str = "rwlgt-iiaaa-aaaaa-aaaaa-cai";
const CONTROLLER: &str = "rdmx6-jaaaa-aaaaa-aaadq-cai";
const LEDGER: &str = "um5iw-rqaaa-aaaaq-qaaba-cai";

fn json_field_values<'a>(value: &'a serde_json::Value, field: &str) -> Vec<&'a serde_json::Value> {
    fn collect<'a>(
        value: &'a serde_json::Value,
        field: &str,
        values: &mut Vec<&'a serde_json::Value>,
    ) {
        match value {
            serde_json::Value::Array(items) => {
                for item in items {
                    collect(item, field, values);
                }
            }
            serde_json::Value::Object(fields) => {
                for (name, value) in fields {
                    if name == field {
                        values.push(value);
                    }
                    collect(value, field, values);
                }
            }
            _ => {}
        }
    }

    let mut values = Vec::new();
    collect(value, field, &mut values);
    values
}

#[derive(Debug, ThisError)]
#[error("simulated lost response")]
pub(super) struct MockError;

#[derive(Clone, Copy, Eq, PartialEq)]
enum MockRootOwnedTopologyPolicy {
    Direct,
    Exact,
}

#[derive(Debug, Eq, PartialEq)]
enum MockFundingRead {
    Balance,
    Effect(EffectState),
}

pub(super) struct MockPlatform {
    observation_calls: usize,
    pub(super) root_management: Option<crate::fleet_ensure::model::RootManagementObservation>,
    pub(super) operator_funding: Option<crate::fleet_ensure::view::OperatorFundingObservation>,
    pub(super) reinstall_authority:
        Option<BTreeMap<String, crate::fleet_ensure::model::RootManagementCanisterObservation>>,
    pub(super) seal_identity: Option<(String, String)>,
    pub(super) seal_reads: Vec<(String, EnsureAction)>,
    pub(super) retained_activation_checks: Vec<(String, String)>,
    pub(super) rejected_activation_root: Option<String>,
    completed: BTreeMap<String, EffectOutcome>,
    created_principals: BTreeMap<String, String>,
    duplicate_create_responses: BTreeMap<String, u32>,
    create_shortfalls: BTreeMap<String, u128>,
    desired: DesiredFleet,
    deferred_created_canisters: BTreeMap<String, String>,
    estate_funding_balance_cycles: Option<u128>,
    fail_once: BTreeSet<String>,
    failed: BTreeSet<String>,
    live: BTreeMap<String, LiveCanister>,
    ledger_fee_cycles: u128,
    mutations: BTreeMap<String, u32>,
    operator_cycles: u128,
    paced_observations: Vec<u32>,
    progress: Vec<FleetEnsureProgress>,
    progress_journal: Option<PathBuf>,
    paced_root_owned_observations: Vec<(String, u32)>,
    protocol_command_only: bool,
    protocol_action: Option<EnsureAction>,
    reviewed_protocol_actions: Vec<EnsureAction>,
    publication_journal: Option<PathBuf>,
    publication_attempts_on_issue: Vec<u32>,
    independent_batches: Vec<Vec<String>>,
    fresh_protocol_actions: Vec<EnsureAction>,
    protocol_pending_waits: u32,
    provisioning_progress: Option<FleetProvisioningProgress>,
    protocol_ready: BTreeSet<String>,
    protocol_retry: EffectRetry,
    root_owned_topology_policy: MockRootOwnedTopologyPolicy,
    root_owned_pending_waits: BTreeMap<String, u32>,
    typed_protocol: bool,
    typed_protocol_burns: Vec<u128>,
    skip_transfer_credit: bool,
    stall_before_mutation: BTreeMap<String, u32>,
    terminal_inventory: TerminalFleetInventory,
    terminal_inventory_expected_operation_id: Option<String>,
    terminal_inventory_operation_ids: Vec<String>,
    version_observation_failures: u8,
    funding_reads: BTreeMap<String, Vec<MockFundingRead>>,
    funding_observation_failures: u8,
    funding_journal: Option<PathBuf>,
}

impl MockPlatform {
    pub(super) fn new(desired: DesiredFleet, live: impl IntoIterator<Item = LiveCanister>) -> Self {
        let ledger_fee_cycles = desired
            .ledger_fee_cycles
            .parse::<Cycles>()
            .map(|cycles| cycles.to_u128())
            .expect("fixture ledger fee");
        Self {
            observation_calls: 0,
            reinstall_authority: None,
            root_management: None,
            operator_funding: None,
            seal_identity: None,
            seal_reads: Vec::new(),
            retained_activation_checks: Vec::new(),
            rejected_activation_root: None,
            completed: BTreeMap::new(),
            created_principals: BTreeMap::new(),
            duplicate_create_responses: BTreeMap::new(),
            create_shortfalls: BTreeMap::new(),
            desired,
            deferred_created_canisters: BTreeMap::new(),
            estate_funding_balance_cycles: None,
            fail_once: BTreeSet::new(),
            failed: BTreeSet::new(),
            live: live
                .into_iter()
                .map(|canister| (canister.principal.clone(), canister))
                .collect(),
            ledger_fee_cycles,
            mutations: BTreeMap::new(),
            operator_cycles: 100_000,
            paced_observations: Vec::new(),
            progress: Vec::new(),
            progress_journal: None,
            paced_root_owned_observations: Vec::new(),
            protocol_command_only: false,
            protocol_action: None,
            reviewed_protocol_actions: Vec::new(),
            publication_journal: None,
            publication_attempts_on_issue: Vec::new(),
            independent_batches: Vec::new(),
            fresh_protocol_actions: Vec::new(),
            protocol_pending_waits: 0,
            provisioning_progress: None,
            protocol_ready: BTreeSet::new(),
            protocol_retry: EffectRetry::None,
            root_owned_topology_policy: MockRootOwnedTopologyPolicy::Direct,
            root_owned_pending_waits: BTreeMap::new(),
            typed_protocol: false,
            typed_protocol_burns: Vec::new(),
            skip_transfer_credit: false,
            stall_before_mutation: BTreeMap::new(),
            terminal_inventory: TerminalFleetInventory::default(),
            terminal_inventory_expected_operation_id: None,
            terminal_inventory_operation_ids: Vec::new(),
            version_observation_failures: 0,
            funding_reads: BTreeMap::new(),
            funding_observation_failures: 0,
            funding_journal: None,
        }
    }

    pub(super) fn defer_created_until_root_install(&mut self, name: &str, root: &str) {
        self.deferred_created_canisters
            .insert(name.to_string(), root.to_string());
    }

    pub(super) fn fail_once(&mut self, action_sha256: String) {
        self.fail_once.insert(action_sha256);
    }

    pub(super) fn set_created_principal(&mut self, name: &str, principal: String) {
        self.created_principals.insert(name.to_string(), principal);
    }

    pub(super) fn require_root_owned_topology(&mut self) {
        self.root_owned_topology_policy = MockRootOwnedTopologyPolicy::Exact;
    }

    pub(super) fn stall_before_mutation(&mut self, action_sha256: String, attempts: u32) {
        self.stall_before_mutation.insert(action_sha256, attempts);
    }

    fn retain_root_owned_lifecycle_for_observations(&mut self, name: &str, observations: u32) {
        self.root_owned_pending_waits
            .insert(name.to_string(), observations);
    }

    pub(super) fn mutation_count(&self, action_sha256: &str) -> u32 {
        self.mutations
            .get(action_sha256)
            .copied()
            .unwrap_or_default()
    }

    pub(super) fn set_fresh_protocol_actions(&mut self, actions: Vec<EnsureAction>) {
        self.fresh_protocol_actions = actions;
    }

    pub(super) fn duplicate_create_response_count(&self, action_sha256: &str) -> u32 {
        self.duplicate_create_responses
            .get(action_sha256)
            .copied()
            .unwrap_or_default()
    }

    pub(super) fn set_operator_cycles(&mut self, cycles: u128) {
        self.operator_cycles = cycles;
    }

    pub(super) fn set_estate_funding_balance(&mut self, cycles: u128) {
        self.estate_funding_balance_cycles = Some(cycles);
    }

    pub(super) const fn operator_cycles(&self) -> u128 {
        self.operator_cycles
    }

    pub(super) fn live_controllers(&self, name: &str) -> Option<&[String]> {
        let principal = self
            .created_principals
            .get(name)
            .cloned()
            .unwrap_or_else(|| format!("created-{name}"));
        self.live
            .get(&principal)
            .map(|live| live.controllers.as_slice())
    }

    fn root_owned_action_authority_is_exact(
        &self,
        action: &EnsureAction,
        state: &FleetEnsureStateRecord,
    ) -> bool {
        if self.root_owned_topology_policy == MockRootOwnedTopologyPolicy::Direct {
            return true;
        }
        let EnsureAction::SetControllers { name, .. } = action else {
            return true;
        };
        let Some(configured) = self.desired.canisters.iter().find(|configured| {
            configured.name == *name && configured.kind == DesiredCanisterKind::Pool
        }) else {
            return true;
        };
        let Some(parent) = configured.parent.as_deref() else {
            return false;
        };
        let retained_principal = |name: &str| {
            state
                .pending_principals
                .get(name)
                .or_else(|| state.principals.get(name))
        };
        let Some(principal) = retained_principal(name) else {
            return false;
        };
        retained_principal(parent).is_some()
            && state.retained_cycles_by_principal.contains_key(principal)
            && state.topology.get(name).is_some_and(|topology| {
                topology.kind == configured.kind && topology.parent.as_deref() == Some(parent)
            })
    }

    fn principal<'a>(
        state: &'a FleetEnsureStateRecord,
        action: &'a EnsureAction,
    ) -> Option<&'a str> {
        let principal = match action {
            EnsureAction::SealAuthority { .. } => {
                panic!("authority sealing uses the production PocketIC adapter")
            }
            EnsureAction::Create { .. } => return None,
            EnsureAction::Delete { principal, .. }
            | EnsureAction::FleetProtocol { principal, .. }
            | EnsureAction::Fund { principal, .. }
            | EnsureAction::FundEstate { principal, .. }
            | EnsureAction::Install { principal, .. }
            | EnsureAction::Protocol { principal, .. }
            | EnsureAction::SetControllers { principal, .. }
            | EnsureAction::Start { principal, .. }
            | EnsureAction::Stop { principal, .. }
            | EnsureAction::Transfer { principal, .. } => principal,
        };
        principal.strip_prefix("created:").map_or_else(
            || Some(principal.as_str()),
            |name| state.pending_principals.get(name).map(String::as_str),
        )
    }

    fn canister_principal<'a>(&'a self, state: &'a FleetEnsureStateRecord, name: &str) -> &'a str {
        state
            .pending_principals
            .get(name)
            .or_else(|| state.principals.get(name))
            .or_else(|| {
                self.desired
                    .canisters
                    .iter()
                    .find(|canister| canister.name == name)
                    .and_then(|canister| canister.principal.as_ref())
            })
            .map(String::as_str)
            .expect("fixture canister Principal")
    }

    fn controllers(
        &self,
        state: &FleetEnsureStateRecord,
        controllers: &[String],
        controller_canisters: &[String],
    ) -> Vec<String> {
        let mut resolved = controllers.to_vec();
        resolved.extend(
            controller_canisters
                .iter()
                .map(|name| self.canister_principal(state, name).to_string()),
        );
        resolved.sort();
        resolved.dedup();
        resolved
    }

    fn create_observation(
        &self,
        action: &EnsureAction,
        record: &EffectRecord,
        state: &FleetEnsureStateRecord,
    ) -> Option<EffectObservation> {
        let EnsureAction::Create {
            name,
            requested_initial_cycles,
            ..
        } = action
        else {
            return None;
        };
        if let Some(root) = self.deferred_created_canisters.get(name)
            && record.created_principal.is_some()
            && state
                .pending_principals
                .get(root)
                .or_else(|| state.principals.get(root))
                .and_then(|principal| self.live.get(principal))
                .is_none_or(|live| live.module_sha256.is_none())
        {
            return Some(EffectObservation {
                provisioning_progress: None,
                provisioning_failure: None,
                applied: false,
                estate_funding_required: None,
                post_cycles: None,
                progress_identity: format!("created:{name}:controller-observation-deferred"),
                retry: EffectRetry::DeferUntilControllerObservation,
            });
        }
        let post_cycles = record
            .created_principal
            .as_deref()
            .and_then(|principal| self.live.get(principal))
            .map(|live| live.cycles);
        let maximum_observation_burn_cycles =
            crate::fleet_ensure::ops::maximum_creation_observation_burn(&self.desired)
                .expect("fixture Create execution and observation burn");
        let applied = create_balance_is_terminal(
            post_cycles,
            *requested_initial_cycles,
            maximum_observation_burn_cycles,
        );
        Some(EffectObservation {
            provisioning_progress: None,
            provisioning_failure: None,
            applied,
            estate_funding_required: None,
            post_cycles,
            progress_identity: format!("created:{:?}", record.created_principal),
            retry: if post_cycles.is_some() && !applied {
                EffectRetry::ReplanRequiredAfterCreateBalanceDrift
            } else {
                EffectRetry::None
            },
        })
    }

    fn estate_funding_observation(
        &self,
        action: &EnsureAction,
        record: &EffectRecord,
    ) -> Option<EffectObservation> {
        let EnsureAction::FundEstate {
            amount,
            expected_post_cycles,
            ledger_fee_cycles,
            ..
        } = action
        else {
            return None;
        };
        let destination_after = self.estate_funding_balance_cycles.unwrap_or_default();
        Some(EffectObservation {
            provisioning_progress: None,
            provisioning_failure: None,
            applied: record.receipt.is_some()
                && crate::fleet_ensure::ops::estate_funding_applied(
                    crate::fleet_ensure::ops::EstateFundingObservation {
                        amount: *amount,
                        destination_after,
                        destination_before: record.destination_pre_cycles,
                        expected_destination_after: *expected_post_cycles,
                        ledger_fee_cycles: *ledger_fee_cycles,
                        source_after: self.operator_cycles,
                        source_before: record.pre_cycles,
                    },
                ),
            estate_funding_required: None,
            post_cycles: Some(self.operator_cycles),
            progress_identity: format!("estate-funding:{:?}:{destination_after}", record.receipt),
            retry: EffectRetry::None,
        })
    }

    fn effect_is_applied(
        &self,
        action: &EnsureAction,
        record: &EffectRecord,
        state: &FleetEnsureStateRecord,
    ) -> bool {
        let principal = Self::principal(state, action);
        match action {
            EnsureAction::SealAuthority { .. } => {
                panic!("authority sealing uses the production PocketIC adapter")
            }
            EnsureAction::Delete { .. } => {
                principal.is_none_or(|value| !self.live.contains_key(value))
            }
            EnsureAction::Install {
                mode, wasm_sha256, ..
            } => principal
                .and_then(|value| self.live.get(value))
                .is_some_and(|live| {
                    crate::fleet_ensure::ops::install_effect_applied(
                        *mode,
                        wasm_sha256,
                        live.module_sha256.as_deref(),
                        record.pre_canister_version,
                        live.canister_version,
                    )
                }),
            EnsureAction::FleetProtocol { name, .. } | EnsureAction::Protocol { name, .. } => {
                self.protocol_ready.contains(name)
            }
            EnsureAction::SetControllers {
                controller_canisters,
                controllers,
                ..
            } => {
                let controllers = self.controllers(state, controllers, controller_canisters);
                principal
                    .and_then(|value| self.live.get(value))
                    .is_some_and(|live| live.controllers == controllers)
            }
            EnsureAction::Start { .. } => principal
                .and_then(|value| self.live.get(value))
                .is_some_and(|live| live.status == CanisterRuntimeStatus::Running),
            EnsureAction::Stop { .. } => principal
                .and_then(|value| self.live.get(value))
                .is_some_and(|live| live.status == CanisterRuntimeStatus::Stopped),
            EnsureAction::Transfer {
                amount,
                maximum_execution_burn_cycles,
                destination,
                ..
            } => {
                let destination = self.canister_principal(state, destination);
                let source = principal
                    .and_then(|value| self.live.get(value))
                    .map(|live| live.cycles)
                    .expect("transfer source");
                let destination = self
                    .live
                    .get(destination)
                    .map(|live| live.cycles)
                    .expect("transfer destination");
                let source_debit = record.pre_cycles.expect("source pre") - source;
                let destination_credit =
                    destination - record.destination_pre_cycles.expect("destination pre");
                source_debit >= *amount
                    && source_debit <= amount + maximum_execution_burn_cycles
                    && destination_credit == *amount
            }
            EnsureAction::Create { .. }
            | EnsureAction::Fund { .. }
            | EnsureAction::FundEstate { .. } => false,
        }
    }

    fn effect_retry(&self, action: &EnsureAction, applied: bool) -> EffectRetry {
        if !applied
            && self.protocol_retry == EffectRetry::ContinuePoolMaintenance
            && is_pool_maintenance(action)
        {
            return EffectRetry::ContinuePoolMaintenance;
        }
        if !applied
            && self.protocol_retry == EffectRetry::ReplayExactIssuedCommand
            && matches!(
                action,
                EnsureAction::FleetProtocol { action, .. }
                    if matches!(
                        action.as_ref(),
                        CurrentFleetProtocolAction::ProvisionComponents { .. }
                    )
            )
        {
            EffectRetry::ReplayExactIssuedCommand
        } else {
            EffectRetry::None
        }
    }

    #[expect(
        clippy::too_many_lines,
        reason = "the deterministic adapter keeps every effect mutation in one exhaustive match"
    )]
    fn mutate(&mut self, action: &EnsureAction, state: &FleetEnsureStateRecord) -> EffectOutcome {
        let principal = Self::principal(state, action).map(str::to_string);
        match action {
            EnsureAction::SealAuthority { .. } => {
                panic!("authority sealing uses the production PocketIC adapter")
            }
            EnsureAction::Create {
                controller_canisters,
                controllers,
                name,
                requested_initial_cycles,
                ..
            } => {
                let creation_fee = self
                    .desired
                    .management_creation_fee_cycles
                    .parse::<Cycles>()
                    .map(|cycles| cycles.to_u128())
                    .expect("creation fee");
                let ledger_fee = self
                    .desired
                    .ledger_fee_cycles
                    .parse::<Cycles>()
                    .map(|cycles| cycles.to_u128())
                    .expect("ledger fee");
                self.operator_cycles -= requested_initial_cycles + creation_fee + ledger_fee;
                let principal = self
                    .created_principals
                    .get(name)
                    .cloned()
                    .unwrap_or_else(|| format!("created-{name}"));
                let retained_cycles = requested_initial_cycles
                    .checked_sub(self.create_shortfalls.get(name).copied().unwrap_or(0))
                    .expect("fixture creation shortfall is bounded");
                self.live.insert(
                    principal.clone(),
                    LiveCanister {
                        canister_version: Some(1),
                        controllers: self.controllers(state, controllers, controller_canisters),
                        cycles: retained_cycles,
                        module_sha256: None,
                        principal: principal.clone(),
                        reinstall_required: false,
                        root_owned_lifecycle: None,
                        status: CanisterRuntimeStatus::Running,
                    },
                );
                EffectOutcome {
                    created_principal: Some(principal),
                    // The production Cycles Ledger response binds the reviewed
                    // creation amount; live observation independently exposes
                    // any post-effect balance drift.
                    post_cycles: Some(*requested_initial_cycles),
                    receipt: Some("create-block".to_string()),
                }
            }
            EnsureAction::Delete { .. } => {
                self.live
                    .remove(principal.as_deref().expect("delete principal"));
                empty_outcome()
            }
            EnsureAction::Fund { amount, .. } => {
                let ledger_fee = self
                    .desired
                    .ledger_fee_cycles
                    .parse::<Cycles>()
                    .map(|cycles| cycles.to_u128())
                    .expect("ledger fee");
                self.operator_cycles -= amount + ledger_fee;
                let live = self
                    .live
                    .get_mut(principal.as_deref().expect("fund principal"))
                    .expect("fund target");
                live.cycles += amount;
                EffectOutcome {
                    created_principal: None,
                    post_cycles: Some(live.cycles),
                    receipt: Some("withdraw-block".to_string()),
                }
            }
            EnsureAction::FundEstate {
                amount,
                ledger_fee_cycles,
                ..
            } => {
                self.operator_cycles -= amount + ledger_fee_cycles;
                self.estate_funding_balance_cycles = Some(
                    self.estate_funding_balance_cycles
                        .unwrap_or_default()
                        .checked_add(*amount)
                        .expect("estate funding balance"),
                );
                EffectOutcome {
                    created_principal: None,
                    post_cycles: None,
                    receipt: Some("estate-transfer-block".to_string()),
                }
            }
            EnsureAction::Install {
                name, wasm_sha256, ..
            } => {
                let live = self
                    .live
                    .get_mut(principal.as_deref().expect("install principal"))
                    .expect("install target");
                live.module_sha256 = Some(wasm_sha256.clone());
                live.canister_version = Some(
                    live.canister_version
                        .unwrap_or_default()
                        .checked_add(1)
                        .expect("fixture canister version"),
                );
                live.reinstall_required = false;
                let imported_pools = self
                    .desired
                    .canisters
                    .iter()
                    .filter(|configured| {
                        configured.kind == DesiredCanisterKind::Pool
                            && configured.parent.as_deref() == Some(name)
                    })
                    .filter_map(|configured| {
                        state.pending_principals.get(&configured.name).cloned()
                    })
                    .collect::<Vec<_>>();
                for principal in imported_pools {
                    if let Some(pool) = self.live.get_mut(&principal) {
                        pool.root_owned_lifecycle = Some(RootOwnedCanisterLifecycle::Idle);
                        pool.status = CanisterRuntimeStatus::Stopped;
                    }
                }
                empty_outcome()
            }
            EnsureAction::FleetProtocol { name, .. } | EnsureAction::Protocol { name, .. } => {
                if !self.protocol_command_only {
                    self.protocol_ready.insert(name.clone());
                }
                if !is_pool_maintenance(action) {
                    self.protocol_retry = EffectRetry::None;
                }
                EffectOutcome {
                    created_principal: None,
                    post_cycles: None,
                    receipt: Some("protocol-receipt".to_string()),
                }
            }
            EnsureAction::SetControllers {
                controller_canisters,
                controllers,
                ..
            } => {
                let controllers = self.controllers(state, controllers, controller_canisters);
                self.live
                    .get_mut(principal.as_deref().expect("settings principal"))
                    .expect("settings target")
                    .controllers = controllers;
                empty_outcome()
            }
            EnsureAction::Start { .. } => {
                self.live
                    .get_mut(principal.as_deref().expect("start principal"))
                    .expect("start target")
                    .status = CanisterRuntimeStatus::Running;
                empty_outcome()
            }
            EnsureAction::Stop { .. } => {
                self.live
                    .get_mut(principal.as_deref().expect("stop principal"))
                    .expect("stop target")
                    .status = CanisterRuntimeStatus::Stopped;
                empty_outcome()
            }
            EnsureAction::Transfer {
                amount,
                destination,
                maximum_execution_burn_cycles,
                ..
            } => {
                let destination = self.canister_principal(state, destination).to_string();
                let post_cycles = {
                    let source = self
                        .live
                        .get_mut(principal.as_deref().expect("transfer principal"))
                        .expect("transfer source");
                    source.cycles -= amount + maximum_execution_burn_cycles;
                    source.cycles
                };
                if !self.skip_transfer_credit {
                    self.live
                        .get_mut(&destination)
                        .expect("transfer destination")
                        .cycles += amount;
                }
                EffectOutcome {
                    created_principal: None,
                    post_cycles: Some(post_cycles),
                    receipt: Some("drain-receipt".to_string()),
                }
            }
        }
    }
}

impl EnsurePlatform for MockPlatform {
    type Error = MockError;

    fn apply_independent_effects(
        &mut self,
        operation_id: &str,
        uploads: &[crate::fleet_ensure::ops::independent_effects::IndependentEffect<'_>],
        state: &FleetEnsureStateRecord,
    ) -> Result<Vec<Result<EffectOutcome, Self::Error>>, Self::Error> {
        if let Some(path) = &self.publication_journal {
            let journal: FleetEnsureJournalRecord =
                serde_json::from_slice(&fs::read(path).unwrap()).unwrap();
            for upload in uploads {
                assert!(
                    journal.effects.contains(upload.record),
                    "all intents precede any submission"
                );
                assert_eq!(upload.record.state, EffectState::Intent);
            }
        }
        self.independent_batches.push(
            uploads
                .iter()
                .map(|upload| action_sha256(upload.action))
                .collect(),
        );
        Ok(uploads
            .iter()
            .map(|upload| self.apply(operation_id, upload.action, upload.record, state))
            .collect())
    }

    fn observe_root_management(
        &mut self,
        _state: &FleetEnsureStateRecord,
        _targets: &BTreeSet<String>,
    ) -> Result<Option<crate::fleet_ensure::model::RootManagementObservation>, Self::Error> {
        Ok(self.root_management.clone())
    }

    fn observe_operator_funding(
        &mut self,
    ) -> Result<Option<crate::fleet_ensure::view::OperatorFundingObservation>, Self::Error> {
        Ok(self.operator_funding.clone())
    }

    fn reinstall_authorities(
        &mut self,
        _state: &FleetEnsureStateRecord,
    ) -> Result<
        Option<BTreeMap<String, crate::fleet_ensure::model::RootManagementCanisterObservation>>,
        Self::Error,
    > {
        Ok(self.reinstall_authority.clone())
    }

    fn authority_sealed(
        &mut self,
        operation: &str,
        action: &EnsureAction,
    ) -> Result<bool, Self::Error> {
        self.seal_reads.push((operation.to_owned(), action.clone()));
        Ok(self.seal_identity.as_ref() == Some(&(operation.to_owned(), action_sha256(action))))
    }

    fn report_progress(&mut self, progress: FleetEnsureProgress) {
        if let Some(path) = &self.progress_journal {
            let journal: FleetEnsureJournalRecord =
                serde_json::from_slice(&fs::read(path).expect("durable progress journal")).unwrap();
            assert_eq!(progress.operation_id, journal.operation_id);
            assert_eq!(progress.plan_sha256, journal.plan_sha256);
            assert_eq!(
                progress.applied_effects as usize,
                journal
                    .effects
                    .iter()
                    .chain(
                        journal
                            .funding_reviews
                            .iter()
                            .filter_map(|review| review.effect.as_ref())
                    )
                    .filter(|effect| effect.state == EffectState::Applied)
                    .count(),
                "progress must describe already persisted receipts"
            );
        }
        self.progress.push(progress);
    }

    fn require_retained_root_activation(
        &mut self,
        operation_id: &str,
        root: &str,
        _state: &FleetEnsureStateRecord,
    ) -> Result<(), Self::Error> {
        self.retained_activation_checks
            .push((operation_id.to_string(), root.to_string()));
        if self.rejected_activation_root.as_deref() == Some(root) {
            Err(MockError)
        } else {
            Ok(())
        }
    }

    fn bind_reviewed_desired(&mut self, desired: &DesiredFleet) -> Result<(), Self::Error> {
        self.desired = desired.clone();
        Ok(())
    }

    fn pace_effect_observation(
        &mut self,
        action: &EnsureAction,
        consecutive_unchanged_observations: u32,
    ) {
        self.paced_observations
            .push(consecutive_unchanged_observations);
        if self.protocol_pending_waits == 0 {
            return;
        }
        self.protocol_pending_waits -= 1;
        if self.protocol_pending_waits == 0 {
            self.protocol_ready.insert(action.name().to_string());
        }
    }

    fn pace_root_owned_observation(
        &mut self,
        target: &str,
        consecutive_retained_observations: u32,
    ) {
        self.paced_root_owned_observations
            .push((target.to_string(), consecutive_retained_observations));
        let Some(remaining) = self.root_owned_pending_waits.get_mut(target) else {
            return;
        };
        *remaining = remaining.saturating_sub(1);
        if *remaining != 0 {
            return;
        }
        let principal = self
            .desired
            .canisters
            .iter()
            .find(|canister| canister.name == target)
            .and_then(|canister| canister.principal.as_ref())
            .expect("retained Root-owned test target Principal");
        self.live
            .get_mut(principal)
            .expect("retained Root-owned live test target")
            .root_owned_lifecycle = Some(RootOwnedCanisterLifecycle::Idle);
    }

    fn observe(
        &mut self,
        _operation_id: &str,
        state: &FleetEnsureStateRecord,
    ) -> Result<FleetObservation, Self::Error> {
        self.observation_calls += 1;
        let canisters = self
            .desired
            .canisters
            .iter()
            .map(|configured| {
                let principal = state
                    .pending_principals
                    .get(&configured.name)
                    .or_else(|| state.principals.get(&configured.name))
                    .map(String::as_str)
                    .or(configured.principal.as_deref());
                (
                    configured.name.clone(),
                    principal.and_then(|principal| self.live.get(principal).cloned()),
                )
            })
            .collect();
        let estate_funding_domains =
            self.estate_funding_balance_cycles
                .map_or_else(BTreeMap::new, |balance_cycles| {
                    self.desired
                        .bootstrap
                        .as_ref()
                        .into_iter()
                        .flat_map(|bootstrap| &bootstrap.roots)
                        .map(|root| {
                            (
                                root.root.clone(),
                                EstateFundingDomainObservation {
                                    balance_cycles: Some(balance_cycles),
                                    cycles_ledger: self.desired.cycles_ledger.clone(),
                                    pool: None,
                                    root_principal: state
                                        .pending_principals
                                        .get(&root.root)
                                        .or_else(|| state.principals.get(&root.root))
                                        .cloned(),
                                },
                            )
                        })
                        .collect()
                });
        Ok(FleetObservation {
            additional_controlled_cycles: BTreeMap::new(),
            canisters,
            estate_funding_domains,
            ledger_fee_cycles: self.ledger_fee_cycles,
            operator_cycles: self.operator_cycles,
            protocol_ready: self
                .desired
                .protocol_steps
                .iter()
                .map(|step| (step.name.clone(), self.protocol_ready.contains(&step.name)))
                .collect(),
        })
    }

    fn protocol_actions(
        &mut self,
        operation_id: &str,
        _state: &FleetEnsureStateRecord,
    ) -> Result<Vec<EnsureAction>, Self::Error> {
        if !self.reviewed_protocol_actions.is_empty() {
            return Ok(self
                .reviewed_protocol_actions
                .iter()
                .filter(|action| !self.protocol_ready.contains(action.name()))
                .cloned()
                .collect());
        }
        if let Some(action) = self.protocol_action.clone() {
            return Ok((!self.protocol_ready.contains(action.name()))
                .then_some(action)
                .into_iter()
                .collect());
        }
        if !self.typed_protocol_burns.is_empty() {
            return Ok(self
                .typed_protocol_burns
                .iter()
                .enumerate()
                .filter_map(|(index, burn)| {
                    let mut action = typed_protocol_action(operation_id);
                    let EnsureAction::FleetProtocol {
                        maximum_execution_burn_cycles,
                        name,
                        ..
                    } = &mut action
                    else {
                        unreachable!("typed fixture action is FleetProtocol");
                    };
                    *maximum_execution_burn_cycles = *burn;
                    *name = format!("fleet-component-provisioning-{index}");
                    (!self.protocol_ready.contains(name.as_str())).then_some(action)
                })
                .collect());
        }
        if !self.typed_protocol || self.protocol_ready.contains("fleet-component-provisioning") {
            return Ok(Vec::new());
        }
        Ok(vec![typed_protocol_action(operation_id)])
    }

    fn fresh_protocol_actions(
        &mut self,
        _operation_id: &str,
        _state: &FleetEnsureStateRecord,
    ) -> Result<Vec<EnsureAction>, Self::Error> {
        Ok(self.fresh_protocol_actions.clone())
    }

    fn terminal_inventory(
        &mut self,
        operation_id: &str,
        _state: &FleetEnsureStateRecord,
    ) -> Result<TerminalFleetInventory, Self::Error> {
        self.terminal_inventory_operation_ids
            .push(operation_id.to_string());
        if self
            .terminal_inventory_expected_operation_id
            .as_deref()
            .is_some_and(|expected| expected != operation_id)
        {
            return Err(MockError);
        }
        Ok(self.terminal_inventory.clone())
    }

    fn observe_effect(
        &mut self,
        _operation_id: &str,
        action: &EnsureAction,
        record: &EffectRecord,
        state: &FleetEnsureStateRecord,
    ) -> Result<EffectObservation, Self::Error> {
        if !self.root_owned_action_authority_is_exact(action, state) {
            return Err(MockError);
        }
        if let Some(observation) = self.create_observation(action, record, state) {
            return Ok(observation);
        }
        if matches!(action, EnsureAction::Fund { .. }) {
            self.funding_reads
                .entry(action_sha256(action))
                .or_default()
                .push(MockFundingRead::Effect(record.state.clone()));
            if self.funding_observation_failures > 0 {
                self.funding_observation_failures -= 1;
                return Err(MockError);
            }
            return Ok(EffectObservation {
                provisioning_progress: None,
                provisioning_failure: None,
                applied: record.receipt.is_some(),
                estate_funding_required: None,
                post_cycles: Self::principal(state, action)
                    .and_then(|principal| self.live.get(principal))
                    .map(|live| live.cycles),
                progress_identity: format!("fund:{:?}", record.receipt),
                retry: EffectRetry::None,
            });
        }
        if let Some(observation) = self.estate_funding_observation(action, record) {
            return Ok(observation);
        }
        let applied = self.effect_is_applied(action, record, state);
        Ok(EffectObservation {
            provisioning_progress: if matches!(action, EnsureAction::FleetProtocol { .. }) {
                self.provisioning_progress.clone()
            } else {
                None
            },
            provisioning_failure: None,
            applied,
            estate_funding_required: None,
            post_cycles: None,
            progress_identity: format!("mock:{action:?}:{applied}"),
            retry: self.effect_retry(action, applied),
        })
    }

    fn action_cycles(
        &mut self,
        action: &EnsureAction,
        state: &FleetEnsureStateRecord,
    ) -> Result<Option<u128>, Self::Error> {
        if matches!(action, EnsureAction::Fund { .. }) {
            self.funding_reads
                .entry(action_sha256(action))
                .or_default()
                .push(MockFundingRead::Balance);
        }
        if !self.root_owned_action_authority_is_exact(action, state) {
            return Err(MockError);
        }
        if matches!(action, EnsureAction::FundEstate { .. }) {
            return Ok(Some(self.operator_cycles));
        }
        Ok(Self::principal(state, action)
            .and_then(|principal| self.live.get(principal))
            .map(|live| live.cycles))
    }

    fn action_destination_cycles(
        &mut self,
        action: &EnsureAction,
        state: &FleetEnsureStateRecord,
    ) -> Result<Option<u128>, Self::Error> {
        if matches!(action, EnsureAction::FundEstate { .. }) {
            return Ok(self.estate_funding_balance_cycles);
        }
        let EnsureAction::Transfer { destination, .. } = action else {
            return Ok(None);
        };
        let destination = self.canister_principal(state, destination);
        Ok(self.live.get(destination).map(|live| live.cycles))
    }

    fn action_canister_version(
        &mut self,
        action: &EnsureAction,
        state: &FleetEnsureStateRecord,
    ) -> Result<Option<u64>, Self::Error> {
        if matches!(action, EnsureAction::Install { .. }) && self.version_observation_failures > 0 {
            self.version_observation_failures -= 1;
            return Err(MockError);
        }
        Ok(Self::principal(state, action)
            .and_then(|principal| self.live.get(principal))
            .and_then(|live| live.canister_version))
    }

    fn apply(
        &mut self,
        _operation_id: &str,
        action: &EnsureAction,
        record: &EffectRecord,
        state: &FleetEnsureStateRecord,
    ) -> Result<EffectOutcome, Self::Error> {
        let hash = crate::fleet_ensure::ops::action_sha256(action);
        if matches!(action, EnsureAction::Fund { .. })
            && let Some(path) = &self.funding_journal
        {
            let journal: FleetEnsureJournalRecord = serde_json::from_slice(
                &fs::read(path).expect("funding intent is durable before withdrawal"),
            )
            .unwrap();
            let retained = journal
                .effects
                .iter()
                .find(|effect| effect.action_sha256 == hash)
                .expect("exact funding intent exists before withdrawal");
            assert_eq!(retained.state, EffectState::Intent);
            assert_eq!(retained.pre_cycles, record.pre_cycles);
            assert!(retained.pre_cycles.is_some());
        }
        if action.fixture_publication_attempt_limit().is_some()
            && let Some(path) = &self.publication_journal
        {
            let journal: FleetEnsureJournalRecord = serde_json::from_slice(
                &fs::read(path).expect("intent is durable before publication issue"),
            )
            .unwrap();
            assert_eq!(
                journal
                    .effects
                    .iter()
                    .find(|effect| effect.action_sha256 == hash),
                Some(record)
            );
            self.publication_attempts_on_issue
                .push(record.publication_attempts);
        }
        if let Some(outcome) = self.completed.get(&hash)
            && !is_pool_maintenance(action)
        {
            if matches!(action, EnsureAction::Create { .. }) {
                *self.duplicate_create_responses.entry(hash).or_default() += 1;
            }
            return Ok(outcome.clone());
        }
        if let Some(remaining) = self.stall_before_mutation.get_mut(&hash)
            && *remaining > 0
        {
            *remaining -= 1;
            return Err(MockError);
        }
        let outcome = self.mutate(action, state);
        *self.mutations.entry(hash.clone()).or_default() += 1;
        self.completed.insert(hash.clone(), outcome.clone());
        if self.fail_once.contains(&hash) && self.failed.insert(hash) {
            return Err(MockError);
        }
        Ok(outcome)
    }
}

#[test]
fn completed_replay_after_operator_spending_requires_fresh_review_without_effects() {
    let mut fixture = fixture();
    let digest = sha256_hex(b"completed replay account activity");
    let planned = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &digest,
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut fixture.platform,
    )
    .unwrap();
    let complete = apply_fixture_plan(&mut fixture, &digest, &planned.plan).unwrap();
    assert!(complete.terminal);
    assert_eq!(
        apply_fixture_plan(&mut fixture, &digest, &planned.plan)
            .unwrap()
            .effects_applied,
        0
    );
    let paths = crate::fleet_ensure::ops::EnsurePaths::under(
        &fixture.root,
        &fixture.desired.environment,
        "test-fleet",
    );
    let journal = fs::read(&paths.journal).unwrap();
    let state = fs::read(&paths.state).unwrap();
    let plan = fs::read(&paths.plan).unwrap();
    let mutations = fixture.platform.mutations.clone();
    fixture.platform.operator_cycles -= 100;
    let changed = apply_fixture_plan(&mut fixture, &digest, &planned.plan).unwrap_err();
    assert!(
        matches!(changed, workflow::EnsureWorkflowError::TerminalReplayBalanceChanged {
        operation_id, plan_sha256, current_operator_cycles, ..
    } if operation_id == planned.plan.operation_id
        && plan_sha256 == planned.plan.plan_sha256
        && current_operator_cycles == fixture.platform.operator_cycles)
    );
    assert_eq!(fs::read(&paths.journal).unwrap(), journal);
    assert_eq!(fs::read(&paths.state).unwrap(), state);
    assert_eq!(fs::read(&paths.plan).unwrap(), plan);
    assert_eq!(fixture.platform.mutations, mutations);

    let fresh = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &digest,
        "test-fleet",
        1_800_000_000_000_000_100,
        &mut fixture.platform,
    )
    .unwrap();
    assert_ne!(fresh.plan.plan_sha256, planned.plan.plan_sha256);
    assert!(workflow::ordered_actions(&fresh.plan).is_empty());
    assert_eq!(fresh.plan.conservation.maximum_operator_debit_cycles, 0);
    let recovered = apply_fixture_plan(&mut fixture, &digest, &fresh.plan).unwrap();
    assert!(recovered.terminal);
    assert_eq!(recovered.effects_applied, 0);
    assert_eq!(
        recovered.actual_conservation.unwrap().operator_debit_cycles,
        0
    );
    assert_eq!(fixture.platform.mutations, mutations);
    fs::remove_dir_all(fixture.root).unwrap();
}

#[test]
fn phase_progress_is_bounded_and_completion_follows_recovered_effects() {
    let mut fixture = fixture();
    fixture.platform.progress_journal = Some(
        crate::fleet_ensure::ops::EnsurePaths::under(&fixture.root, "local", "test-fleet").journal,
    );
    let desired_sha256 = "e1".repeat(32);
    let planned = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut fixture.platform,
    )
    .expect("plan progress journey");
    let actions = workflow::ordered_actions(&planned.plan);
    fixture.platform.fail_once.insert(action_sha256(actions[0]));
    let interrupted = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        &planned.plan.plan_sha256,
        &mut fixture.platform,
    );
    assert!(matches!(
        interrupted,
        Err(workflow::EnsureWorkflowError::Platform(_))
    ));
    assert_eq!(fixture.platform.progress.len(), 1);
    assert_eq!(fixture.platform.progress[0].applied_effects, 0);
    assert_eq!(
        fixture.platform.progress[0]
            .next_action
            .as_ref()
            .unwrap()
            .target,
        actions[0].name()
    );
    assert_eq!(
        fixture.platform.progress[0].phase,
        FleetEnsurePhase::Infrastructure
    );
    fixture.platform.progress.clear();
    let completed = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        &planned.plan.plan_sha256,
        &mut fixture.platform,
    )
    .expect("recover the effect before publishing completion progress");
    assert!(completed.terminal);
    let progress = &fixture.platform.progress;
    assert_durable_progress_counts(
        progress,
        &planned.plan,
        actions.len(),
        completed.effects_applied,
    );
    assert!(
        progress
            .windows(2)
            .all(|pair| pair[0].applied_effects <= pair[1].applied_effects)
    );
    assert_eq!(
        progress.last().expect("terminal event").phase,
        FleetEnsurePhase::Complete
    );
    assert_eq!(progress.last().unwrap().next_action, None);
    assert_unfinished_action_projection(&fixture.root, &actions);
    assert_eq!(
        progress.last().expect("terminal event").applied_effects,
        completed.effects_applied
    );
    let mutations = fixture.platform.mutations.clone();
    fixture.platform.progress.clear();
    let replay = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        &planned.plan.plan_sha256,
        &mut fixture.platform,
    )
    .expect("terminal replay");
    assert!(replay.terminal);
    assert_eq!(fixture.platform.mutations, mutations);
    assert!(fixture.platform.progress.iter().all(|event| {
        event.applied_effects == completed.effects_applied
            && event.phase != FleetEnsurePhase::Infrastructure
    }));
    fs::remove_dir_all(fixture.root).expect("remove progress fixture");
}

fn assert_durable_progress_counts(
    progress: &[FleetEnsureProgress],
    plan: &FleetEnsurePlan,
    action_count: usize,
    applied: u32,
) {
    // One event per durable effect, plus initial phase and terminal boundaries.
    assert!(progress.len() <= action_count + 3);
    let advancing = progress
        .iter()
        .filter(|event| event.phase == FleetEnsurePhase::Infrastructure)
        .map(|event| event.applied_effects)
        .collect::<Vec<_>>();
    assert_eq!(advancing, (0..=applied).collect::<Vec<_>>());
    assert!(progress.iter().all(|event| {
        event.operation_id == plan.operation_id
            && event.plan_sha256 == plan.plan_sha256
            && event.reviewed_effects == action_count
            && event.applied_effects as usize <= event.reviewed_effects
    }));
}

fn assert_unfinished_action_projection(root: &std::path::Path, actions: &[&EnsureAction]) {
    let paths = crate::fleet_ensure::ops::EnsurePaths::under(root, "local", "test-fleet");
    let mut journal: FleetEnsureJournalRecord =
        serde_json::from_slice(&fs::read(&paths.journal).unwrap()).unwrap();
    // Concurrent completion can leave a hole; aggregate applied counts cannot locate it.
    journal.effects[0].state = EffectState::Intent;
    let pending = crate::fleet_ensure::ops::progress::next_action(actions, &journal).unwrap();
    assert_eq!(pending.target, actions[0].name());
    journal.effects[0].state = EffectState::Applied;
    assert_eq!(
        crate::fleet_ensure::ops::progress::next_action(actions, &journal),
        None
    );
}

fn is_pool_maintenance(action: &EnsureAction) -> bool {
    matches!(action, EnsureAction::FleetProtocol { action, .. }
        if matches!(action.as_ref(), CurrentFleetProtocolAction::MaintainPoolReadiness { .. }))
}

#[test]
#[expect(
    clippy::too_many_lines,
    reason = "one recovery proof retains the reviewed attempt budget through response loss and process restart"
)]
fn pool_maintenance_attempt_bound_survives_lost_response_and_restart() {
    let mut fixture = fixture();
    fixture.desired.protocol = Some(DesiredFleetProtocol {
        app_config: "canic.toml".to_string(),
        component_group_placements: Vec::new(),
        coordinator_candid: "coordinator.did".to_string(),
        root_candid: "root.did".to_string(),
        store_candid: "store.did".to_string(),
    });
    fixture
        .desired
        .canisters
        .iter_mut()
        .find(|canister| canister.kind == DesiredCanisterKind::Coordinator)
        .expect("fixture Coordinator")
        .wasm = None;
    let mut action = typed_protocol_action(&"19".repeat(32));
    let EnsureAction::FleetProtocol {
        action: current,
        name,
        ..
    } = &mut action
    else {
        unreachable!("typed fixture protocol");
    };
    **current = CurrentFleetProtocolAction::MaintainPoolReadiness {
        maximum_updates: 4,
        minimum_ready: 1,
        readiness_floor: Cycles::new(1),
    };
    *name = "pool-readiness".to_string();
    let action_hash = crate::fleet_ensure::ops::action_sha256(&action);
    fixture.platform.desired = fixture.desired.clone();
    fixture.platform.protocol_command_only = true;
    fixture.platform.protocol_retry = EffectRetry::ContinuePoolMaintenance;
    fixture.platform.fail_once.insert(action_hash.clone());
    let source = "29".repeat(32);
    let mut planned = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut fixture.platform,
    )
    .expect("review maintenance with four calls maximum");
    // Exercise the workflow from its durable reviewed-journal boundary.
    // Root policy admission is covered independently by the policy tests.
    planned.plan.protocol_actions.push(action.clone());
    planned.plan.plan_sha256 = crate::fleet_ensure::policy::expected_plan_sha256(&planned.plan);
    let paths = crate::fleet_ensure::ops::EnsurePaths::under(
        &fixture.root,
        &fixture.desired.environment,
        "test-fleet",
    );
    crate::fleet_ensure::ops::write_plan(&paths, &planned.plan)
        .expect("retain reviewed maintenance plan");
    crate::fleet_ensure::ops::write_journal(
        &paths,
        &FleetEnsureJournalRecord {
            funding_observations: BTreeMap::new(),
            funding_reviews: Vec::new(),
            successor_phases: Vec::new(),
            completion: FleetEnsureCompletion::InProgress,
            estate_funding_required: None,
            effects: Vec::new(),
            fleet: "test-fleet".to_string(),
            initial_controlled_cycles: planned.plan.conservation.observed_controlled_cycles,
            initial_estate_funding_cycles_by_root: BTreeMap::new(),
            initial_operator_cycles: fixture.platform.operator_cycles,
            operation_id: planned.plan.operation_id.clone(),
            plan_sha256: planned.plan.plan_sha256.clone(),
            schema_version: FLEET_ENSURE_SCHEMA_VERSION,
            stalled_observations: 0,
        },
    )
    .expect("retain reviewed operation before any call");
    fixture.platform.protocol_action = Some(action.clone());
    let lost = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        &planned.plan.plan_sha256,
        &mut fixture.platform,
    );
    assert!(matches!(
        lost,
        Err(workflow::EnsureWorkflowError::Platform(_))
    ));
    assert_eq!(fixture.platform.mutations.get(&action_hash), Some(&1));
    let journal = crate::fleet_ensure::ops::read_journal(&paths)
        .expect("read response-loss journal")
        .expect("durable journal");
    let retained = journal
        .effects
        .iter()
        .find(|effect| effect.action_sha256 == action_hash)
        .expect("maintenance intent");
    assert_eq!(retained.maintenance_attempts, 1);
    assert_eq!(retained.state, EffectState::Intent);

    let mut resumed = MockPlatform::new(
        fixture.desired.clone(),
        fixture.platform.live.values().cloned(),
    );
    resumed.operator_cycles = fixture.platform.operator_cycles;
    resumed.protocol_action = Some(action);
    resumed.protocol_command_only = true;
    resumed.protocol_retry = EffectRetry::ContinuePoolMaintenance;
    for _ in 0..2 {
        let exhausted = workflow::apply(
            &fixture.root,
            &fixture.desired,
            &source,
            "test-fleet",
            &planned.plan.plan_sha256,
            &mut resumed,
        );
        assert!(matches!(
            exhausted,
            Err(workflow::EnsureWorkflowError::PoolMaintenanceBound {
                maximum_updates: 4,
                ..
            })
        ));
        assert_eq!(resumed.mutations.get(&action_hash), Some(&3));
    }
    let journal = crate::fleet_ensure::ops::read_journal(&paths)
        .expect("read exhausted journal")
        .expect("durable journal");
    assert_eq!(
        journal
            .effects
            .iter()
            .find(|effect| effect.action_sha256 == action_hash)
            .expect("maintenance effect")
            .maintenance_attempts,
        4
    );

    assert_eq!(
        resumed.paced_observations,
        vec![0, 1, 2],
        "unchanged successful maintenance calls must retain backoff across retries"
    );
    resumed.protocol_ready.insert("pool-readiness".to_string());
    let terminal = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        &planned.plan.plan_sha256,
        &mut resumed,
    )
    .expect("terminal protected observation remains available at the attempt bound");
    assert!(terminal.terminal);
    assert_eq!(resumed.mutations.get(&action_hash), Some(&3));
    fs::remove_dir_all(fixture.root).expect("remove maintenance fixture");
}

#[test]
fn successful_create_response_retains_exact_applied_evidence_and_replays_without_effect() {
    assert_create_response_journey(false);
}

#[test]
fn duplicate_create_response_with_principal_retains_exact_applied_evidence_and_replays_without_effect()
 {
    assert_create_response_journey(true);
}

#[test]
fn first_create_balance_within_reviewed_observation_burn_converges() {
    let mut fixture = fixture();
    fixture.desired.maximum_observation_burn_cycles = "2".to_string();
    fixture
        .desired
        .canisters
        .iter_mut()
        .find(|canister| canister.name == "created")
        .expect("created desired canister")
        .minimum_cycles = "18".to_string();
    fixture
        .platform
        .create_shortfalls
        .insert("created".to_string(), 1);
    let desired_sha256 = "113".repeat(21) + "1";
    let planned = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut fixture.platform,
    )
    .expect("plan burn-bounded Create journey");

    let applied = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        &planned.plan.plan_sha256,
        &mut fixture.platform,
    )
    .expect("accept first live balance inside reviewed observation burn");
    assert!(applied.terminal);

    let paths = crate::fleet_ensure::ops::EnsurePaths::under(
        &fixture.root,
        &fixture.desired.environment,
        "test-fleet",
    );
    assert_retained_create_balance(&paths, 19);
    fs::remove_dir_all(fixture.root).expect("remove test directory");
}

#[test]
fn create_balance_terminal_predicate_is_exact_and_bounded() {
    let burn = crate::fleet_ensure::model::creation_observation_burn(1, 1).unwrap();
    assert!(create_balance_is_terminal(Some(998), 1_000, burn));
    assert!(!create_balance_is_terminal(Some(997), 1_000, burn));
    assert_eq!(
        crate::fleet_ensure::model::creation_observation_burn(u128::MAX, 1),
        None
    );
    assert!(create_balance_is_terminal(Some(1_000), 1_000, 0));
    assert!(create_balance_is_terminal(Some(999), 1_000, 1));
    assert!(!create_balance_is_terminal(Some(998), 1_000, 1));
    assert!(create_balance_is_terminal(Some(1_001), 1_000, 1));
    assert!(!create_balance_is_terminal(None, 1_000, 1));
}

fn assert_create_response_journey(response_is_duplicate: bool) {
    let mut fixture = fixture();
    let desired_sha256 = "102".repeat(21) + "1";
    let planned = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut fixture.platform,
    )
    .expect("plan production-shaped Create journey");
    let create = workflow::ordered_actions(&planned.plan)
        .into_iter()
        .find(|action| matches!(action, EnsureAction::Create { name, .. } if name == "created"))
        .expect("reviewed Create action");
    let create_hash = crate::fleet_ensure::ops::action_sha256(create);
    let EnsureAction::Create {
        requested_initial_cycles,
        ..
    } = create
    else {
        unreachable!("selected action is Create");
    };
    let requested_initial_cycles = *requested_initial_cycles;
    if response_is_duplicate {
        fixture.platform.fail_once.insert(create_hash.clone());
        let error = workflow::apply(
            &fixture.root,
            &fixture.desired,
            &desired_sha256,
            "test-fleet",
            &planned.plan.plan_sha256,
            &mut fixture.platform,
        )
        .expect_err("first Create response is lost after the Ledger effect");
        assert!(matches!(error, workflow::EnsureWorkflowError::Platform(_)));
    }

    let applied = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        &planned.plan.plan_sha256,
        &mut fixture.platform,
    )
    .expect("apply successful or duplicate-with-Principal Create response");
    assert!(applied.terminal);
    assert_eq!(fixture.platform.mutations.get(&create_hash), Some(&1));
    assert_eq!(
        fixture
            .platform
            .duplicate_create_responses
            .get(&create_hash)
            .copied()
            .unwrap_or_default(),
        u32::from(response_is_duplicate)
    );

    let paths = crate::fleet_ensure::ops::EnsurePaths::under(
        &fixture.root,
        &fixture.desired.environment,
        "test-fleet",
    );
    let journal = crate::fleet_ensure::ops::read_journal(&paths)
        .expect("read Create journal")
        .expect("retained Create journal");
    let effect = journal
        .effects
        .iter()
        .find(|effect| effect.action_sha256 == create_hash)
        .expect("retained Create effect");
    assert_eq!(effect.state, EffectState::Applied);
    assert_eq!(effect.created_principal.as_deref(), Some("created-created"));
    assert_eq!(effect.receipt.as_deref(), Some("create-block"));
    assert_eq!(effect.post_cycles, Some(requested_initial_cycles));
    assert_retained_create_balance(&paths, requested_initial_cycles);

    let mutations = fixture.platform.mutations.clone();
    let replay_plan = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        1_800_000_000_000_000_100,
        &mut fixture.platform,
    )
    .expect("plan terminal Create replay");
    assert!(workflow::ordered_actions(&replay_plan.plan).is_empty());
    let replay = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        &replay_plan.plan.plan_sha256,
        &mut fixture.platform,
    )
    .expect("apply effect-free Create replay");
    assert!(replay.terminal);
    assert_eq!(replay.effects_applied, 0);
    assert_eq!(fixture.platform.mutations, mutations);

    fs::remove_dir_all(fixture.root).expect("remove test directory");
}

fn completed_create_journey() -> (Fixture, String, FleetEnsurePlan, String, u128) {
    let mut fixture = fixture();
    let desired_sha256 = "102".repeat(21) + "2";
    let planned = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut fixture.platform,
    )
    .expect("plan retained Create journey");
    let create = workflow::ordered_actions(&planned.plan)
        .into_iter()
        .find(|action| matches!(action, EnsureAction::Create { name, .. } if name == "created"))
        .expect("reviewed Create action");
    let create_hash = crate::fleet_ensure::ops::action_sha256(create);
    let EnsureAction::Create {
        requested_initial_cycles,
        ..
    } = create
    else {
        unreachable!("selected action is Create");
    };
    let requested_initial_cycles = *requested_initial_cycles;
    workflow::apply(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        &planned.plan.plan_sha256,
        &mut fixture.platform,
    )
    .expect("complete retained Create journey");
    (
        fixture,
        desired_sha256,
        planned.plan,
        create_hash,
        requested_initial_cycles,
    )
}

#[test]
fn terminal_create_publication_rejects_a_conflicting_retained_balance() {
    let (mut fixture, desired_sha256, plan, create_hash, requested_initial_cycles) =
        completed_create_journey();
    let paths = crate::fleet_ensure::ops::EnsurePaths::under(
        &fixture.root,
        &fixture.desired.environment,
        "test-fleet",
    );
    let mut journal = crate::fleet_ensure::ops::read_journal(&paths)
        .expect("read completed Create journal")
        .expect("completed Create journal");
    journal.completion = FleetEnsureCompletion::InProgress;
    crate::fleet_ensure::ops::write_journal(&paths, &journal)
        .expect("reopen exact Create operation");
    let mut state = crate::fleet_ensure::ops::read_state(&paths, "test-fleet")
        .expect("read completed Create state");
    state
        .retained_cycles_by_principal
        .insert("created-created".to_string(), requested_initial_cycles - 1);
    crate::fleet_ensure::ops::write_state(&paths, &state)
        .expect("retain conflicting Create balance");
    let mutations = fixture.platform.mutations.clone();

    let error = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        &plan.plan_sha256,
        &mut fixture.platform,
    )
    .expect_err("conflicting retained Create balance must fail closed");
    assert!(matches!(
        error,
        workflow::EnsureWorkflowError::JournalIntegrity
    ));
    assert_eq!(fixture.platform.mutations, mutations);
    let retained_state = crate::fleet_ensure::ops::read_state(&paths, "test-fleet")
        .expect("read rejected Create state");
    assert_eq!(retained_state, state);
    let retained_journal = crate::fleet_ensure::ops::read_journal(&paths)
        .expect("read rejected Create journal")
        .expect("rejected Create journal");
    assert_eq!(retained_journal, journal);
    assert!(
        retained_journal
            .effects
            .iter()
            .any(|effect| effect.action_sha256 == create_hash)
    );

    fs::remove_dir_all(fixture.root).expect("remove test directory");
}

fn assert_retained_create_balance(paths: &crate::fleet_ensure::ops::EnsurePaths, expected: u128) {
    let state = crate::fleet_ensure::ops::read_state(paths, "test-fleet")
        .expect("read terminal Create state");
    assert_eq!(
        state.retained_cycles_by_principal.get("created-created"),
        Some(&expected)
    );
}

fn configure_fixture_publication(fixture: &mut Fixture) {
    let root = Principal::from_slice(&[71]);
    let store = Principal::from_slice(&[72]);
    let wasm = fixture.root.join("app.wasm");
    for (name, principal, kind) in [
        ("fixture-root", root, DesiredCanisterKind::Root),
        ("fixture-store", store, DesiredCanisterKind::Store),
    ] {
        let text = principal.to_text();
        let mut canister = desired_canister(name, Some(&text), false, &wasm, None);
        canister.kind = kind;
        canister.parent = Some(
            if kind == DesiredCanisterKind::Root {
                "treasury"
            } else {
                "fixture-root"
            }
            .to_string(),
        );
        canister.wasm = None;
        fixture.desired.canisters.push(canister);
        fixture.platform.live.insert(
            text.clone(),
            live(
                &text,
                500,
                Some(&sha256_hex(b"current-wasm")),
                true,
                &[CONTROLLER],
            ),
        );
    }
    fixture
        .desired
        .canisters
        .iter_mut()
        .find(|canister| canister.kind == DesiredCanisterKind::Coordinator)
        .unwrap()
        .wasm = None;
    fixture.desired.protocol = Some(DesiredFleetProtocol {
        app_config: "canic.toml".into(),
        component_group_placements: Vec::new(),
        coordinator_candid: "coordinator.did".into(),
        root_candid: "root.did".into(),
        store_candid: "store.did".into(),
    });
    fixture.platform.reviewed_protocol_actions = fixture_publication_actions(root, store);
}

fn fixture_publication_actions(root: Principal, store: Principal) -> Vec<EnsureAction> {
    let chunks = [b"first rows".as_slice(), b"last rows".as_slice()];
    let descriptor = canic_core::dto::fixture_provisioning::FixtureDescriptor {
        schema_version: 1,
        format_hash: [1; 32],
        completion_summary: [2; 32],
        encoded_length: chunks.iter().map(|bytes| bytes.len() as u64).sum(),
        chunks: chunks
            .iter()
            .map(
                |bytes| canic_core::dto::fixture_provisioning::FixtureChunkDescriptor {
                    digest: canic_core::cdk::utils::hash::wasm_hash(bytes)
                        .try_into()
                        .unwrap(),
                    length: u32::try_from(bytes.len()).unwrap(),
                },
            )
            .collect(),
    };
    let content_id =
        canic_control_plane::api::fixture_content::FixtureContentApi::content_id(&descriptor)
            .unwrap();
    let role = canic_core::ids::CanisterRole::new("app");
    let total = descriptor.encoded_length;
    let mut actions = vec![fleet_protocol_action(
        "prepare-fixture",
        CurrentFleetProtocolAction::PrepareStoreFixture {
            maximum_attempts: 1,
            request: canic_core::dto::root_store::RootStoreFixturePrepareRequest {
                bootstrap: canic_core::dto::root_store::RootStoreBootstrapRequest {
                    operation_id: [8; 32],
                    manifest_payload_size_bytes: 100,
                },
                role: role.clone(),
            },
            source: canic_core::dto::root_store::RootStoreFixture {
                role,
                content_id,
                descriptor,
            },
            store,
        },
    )];
    let mut received_bytes = 0;
    for (index, bytes) in chunks.iter().enumerate() {
        received_bytes += bytes.len() as u64;
        let action = CurrentFleetProtocolAction::PublishStoreFixtureChunk {
            maximum_attempts: 1,
            request: canic_core::dto::fixture_provisioning::FixtureChunkUpload {
                content_id,
                index: u32::try_from(index).unwrap(),
                bytes: bytes.to_vec(),
            },
            expected: canic_core::dto::fixture_provisioning::FixtureSourceStatus {
                content_id,
                next_chunk: u32::try_from(index).unwrap() + 1,
                chunk_count: u32::try_from(chunks.len()).unwrap(),
                received_bytes,
                complete: index + 1 == chunks.len(),
            },
            source_bytes: total,
        };
        actions.push(fleet_protocol_action(
            &format!("publish-fixture-{index}"),
            action,
        ));
    }
    for outer in &mut actions {
        if let EnsureAction::FleetProtocol {
            action,
            principal,
            candid,
            ..
        } = outer
        {
            let is_root = action.target_kind() == DesiredCanisterKind::Root;
            *principal = if is_root { root } else { store }.to_text();
            *candid = if is_root { "root.did" } else { "store.did" }.to_string();
        }
    }
    actions
}

fn publication_retry_fixture(maximum_attempts: u32) -> Fixture {
    let mut fixture = fixture();
    fixture
        .desired
        .canisters
        .retain(|canister| canister.name == "treasury");
    fixture
        .platform
        .live
        .retain(|principal, _| principal == TREASURY);
    configure_fixture_publication(&mut fixture);
    for outer in &mut fixture.platform.reviewed_protocol_actions {
        let EnsureAction::FleetProtocol {
            action,
            maximum_execution_burn_cycles,
            ..
        } = outer
        else {
            unreachable!()
        };
        match action.as_mut() {
            CurrentFleetProtocolAction::PrepareStoreFixture {
                maximum_attempts: limit,
                ..
            }
            | CurrentFleetProtocolAction::PublishStoreFixtureChunk {
                maximum_attempts: limit,
                ..
            } => *limit = maximum_attempts,
            _ => unreachable!(),
        }
        *maximum_execution_burn_cycles = u128::from(maximum_attempts);
    }
    fixture.platform.desired = fixture.desired.clone();
    fixture
}

#[test]
fn fixture_publication_attempt_bound_survives_failure_and_restart() {
    // Exercise both preparation and a later chunk after an earlier committed prefix.
    for selected in [0, 2] {
        let mut fixture = publication_retry_fixture(2);
        let planned = workflow::plan(
            &fixture.root,
            &fixture.desired,
            "publication",
            "test-fleet",
            1,
            &mut fixture.platform,
        )
        .unwrap();
        let paths =
            crate::fleet_ensure::ops::EnsurePaths::under(&fixture.root, "local", "test-fleet");
        let action = planned.plan.protocol_actions[selected].clone();
        let hash = action_sha256(&action);
        for attempt in 1..=2 {
            // Reconstruct the process; only observed external state and the on-disk journal survive.
            let mut resumed = MockPlatform::new(
                fixture.desired.clone(),
                fixture.platform.live.values().cloned(),
            );
            resumed.protocol_ready = fixture.platform.protocol_ready.clone();
            resumed.reviewed_protocol_actions = fixture.platform.reviewed_protocol_actions.clone();
            resumed.publication_journal = Some(paths.journal.clone());
            resumed.stall_before_mutation(hash.clone(), 1);
            let result = workflow::apply(
                &fixture.root,
                &fixture.desired,
                "publication",
                "test-fleet",
                &planned.plan.plan_sha256,
                &mut resumed,
            );
            assert!(matches!(
                result,
                Err(workflow::EnsureWorkflowError::Platform(_)
                    | workflow::EnsureWorkflowError::Stalled { .. })
            ));
            assert_eq!(resumed.publication_attempts_on_issue.last(), Some(&attempt));
            let journal = crate::fleet_ensure::ops::read_journal(&paths)
                .unwrap()
                .unwrap();
            let record = journal
                .effects
                .iter()
                .find(|effect| effect.action_sha256 == hash)
                .unwrap();
            assert_eq!(record.publication_attempts, attempt);
            assert_eq!(record.state, EffectState::Intent);
            fixture.platform = resumed;
        }
        fixture.platform.publication_attempts_on_issue.clear();
        for _ in 0..2 {
            assert!(matches!(
                workflow::apply(
                    &fixture.root,
                    &fixture.desired,
                    "publication",
                    "test-fleet",
                    &planned.plan.plan_sha256,
                    &mut fixture.platform
                ),
                Err(workflow::EnsureWorkflowError::FixturePublicationBound {
                    maximum_attempts: 2,
                    ..
                })
            ));
            assert!(fixture.platform.publication_attempts_on_issue.is_empty());
            assert!(!fixture.platform.mutations.contains_key(&hash));
        }
        fs::remove_dir_all(fixture.root).unwrap();
    }
}

#[test]
fn fixture_publication_last_attempt_lost_reply_reconciles_without_another_call() {
    let mut fixture = publication_retry_fixture(1);
    let planned = workflow::plan(
        &fixture.root,
        &fixture.desired,
        "publication",
        "test-fleet",
        1,
        &mut fixture.platform,
    )
    .unwrap();
    let paths = crate::fleet_ensure::ops::EnsurePaths::under(&fixture.root, "local", "test-fleet");
    fixture.platform.publication_journal = Some(paths.journal.clone());
    fixture.platform.fail_once = planned
        .plan
        .protocol_actions
        .iter()
        .map(action_sha256)
        .collect();
    loop {
        match workflow::apply(
            &fixture.root,
            &fixture.desired,
            "publication",
            "test-fleet",
            &planned.plan.plan_sha256,
            &mut fixture.platform,
        ) {
            Ok(report) => {
                assert!(report.terminal);
                break;
            }
            Err(workflow::EnsureWorkflowError::Platform(_)) => {
                let mut resumed = MockPlatform::new(
                    fixture.desired.clone(),
                    fixture.platform.live.values().cloned(),
                );
                resumed.protocol_ready = fixture.platform.protocol_ready.clone();
                resumed.reviewed_protocol_actions =
                    fixture.platform.reviewed_protocol_actions.clone();
                resumed.fail_once = fixture.platform.fail_once.clone();
                resumed.publication_journal = Some(paths.journal.clone());
                fixture.platform = resumed;
            }
            Err(error) => panic!("unexpected publication recovery: {error}"),
        }
    }
    let journal = crate::fleet_ensure::ops::read_journal(&paths)
        .unwrap()
        .unwrap();
    assert!(!journal.effects.is_empty());
    assert!(
        journal
            .effects
            .iter()
            .all(|effect| effect.publication_attempts == 1 && effect.state == EffectState::Applied)
    );
    assert!(fixture.platform.publication_attempts_on_issue.is_empty());
    assert!(
        workflow::apply(
            &fixture.root,
            &fixture.desired,
            "publication",
            "test-fleet",
            &planned.plan.plan_sha256,
            &mut fixture.platform
        )
        .unwrap()
        .terminal
    );
    assert!(fixture.platform.publication_attempts_on_issue.is_empty());
    fs::remove_dir_all(fixture.root).unwrap();
}

#[test]
fn fixture_publication_review_accounts_for_each_retry_and_rejects_zero_attempts() {
    let mut fixture = publication_retry_fixture(1);
    fixture.desired.maximum_observation_burn_cycles = "1".into();
    let one = workflow::plan(
        &fixture.root,
        &fixture.desired,
        "publication",
        "test-fleet",
        1,
        &mut fixture.platform,
    )
    .unwrap();
    let mut retry = publication_retry_fixture(3);
    retry.desired.maximum_observation_burn_cycles = "1".into();
    let three = workflow::plan(
        &retry.root,
        &retry.desired,
        "publication",
        "test-fleet",
        1,
        &mut retry.platform,
    )
    .unwrap();
    let count = one.plan.protocol_actions.len() as u128;
    // Two extra update calls and their three observations per action.
    assert_eq!(
        three.plan.conservation.maximum_execution_burn_cycles
            - one.plan.conservation.maximum_execution_burn_cycles,
        count * 2 * 4
    );
    assert_ne!(one.plan.plan_sha256, three.plan.plan_sha256);
    let mut invalid = publication_retry_fixture(0);
    assert!(matches!(
        workflow::plan(
            &invalid.root,
            &invalid.desired,
            "publication",
            "test-fleet",
            1,
            &mut invalid.platform
        ),
        Err(workflow::EnsureWorkflowError::Policy(
            crate::fleet_ensure::policy::EnsurePolicyError::InvalidProtocolStep(_)
        ))
    ));
    for root in [fixture.root, retry.root, invalid.root] {
        fs::remove_dir_all(root).unwrap();
    }
}

#[test]
#[expect(
    clippy::too_many_lines,
    reason = "one durable journal journey covers every lost effect response and terminal replay"
)]
fn interruption_at_every_effect_converges_once_and_second_run_has_zero_effects() {
    let mut fixture = fixture();
    configure_fixture_publication(&mut fixture);
    fixture
        .desired
        .protocol_steps
        .push(protocol_step(&fixture.root));
    fixture.platform.desired = fixture.desired.clone();
    let root = fixture.root.clone();
    let desired_sha256 = "a".repeat(64);
    let mut platform = fixture.platform;
    let planned = workflow::plan(
        &root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut platform,
    )
    .expect("compile reviewed plan");
    let actions = planned
        .plan
        .canisters
        .iter()
        .flat_map(|canister| canister.actions.iter())
        .chain(planned.plan.protocol_actions.iter())
        .collect::<Vec<_>>();
    assert!(!actions.is_empty());
    let hashes = actions
        .iter()
        .map(|action| action_sha256(action))
        .collect::<BTreeSet<_>>();
    assert_eq!(hashes.len(), actions.len());
    assert!(actions.iter().any(|action| matches!(action,
        EnsureAction::FleetProtocol { action, .. }
            if matches!(action.as_ref(), CurrentFleetProtocolAction::PublishStoreFixtureChunk { .. }))));
    platform.fail_once = actions
        .iter()
        .map(|action| crate::fleet_ensure::ops::action_sha256(action))
        .collect();

    let report = loop {
        match workflow::apply(
            &root,
            &fixture.desired,
            &desired_sha256,
            "test-fleet",
            &planned.plan.plan_sha256,
            &mut platform,
        ) {
            Ok(report) => break report,
            Err(workflow::EnsureWorkflowError::Platform(MockError)) => {
                assert!(matches!(
                    crate::fleet_ensure::read_last_converged_fleet_inventory(
                        &root,
                        &fixture.desired.environment,
                        "test-fleet",
                    ),
                    Err(crate::fleet_ensure::CurrentFleetInventoryError::NotConverged { .. })
                ));
            }
            Err(error) => panic!("unexpected resume result: {error}"),
        }
    };
    assert!(report.terminal);
    let inventory = crate::fleet_ensure::read_last_converged_fleet_inventory(
        &root,
        &fixture.desired.environment,
        "test-fleet",
    )
    .expect("terminal ensure inventory");
    assert_eq!(
        inventory.entries.len(),
        fixture
            .desired
            .canisters
            .iter()
            .filter(|canister| canister.presence == DesiredPresence::Present)
            .count()
    );
    let conservation = report
        .actual_conservation
        .expect("terminal conservation proof");
    assert_eq!(
        conservation.observed_starting_cycles + conservation.received_new_funding_cycles
            - conservation.observed_net_cycle_debit_cycles,
        conservation.final_controlled_cycles
    );
    assert!(platform.mutations.values().all(|count| *count == 1));
    let mutation_count = platform.mutations.values().sum::<u32>();

    let no_effect_plan = workflow::plan(
        &root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        1_800_000_000_000_000_100,
        &mut platform,
    )
    .expect("plan converged Fleet");
    assert!(
        no_effect_plan
            .plan
            .canisters
            .iter()
            .all(|canister| canister.actions.is_empty())
    );
    assert!(no_effect_plan.plan.protocol_actions.is_empty());
    let terminal = workflow::apply(
        &root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        &no_effect_plan.plan.plan_sha256,
        &mut platform,
    )
    .expect("effect-free terminal replay");
    assert!(terminal.terminal);
    assert_eq!(platform.mutations.values().sum::<u32>(), mutation_count);
}

#[test]
#[expect(
    clippy::too_many_lines,
    reason = "one proof keeps the retained thirteen-applied/one-issued journal and replay assertions visible"
)]
fn retryable_provisioning_failure_replays_only_the_exact_retained_issued_command() {
    let mut fixture = fixture();
    fixture.desired.protocol = Some(DesiredFleetProtocol {
        app_config: "canic.toml".to_string(),
        component_group_placements: Vec::new(),
        coordinator_candid: "coordinator.did".to_string(),
        root_candid: "root.did".to_string(),
        store_candid: "store.did".to_string(),
    });
    fixture
        .desired
        .canisters
        .iter_mut()
        .find(|canister| canister.kind == DesiredCanisterKind::Coordinator)
        .expect("fixture Coordinator")
        .wasm = None;
    fixture.platform.desired = fixture.desired.clone();
    let desired_sha256 = "45".repeat(32);
    let initial = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut fixture.platform,
    )
    .expect("initial plan");
    workflow::apply(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        &initial.plan.plan_sha256,
        &mut fixture.platform,
    )
    .expect("initial convergence");

    fixture.platform.completed.clear();
    fixture.platform.mutations.clear();
    fixture.platform.typed_protocol = true;
    let pending = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        1_800_000_000_000_000_100,
        &mut fixture.platform,
    )
    .expect("typed provisioning plan");
    assert!(
        pending
            .plan
            .canisters
            .iter()
            .all(|canister| canister.actions.is_empty())
    );
    let [issued] = pending.plan.protocol_actions.as_slice() else {
        panic!("expected one typed provisioning action");
    };
    let issued = issued.clone();
    let mut retained = pending.plan;
    retained.protocol_actions.clear();
    for index in 0..13 {
        let mut applied = issued.clone();
        let EnsureAction::FleetProtocol { name, .. } = &mut applied else {
            panic!("typed provisioning action");
        };
        *name = format!("applied-{index:02}");
        retained.protocol_actions.push(applied);
    }
    retained.protocol_actions.push(issued.clone());
    retained.plan_sha256 = crate::fleet_ensure::policy::expected_plan_sha256(&retained);

    let paths = crate::fleet_ensure::ops::EnsurePaths::under(
        &fixture.root,
        &fixture.desired.environment,
        "test-fleet",
    );
    crate::fleet_ensure::ops::write_plan(&paths, &retained).expect("write retained plan");
    let effects = retained
        .protocol_actions
        .iter()
        .enumerate()
        .map(|(index, action)| EffectRecord {
            publication_attempts: 0,
            maintenance_attempts: 0,
            action_sha256: crate::fleet_ensure::ops::action_sha256(action),
            created_principal: None,
            destination_post_cycles: None,
            destination_pre_cycles: None,
            post_cycles: None,
            pre_cycles: None,
            pre_canister_version: None,
            progress_identity: Some(format!("retained-{index}")),
            receipt: Some("protocol-receipt".to_string()),
            state: if index == 13 {
                EffectState::Issued
            } else {
                EffectState::Applied
            },
        })
        .collect();
    crate::fleet_ensure::ops::write_journal(
        &paths,
        &FleetEnsureJournalRecord {
            funding_observations: BTreeMap::new(),
            funding_reviews: Vec::new(),
            successor_phases: Vec::new(),
            completion: FleetEnsureCompletion::InProgress,
            estate_funding_required: None,
            effects,
            fleet: "test-fleet".to_string(),
            initial_controlled_cycles: retained.conservation.observed_controlled_cycles,
            initial_estate_funding_cycles_by_root: BTreeMap::new(),
            initial_operator_cycles: fixture.platform.operator_cycles,
            operation_id: retained.operation_id.clone(),
            plan_sha256: retained.plan_sha256.clone(),
            schema_version: FLEET_ENSURE_SCHEMA_VERSION,
            stalled_observations: 0,
        },
    )
    .expect("write retained issued journal");
    fixture.platform.protocol_retry = EffectRetry::ReplayExactIssuedCommand;
    let identities = fixture
        .platform
        .live
        .keys()
        .cloned()
        .collect::<BTreeSet<_>>();
    let cycles = fixture
        .platform
        .live
        .iter()
        .map(|(principal, live)| (principal.clone(), live.cycles))
        .collect::<BTreeMap<_, _>>();
    let operator_cycles = fixture.platform.operator_cycles;

    let report = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        &retained.plan_sha256,
        &mut fixture.platform,
    )
    .expect("resume retained issued provisioning");

    assert!(report.terminal);
    let issued_hash = crate::fleet_ensure::ops::action_sha256(&issued);
    assert_eq!(fixture.platform.mutations.get(&issued_hash), Some(&1));
    assert_eq!(fixture.platform.mutations.values().sum::<u32>(), 1);
    assert_eq!(
        fixture
            .platform
            .live
            .keys()
            .cloned()
            .collect::<BTreeSet<_>>(),
        identities
    );
    assert_eq!(
        fixture
            .platform
            .live
            .iter()
            .map(|(principal, live)| (principal.clone(), live.cycles))
            .collect::<BTreeMap<_, _>>(),
        cycles
    );
    assert_eq!(fixture.platform.operator_cycles, operator_cycles);
    let journal = crate::fleet_ensure::ops::read_journal(&paths)
        .expect("read terminal journal")
        .expect("terminal journal");
    assert_eq!(journal.effects.len(), 14);
    assert!(
        journal
            .effects
            .iter()
            .all(|effect| effect.state == EffectState::Applied)
    );
}

#[test]
#[expect(
    clippy::too_many_lines,
    reason = "one proof keeps same-ID Start, cycle conservation, and effect-free replay together"
)]
fn stopped_retained_coordinator_starts_same_id_then_replays_without_effect() {
    let mut fixture = fixture();
    let desired_sha256 = "46".repeat(32);
    let initial = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut fixture.platform,
    )
    .expect("initial plan");
    workflow::apply(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        &initial.plan.plan_sha256,
        &mut fixture.platform,
    )
    .expect("initial convergence");
    fixture.platform.completed.clear();
    fixture.platform.mutations.clear();
    fixture
        .platform
        .live
        .get_mut(TREASURY)
        .expect("retained Coordinator")
        .status = CanisterRuntimeStatus::Stopped;
    let identities = fixture
        .platform
        .live
        .keys()
        .cloned()
        .collect::<BTreeSet<_>>();
    let cycles = fixture
        .platform
        .live
        .iter()
        .map(|(principal, live)| (principal.clone(), live.cycles))
        .collect::<BTreeMap<_, _>>();

    let start = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        1_800_000_000_000_000_100,
        &mut fixture.platform,
    )
    .expect("same-ID Start plan");
    let actions = workflow::ordered_actions(&start.plan);
    let [action] = actions.as_slice() else {
        panic!("expected one same-ID Start action");
    };
    assert!(matches!(
        action,
        EnsureAction::Start { principal, .. } if principal == TREASURY
    ));
    let action_hash = crate::fleet_ensure::ops::action_sha256(action);
    workflow::apply(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        &start.plan.plan_sha256,
        &mut fixture.platform,
    )
    .expect("same-ID Start convergence");
    assert_eq!(fixture.platform.mutations.get(&action_hash), Some(&1));
    assert_eq!(fixture.platform.mutations.values().sum::<u32>(), 1);
    assert_eq!(
        fixture
            .platform
            .live
            .keys()
            .cloned()
            .collect::<BTreeSet<_>>(),
        identities
    );
    assert_eq!(
        fixture
            .platform
            .live
            .iter()
            .map(|(principal, live)| (principal.clone(), live.cycles))
            .collect::<BTreeMap<_, _>>(),
        cycles
    );

    let replay = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        1_800_000_000_000_000_200,
        &mut fixture.platform,
    )
    .expect("effect-free replay plan");
    assert!(workflow::ordered_actions(&replay.plan).is_empty());
    workflow::apply(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        &replay.plan.plan_sha256,
        &mut fixture.platform,
    )
    .expect("effect-free replay");
    assert_eq!(fixture.platform.mutations.values().sum::<u32>(), 1);
}

#[test]
fn conservation_equation_accounts_for_funding_fees_and_burn_separately() {
    let fixture = fixture();
    let mut platform = fixture.platform;
    let report = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &"b".repeat(64),
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut platform,
    )
    .expect("compile plan");
    let proof = report.plan.conservation;
    assert_eq!(
        proof.observed_controlled_cycles + proof.maximum_operator_debit_cycles
            - proof.maximum_unavoidable_fee_cycles
            - proof.maximum_execution_burn_cycles,
        proof.expected_post_operation_cycles
    );
    assert!(proof.scheduled_transfer_cycles > 0);
    assert!(proof.retained_in_reused_canisters_cycles > 0);
    assert!(proof.maximum_new_funding_cycles > 0);
}

#[test]
fn infrastructure_install_order_keeps_store_before_root_initialization() {
    let coordinator = install_action(
        "coordinator",
        DesiredCanisterInit::Coordinator,
        crate::fleet_ensure::model::InstallMode::Reinstall,
    );
    let store = install_action(
        "store",
        DesiredCanisterInit::Store {
            root: "root".to_string(),
        },
        crate::fleet_ensure::model::InstallMode::Reinstall,
    );
    let root = install_action(
        "root",
        DesiredCanisterInit::Root {
            root: "root".to_string(),
        },
        crate::fleet_ensure::model::InstallMode::Reinstall,
    );

    assert!(workflow::action_order(&coordinator) < workflow::action_order(&store));
    assert!(workflow::action_order(&store) < workflow::action_order(&root));
}

#[test]
#[expect(
    clippy::too_many_lines,
    reason = "one interruption journey checks Registry authority before and after all infrastructure reinstalls"
)]
fn active_registry_is_retired_only_after_every_infrastructure_reinstall_is_applied() {
    let mut fixture = fixture();
    let mut plan = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &"70".repeat(32),
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut fixture.platform,
    )
    .expect("compile fixture plan")
    .plan;
    fixture.desired.protocol = Some(DesiredFleetProtocol {
        app_config: "canic.toml".to_string(),
        component_group_placements: Vec::new(),
        coordinator_candid: "coordinator.did".to_string(),
        root_candid: "root.did".to_string(),
        store_candid: "store.did".to_string(),
    });
    let roles = [
        (
            "treasury",
            DesiredCanisterKind::Coordinator,
            DesiredCanisterInit::Coordinator,
        ),
        (
            "app",
            DesiredCanisterKind::Root,
            DesiredCanisterInit::Root {
                root: "app".to_string(),
            },
        ),
        (
            "created",
            DesiredCanisterKind::Store,
            DesiredCanisterInit::Store {
                root: "app".to_string(),
            },
        ),
    ];
    let mut planned = Vec::new();
    let mut effects = Vec::new();
    for (name, kind, init) in roles {
        fixture
            .desired
            .canisters
            .iter_mut()
            .find(|canister| canister.name == name)
            .expect("fixture infrastructure role")
            .kind = kind;
        let action = install_action(
            name,
            init,
            crate::fleet_ensure::model::InstallMode::Reinstall,
        );
        effects.push(EffectRecord {
            publication_attempts: 0,
            maintenance_attempts: 0,
            action_sha256: crate::fleet_ensure::ops::action_sha256(&action),
            created_principal: None,
            destination_post_cycles: None,
            destination_pre_cycles: None,
            post_cycles: None,
            pre_cycles: Some(1),
            pre_canister_version: Some(1),
            progress_identity: Some(format!("applied-{name}")),
            receipt: None,
            state: EffectState::Applied,
        });
        planned.push(crate::fleet_ensure::model::CanisterPlan {
            actions: vec![action],
            disposition: CanisterDisposition::Reinstall,
            name: name.to_string(),
            observed_cycles: 1,
            principal: Some(Principal::from_slice(&[99; 29]).to_text()),
        });
    }
    plan.canisters = planned;
    plan.protocol_actions.clear();
    let mut journal = FleetEnsureJournalRecord {
        funding_observations: BTreeMap::new(),
        funding_reviews: Vec::new(),
        successor_phases: Vec::new(),
        completion: FleetEnsureCompletion::InProgress,
        estate_funding_required: None,
        effects,
        fleet: plan.fleet.clone(),
        initial_controlled_cycles: 3,
        initial_estate_funding_cycles_by_root: BTreeMap::new(),
        initial_operator_cycles: 0,
        operation_id: plan.operation_id.clone(),
        plan_sha256: plan.plan_sha256.clone(),
        schema_version: FLEET_ENSURE_SCHEMA_VERSION,
        stalled_observations: 0,
    };

    assert!(workflow::completed_infrastructure_reinstall(
        &fixture.desired,
        &plan,
        &journal
    ));
    journal.effects[2].state = EffectState::Issued;
    assert!(!workflow::completed_infrastructure_reinstall(
        &fixture.desired,
        &plan,
        &journal
    ));
}

#[test]
fn same_module_reinstall_requires_a_newer_canister_version() {
    use crate::fleet_ensure::{model::InstallMode, ops::install_effect_applied};

    assert!(!install_effect_applied(
        InstallMode::Reinstall,
        "same",
        Some("same"),
        Some(7),
        Some(7),
    ));
    assert!(install_effect_applied(
        InstallMode::Reinstall,
        "same",
        Some("same"),
        Some(7),
        Some(8),
    ));
    assert!(!install_effect_applied(
        InstallMode::Reinstall,
        "same",
        Some("different"),
        Some(7),
        Some(8),
    ));
}

#[test]
fn ledger_withdraw_completion_uses_the_reviewed_burn_aware_lower_bound() {
    use crate::fleet_ensure::ops::{NativeFundingObservation, native_funding_applied};

    let observation = NativeFundingObservation {
        amount: 1_000_000_310_113,
        expected_post_cycles: 2_900_000_000_000,
        funding_deficit_cycles: 310_113,
        funding_margin_cycles: 1_000_000_000_000,
        live_cycles: Some(2_898_749_313_788),
        pre_cycles: Some(1_899_999_689_887),
    };
    assert!(native_funding_applied(observation));
    assert!(!native_funding_applied(NativeFundingObservation {
        live_cycles: Some(1_899_999_689_887),
        ..observation
    }));
    assert!(!native_funding_applied(NativeFundingObservation {
        live_cycles: None,
        ..observation
    }));
    assert!(!native_funding_applied(NativeFundingObservation {
        funding_deficit_cycles: observation.funding_deficit_cycles + 1,
        ..observation
    }));
}

#[test]
fn ledger_withdraw_completion_includes_burn_between_review_and_intent() {
    use crate::fleet_ensure::ops::{NativeFundingObservation, native_funding_applied};

    let observation = NativeFundingObservation {
        amount: 3_500_004_258_258,
        expected_post_cycles: 8_000_000_000_000,
        funding_deficit_cycles: 1_500_004_258_258,
        funding_margin_cycles: 2_000_000_000_000,
        live_cycles: Some(7_999_998_049_258),
        pre_cycles: Some(4_499_995_491_646),
    };
    assert!(native_funding_applied(observation));
    for live_cycles in [6_000_000_000_000, 7_999_999_749_904, 9_000_000_000_000] {
        assert!(native_funding_applied(NativeFundingObservation {
            live_cycles: Some(live_cycles),
            ..observation
        }));
    }
    assert!(!native_funding_applied(NativeFundingObservation {
        live_cycles: Some(5_999_999_999_999),
        ..observation
    }));
    assert!(native_funding_applied(NativeFundingObservation {
        pre_cycles: Some(4_499_995_741_743),
        ..observation
    }));
}

#[test]
fn issued_funding_reconciles_burn_and_replays_without_a_second_withdrawal() {
    let mut fixture = fixture();
    fixture.platform.funding_journal = Some(
        crate::fleet_ensure::ops::EnsurePaths::under(&fixture.root, "local", "test-fleet").journal,
    );
    let desired_sha256 = "116".repeat(21) + "1";
    let planned = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut fixture.platform,
    )
    .expect("plan funding reconciliation journey");
    let funding = workflow::ordered_actions(&planned.plan)
        .into_iter()
        .find(|action| matches!(action, EnsureAction::Fund { name, .. } if name == "app"))
        .expect("reviewed App funding action");
    let funding_hash = crate::fleet_ensure::ops::action_sha256(funding);
    let EnsureAction::Fund {
        expected_post_cycles,
        funding_margin_cycles,
        principal,
        ..
    } = funding
    else {
        unreachable!("selected action is Fund");
    };
    assert!(*funding_margin_cycles > 1);
    fixture.platform.fail_once.insert(funding_hash.clone());

    let first = apply_fixture_plan(&mut fixture, &desired_sha256, &planned.plan)
        .expect_err("lose the first Ledger response after its exact effect");
    assert!(matches!(first, workflow::EnsureWorkflowError::Platform(_)));
    assert_eq!(
        fixture.platform.funding_reads[&funding_hash],
        [MockFundingRead::Effect(EffectState::Intent)]
    );
    let live = fixture
        .platform
        .live
        .get_mut(principal)
        .expect("funded App remains controlled");
    live.cycles = expected_post_cycles - 1;

    let resumed = apply_fixture_plan(&mut fixture, &desired_sha256, &planned.plan)
        .expect("adopt exact duplicate receipt and burn-aware live balance");
    assert!(resumed.terminal);
    assert_eq!(fixture.platform.mutations.get(&funding_hash), Some(&1));
    assert_eq!(
        fixture.platform.funding_reads[&funding_hash],
        [
            MockFundingRead::Effect(EffectState::Intent),
            MockFundingRead::Effect(EffectState::Intent),
            MockFundingRead::Effect(EffectState::Issued),
        ]
    );

    let paths = crate::fleet_ensure::ops::EnsurePaths::under(
        &fixture.root,
        &fixture.desired.environment,
        "test-fleet",
    );
    let journal = crate::fleet_ensure::ops::read_journal(&paths)
        .expect("read funding journal")
        .expect("retained funding journal");
    let effect = journal
        .effects
        .iter()
        .find(|effect| effect.action_sha256 == funding_hash)
        .expect("retained funding effect");
    assert_eq!(effect.state, EffectState::Applied);
    assert_eq!(effect.receipt.as_deref(), Some("withdraw-block"));
    assert_eq!(effect.post_cycles, Some(expected_post_cycles - 1));
    let state =
        crate::fleet_ensure::ops::read_state(&paths, "test-fleet").expect("read funding state");
    assert_eq!(
        state.retained_cycles_by_principal.get(principal),
        Some(&(expected_post_cycles - 1))
    );

    let mutations = fixture.platform.mutations.clone();
    let replay_plan = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        1_800_000_000_000_000_100,
        &mut fixture.platform,
    )
    .expect("plan effect-free funding replay");
    assert!(workflow::ordered_actions(&replay_plan.plan).is_empty());
    let replay = apply_fixture_plan(&mut fixture, &desired_sha256, &replay_plan.plan)
        .expect("terminal funding replay");
    assert!(replay.terminal);
    assert_eq!(fixture.platform.mutations, mutations);

    fs::remove_dir_all(fixture.root).expect("remove test directory");
}

#[test]
fn operator_shortfall_rejects_before_journalling_and_funded_retry_reconciles() {
    let mut fixture = fixture();
    let source = sha256_hex(b"operator shortfall admission");
    let planned = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut fixture.platform,
    )
    .unwrap();
    let required = planned.plan.conservation.maximum_operator_debit_cycles;
    let paths = crate::fleet_ensure::ops::EnsurePaths::under(&fixture.root, "local", "test-fleet");
    let original = fs::read(&paths.plan).unwrap();
    for actual in [0, required - 1] {
        fixture.platform.set_operator_cycles(actual);
        assert!(matches!(
            apply_fixture_plan(&mut fixture, &source, &planned.plan),
            Err(workflow::EnsureWorkflowError::InsufficientOperatorCycles {
                actual: observed, required: debit,
            }) if observed == actual && debit == required
        ));
        assert!(!paths.journal.exists());
        assert_eq!(fs::read(&paths.plan).unwrap(), original);
        assert!(fixture.platform.mutations.is_empty());
    }
    let funding = workflow::ordered_actions(&planned.plan)
        .into_iter()
        .find(|action| matches!(action, EnsureAction::Fund { .. }))
        .unwrap();
    let funding_hash = action_sha256(funding);
    fixture.platform.fail_once(funding_hash.clone());
    fixture.platform.set_operator_cycles(required);
    assert!(matches!(
        apply_fixture_plan(&mut fixture, &source, &planned.plan),
        Err(workflow::EnsureWorkflowError::Platform(_))
    ));
    assert!(fixture.platform.operator_cycles < required);
    let resumed = apply_fixture_plan(&mut fixture, &source, &planned.plan).unwrap();
    assert!(resumed.terminal);
    assert_eq!(fixture.platform.mutation_count(&funding_hash), 1);
    let journal = crate::fleet_ensure::ops::read_journal(&paths)
        .unwrap()
        .unwrap();
    assert_eq!(journal.initial_operator_cycles, required);
    let mutations = fixture.platform.mutations.clone();
    let replay = apply_fixture_plan(&mut fixture, &source, &planned.plan).unwrap();
    assert!(replay.terminal);
    assert_eq!(fixture.platform.mutations, mutations);
    fs::remove_dir_all(fixture.root).unwrap();
}

#[test]
fn failed_initial_funding_observation_never_persists_or_pays_that_intent() {
    let mut fixture = fixture();
    let source = "f4".repeat(32);
    let planned = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut fixture.platform,
    )
    .unwrap();
    let funding = workflow::ordered_actions(&planned.plan)
        .into_iter()
        .find(|action| matches!(action, EnsureAction::Fund { .. }))
        .unwrap();
    let funding_hash = action_sha256(funding);
    let paths = crate::fleet_ensure::ops::EnsurePaths::under(&fixture.root, "local", "test-fleet");
    fixture.platform.funding_observation_failures = 1;
    fixture.platform.funding_journal = Some(paths.journal.clone());
    assert!(matches!(
        apply_fixture_plan(&mut fixture, &source, &planned.plan),
        Err(workflow::EnsureWorkflowError::Platform(_))
    ));
    let journal = crate::fleet_ensure::ops::read_journal(&paths)
        .unwrap()
        .unwrap();
    assert!(
        !journal
            .effects
            .iter()
            .any(|effect| effect.action_sha256 == funding_hash)
    );
    assert_eq!(fixture.platform.mutation_count(&funding_hash), 0);
    assert_eq!(
        fixture.platform.funding_reads[&funding_hash],
        [MockFundingRead::Effect(EffectState::Intent)]
    );
    let resumed = apply_fixture_plan(&mut fixture, &source, &planned.plan).unwrap();
    assert!(resumed.terminal);
    assert_eq!(fixture.platform.mutation_count(&funding_hash), 1);
    assert_eq!(
        fixture.platform.funding_reads[&funding_hash],
        [
            MockFundingRead::Effect(EffectState::Intent),
            MockFundingRead::Effect(EffectState::Intent),
            MockFundingRead::Effect(EffectState::Issued),
        ]
    );
    fs::remove_dir_all(fixture.root).unwrap();
}

#[test]
#[expect(
    clippy::too_many_lines,
    reason = "one retained-withdrawal journey checks immutable authority, lost reply, conservation and terminal replay"
)]
fn operator_mint_review_preserves_an_underfunded_original_withdrawal() {
    let mut fixture = fixture();
    fixture
        .desired
        .canisters
        .retain(|canister| canister.name == "treasury");
    fixture.desired.canisters[0].minimum_cycles = "10000".into();
    fixture.desired.canisters[0].initial_cycles = "10000".into();
    fixture.platform = MockPlatform::new(
        fixture.desired.clone(),
        [live(
            TREASURY,
            500,
            Some(&sha256_hex(b"current-wasm")),
            true,
            &[CONTROLLER],
        )],
    );
    let source = sha256_hex(b"retained initial operator shortfall");
    let mut report = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut fixture.platform,
    )
    .unwrap();
    let action = workflow::ordered_actions(&report.plan)[0].clone();
    assert!(matches!(action, EnsureAction::Fund { .. }));
    let paths = crate::fleet_ensure::ops::EnsurePaths::under(&fixture.root, "local", "test-fleet");
    // Reproduce the released admission bug from an interrupted fresh journal;
    // the current entrypoint itself correctly refuses this starting balance.
    fixture.platform.funding_observation_failures = 1;
    assert!(matches!(
        apply_fixture_plan(&mut fixture, &source, &report.plan),
        Err(workflow::EnsureWorkflowError::Platform(_))
    ));
    let mut journal = crate::fleet_ensure::ops::read_journal(&paths)
        .unwrap()
        .unwrap();
    assert!(journal.effects.is_empty());
    let state = crate::fleet_ensure::ops::read_state(&paths, "test-fleet").unwrap();
    let prepared = crate::fleet_ensure::ops::effect_preparation::prepare_effect(
        &mut fixture.platform,
        &report.plan.operation_id,
        &action,
        &state,
    )
    .unwrap();
    journal.effects.push(prepared.record);
    journal.initial_operator_cycles = 0;
    let live = fixture.platform.live[TREASURY].clone();
    retain_recorded_retirement(
        &mut report.plan,
        vec![crate::fleet_ensure::model::RootManagementBinding {
            controllers: live.controllers.clone(),
            module_sha256: live.module_sha256.clone().unwrap(),
            name: "treasury".into(),
            principal: TREASURY.into(),
            subnet: SUBNET.into(),
        }],
    );
    fixture.platform.reinstall_authority = Some(BTreeMap::from([(
        "treasury".into(),
        crate::fleet_ensure::model::RootManagementCanisterObservation {
            live,
            name: "treasury".into(),
            subnet: SUBNET.into(),
        },
    )]));
    journal.plan_sha256.clone_from(&report.plan.plan_sha256);
    crate::fleet_ensure::ops::write_plan(&paths, &report.plan).unwrap();
    crate::fleet_ensure::ops::write_journal(&paths, &journal).unwrap();
    fixture.platform.set_operator_cycles(0);
    fixture.platform.operator_funding =
        Some(crate::fleet_ensure::view::OperatorFundingObservation {
            cycles_ledger: LEDGER.into(),
            ledger_fee_cycles: 10,
            operator_cycles: 0,
        });
    let original_plan = fs::read(&paths.plan).unwrap();
    assert_eq!(
        workflow::retained_in_progress_plan::<MockError>(&fixture.root, "local", "test-fleet")
            .unwrap(),
        Some(report.plan.clone())
    );
    assert!(!workflow::operator_mint::fresh_quote_available(&paths).unwrap());
    assert!(workflow::operator_mint::status(&paths).unwrap().is_none());
    let mut tampered: serde_json::Value = serde_json::from_slice(&original_plan).unwrap();
    tampered["reinstall"]["source"]["terminal_retirement"]["conservation"]["measured_execution_burn_cycles"] =
        serde_json::json!("0");
    fs::write(&paths.plan, serde_json::to_vec(&tampered).unwrap()).unwrap();
    assert!(matches!(
        workflow::retained_in_progress_plan::<MockError>(&fixture.root, "local", "test-fleet"),
        Err(workflow::EnsureWorkflowError::PlanIntegrity)
    ));
    assert!(matches!(
        workflow::operator_mint::status(&paths),
        Err(workflow::EnsureWorkflowError::PlanIntegrity)
    ));
    fs::write(&paths.plan, &original_plan).unwrap();
    assert!(matches!(
        workflow::plan_reinstall(
            &fixture.root, &fixture.desired, "new-release", "test-fleet", 2,
            &mut fixture.platform,
        ),
        Err(workflow::EnsureWorkflowError::RetainedOperationRecoveryRequired {
            operation_id, plan_sha256,
        }) if operation_id == report.plan.operation_id && plan_sha256 == report.plan.plan_sha256
    ));
    assert_eq!(
        crate::fleet_ensure::ops::read_journal(&paths).unwrap(),
        Some(journal.clone())
    );
    let review = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        1_800_000_000_000_000_001,
        &mut fixture.platform,
    )
    .unwrap()
    .funding_review
    .unwrap();
    assert!(matches!(
        review.pause,
        crate::fleet_ensure::model::FundingPauseRecord::Operator(_)
    ));
    assert_eq!(review.action, action);
    assert!(review.effect.is_none());
    let retained = crate::fleet_ensure::ops::read_journal(&paths)
        .unwrap()
        .unwrap();
    assert_eq!(retained.effects, journal.effects);
    assert_eq!(retained.initial_operator_cycles, 0);
    assert_eq!(
        retained.initial_controlled_cycles,
        journal.initial_controlled_cycles
    );
    assert_eq!(fs::read(&paths.plan).unwrap(), original_plan);
    assert!(fixture.platform.mutations.is_empty());
    assert!(matches!(
        workflow::apply(
            &fixture.root,
            &fixture.desired,
            &source,
            "test-fleet",
            &review.review_sha256,
            &mut fixture.platform
        ),
        Err(workflow::EnsureWorkflowError::OperatorFundingRequired { .. })
    ));
    assert!(fixture.platform.mutations.is_empty());
    // This native accounting fixture starts after durable credit admission. The
    // production-Ledger PocketIC case separately authenticates the two receipts.
    let credit = report.plan.conservation.maximum_operator_debit_cycles + 100;
    retain_operator_credit_fixture(&paths, &report.plan, &review, credit);
    fixture.platform.set_operator_cycles(credit);
    let action_hash = action_sha256(&action);
    fixture.platform.fail_once(action_hash.clone());
    assert!(matches!(
        apply_fixture_plan(&mut fixture, &source, &report.plan),
        Err(workflow::EnsureWorkflowError::Platform(_))
    ));
    let completed = apply_fixture_plan(&mut fixture, &source, &report.plan).unwrap();
    assert!(completed.terminal);
    assert_eq!(
        completed.actual_conservation.unwrap().operator_debit_cycles,
        report.plan.conservation.maximum_operator_debit_cycles
    );
    assert_eq!(fixture.platform.mutation_count(&action_hash), 1);
    let retained = crate::fleet_ensure::ops::read_journal(&paths)
        .unwrap()
        .unwrap();
    assert_eq!(retained.initial_operator_cycles, 0);
    assert_eq!(
        retained.effects[0].action_sha256,
        journal.effects[0].action_sha256
    );
    let mutations = fixture.platform.mutations.clone();
    assert!(
        apply_fixture_plan(&mut fixture, &source, &report.plan)
            .unwrap()
            .terminal
    );
    assert_eq!(fixture.platform.mutations, mutations);
    assert_eq!(fs::read(&paths.plan).unwrap(), original_plan);
    fs::remove_dir_all(fixture.root).unwrap();
}

/// Attach immutable retirement history to the native funding fixture. This is
/// evidence-only coverage; managed reset/seal journeys retain their own fixtures.
pub(super) fn retain_recorded_retirement(
    plan: &mut FleetEnsurePlan,
    authorities: Vec<crate::fleet_ensure::model::RootManagementBinding>,
) {
    let source_operation = sha256_hex(b"completed retirement source");
    plan.reinstall = Some(Box::new(FleetReinstallRecord {
        target_artifacts_sha256: None,
        activation_reset: None,
        operation_id: plan.operation_id.clone(),
        source_operation_id: source_operation.clone(),
        authorities,
        assets: Vec::new(),
        source: Some(Box::new(FleetReinstallSourceRecord {
            reviewed_desired: *plan.reviewed_desired.clone().unwrap(),
            wasm_sha256_by_canister: BTreeMap::new(),
            candid_sha256_by_path: BTreeMap::new(),
            terminal_retirement: Some(Box::new(FleetTerminalRetirementRecord {
                source: FleetTerminalSourceRecord {
                    operation_id: source_operation,
                    plan_sha256: sha256_hex(b"source plan"),
                    plan_document_sha256: sha256_hex(b"source plan bytes"),
                    journal_document_sha256: sha256_hex(b"source journal bytes"),
                    state_document_sha256: sha256_hex(b"source state bytes"),
                    phase_document_sha256: BTreeMap::new(),
                },
                conservation: serde_json::from_value(serde_json::json!({
                    "estate_funding_cycles": "0", "exact_estate_creation_fee_cycles": "0",
                    "exact_unavoidable_fee_cycles": "10", "final_controlled_cycles": "600",
                    "measured_execution_burn_cycles": "10", "observed_starting_cycles": "500",
                    "observed_settlement_credit_cycles": "0", "operator_debit_cycles": "120",
                    "received_new_funding_cycles": "110",
                }))
                .unwrap(),
            })),
        })),
    }));
    plan.plan_sha256 = crate::fleet_ensure::policy::expected_plan_sha256(plan);
}

fn retain_operator_credit_fixture(
    paths: &crate::fleet_ensure::ops::EnsurePaths,
    plan: &FleetEnsurePlan,
    funding: &crate::fleet_ensure::model::FundingReviewRecord,
    credit: u128,
) {
    let digest = |text: &str| {
        canic_core::cdk::utils::hash::decode_hex(text)
            .unwrap()
            .try_into()
            .unwrap()
    };
    let intent = prepare_intent(
        OperatorMintAuthority {
            operation_id: digest(&plan.operation_id),
            plan_sha256: digest(&plan.plan_sha256),
            funding_review_sha256: digest(&funding.review_sha256),
            network_identity_sha256: [8; 32],
            operator: CONTROLLER.parse().unwrap(),
            icp_ledger: OLD_APP.parse().unwrap(),
            cmc: "rkp4c-7iaaa-aaaaa-aaaca-cai".parse().unwrap(),
            cycles_ledger: LEDGER.parse().unwrap(),
        },
        100_000_000,
        10_000,
        1_800_000_000_000_000_010,
    )
    .unwrap();
    let review = records::approve(&records::review(intent.clone()).unwrap()).unwrap();
    let review = records::transfer_outcome(
        &review,
        OperatorMintTransferOutcomeRecord::Accepted { block_index: 1 },
    );
    let review = records::notification(&review, 1).unwrap();
    let mut review = records::notification_outcome(
        &review,
        OperatorMintNotificationOutcomeRecord::Minted {
            deposit_block_index: 2,
            gross_minted_cycles: credit + 1,
            historical_balance_cycles: credit,
        },
    );
    review.receipt = Some(OperatorMintReceiptRecord {
        destination_owner: intent.authority.operator,
        destination_subaccount: None,
        deposit_memo: intent.deposit_memo,
        intent,
        icp_block_index: 1,
        deposit_block_index: 2,
        gross_minted_cycles: credit + 1,
        deposit_fee_cycles: 1,
        net_credit_cycles: credit,
    });
    let mut journal = crate::fleet_ensure::ops::read_journal(paths)
        .unwrap()
        .unwrap();
    records::retain(&mut journal, Some(review));
    crate::fleet_ensure::ops::write_journal(paths, &journal).unwrap();
}

#[test]
fn lost_estate_funding_response_reuses_the_exact_ledger_transfer() {
    let fixture = fixture();
    let mut platform = fixture.platform;
    platform.set_operator_cycles(100);
    platform.set_estate_funding_balance(40);
    let action = EnsureAction::FundEstate {
        amount: 50,
        created_at_time: 1_800_000_000_000_000_000,
        expected_post_cycles: 90,
        ledger: LEDGER.to_string(),
        ledger_fee_cycles: 10,
        name: "root-0".to_string(),
        principal: TREASURY.to_string(),
    };
    let action_hash = crate::fleet_ensure::ops::action_sha256(&action);
    platform.fail_once(action_hash.clone());
    let state = crate::fleet_ensure::ops::read_state(
        &crate::fleet_ensure::ops::EnsurePaths::under(
            &fixture.root,
            &platform.desired.environment,
            &platform.desired.fleet,
        ),
        &platform.desired.fleet,
    )
    .expect("construct current state");
    let mut record = EffectRecord {
        publication_attempts: 0,
        maintenance_attempts: 0,
        action_sha256: action_hash.clone(),
        created_principal: None,
        destination_post_cycles: Some(40),
        destination_pre_cycles: Some(40),
        post_cycles: None,
        pre_cycles: Some(100),
        pre_canister_version: None,
        progress_identity: None,
        receipt: None,
        state: EffectState::Intent,
    };

    assert!(
        platform
            .apply("operation", &action, &record, &state)
            .is_err()
    );
    let replay = platform
        .apply("operation", &action, &record, &state)
        .expect("same transfer identity returns its retained receipt");
    assert_eq!(platform.mutation_count(&action_hash), 1);
    record.receipt = replay.receipt;
    record.state = EffectState::Issued;
    let observed = platform
        .observe_effect("operation", &action, &record, &state)
        .expect("observe both sides of the retained transfer");
    assert!(observed.applied);
    assert_eq!(observed.post_cycles, Some(40));

    fs::remove_dir_all(fixture.root).expect("remove test directory");
}

fn assert_same_plan_replay(
    root: &Path,
    desired: &DesiredFleet,
    plan: &FleetEnsurePlan,
    platform: &mut MockPlatform,
) {
    let paths =
        crate::fleet_ensure::ops::EnsurePaths::under(root, &desired.environment, &plan.fleet);
    let retained_journal_bytes = fs::read(&paths.journal).unwrap();
    let mutations = platform.mutations.clone();
    let observations = platform.observation_calls;
    let same_plan = workflow::apply(
        root,
        desired,
        &plan.desired_sha256,
        &plan.fleet,
        &plan.plan_sha256,
        platform,
    )
    .expect("replay the completed review without generating another plan");
    assert!(same_plan.terminal);
    assert_eq!(platform.observation_calls - observations, 1);
    assert_eq!(same_plan.effects_applied, 0);
    assert_eq!(platform.mutations, mutations);
    assert_eq!(fs::read(&paths.journal).unwrap(), retained_journal_bytes);
}

#[test]
fn same_module_reinstall_runs_once_and_replay_is_effect_free() {
    let mut fixture = fixture();
    fixture.desired.canisters.truncate(1);
    fixture.platform.desired = fixture.desired.clone();
    let treasury = fixture
        .platform
        .live
        .get_mut(TREASURY)
        .expect("live Coordinator");
    treasury.reinstall_required = true;
    let before = treasury.canister_version;
    let desired_sha256 = "7".repeat(64);

    let planned = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut fixture.platform,
    )
    .expect("plan same-module reinstall");
    assert!(matches!(
        workflow::ordered_actions(&planned.plan).as_slice(),
        [EnsureAction::Install {
            mode: crate::fleet_ensure::model::InstallMode::Reinstall,
            ..
        }]
    ));
    fixture.platform.version_observation_failures = 1;
    assert!(matches!(
        workflow::apply(
            &fixture.root,
            &fixture.desired,
            &desired_sha256,
            "test-fleet",
            &planned.plan.plan_sha256,
            &mut fixture.platform,
        ),
        Err(workflow::EnsureWorkflowError::Platform(MockError))
    ));
    let paths = crate::fleet_ensure::ops::EnsurePaths::under(
        &fixture.root,
        &fixture.desired.environment,
        "test-fleet",
    );
    let interrupted = crate::fleet_ensure::ops::read_journal(&paths)
        .expect("read interrupted reinstall journal")
        .expect("interrupted reinstall journal");
    assert_eq!(interrupted.plan_sha256, planned.plan.plan_sha256);
    assert!(interrupted.effects.is_empty());
    assert_eq!(fixture.platform.mutations.values().sum::<u32>(), 0);

    let applied = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        &planned.plan.plan_sha256,
        &mut fixture.platform,
    )
    .expect("apply same-module reinstall");
    assert!(applied.terminal);
    assert_eq!(applied.effects_applied, 1);
    let completed = crate::fleet_ensure::ops::read_journal(&paths)
        .expect("read completed reinstall journal")
        .expect("completed reinstall journal");
    assert_eq!(completed.effects.len(), 1);
    assert_eq!(completed.effects[0].pre_canister_version, before);
    assert_eq!(completed.effects[0].state, EffectState::Applied);
    let after = fixture.platform.live[TREASURY].canister_version;
    assert!(after > before);

    assert_same_plan_replay(
        &fixture.root,
        &fixture.desired,
        &planned.plan,
        &mut fixture.platform,
    );

    let replay_plan = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        1_800_000_000_000_000_001,
        &mut fixture.platform,
    )
    .expect("plan terminal replay");
    assert!(workflow::ordered_actions(&replay_plan.plan).is_empty());
    let replay = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        &replay_plan.plan.plan_sha256,
        &mut fixture.platform,
    )
    .expect("apply terminal replay");
    assert!(replay.terminal);
    assert_eq!(replay.effects_applied, 0);
}

#[test]
fn reviewed_plan_accepts_bounded_bidirectional_balance_movement_with_truthful_start() {
    let mut fixture = fixture();
    let desired_sha256 = "76".repeat(32);
    let initial = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut fixture.platform,
    )
    .expect("plan initial convergence");
    workflow::apply(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        &initial.plan.plan_sha256,
        &mut fixture.platform,
    )
    .expect("apply initial convergence");

    fixture.desired.maximum_observation_burn_cycles = "10".to_string();
    fixture.platform.desired = fixture.desired.clone();
    fixture.platform.mutations.clear();
    let reviewed = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        1_800_000_000_000_000_100,
        &mut fixture.platform,
    )
    .expect("plan effect-free replay");
    assert!(workflow::ordered_actions(&reviewed.plan).is_empty());
    let reviewed_cycles = reviewed.plan.conservation.observed_controlled_cycles;
    fixture
        .platform
        .live
        .get_mut(TREASURY)
        .expect("retained Coordinator")
        .cycles -= 4;
    fixture
        .platform
        .live
        .get_mut(OLD_APP)
        .expect("retained App")
        .cycles += 3;

    let applied = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        &reviewed.plan.plan_sha256,
        &mut fixture.platform,
    )
    .expect("accept bounded decrease and refund");
    assert!(applied.terminal);
    assert_eq!(applied.effects_applied, 0);
    assert_eq!(fixture.platform.mutations.values().sum::<u32>(), 0);
    assert_eq!(
        applied
            .actual_conservation
            .expect("terminal conservation")
            .observed_starting_cycles,
        reviewed_cycles - 1
    );

    let replay = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        1_800_000_000_000_000_200,
        &mut fixture.platform,
    )
    .expect("replan terminal state");
    assert!(workflow::ordered_actions(&replay.plan).is_empty());
}

#[test]
fn reviewed_plan_rejects_balance_loss_beyond_its_bound_before_effects() {
    let mut fixture = fixture();
    let desired_sha256 = "77".repeat(32);
    let initial = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut fixture.platform,
    )
    .expect("plan initial convergence");
    workflow::apply(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        &initial.plan.plan_sha256,
        &mut fixture.platform,
    )
    .expect("apply initial convergence");
    fixture.desired.maximum_observation_burn_cycles = "10".to_string();
    fixture.platform.desired = fixture.desired.clone();
    fixture.platform.mutations.clear();
    let reviewed = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        1_800_000_000_000_000_100,
        &mut fixture.platform,
    )
    .expect("plan bounded estate");
    fixture
        .platform
        .live
        .get_mut(TREASURY)
        .expect("retained Coordinator")
        .cycles -= 11;

    let error = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        &reviewed.plan.plan_sha256,
        &mut fixture.platform,
    )
    .expect_err("reject movement beyond reviewed bound");
    assert!(matches!(
        error,
        workflow::EnsureWorkflowError::DriftedBeforeApply
    ));
    assert_eq!(fixture.platform.mutations.values().sum::<u32>(), 0);
}

#[test]
fn funding_margin_is_bounded_by_the_target_observation_only() {
    let mut fixture = fixture();
    fixture.desired.maximum_observation_burn_cycles = "10".to_string();
    fixture.platform.desired = fixture.desired.clone();

    let report = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &"3".repeat(64),
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut fixture.platform,
    )
    .expect("compile target-bounded funding plan");
    let funding = report
        .plan
        .canisters
        .iter()
        .find(|canister| canister.name == "app")
        .and_then(|canister| {
            canister.actions.iter().find_map(|action| match action {
                EnsureAction::SealAuthority { .. } => {
                    panic!("authority sealing uses the production PocketIC adapter")
                }
                EnsureAction::Fund {
                    amount,
                    expected_post_cycles,
                    funding_deficit_cycles,
                    funding_margin_cycles,
                    ..
                } => Some((
                    *amount,
                    *expected_post_cycles,
                    *funding_deficit_cycles,
                    *funding_margin_cycles,
                )),
                _ => None,
            })
        })
        .expect("underfunded App has one funding action");

    assert_eq!(funding, (28, 33, 15, 13));
}

#[test]
fn mixed_subnet_creation_fees_reject_before_plan_or_effect() {
    let mut fixture = fixture();
    let other_subnet = Principal::from_slice(&[77]).to_text();
    fixture.desired.canisters[1]
        .subnet
        .clone_from(&other_subnet);
    fixture.platform.desired = fixture.desired.clone();
    let error = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &"14".repeat(32),
        "test-fleet",
        100,
        &mut fixture.platform,
    )
    .expect_err("two creation subnets cannot share scalar fee authority");
    let mut expected = vec![SUBNET.to_string(), other_subnet];
    expected.sort();
    assert!(matches!(
        error,
        workflow::EnsureWorkflowError::Policy(
            crate::fleet_ensure::policy::EnsurePolicyError::MixedSubnetCreationFees { subnets }
        ) if subnets == expected
    ));
    assert!(fixture.platform.mutations.is_empty());
    let paths = crate::fleet_ensure::ops::EnsurePaths::under(&fixture.root, "local", "test-fleet");
    assert!(
        crate::fleet_ensure::ops::read_plan(&paths)
            .unwrap()
            .is_none()
    );
    assert!(
        crate::fleet_ensure::ops::read_journal(&paths)
            .unwrap()
            .is_none()
    );
    fs::remove_dir_all(fixture.root).unwrap();
}

#[test]
fn retained_other_subnet_does_not_block_single_subnet_creation() {
    let mut fixture = fixture();
    let other_subnet = Principal::from_slice(&[77]).to_text();
    fixture.desired.canisters[0]
        .subnet
        .clone_from(&other_subnet);
    fixture.platform.desired = fixture.desired.clone();
    let planned = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &"14".repeat(32),
        "test-fleet",
        100,
        &mut fixture.platform,
    )
    .expect("retained treasury consumes no creation fee");
    assert_eq!(
        planned.plan.conservation.maximum_unavoidable_fee_cycles,
        130
    );
    assert!(fixture.platform.mutations.is_empty());
    fs::remove_dir_all(fixture.root).unwrap();
}

#[test]
fn retained_mixed_subnet_creation_plan_rejects_before_apply() {
    let mut fixture = fixture();
    let desired_sha256 = "14".repeat(32);
    let mut plan = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        100,
        &mut fixture.platform,
    )
    .unwrap()
    .plan;
    // Model an already reviewed document admitted before the fee-scope guard.
    for action in plan
        .canisters
        .iter_mut()
        .flat_map(|canister| &mut canister.actions)
    {
        if let EnsureAction::Create { name, subnet, .. } = action
            && name == "created"
        {
            *subnet = Principal::from_slice(&[77]).to_text();
        }
    }
    plan.plan_sha256 = crate::fleet_ensure::policy::expected_plan_sha256(&plan);
    let paths = crate::fleet_ensure::ops::EnsurePaths::under(&fixture.root, "local", "test-fleet");
    crate::fleet_ensure::ops::write_plan(&paths, &plan).unwrap();
    let error = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        &plan.plan_sha256,
        &mut fixture.platform,
    )
    .expect_err("retained review cannot bypass fee authority");
    assert!(matches!(
        error,
        workflow::EnsureWorkflowError::Policy(
            crate::fleet_ensure::policy::EnsurePolicyError::MixedSubnetCreationFees { .. }
        )
    ));
    assert!(fixture.platform.mutations.is_empty());
    assert_eq!(
        crate::fleet_ensure::ops::read_plan(&paths)
            .unwrap()
            .unwrap(),
        plan
    );
    assert!(
        crate::fleet_ensure::ops::read_journal(&paths)
            .unwrap()
            .is_none()
    );
    fs::remove_dir_all(fixture.root).unwrap();
}

#[test]
fn live_ledger_fee_drift_rejects_before_intent_or_effect() {
    let mut fixture = fixture();
    fixture.platform.ledger_fee_cycles += 1;
    let error = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &"9".repeat(64),
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut fixture.platform,
    )
    .expect_err("changed live Ledger fee requires a new desired plan");
    assert!(matches!(
        error,
        workflow::EnsureWorkflowError::Policy(
            crate::fleet_ensure::policy::EnsurePolicyError::LedgerFeeDrift { .. }
        )
    ));
    assert_eq!(fixture.platform.mutations.values().sum::<u32>(), 0);
}

#[test]
fn missing_exact_retained_identity_rejects_instead_of_creating_replacement() {
    let mut fixture = fixture();
    fixture.platform.live.remove(OLD_APP);
    let error = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &"8".repeat(64),
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut fixture.platform,
    )
    .expect_err("missing seeded identity must fail closed");
    assert!(matches!(
        error,
        workflow::EnsureWorkflowError::Policy(
            crate::fleet_ensure::policy::EnsurePolicyError::MissingObservation { name }
        ) if name == "app"
    ));
    assert_eq!(fixture.platform.mutations.values().sum::<u32>(), 0);
}

#[test]
fn anonymous_operator_and_treasury_replacement_reject_before_effects() {
    let mut anonymous = fixture();
    anonymous.desired.operator = Principal::anonymous().to_text();
    anonymous.platform.desired = anonymous.desired.clone();
    let error = workflow::plan(
        &anonymous.root,
        &anonymous.desired,
        &"7".repeat(64),
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut anonymous.platform,
    )
    .expect_err("anonymous operator rejects");
    assert!(matches!(
        error,
        workflow::EnsureWorkflowError::Policy(
            crate::fleet_ensure::policy::EnsurePolicyError::InvalidPrincipal { .. }
        )
    ));

    let mut treasury = fixture();
    treasury
        .desired
        .canisters
        .iter_mut()
        .find(|canister| canister.name == "treasury")
        .expect("treasury config")
        .replace = true;
    treasury.platform.desired = treasury.desired.clone();
    let error = workflow::plan(
        &treasury.root,
        &treasury.desired,
        &"6".repeat(64),
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut treasury.platform,
    )
    .expect_err("treasury replacement rejects");
    assert!(matches!(
        error,
        workflow::EnsureWorkflowError::Policy(
            crate::fleet_ensure::policy::EnsurePolicyError::TreasuryReplacement { .. }
        )
    ));
    assert_eq!(anonymous.platform.mutations.values().sum::<u32>(), 0);
    assert_eq!(treasury.platform.mutations.values().sum::<u32>(), 0);

    let mut unsafe_name = fixture();
    unsafe_name.desired.fleet = "../escape".to_string();
    unsafe_name.platform.desired = unsafe_name.desired.clone();
    let error = workflow::plan(
        &unsafe_name.root,
        &unsafe_name.desired,
        &"5".repeat(64),
        "../escape",
        1_800_000_000_000_000_000,
        &mut unsafe_name.platform,
    )
    .expect_err("unsafe Fleet path label rejects before state access");
    assert!(matches!(
        error,
        workflow::EnsureWorkflowError::Policy(
            crate::fleet_ensure::policy::EnsurePolicyError::UnsafePathLabel { .. }
        )
    ));
    assert!(!unsafe_name.root.join(".canic").exists());
}

#[test]
fn managed_topology_without_typed_protocol_intent_rejects_before_effects() {
    let mut fixture = fixture();
    let coordinator = fixture
        .desired
        .canisters
        .iter()
        .find(|canister| canister.kind == DesiredCanisterKind::Coordinator)
        .expect("fixture Coordinator")
        .name
        .clone();
    let mut root = desired_canister(
        "managed-root",
        None,
        false,
        &fixture.root.join("app.wasm"),
        None,
    );
    root.kind = DesiredCanisterKind::Root;
    root.parent = Some(coordinator);
    fixture.desired.canisters.push(root);
    fixture.platform.desired = fixture.desired.clone();

    let error = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &"f".repeat(64),
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut fixture.platform,
    )
    .expect_err("managed topology cannot omit typed protocol intent");

    assert!(matches!(
        error,
        workflow::EnsureWorkflowError::Policy(
            crate::fleet_ensure::policy::EnsurePolicyError::InvalidTopology {
                reason: "managed Fleet roles require complete typed protocol intent",
                ..
            }
        )
    ));
    assert_eq!(fixture.platform.mutations.values().sum::<u32>(), 0);
}

#[test]
fn terminal_inventory_rejects_missing_and_duplicate_principals() {
    let mut state = FleetEnsureStateRecord {
        active_registry: None,
        completed_reinstall_action_sha256: BTreeMap::new(),
        completed_reinstall_operation_id: None,
        completed_reinstalls: BTreeMap::new(),
        fleet: "inventory-test".to_string(),
        pending_principals: BTreeMap::new(),
        principals: BTreeMap::new(),
        retained_cycles_by_principal: BTreeMap::new(),
        schema_version: FLEET_ENSURE_SCHEMA_VERSION,
        topology: BTreeMap::from([(
            "coordinator".to_string(),
            crate::fleet_ensure::model::FleetEnsureTopologyRecord {
                kind: DesiredCanisterKind::Coordinator,
                module_hash: None,
                parent: None,
                protocol_binding: None,
                role: None,
            },
        )]),
    };
    assert!(matches!(
        crate::fleet_ensure::inventory::project_current_fleet_inventory(&state),
        Err(crate::fleet_ensure::CurrentFleetInventoryError::MissingPrincipal(name))
            if name == "coordinator"
    ));

    state
        .principals
        .insert("coordinator".to_string(), TREASURY.to_string());
    state.topology.insert(
        "duplicate".to_string(),
        crate::fleet_ensure::model::FleetEnsureTopologyRecord {
            kind: DesiredCanisterKind::Auxiliary,
            module_hash: None,
            parent: None,
            protocol_binding: None,
            role: None,
        },
    );
    state
        .principals
        .insert("duplicate".to_string(), TREASURY.to_string());
    assert!(matches!(
        crate::fleet_ensure::inventory::project_current_fleet_inventory(&state),
        Err(crate::fleet_ensure::CurrentFleetInventoryError::DuplicatePrincipal)
    ));
}

#[test]
#[expect(
    clippy::too_many_lines,
    reason = "one end-to-end successor-plan crash boundary retains the complete authority setup and outcome"
)]
fn terminal_protocol_inventory_survives_an_effect_free_successor_plan() {
    let fixture = fixture();
    let mut platform = fixture.platform;
    let component = Principal::from_slice(&[29; 29]).to_text();
    let module_hash = "ab".repeat(32);
    let registry = empty_active_registry();
    platform.terminal_inventory = TerminalFleetInventory {
        active_registry: Some(registry.clone()),
        controlled_cycles_by_principal: BTreeMap::from([(component.clone(), 0)]),
        entries: vec![RegistryEntry {
            module_hash: Some(module_hash.clone()),
            parent_pid: Some(TREASURY.to_string()),
            pid: component.clone(),
            protocol_binding: None,
            role: Some("managed_component".to_string()),
        }],
    };
    let source = "4".repeat(64);
    let first = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut platform,
    )
    .expect("compile first plan");
    workflow::apply(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        &first.plan.plan_sha256,
        &mut platform,
    )
    .expect("converge first plan");

    platform
        .live
        .get_mut(TREASURY)
        .expect("terminal treasury")
        .cycles -= 25;
    platform
        .terminal_inventory
        .controlled_cycles_by_principal
        .insert(component.clone(), 25);

    platform.terminal_inventory_expected_operation_id = Some(first.plan.operation_id.clone());
    let second = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        1_800_000_000_000_000_100,
        &mut platform,
    )
    .expect("compile effect-free successor");
    assert!(second.plan.protocol_actions.is_empty());
    assert_eq!(
        second.plan.terminal_inventory_operation_id.as_deref(),
        Some(first.plan.operation_id.as_str())
    );
    let backup_inventory = crate::fleet_ensure::read_last_converged_fleet_inventory(
        &fixture.root,
        &fixture.desired.environment,
        "test-fleet",
    )
    .expect("last terminal inventory remains readable after successor planning");
    assert!(backup_inventory.entries.iter().any(|entry| {
        entry.pid == component
            && entry.parent_pid.as_deref() == Some(TREASURY)
            && entry.module_hash.as_deref() == Some(module_hash.as_str())
    }));
    assert!(matches!(
        crate::fleet_ensure::resolve_current_fleet(
            &fixture.root,
            &fixture.desired.environment,
            "test-fleet",
        ),
        Err(crate::fleet_ensure::CurrentFleetInventoryError::PlanAuthorityConflict { .. })
    ));
    workflow::apply(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        &second.plan.plan_sha256,
        &mut platform,
    )
    .expect("apply effect-free successor");
    assert_eq!(
        second.plan.conservation.observed_controlled_cycles,
        platform
            .live
            .values()
            .map(|canister| canister.cycles)
            .sum::<u128>()
            + 25
    );
    apply_effect_free_successor(
        &fixture.root,
        &fixture.desired,
        &mut platform,
        &source,
        1_800_000_000_000_000_200,
        &first.plan.operation_id,
    );

    let current = crate::fleet_ensure::resolve_current_fleet(
        &fixture.root,
        &fixture.desired.environment,
        "test-fleet",
    )
    .expect("resolve terminal current Fleet");
    assert_eq!(
        current
            .initial_active_registry("test-fleet")
            .expect("terminal Registry authority"),
        &registry
    );
    assert!(current.registry.entries.iter().any(|entry| {
        entry.pid == component
            && entry.parent_pid.as_deref() == Some(TREASURY)
            && entry.module_hash.as_deref() == Some(module_hash.as_str())
    }));
}

fn apply_effect_free_successor(
    root: &Path,
    desired: &DesiredFleet,
    platform: &mut MockPlatform,
    source: &str,
    planned_at_time: u64,
    terminal_operation_id: &str,
) -> FleetEnsurePlan {
    let report = workflow::plan(
        root,
        desired,
        source,
        "test-fleet",
        planned_at_time,
        platform,
    )
    .expect("compile effect-free successor");
    assert!(report.plan.protocol_actions.is_empty());
    assert_eq!(
        report.plan.terminal_inventory_operation_id.as_deref(),
        Some(terminal_operation_id)
    );
    assert_eq!(
        platform
            .terminal_inventory_operation_ids
            .last()
            .map(String::as_str),
        Some(terminal_operation_id)
    );
    workflow::apply(
        root,
        desired,
        source,
        "test-fleet",
        &report.plan.plan_sha256,
        platform,
    )
    .expect("apply effect-free successor");
    report.plan
}

#[test]
#[expect(
    clippy::too_many_lines,
    reason = "the regression proves retained identity and conservation across the complete short-result replay"
)]
fn post_effect_balance_drift_preserves_the_nonterminal_journal_and_inventory() {
    let fixture = fixture();
    let mut platform = fixture.platform;
    platform.create_shortfalls.insert("created".to_string(), 2);
    let source = "8".repeat(64);
    let planned = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut platform,
    )
    .expect("compile reviewed plan");
    let error = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        &planned.plan.plan_sha256,
        &mut platform,
    )
    .expect_err("short creation balance must stop before later effects");
    assert!(
        matches!(
            error,
            workflow::EnsureWorkflowError::ReplanRequiredAfterCreateBalanceDrift {
                actual_cycles: 18,
                configured_fee_cycles: 50,
                deficit_cycles: 2,
                requested_cycles: 20,
                ..
            }
        ),
        "unexpected short-create error: {error:?}"
    );
    let creation_hash = planned
        .plan
        .canisters
        .iter()
        .find(|canister| canister.name == "created")
        .and_then(|canister| canister.actions.first())
        .map(crate::fleet_ensure::ops::action_sha256)
        .expect("created action identity");
    assert_eq!(platform.mutations.get(&creation_hash), Some(&1));
    assert_eq!(platform.mutations.values().sum::<u32>(), 1);

    let paths = crate::fleet_ensure::ops::EnsurePaths::under(
        &fixture.root,
        &fixture.desired.environment,
        "test-fleet",
    );
    let journal = crate::fleet_ensure::ops::read_journal(&paths)
        .expect("read journal")
        .expect("retained journal");
    assert_eq!(
        journal.completion,
        crate::fleet_ensure::model::FleetEnsureCompletion::ReplanRequired
    );
    assert_eq!(journal.effects.len(), 1);
    assert_eq!(journal.effects[0].state, EffectState::Applied);
    assert!(matches!(
        crate::fleet_ensure::read_last_converged_fleet_inventory(
            &fixture.root,
            &fixture.desired.environment,
            &fixture.desired.fleet,
        ),
        Err(crate::fleet_ensure::CurrentFleetInventoryError::NotConverged { .. })
    ));
    let state = crate::fleet_ensure::ops::read_state(&paths, "test-fleet").expect("read state");
    assert_eq!(
        state.principals.get("created").map(String::as_str),
        Some("created-created")
    );
    let retained_topology = state
        .topology
        .get("created")
        .expect("created topology retained before replan");
    let reviewed_created = fixture
        .desired
        .canisters
        .iter()
        .find(|canister| canister.name == "created")
        .expect("reviewed created canister");
    assert_eq!(retained_topology.kind, reviewed_created.kind);
    assert_eq!(retained_topology.parent, reviewed_created.parent);
    assert_eq!(
        retained_topology.protocol_binding,
        reviewed_created.protocol_binding
    );
    assert_eq!(
        state.retained_cycles_by_principal.get("created-created"),
        Some(&18)
    );

    let mut successor_platform =
        MockPlatform::new(fixture.desired.clone(), platform.live.values().cloned());
    successor_platform.operator_cycles = platform.operator_cycles;
    successor_platform.terminal_inventory = platform.terminal_inventory.clone();
    let successor = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        1_800_000_000_000_000_100,
        &mut successor_platform,
    )
    .expect("a closed nonconverged operation permits a freshly reviewed plan");
    assert!(successor.plan.canisters.iter().all(|canister| {
        canister.name != "created"
            || canister
                .actions
                .iter()
                .all(|action| !matches!(action, EnsureAction::Create { .. }))
    }));
    assert!(successor.plan.canisters.iter().any(|canister| {
        canister.name == "created"
            && canister
                .actions
                .iter()
                .any(|action| matches!(action, EnsureAction::Fund { .. }))
    }));
    let terminal = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        &successor.plan.plan_sha256,
        &mut successor_platform,
    )
    .expect("successor plan converges the retained created canister");
    assert!(terminal.terminal);
    assert!(
        !successor_platform.mutations.contains_key(&creation_hash),
        "a fresh host process must not repeat the retained creation"
    );
    let state = crate::fleet_ensure::ops::read_state(&paths, "test-fleet").expect("read state");
    assert!(state.pending_principals.is_empty());
    assert_eq!(
        state.principals.get("created").map(String::as_str),
        Some("created-created")
    );
    let inventory = crate::fleet_ensure::resolve_current_fleet(
        &fixture.root,
        &fixture.desired.environment,
        &fixture.desired.fleet,
    )
    .expect("terminal current inventory is available to operator commands");
    assert_eq!(inventory.topology.coordinator_canister_id, TREASURY);
    assert_eq!(inventory.plan.plan_sha256, successor.plan.plan_sha256);
    assert_eq!(inventory.plan.operation_id, successor.plan.operation_id);
}

#[test]
fn consecutive_stalls_are_bounded_and_real_progress_resets_the_budget() {
    let fixture = fixture();
    let mut platform = fixture.platform;
    let source = "f".repeat(64);
    let planned = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut platform,
    )
    .expect("compile reviewed plan");
    let first_action = planned
        .plan
        .canisters
        .iter()
        .flat_map(|canister| &canister.actions)
        .next()
        .expect("fixture has an action");
    let first_hash = crate::fleet_ensure::ops::action_sha256(first_action);
    platform.stall_before_mutation.insert(first_hash.clone(), 2);

    let first = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        &planned.plan.plan_sha256,
        &mut platform,
    )
    .expect_err("first stalled observation remains retryable");
    assert!(matches!(
        first,
        workflow::EnsureWorkflowError::Platform(MockError)
    ));

    let second = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        &planned.plan.plan_sha256,
        &mut platform,
    )
    .expect_err("second consecutive stall reaches the configured bound");
    assert!(matches!(
        second,
        workflow::EnsureWorkflowError::Stalled {
            observations: 2,
            ..
        }
    ));
    assert_eq!(platform.mutations.values().sum::<u32>(), 0);

    let terminal = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        &planned.plan.plan_sha256,
        &mut platform,
    )
    .expect("genuine progress resumes the retained operation");
    assert!(terminal.terminal);
    assert_eq!(platform.mutations.get(&first_hash), Some(&1));
}

#[test]
fn retirement_without_exact_drain_authority_leaves_material_cycles_untouched() {
    let mut fixture = fixture();
    fixture
        .desired
        .canisters
        .iter_mut()
        .find(|canister| canister.name == "retired")
        .expect("retired config")
        .drain = None;
    fixture.platform.desired = fixture.desired.clone();
    let error = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &"c".repeat(64),
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut fixture.platform,
    )
    .expect_err("material cycles require drain");
    assert!(matches!(
        error,
        workflow::EnsureWorkflowError::Policy(
            crate::fleet_ensure::policy::EnsurePolicyError::NoSafeDrain { .. }
        )
    ));
    assert_eq!(fixture.platform.mutations.values().sum::<u32>(), 0);
}

#[test]
fn retirement_never_stops_or_deletes_until_treasury_credit_is_observed() {
    let fixture = fixture();
    let mut platform = fixture.platform;
    platform.skip_transfer_credit = true;
    let source = "7".repeat(64);
    let planned = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut platform,
    )
    .expect("compile reviewed plan");
    let retired = planned
        .plan
        .canisters
        .iter()
        .find(|canister| canister.name == "retired")
        .expect("retired canister plan");
    let transfer = retired
        .actions
        .iter()
        .find(|action| matches!(action, EnsureAction::Transfer { .. }))
        .map(crate::fleet_ensure::ops::action_sha256)
        .expect("transfer action");
    let stop = retired
        .actions
        .iter()
        .find(|action| matches!(action, EnsureAction::Stop { .. }))
        .map(crate::fleet_ensure::ops::action_sha256)
        .expect("stop action");
    let delete = retired
        .actions
        .iter()
        .find(|action| matches!(action, EnsureAction::Delete { .. }))
        .map(crate::fleet_ensure::ops::action_sha256)
        .expect("delete action");

    let error = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        &planned.plan.plan_sha256,
        &mut platform,
    )
    .expect_err("missing treasury credit blocks retirement");

    assert!(matches!(
        error,
        workflow::EnsureWorkflowError::Stalled { .. }
    ));
    assert_eq!(platform.mutations.get(&transfer), Some(&1));
    assert_eq!(platform.mutations.get(&stop), None);
    assert_eq!(platform.mutations.get(&delete), None);
    assert!(platform.live.contains_key(RETIRED));
}

#[test]
fn protocol_response_is_only_issuance_and_terminal_status_gates_later_actions() {
    let mut fixture = fixture();
    fixture
        .desired
        .protocol_steps
        .push(protocol_step(&fixture.root));
    fixture.platform.desired = fixture.desired.clone();
    fixture.platform.protocol_command_only = true;
    let source = "8".repeat(64);
    let mut platform = fixture.platform;
    let planned = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut platform,
    )
    .expect("compile reviewed plan");
    let protocol = planned
        .plan
        .protocol_actions
        .first()
        .map(crate::fleet_ensure::ops::action_sha256)
        .expect("protocol action");
    let later_transfer = planned
        .plan
        .canisters
        .iter()
        .flat_map(|canister| &canister.actions)
        .find(|action| matches!(action, EnsureAction::Transfer { .. }))
        .map(crate::fleet_ensure::ops::action_sha256)
        .expect("later transfer");

    let error = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        &planned.plan.plan_sha256,
        &mut platform,
    )
    .expect_err("issued protocol remains nonterminal");
    assert!(matches!(
        error,
        workflow::EnsureWorkflowError::Stalled { .. }
    ));
    assert_eq!(platform.mutations.get(&protocol), Some(&1));
    assert_eq!(platform.mutations.get(&later_transfer), None);

    platform
        .protocol_ready
        .insert("fleet-catalog-terminal".to_string());
    let terminal = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        &planned.plan.plan_sha256,
        &mut platform,
    )
    .expect("terminal observation resumes without reissuing protocol command");
    assert!(terminal.terminal);
    assert_eq!(platform.mutations.get(&protocol), Some(&1));
}

#[test]
fn in_progress_operation_resumes_reviewed_desired_before_newer_input() {
    let mut fixture = fixture();
    fixture.desired.protocol = Some(DesiredFleetProtocol {
        app_config: "canic.toml".to_string(),
        component_group_placements: Vec::new(),
        coordinator_candid: "coordinator.did".to_string(),
        root_candid: "root.did".to_string(),
        store_candid: "store.did".to_string(),
    });
    fixture
        .desired
        .canisters
        .iter_mut()
        .find(|canister| canister.kind == DesiredCanisterKind::Coordinator)
        .expect("fixture Coordinator")
        .wasm = None;
    fixture.platform.desired = fixture.desired.clone();
    fixture.platform.protocol_command_only = true;
    fixture.platform.typed_protocol = true;
    let reviewed_sha256 = "31".repeat(32);
    let mut platform = fixture.platform;
    let planned = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &reviewed_sha256,
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut platform,
    )
    .expect("compile reviewed operation");
    let protocol = planned
        .plan
        .protocol_actions
        .first()
        .map(crate::fleet_ensure::ops::action_sha256)
        .expect("protocol action");

    let error = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &reviewed_sha256,
        "test-fleet",
        &planned.plan.plan_sha256,
        &mut platform,
    )
    .expect_err("issued protocol remains nonterminal");
    assert!(matches!(
        error,
        workflow::EnsureWorkflowError::Stalled { .. }
    ));
    assert_eq!(platform.mutations.get(&protocol), Some(&1));

    let mut newer = fixture.desired.clone();
    newer.maximum_stalled_observations = 17;
    let newer_sha256 = "32".repeat(32);
    platform.desired = newer.clone();
    let retained = workflow::plan(
        &fixture.root,
        &newer,
        &newer_sha256,
        "test-fleet",
        1_800_000_000_000_000_100,
        &mut platform,
    )
    .expect("return the immutable in-progress operation");
    assert_eq!(retained.plan.plan_sha256, planned.plan.plan_sha256);
    assert_eq!(retained.plan.desired_sha256, reviewed_sha256);
    assert_eq!(reviewed_desired(&retained.plan), Some(&fixture.desired));

    platform
        .protocol_ready
        .insert("fleet-component-provisioning".to_string());
    let terminal = workflow::apply(
        &fixture.root,
        &newer,
        &newer_sha256,
        "test-fleet",
        &retained.plan.plan_sha256,
        &mut platform,
    )
    .expect("resume the retained desired operation");
    assert!(terminal.terminal);
    assert_eq!(platform.mutations.get(&protocol), Some(&1));
    assert_eq!(platform.desired, fixture.desired);

    platform.desired = newer.clone();
    let successor = workflow::plan(
        &fixture.root,
        &newer,
        &newer_sha256,
        "test-fleet",
        1_800_000_000_000_000_200,
        &mut platform,
    )
    .expect("consider newer desired only after terminal operation");
    assert_eq!(successor.plan.desired_sha256, newer_sha256);
    assert_eq!(reviewed_desired(&successor.plan), Some(&newer));
}

#[test]
fn typed_fleet_protocol_is_issued_once_and_requires_terminal_status() {
    let mut fixture = fixture();
    fixture.desired.protocol = Some(DesiredFleetProtocol {
        app_config: "canic.toml".to_string(),
        component_group_placements: Vec::new(),
        coordinator_candid: "coordinator.did".to_string(),
        root_candid: "root.did".to_string(),
        store_candid: "store.did".to_string(),
    });
    fixture
        .desired
        .canisters
        .iter_mut()
        .find(|canister| canister.kind == DesiredCanisterKind::Coordinator)
        .expect("fixture Coordinator")
        .wasm = None;
    fixture.platform.desired = fixture.desired.clone();
    fixture.platform.protocol_command_only = true;
    fixture.platform.typed_protocol = true;
    let source = "a".repeat(64);
    let mut platform = fixture.platform;
    let planned = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut platform,
    )
    .expect("compile reviewed typed protocol plan");
    let action = planned
        .plan
        .protocol_actions
        .first()
        .expect("typed protocol action");
    assert!(matches!(action, EnsureAction::FleetProtocol { .. }));
    let action_hash = crate::fleet_ensure::ops::action_sha256(action);

    let error = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        &planned.plan.plan_sha256,
        &mut platform,
    )
    .expect_err("issued typed protocol remains nonterminal");
    assert!(matches!(
        error,
        workflow::EnsureWorkflowError::Stalled { .. }
    ));
    assert_eq!(platform.mutations.get(&action_hash), Some(&1));

    platform
        .protocol_ready
        .insert("fleet-component-provisioning".to_string());
    let terminal = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        &planned.plan.plan_sha256,
        &mut platform,
    )
    .expect("terminal typed status resumes without a second command");
    assert!(terminal.terminal);
    assert_eq!(platform.mutations.get(&action_hash), Some(&1));

    let mutation_count = platform.mutations.values().sum::<u32>();
    let successor = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        1_800_000_000_000_000_100,
        &mut platform,
    )
    .expect("plan immediate terminal replay");
    let replay = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        &successor.plan.plan_sha256,
        &mut platform,
    )
    .expect("apply immediate terminal replay");
    assert!(replay.terminal);
    assert_eq!(platform.mutations.get(&action_hash), Some(&1));
    assert_eq!(platform.mutations.values().sum::<u32>(), mutation_count);
}

#[test]
fn long_running_component_provisioning_is_paced_past_eight_observations_without_reissue() {
    let mut fixture = fixture();
    fixture.desired.protocol = Some(DesiredFleetProtocol {
        app_config: "canic.toml".to_string(),
        component_group_placements: Vec::new(),
        coordinator_candid: "coordinator.did".to_string(),
        root_candid: "root.did".to_string(),
        store_candid: "store.did".to_string(),
    });
    fixture.desired.maximum_stalled_observations = 12;
    fixture
        .desired
        .canisters
        .iter_mut()
        .find(|canister| canister.kind == DesiredCanisterKind::Coordinator)
        .expect("fixture Coordinator")
        .wasm = None;
    fixture.platform.desired = fixture.desired.clone();
    fixture.platform.protocol_command_only = true;
    fixture.platform.protocol_pending_waits = 10;
    let summary = FleetProvisioningProgress {
        pending_root_failure: None,
        phase: canic_core::dto::component_provisioning::FleetComponentProvisioningPhase::ActivatingRuntimes,
        root_batch_count: 1,
        accepted_root_count: 1,
        provisioned_root_count: 1,
        directory_confirmed_root_count: 1,
        directory_confirmation_root_count: 1,
        runtime_activated_root_count: 0,
        component_count: 3,
    };
    fixture.platform.provisioning_progress = Some(summary.clone());
    fixture.platform.typed_protocol = true;
    let source = "b".repeat(64);
    let mut platform = fixture.platform;
    let planned = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut platform,
    )
    .expect("compile reviewed long-running protocol plan");
    let action_hash = planned
        .plan
        .protocol_actions
        .first()
        .map(crate::fleet_ensure::ops::action_sha256)
        .expect("typed protocol action");

    let terminal = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        &planned.plan.plan_sha256,
        &mut platform,
    )
    .expect("paced observations reach terminal protocol status");

    assert!(terminal.terminal);
    assert_eq!(platform.paced_observations, (1..=10).collect::<Vec<_>>());
    assert_eq!(platform.mutations.get(&action_hash), Some(&1));
    assert_provisioning_wait_progress(&platform, &planned.plan, &summary);

    let mutation_count = platform.mutations.values().sum::<u32>();
    let successor = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        1_800_000_000_000_000_100,
        &mut platform,
    )
    .expect("plan immediate terminal replay");
    let replay = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        &successor.plan.plan_sha256,
        &mut platform,
    )
    .expect("terminal replay is effect-free");
    assert!(replay.terminal);
    assert_eq!(platform.paced_observations, (1..=10).collect::<Vec<_>>());
    assert_eq!(platform.mutations.get(&action_hash), Some(&1));
    assert_eq!(platform.mutations.values().sum::<u32>(), mutation_count);
}

fn assert_provisioning_wait_progress(
    platform: &MockPlatform,
    plan: &FleetEnsurePlan,
    summary: &FleetProvisioningProgress,
) {
    let waits = platform
        .progress
        .iter()
        .filter_map(|event| {
            if let FleetEnsureProgressState::AwaitingProgress {
                elapsed_seconds,
                provisioning,
            } = &event.state
            {
                assert_eq!(event.phase, FleetEnsurePhase::WorkloadProvisioning);
                assert_eq!(event.operation_id, plan.operation_id);
                assert_eq!(event.plan_sha256, plan.plan_sha256);
                assert_eq!(provisioning.as_ref(), Some(summary));
                assert!(event.applied_effects < u32::try_from(event.reviewed_effects).unwrap());
                Some(*elapsed_seconds)
            } else {
                None
            }
        })
        .collect::<Vec<_>>();
    assert_eq!(waits.len(), platform.paced_observations.len());
    assert!(waits.windows(2).all(|pair| pair[0] <= pair[1]));
    assert!(matches!(
        platform.progress.last().unwrap().state,
        FleetEnsureProgressState::Complete
    ));
}

#[test]
#[expect(
    clippy::too_many_lines,
    reason = "one focused regression keeps the exact Registry cycle projection and identity proof together"
)]
fn current_plan_round_trips_registry_actions_with_bounded_decimal_cycles() {
    let root = temp_dir("canic-fleet-ensure-json-round-trip");
    let paths = crate::fleet_ensure::ops::EnsurePaths::under(&root, "staging", "json-round-trip");
    let base_registry = empty_active_registry();
    let base_version = FleetRegistryVersion {
        authority: base_registry.authority.clone(),
        revision: base_registry.revision,
        content_hash: [12; 32],
    };
    let root_entry = FleetSubnetRootEntry {
        placement_subnet: SubnetId::from_principal(
            SUBNET.parse().expect("fixture Subnet Principal"),
        ),
        fleet_subnet_root: OLD_APP.parse().expect("fixture Root Principal"),
        component_admissions: Vec::new(),
        component_topology_digest: ComponentTopologyDigest::from_bytes([13; 32]),
        active_release_set: FleetSubnetRootReleaseSet {
            release_build_id: ReleaseBuildId::from_nonce(ReleaseBuildNonce::from_random_bytes(
                [14; 32],
            )),
            manifest_digest: ReleaseSetDigest::from_bytes([15; 32]),
        },
        limits: FleetSubnetRootLimits {
            maximum_component_instances: 1,
            maximum_registry_bytes: 16_777_216,
            maximum_wasm_store_bytes: 40_000_000,
            canister_pool: FleetSubnetCanisterPoolConfig {
                minimum_size: 1,
                maximum_size: 2,
                canister_cycles: Cycles::new(u128::MAX),
                creation_execution_margin: Cycles::new(1),
            },
            cycles_funding: CyclesFundingBudget {
                window_secs: 3_600,
                maximum_cycles: Cycles::new(u128::MAX),
            },
            maximum_group_placements: 1,
        },
        funding: crate::test_support::fleet_subnet_root_funding_authority(),
        status: FleetSubnetRootStatus::Joining,
    };
    let mut joined_registry = base_registry;
    joined_registry.revision = 2;
    joined_registry.fleet_subnet_roots.push(root_entry.clone());
    let joined_version = FleetRegistryVersion {
        authority: joined_registry.authority.clone(),
        revision: joined_registry.revision,
        content_hash: [16; 32],
    };
    let actions = vec![
        fleet_protocol_action(
            "registry-join",
            CurrentFleetProtocolAction::JoinRoot {
                expected_registry: joined_registry.clone(),
                expected_version: joined_version.clone(),
                request: FleetSubnetRootJoinRequest {
                    expected_registry: base_version,
                    entry: root_entry,
                },
            },
        ),
        fleet_protocol_action(
            "registry-activate",
            CurrentFleetProtocolAction::ActivateRegistry {
                expected_registry: joined_registry,
                expected_version: joined_version.clone(),
                request: FleetRegistryActivationRequest {
                    expected_registry: joined_version,
                },
            },
        ),
    ];
    let action_hashes = actions
        .iter()
        .map(crate::fleet_ensure::ops::action_sha256)
        .collect::<Vec<_>>();
    let mut plan = FleetEnsurePlan {
        continuation: None,
        canisters: Vec::new(),
        conservation: CycleConservation {
            estate_funding_domains: Vec::new(),
            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,
        },
        desired_sha256: "17".repeat(32),
        environment: "staging".to_string(),
        fleet: "json-round-trip".to_string(),
        operation_id: "18".repeat(32),
        plan_sha256: String::new(),
        planned_at_time: 1,
        protocol_actions: actions,
        recovery_review: None,
        reinstall: None,
        root_reinstall_bindings: Vec::new(),
        root_start_authority: None,
        reviewed_desired: None,
        schema_version: FLEET_ENSURE_SCHEMA_VERSION,
        scope: crate::fleet_ensure::model::FleetEnsurePlanScope::Full,
        terminal_inventory_operation_id: None,
    };
    plan.plan_sha256 = crate::fleet_ensure::policy::expected_plan_sha256(&plan);

    crate::fleet_ensure::ops::write_plan(&paths, &plan).expect("write current plan");
    let encoded = fs::read(&paths.plan).expect("read current plan JSON");
    let encoded: serde_json::Value =
        serde_json::from_slice(&encoded).expect("decode current plan JSON");
    let maximum = serde_json::Value::String(u128::MAX.to_string());
    for field in ["canister_cycles", "maximum_cycles"] {
        let values = json_field_values(&encoded, field);
        assert!(!values.is_empty(), "current plan omits {field}");
        assert!(
            values.contains(&&maximum),
            "current plan omits the maximum decimal {field}"
        );
    }
    let reopened = crate::fleet_ensure::ops::read_plan(&paths)
        .expect("read current plan")
        .expect("retained current plan");

    assert_eq!(reopened, plan);
    assert_eq!(
        crate::fleet_ensure::policy::expected_plan_sha256(&reopened),
        plan.plan_sha256
    );
    assert_eq!(
        reopened
            .protocol_actions
            .iter()
            .map(crate::fleet_ensure::ops::action_sha256)
            .collect::<Vec<_>>(),
        action_hashes
    );

    fs::remove_dir_all(root).expect("remove Fleet ensure JSON fixture");
}

#[test]
#[expect(
    clippy::too_many_lines,
    reason = "one focused proof keeps hash retention, partial progress, and content drift together"
)]
fn current_plan_retains_store_chunks_by_hash_instead_of_inline_bytes() {
    let root = temp_dir("canic-fleet-ensure-content-addressed-plan");
    let paths =
        crate::fleet_ensure::ops::EnsurePaths::under(&root, "staging", "content-addressed-plan");
    let template_id = TemplateId::owned("component:app".to_string());
    let version = TemplateVersion::owned("22".repeat(32));
    let bytes = vec![42; 4_096];
    let chunk_hash = canic_core::cdk::utils::hash::wasm_hash(&bytes);
    let actions = vec![fleet_protocol_action(
        "publish-store-chunk",
        CurrentFleetProtocolAction::PublishStoreChunk {
            request: TemplateChunkInput {
                preparation: Some(TemplateChunkSetPrepareInput {
                    manifest: None,
                    template_id: template_id.clone(),
                    version: version.clone(),
                    payload_hash: chunk_hash.clone(),
                    payload_size_bytes: bytes.len() as u64,
                    chunk_hashes: vec![chunk_hash.clone()],
                }),
                template_id,
                version,
                chunk_index: 0,
                bytes: bytes.clone(),
            },
        },
    )];
    let mut plan = FleetEnsurePlan {
        continuation: None,
        canisters: Vec::new(),
        conservation: CycleConservation {
            estate_funding_domains: Vec::new(),
            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,
        },
        desired_sha256: "23".repeat(32),
        environment: "staging".to_string(),
        fleet: "content-addressed-plan".to_string(),
        operation_id: "24".repeat(32),
        plan_sha256: String::new(),
        planned_at_time: 1,
        protocol_actions: actions,
        recovery_review: None,
        reinstall: None,
        root_reinstall_bindings: Vec::new(),
        root_start_authority: None,
        reviewed_desired: None,
        schema_version: FLEET_ENSURE_SCHEMA_VERSION,
        scope: crate::fleet_ensure::model::FleetEnsurePlanScope::Full,
        terminal_inventory_operation_id: None,
    };
    plan.plan_sha256 = crate::fleet_ensure::policy::expected_plan_sha256(&plan);

    fs::create_dir_all(paths.plan.parent().expect("plan state directory"))
        .expect("create inline-plan directory");
    fs::write(
        &paths.plan,
        crate::fleet_ensure::json::to_vec(&plan).expect("encode former inline plan"),
    )
    .expect("retain former inline plan");
    let inline_size = fs::metadata(&paths.plan)
        .expect("inspect former inline plan")
        .len();
    let inline = crate::fleet_ensure::ops::read_plan(&paths)
        .expect("read former inline current plan")
        .expect("former inline current plan");
    assert_eq!(inline, plan);
    assert!(
        crate::fleet_ensure::ops::compact_inline_plan(&paths, &inline)
            .expect("compact former inline current plan")
    );
    let encoded = fs::read(&paths.plan).expect("read hash-only current plan");
    assert!(encoded.len() as u64 * 2 < inline_size);
    let encoded: serde_json::Value =
        serde_json::from_slice(&encoded).expect("decode hash-only current plan");
    assert!(json_field_values(&encoded, "bytes").is_empty());
    assert_eq!(
        json_field_values(&encoded, "bytes_sha256"),
        [&serde_json::Value::String(sha256_hex(&bytes))]
    );
    assert_eq!(json_field_values(&encoded, "chunk_hashes").len(), 1);
    let object = paths
        .content
        .join(canic_core::cdk::utils::hash::hex_bytes(&chunk_hash));
    assert_eq!(
        fs::read(&object).expect("read retained chunk object"),
        bytes
    );

    let reopened = crate::fleet_ensure::ops::read_plan(&paths)
        .expect("read hash-only current plan")
        .expect("retained hash-only current plan");
    assert_eq!(reopened, plan);
    assert_eq!(
        crate::fleet_ensure::policy::expected_plan_sha256(&reopened),
        plan.plan_sha256
    );
    assert!(
        !crate::fleet_ensure::ops::compact_inline_plan(&paths, &reopened)
            .expect("leave canonical current plan unchanged")
    );

    let partial_paths = crate::fleet_ensure::ops::EnsurePaths::under(
        &root,
        "staging",
        "content-addressed-partial-plan",
    );
    let mut partial = plan;
    partial.fleet = "content-addressed-partial-plan".to_string();
    let EnsureAction::FleetProtocol { action, .. } = &mut partial.protocol_actions[0] else {
        panic!("Store action");
    };
    let CurrentFleetProtocolAction::PublishStoreChunk { request } = action.as_mut() else {
        panic!("Store chunk");
    };
    request.preparation = None;
    partial.plan_sha256 = crate::fleet_ensure::policy::expected_plan_sha256(&partial);
    crate::fleet_ensure::ops::write_plan(&partial_paths, &partial)
        .expect("write publish-only partial-progress plan");
    assert_eq!(
        crate::fleet_ensure::ops::read_plan(&partial_paths)
            .expect("read publish-only partial-progress plan")
            .expect("retained publish-only partial-progress plan"),
        partial
    );

    fs::remove_file(&object).expect("remove isolated chunk object");
    assert!(matches!(
        crate::fleet_ensure::ops::read_plan(&paths),
        Err(crate::fleet_ensure::ops::EnsureStateError::StoreChunkUnavailable { path })
            if path == object
    ));
    fs::write(&object, [9]).expect("tamper isolated chunk object");
    assert!(matches!(
        crate::fleet_ensure::ops::read_plan(&paths),
        Err(crate::fleet_ensure::ops::EnsureStateError::StoreChunkMismatch { path })
            if path == object
    ));
    fs::remove_dir_all(root).expect("remove content-addressed plan fixture");
}

#[test]
#[ignore = "requires an explicit read-only current Fleet Ensure evidence directory"]
#[expect(
    clippy::too_many_lines,
    reason = "one evidence regression keeps isolation, all action variants, and immutable identity checks together"
)]
fn retained_current_plan_and_issued_journal_round_trip_from_an_isolated_copy() {
    let source = PathBuf::from(
        env::var_os("CANIC_CURRENT_FLEET_ENSURE_EVIDENCE")
            .expect("set CANIC_CURRENT_FLEET_ENSURE_EVIDENCE"),
    );
    let source_plan = source.join("plan.json");
    let source_journal = source.join("journal.json");
    let original_plan = fs::read(&source_plan).expect("read retained plan evidence");
    let original_journal = fs::read(&source_journal).expect("read retained journal evidence");
    let scratch = temp_dir("canic-retained-fleet-ensure-json");
    let paths = crate::fleet_ensure::ops::EnsurePaths::under(&scratch, "evidence", "retained");
    fs::create_dir_all(
        paths
            .plan
            .parent()
            .expect("Fleet Ensure evidence directory"),
    )
    .expect("create isolated evidence directory");
    fs::write(&paths.plan, &original_plan).expect("copy retained plan evidence");
    fs::write(&paths.journal, &original_journal).expect("copy retained journal evidence");

    let mut plan = crate::fleet_ensure::ops::read_plan(&paths)
        .expect("decode retained current plan")
        .expect("retained current plan");
    let journal = crate::fleet_ensure::ops::read_journal(&paths)
        .expect("decode retained current journal")
        .expect("retained current journal");
    assert_eq!(
        crate::fleet_ensure::policy::expected_plan_sha256(&plan),
        plan.plan_sha256
    );
    assert_eq!(journal.plan_sha256, plan.plan_sha256);
    assert_eq!(journal.operation_id, plan.operation_id);
    assert!(matches!(
        journal.effects.last().map(|effect| &effect.state),
        Some(crate::fleet_ensure::model::EffectState::Issued)
    ));
    let retained_action_hashes = crate::fleet_ensure::workflow::ordered_actions(&plan)
        .into_iter()
        .take(journal.effects.len())
        .map(crate::fleet_ensure::ops::action_sha256)
        .collect::<Vec<_>>();
    assert_eq!(
        retained_action_hashes,
        journal
            .effects
            .iter()
            .map(|effect| effect.action_sha256.clone())
            .collect::<Vec<_>>()
    );
    assert!(
        crate::fleet_ensure::ops::compact_inline_plan(&paths, &plan)
            .expect("compact isolated retained plan before resumed effects")
    );
    let compacted_retained = fs::read(&paths.plan).expect("read compacted retained plan");
    assert!(compacted_retained.len() < original_plan.len() / 10);
    assert_eq!(
        crate::fleet_ensure::ops::read_plan(&paths)
            .expect("reopen compacted retained plan")
            .expect("compacted retained plan"),
        plan
    );
    assert_eq!(
        fs::read(&paths.journal).expect("reread isolated retained journal"),
        original_journal
    );

    let (registry_authority, root_entry) = plan
        .protocol_actions
        .iter()
        .find_map(|action| match action {
            EnsureAction::SealAuthority { .. } => {
                panic!("authority sealing uses the production PocketIC adapter")
            }
            EnsureAction::FleetProtocol { action, .. } => match action.as_ref() {
                CurrentFleetProtocolAction::JoinRoot {
                    expected_registry,
                    request,
                    ..
                } => Some((expected_registry.authority.clone(), request.entry.clone())),
                _ => None,
            },
            _ => None,
        })
        .expect("retained current plan Registry join");
    let store = plan
        .protocol_actions
        .iter()
        .find_map(|action| match action {
            EnsureAction::SealAuthority { .. } => {
                panic!("authority sealing uses the production PocketIC adapter")
            }
            EnsureAction::FleetProtocol { action, .. } => match action.as_ref() {
                CurrentFleetProtocolAction::BootstrapStore { expected, .. } => {
                    Some(expected.clone())
                }
                _ => None,
            },
            _ => None,
        })
        .expect("retained current plan Store bootstrap");
    plan.protocol_actions.push(fleet_protocol_action(
        "retained-evidence-adopt-store",
        CurrentFleetProtocolAction::AdoptStore {
            request: FleetSubnetWasmStoreAdoptionRequest {
                operation_id: [20; 32],
                authority: FleetSubnetWasmStoreAuthority {
                    authority: registry_authority,
                    placement_subnet: root_entry.placement_subnet,
                    fleet_subnet_root: root_entry.fleet_subnet_root,
                    wasm_store: store.wasm_store,
                    installation_controller: Principal::anonymous(),
                    release_build_id: store.release_set.release_build_id,
                    wasm_module_hash: [21; 32],
                },
            },
        },
    ));
    plan.plan_sha256 = crate::fleet_ensure::policy::expected_plan_sha256(&plan);
    let action_hashes = crate::fleet_ensure::workflow::ordered_actions(&plan)
        .into_iter()
        .map(crate::fleet_ensure::ops::action_sha256)
        .collect::<Vec<_>>();
    let expected_variants = BTreeSet::from([
        "activate_registry",
        "activate_registry_mirror",
        "adopt_store",
        "bootstrap_store",
        "join_root",
        "prepare_component_registry",
        "provision_components",
        "publish_store_chunk",
        "synchronize_registry",
    ]);
    assert_eq!(current_protocol_variants(&plan), expected_variants);

    crate::fleet_ensure::ops::write_plan(&paths, &plan).expect("write isolated current plan");
    let reopened = crate::fleet_ensure::ops::read_plan(&paths)
        .expect("reopen isolated current plan")
        .expect("isolated current plan");
    assert_eq!(reopened, plan);
    assert_eq!(
        crate::fleet_ensure::policy::expected_plan_sha256(&reopened),
        plan.plan_sha256
    );
    assert_eq!(
        crate::fleet_ensure::workflow::ordered_actions(&reopened)
            .into_iter()
            .map(crate::fleet_ensure::ops::action_sha256)
            .collect::<Vec<_>>(),
        action_hashes
    );
    let canonical = fs::read(&paths.plan).expect("read canonical current plan");
    assert!(canonical.len() < original_plan.len() / 10);
    assert!(
        !canonical
            .windows(b"\"bytes\"".len())
            .any(|window| window == b"\"bytes\"")
    );
    let second = crate::fleet_ensure::ops::EnsurePaths::under(&scratch, "evidence", "second");
    crate::fleet_ensure::ops::write_plan(&second, &reopened)
        .expect("write second canonical current plan");
    assert_eq!(
        fs::read(&second.plan).expect("read second canonical current plan"),
        canonical
    );

    assert_eq!(
        fs::read(source_plan).expect("reread source plan"),
        original_plan
    );
    assert_eq!(
        fs::read(source_journal).expect("reread source journal"),
        original_journal
    );
    fs::remove_dir_all(scratch).expect("remove isolated retained evidence");
}

#[test]
fn tampered_reviewed_plan_fails_before_any_effect() {
    let fixture = fixture();
    let mut platform = fixture.platform;
    let source = "e".repeat(64);
    let planned = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut platform,
    )
    .expect("compile reviewed plan");
    let paths = crate::fleet_ensure::ops::EnsurePaths::under(
        &fixture.root,
        &fixture.desired.environment,
        "test-fleet",
    );
    let mut retained = serde_json::from_slice::<crate::fleet_ensure::model::FleetEnsurePlan>(
        &fs::read(&paths.plan).expect("read retained plan"),
    )
    .expect("decode retained plan");
    retained.conservation.maximum_new_funding_cycles += 1;
    fs::write(
        &paths.plan,
        serde_json::to_vec_pretty(&retained).expect("encode tampered plan"),
    )
    .expect("tamper retained plan");

    let error = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &source,
        "test-fleet",
        &planned.plan.plan_sha256,
        &mut platform,
    )
    .expect_err("tampered retained plan rejects");
    assert!(matches!(
        error,
        workflow::EnsureWorkflowError::PlanIntegrity
    ));
    assert_eq!(platform.mutations.values().sum::<u32>(), 0);
}

#[test]
fn unreadable_plan_without_recovery_evidence_preserves_bytes_and_refuses_effects() {
    let fixture = fixture();
    let mut platform = fixture.platform;
    let digest = "e".repeat(64);
    let planned = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &digest,
        "test-fleet",
        1,
        &mut platform,
    )
    .expect("retain current plan");
    let paths = crate::fleet_ensure::ops::EnsurePaths::under(
        &fixture.root,
        &fixture.desired.environment,
        "test-fleet",
    );
    for (bytes, category) in [
        (b"{".as_slice(), serde_json::error::Category::Eof),
        (
            br#"{"protocol_actions":[]}"#.as_slice(),
            serde_json::error::Category::Data,
        ),
    ] {
        fs::write(&paths.plan, bytes).expect("retain damaged plan");
        let failures = [
            workflow::plan(
                &fixture.root,
                &fixture.desired,
                &digest,
                "test-fleet",
                2,
                &mut platform,
            )
            .expect_err("planning must preserve unreadable authority"),
            workflow::apply(
                &fixture.root,
                &fixture.desired,
                &digest,
                "test-fleet",
                &planned.plan.plan_sha256,
                &mut platform,
            )
            .expect_err("apply must refuse unreadable authority"),
        ];
        for error in failures {
            let workflow::EnsureWorkflowError::RetainedPlanUnreadable { source } = error else {
                panic!("expected an unreadable plan diagnostic");
            };
            assert!(matches!(*source,
                crate::fleet_ensure::ops::EnsureStateError::Decode { path, source }
                if path == paths.plan && source.classify() == category
            ));
        }
        assert_eq!(fs::read(&paths.plan).expect("retained bytes"), bytes);
        assert!(!paths.journal.exists());
    }
    assert_eq!(platform.mutations.values().sum::<u32>(), 0);
    fs::remove_dir_all(fixture.root).expect("remove fixture");
}

#[test]
#[ignore = "the workspace runner supplies one shared PocketIC server and serial execution"]
#[expect(
    clippy::too_many_lines,
    reason = "one governed journey keeps literal-zero planning, lost-Create resume, convergence and second-run proof together"
)]
fn governed_pocketic_fresh_estate_recovers_creation_and_replays_without_effects() {
    use ic_testkit::pic::PocketIcBuilder;
    use ic_testkit::pocket_ic::{CanisterSettings, CreateCanisterParams, PocketIc};

    struct PocketPlatform {
        create_outcomes: BTreeMap<String, EffectOutcome>,
        desired: DesiredFleet,
        fail_after_effect: Option<String>,
        known: BTreeSet<String>,
        mutations: BTreeMap<String, u32>,
        operator_cycles: u128,
        pic: PocketIc,
        protocol_ready: BTreeSet<String>,
        reinstall_required: BTreeSet<String>,
    }

    impl PocketPlatform {
        fn principal<'a>(state: &'a FleetEnsureStateRecord, principal: &'a str) -> Option<&'a str> {
            principal
                .strip_prefix("created:")
                .map_or(Some(principal), |name| {
                    state.pending_principals.get(name).map(String::as_str)
                })
        }

        fn live(&self, principal: &str) -> Option<LiveCanister> {
            if !self.known.contains(principal) {
                return None;
            }
            let id = principal.parse().expect("PocketIC Principal");
            let status = self
                .pic
                .canister_status(id, Some(CONTROLLER.parse().expect("controller Principal")))
                .expect("PocketIC status");
            Some(LiveCanister {
                canister_version: Some(status.version),
                controllers: status
                    .settings
                    .controllers
                    .iter()
                    .map(ToString::to_string)
                    .collect(),
                cycles: self.pic.cycle_balance(id),
                module_sha256: status
                    .module_hash
                    .map(canic_core::cdk::utils::hash::hex_bytes),
                principal: principal.to_string(),
                reinstall_required: self.reinstall_required.contains(principal),
                root_owned_lifecycle: None,
                status: match format!("{:?}", status.status).as_str() {
                    "Running" => CanisterRuntimeStatus::Running,
                    "Stopped" => CanisterRuntimeStatus::Stopped,
                    _ => CanisterRuntimeStatus::Stopping,
                },
            })
        }
    }

    impl EnsurePlatform for PocketPlatform {
        type Error = std::io::Error;

        fn bind_reviewed_desired(&mut self, desired: &DesiredFleet) -> Result<(), Self::Error> {
            self.desired = desired.clone();
            Ok(())
        }

        fn pace_effect_observation(
            &mut self,
            _action: &EnsureAction,
            _consecutive_unchanged_observations: u32,
        ) {
        }

        fn observe(
            &mut self,
            _operation_id: &str,
            state: &FleetEnsureStateRecord,
        ) -> Result<FleetObservation, Self::Error> {
            let canisters = self
                .desired
                .canisters
                .iter()
                .map(|configured| {
                    let principal = state
                        .principals
                        .get(&configured.name)
                        .map(String::as_str)
                        .or(configured.principal.as_deref());
                    (
                        configured.name.clone(),
                        principal.and_then(|principal| self.live(principal)),
                    )
                })
                .collect();
            Ok(FleetObservation {
                additional_controlled_cycles: BTreeMap::new(),
                canisters,
                estate_funding_domains: BTreeMap::new(),
                ledger_fee_cycles: self
                    .desired
                    .ledger_fee_cycles
                    .parse::<Cycles>()
                    .map(|cycles| cycles.to_u128())
                    .expect("PocketIC fixture ledger fee"),
                operator_cycles: self.operator_cycles,
                protocol_ready: self
                    .desired
                    .protocol_steps
                    .iter()
                    .map(|step| (step.name.clone(), self.protocol_ready.contains(&step.name)))
                    .collect(),
            })
        }

        fn observe_effect(
            &mut self,
            _operation_id: &str,
            action: &EnsureAction,
            record: &EffectRecord,
            state: &FleetEnsureStateRecord,
        ) -> Result<EffectObservation, Self::Error> {
            if let EnsureAction::Create {
                requested_initial_cycles,
                ..
            } = action
            {
                let post_cycles = record
                    .created_principal
                    .as_deref()
                    .and_then(|principal| self.live(principal))
                    .map(|live| live.cycles);
                let maximum_observation_burn_cycles =
                    crate::fleet_ensure::ops::maximum_creation_observation_burn(&self.desired)
                        .expect("fixture Create execution and observation burn");
                let applied = create_balance_is_terminal(
                    post_cycles,
                    *requested_initial_cycles,
                    maximum_observation_burn_cycles,
                );
                return Ok(EffectObservation {
                    provisioning_progress: None,
                    provisioning_failure: None,
                    applied,
                    estate_funding_required: None,
                    post_cycles,
                    progress_identity: format!("created:{:?}", record.created_principal),
                    retry: if post_cycles.is_some() && !applied {
                        EffectRetry::ReplanRequiredAfterCreateBalanceDrift
                    } else {
                        EffectRetry::None
                    },
                });
            }
            if matches!(action, EnsureAction::Fund { .. }) {
                return Ok(EffectObservation {
                    provisioning_progress: None,
                    provisioning_failure: None,
                    applied: record.receipt.is_some(),
                    estate_funding_required: None,
                    post_cycles: None,
                    progress_identity: format!("fund:{:?}", record.receipt),
                    retry: EffectRetry::None,
                });
            }
            let principal = match action {
                EnsureAction::SealAuthority { .. } => {
                    panic!("authority sealing uses the production PocketIC adapter")
                }
                EnsureAction::Delete { principal, .. }
                | EnsureAction::FleetProtocol { principal, .. }
                | EnsureAction::FundEstate { principal, .. }
                | EnsureAction::Install { principal, .. }
                | EnsureAction::Protocol { principal, .. }
                | EnsureAction::SetControllers { principal, .. }
                | EnsureAction::Start { principal, .. }
                | EnsureAction::Stop { principal, .. }
                | EnsureAction::Transfer { principal, .. } => Self::principal(state, principal),
                EnsureAction::Create { .. } | EnsureAction::Fund { .. } => None,
            };
            let applied = match action {
                EnsureAction::SealAuthority { .. } => {
                    panic!("authority sealing uses the production PocketIC adapter")
                }
                EnsureAction::Install {
                    mode, wasm_sha256, ..
                } => principal
                    .and_then(|value| self.live(value))
                    .is_some_and(|live| {
                        crate::fleet_ensure::ops::install_effect_applied(
                            *mode,
                            wasm_sha256,
                            live.module_sha256.as_deref(),
                            record.pre_canister_version,
                            live.canister_version,
                        )
                    }),
                EnsureAction::Start { .. } => principal
                    .and_then(|value| self.live(value))
                    .is_some_and(|live| live.status == CanisterRuntimeStatus::Running),
                EnsureAction::SetControllers { controllers, .. } => principal
                    .and_then(|value| self.live(value))
                    .is_some_and(|live| live.controllers == *controllers),
                EnsureAction::Delete { .. } => {
                    principal.is_none_or(|value| self.live(value).is_none())
                }
                EnsureAction::Stop { .. } => principal
                    .and_then(|value| self.live(value))
                    .is_some_and(|live| live.status == CanisterRuntimeStatus::Stopped),
                EnsureAction::FleetProtocol { name, .. } | EnsureAction::Protocol { name, .. } => {
                    self.protocol_ready.contains(name)
                }
                EnsureAction::Create { .. }
                | EnsureAction::Fund { .. }
                | EnsureAction::FundEstate { .. }
                | EnsureAction::Transfer { .. } => false,
            };
            Ok(EffectObservation {
                provisioning_progress: None,
                provisioning_failure: None,
                applied,
                estate_funding_required: None,
                post_cycles: None,
                progress_identity: format!("pocketic:{action:?}:{applied}"),
                retry: EffectRetry::None,
            })
        }

        fn action_cycles(
            &mut self,
            action: &EnsureAction,
            state: &FleetEnsureStateRecord,
        ) -> Result<Option<u128>, Self::Error> {
            let principal = match action {
                EnsureAction::SealAuthority { .. } => {
                    panic!("authority sealing uses the production PocketIC adapter")
                }
                EnsureAction::Create { .. } => return Ok(None),
                EnsureAction::Delete { principal, .. }
                | EnsureAction::FleetProtocol { principal, .. }
                | EnsureAction::Fund { principal, .. }
                | EnsureAction::FundEstate { principal, .. }
                | EnsureAction::Install { principal, .. }
                | EnsureAction::Protocol { principal, .. }
                | EnsureAction::SetControllers { principal, .. }
                | EnsureAction::Start { principal, .. }
                | EnsureAction::Stop { principal, .. }
                | EnsureAction::Transfer { principal, .. } => Self::principal(state, principal),
            };
            Ok(principal.and_then(|value| self.live(value).map(|live| live.cycles)))
        }

        fn action_destination_cycles(
            &mut self,
            _action: &EnsureAction,
            _state: &FleetEnsureStateRecord,
        ) -> Result<Option<u128>, Self::Error> {
            Ok(None)
        }

        fn action_canister_version(
            &mut self,
            action: &EnsureAction,
            state: &FleetEnsureStateRecord,
        ) -> Result<Option<u64>, Self::Error> {
            let principal = match action {
                EnsureAction::SealAuthority { .. } => {
                    panic!("authority sealing uses the production PocketIC adapter")
                }
                EnsureAction::Install { principal, .. } => Self::principal(state, principal),
                _ => None,
            };
            Ok(principal
                .and_then(|value| self.live(value))
                .and_then(|live| live.canister_version))
        }

        fn apply(
            &mut self,
            _operation_id: &str,
            action: &EnsureAction,
            _record: &EffectRecord,
            state: &FleetEnsureStateRecord,
        ) -> Result<EffectOutcome, Self::Error> {
            let principal = |value: &str| {
                Self::principal(state, value)
                    .expect("created Principal exists")
                    .parse()
                    .expect("PocketIC Principal")
            };
            let action_hash = action_sha256(action);
            if let Some(outcome) = self.create_outcomes.get(&action_hash) {
                return Ok(outcome.clone());
            }
            let outcome = match action {
                EnsureAction::SealAuthority { .. } => {
                    panic!("authority sealing uses the production PocketIC adapter")
                }
                EnsureAction::Create {
                    requested_initial_cycles,
                    ..
                } => {
                    let id = self
                        .pic
                        .create_canister_with_params(
                            None,
                            CreateCanisterParams {
                                cycles: Some(*requested_initial_cycles),
                                settings: Some(CanisterSettings {
                                    controllers: Some(vec![
                                        CONTROLLER.parse().expect("controller Principal"),
                                    ]),
                                    ..CanisterSettings::default()
                                }),
                                ..CreateCanisterParams::default()
                            },
                        )
                        .map_err(std::io::Error::other)?;
                    self.operator_cycles -= requested_initial_cycles;
                    self.known.insert(id.to_string());
                    let outcome = EffectOutcome {
                        created_principal: Some(id.to_string()),
                        post_cycles: Some(*requested_initial_cycles),
                        receipt: Some(format!("pocketic-create-{id}")),
                    };
                    self.create_outcomes
                        .insert(action_hash.clone(), outcome.clone());
                    Ok(outcome)
                }
                EnsureAction::Fund {
                    amount,
                    principal: target,
                    ..
                } => {
                    let id = principal(target);
                    self.pic.add_cycles(id, *amount);
                    self.operator_cycles -= amount;
                    Ok(EffectOutcome {
                        created_principal: None,
                        post_cycles: Some(self.pic.cycle_balance(id)),
                        receipt: Some(format!("pocketic-fund-{id}")),
                    })
                }
                EnsureAction::FundEstate { .. } => Err(std::io::Error::other(
                    "governed current-state journey has no estate funding domain",
                )),
                EnsureAction::Install {
                    mode,
                    principal: target,
                    wasm,
                    ..
                } => {
                    let id = principal(target);
                    let bytes = fs::read(wasm).expect("read PocketIC Wasm");
                    match mode {
                        crate::fleet_ensure::model::InstallMode::Install => {
                            self.pic.install_canister(
                                id,
                                bytes,
                                Vec::new(),
                                Some(CONTROLLER.parse().expect("controller Principal")),
                            );
                        }
                        crate::fleet_ensure::model::InstallMode::Reinstall => {
                            self.pic
                                .reinstall_canister(
                                    id,
                                    bytes,
                                    Vec::new(),
                                    Some(CONTROLLER.parse().expect("controller Principal")),
                                )
                                .expect("reinstall PocketIC canister");
                            self.reinstall_required.remove(&id.to_string());
                        }
                    }
                    Ok(empty_outcome())
                }
                EnsureAction::FleetProtocol {
                    name,
                    principal: target,
                    ..
                }
                | EnsureAction::Protocol {
                    name,
                    principal: target,
                    ..
                } => {
                    self.protocol_ready.insert(name.clone());
                    Ok(EffectOutcome {
                        created_principal: None,
                        post_cycles: Some(self.pic.cycle_balance(principal(target))),
                        receipt: Some("pocketic-protocol".to_string()),
                    })
                }
                EnsureAction::SetControllers {
                    controllers,
                    principal: target,
                    ..
                } => {
                    self.pic
                        .set_controllers(
                            principal(target),
                            Some(CONTROLLER.parse().expect("controller Principal")),
                            controllers
                                .iter()
                                .map(|value| value.parse().expect("controller Principal"))
                                .collect(),
                        )
                        .expect("set PocketIC controllers");
                    Ok(empty_outcome())
                }
                EnsureAction::Start {
                    principal: target, ..
                } => {
                    self.pic
                        .start_canister(
                            principal(target),
                            Some(CONTROLLER.parse().expect("controller Principal")),
                        )
                        .expect("start PocketIC canister");
                    Ok(empty_outcome())
                }
                EnsureAction::Delete { .. }
                | EnsureAction::Stop { .. }
                | EnsureAction::Transfer { .. } => Err(std::io::Error::other(
                    "governed current-state journey does not retire canisters",
                )),
            }?;
            *self.mutations.entry(action_hash.clone()).or_default() += 1;
            if self.fail_after_effect.as_deref() == Some(&action_hash) {
                self.fail_after_effect = None;
                return Err(std::io::Error::other(
                    "simulated process loss after the exact remote effect",
                ));
            }
            Ok(outcome)
        }
    }

    let root = temp_dir("canic-fleet-ensure-pocketic");
    fs::create_dir_all(&root).expect("create PocketIC fixture root");
    let wasm = root.join("current.wasm");
    fs::write(&wasm, b"\0asm\x01\0\0\0").expect("write minimal Wasm");
    let pic = start_pocket_ic(PocketIcBuilder::new().with_application_subnet());
    let mut canisters = vec![DesiredCanister {
        canic_init: None,
        controller_canisters: Vec::new(),
        controllers: vec![CONTROLLER.to_string()],
        drain: None,
        initial_cycles: "1000000000000".to_string(),
        init_arg: None,
        init_candid: None,
        kind: DesiredCanisterKind::Auxiliary,
        minimum_cycles: "500000000000".to_string(),
        name: "treasury".to_string(),
        parent: None,
        presence: DesiredPresence::Present,
        principal: None,
        protocol_binding: None,
        replace: false,
        subnet: SUBNET.to_string(),
        wasm: Some(wasm.display().to_string()),
    }];
    let mut root_desired = desired_canister("root", None, false, &wasm, None);
    root_desired.kind = DesiredCanisterKind::Root;
    root_desired.parent = Some("coordinator".to_string());
    root_desired.initial_cycles = "1000000000000".to_string();
    root_desired.minimum_cycles = "500000000000".to_string();
    canisters.push(root_desired);
    for role in ["coordinator", "service"] {
        let mut canister = desired_canister(role, None, false, &wasm, None);
        canister.kind = if role == "coordinator" {
            DesiredCanisterKind::Coordinator
        } else {
            DesiredCanisterKind::Auxiliary
        };
        canister.parent = None;
        canister.initial_cycles = "1000000000000".to_string();
        canister.minimum_cycles = "500000000000".to_string();
        canisters.push(canister);
    }
    let desired = DesiredFleet {
        bootstrap: None,
        canisters,
        cycles_ledger: LEDGER.to_string(),
        environment: "local".to_string(),
        fleet: "synthetic".to_string(),
        ledger_fee_cycles: "0".to_string(),
        management_creation_fee_cycles: "0".to_string(),
        material_cycle_threshold: "1000000".to_string(),
        maximum_observation_burn_cycles: "10000000".to_string(),
        maximum_stalled_observations: 8,
        maximum_update_burn_cycles: "100000000000".to_string(),
        operator: CONTROLLER.to_string(),
        protocol: None,
        protocol_steps: [
            ("prepare-coordinator", "coordinator"),
            ("prepare-root", "root"),
            ("activate-service", "service"),
        ]
        .into_iter()
        .map(|(name, canister)| protocol_step_for(&root, name, canister))
        .collect(),
        schema_version: FLEET_ENSURE_SCHEMA_VERSION,
        treasury: "treasury".to_string(),
    };
    let mut platform = PocketPlatform {
        create_outcomes: BTreeMap::new(),
        desired: desired.clone(),
        fail_after_effect: None,
        known: BTreeSet::new(),
        mutations: BTreeMap::new(),
        operator_cycles: 10_000_000_000_000,
        pic,
        protocol_ready: BTreeSet::new(),
        reinstall_required: BTreeSet::new(),
    };
    let source = "d".repeat(64);
    let planned = workflow::plan(
        &root,
        &desired,
        &source,
        "synthetic",
        1_800_000_000_000_000_000,
        &mut platform,
    )
    .expect("plan literal zero PocketIC estate");
    let first_create = workflow::ordered_actions(&planned.plan)
        .into_iter()
        .find(|action| matches!(action, EnsureAction::Create { .. }))
        .map(action_sha256)
        .expect("literal zero-estate plan contains one Create");
    platform.fail_after_effect = Some(first_create.clone());
    assert!(matches!(
        workflow::apply(
            &root,
            &desired,
            &source,
            "synthetic",
            &planned.plan.plan_sha256,
            &mut platform,
        ),
        Err(workflow::EnsureWorkflowError::Platform(_))
    ));
    let paths = crate::fleet_ensure::ops::EnsurePaths::under(&root, "local", "synthetic");
    let interrupted = crate::fleet_ensure::ops::read_journal(&paths)
        .expect("read interrupted zero-estate journal")
        .expect("interrupted zero-estate journal");
    assert_eq!(interrupted.completion, FleetEnsureCompletion::InProgress);
    assert_eq!(platform.mutations.get(&first_create), Some(&1));

    // An outside top-up arrives after creation paid but before its lost reply is
    // reconciled. It must not invalidate the exact retained creation identity.
    let donated_to: Principal = platform.known.iter().next().unwrap().parse().unwrap();
    let donation = 5_000_000_000_000;
    platform.pic.add_cycles(donated_to, donation);

    // Reconstruct the host adapter around the same live PocketIC estate. The
    // reviewed plan, journal and state are reopened from disk; only remote
    // idempotency evidence crosses this simulated process boundary.
    let PocketPlatform {
        create_outcomes,
        known,
        mutations,
        operator_cycles,
        pic,
        protocol_ready,
        reinstall_required,
        ..
    } = platform;
    let mut restarted = PocketPlatform {
        create_outcomes,
        desired: desired.clone(),
        fail_after_effect: None,
        known,
        mutations,
        operator_cycles,
        pic,
        protocol_ready,
        reinstall_required,
    };
    let applied = workflow::apply(
        &root,
        &desired,
        &source,
        "synthetic",
        &planned.plan.plan_sha256,
        &mut restarted,
    )
    .expect("resume and converge literal zero PocketIC estate");
    assert!(applied.terminal);
    let actual = applied.actual_conservation.as_ref().unwrap();
    assert!(actual.observed_net_cycle_credit_cycles > 0);
    assert_eq!(actual.observed_net_cycle_debit_cycles, 0);
    assert_eq!(
        actual.observed_starting_cycles,
        interrupted.initial_controlled_cycles
    );
    assert_eq!(restarted.mutations.get(&first_create), Some(&1));
    assert_eq!(restarted.known.len(), desired.canisters.len());

    let paid_before_replay = restarted.mutations.clone();
    restarted.pic.add_cycles(donated_to, donation);
    let replay = workflow::apply(
        &root,
        &desired,
        &source,
        "synthetic",
        &planned.plan.plan_sha256,
        &mut restarted,
    )
    .expect("original operation replay accepts another outside donation");
    assert!(replay.terminal);
    assert_eq!(restarted.mutations, paid_before_replay);

    let second = workflow::plan(
        &root,
        &desired,
        &source,
        "synthetic",
        1_800_000_000_000_000_100,
        &mut restarted,
    )
    .expect("plan converged PocketIC estate");
    assert!(
        second
            .plan
            .canisters
            .iter()
            .all(|canister| canister.actions.is_empty())
    );
    let mutations_before_replay = restarted.mutations.clone();
    let replay = workflow::apply(
        &root,
        &desired,
        &source,
        "synthetic",
        &second.plan.plan_sha256,
        &mut restarted,
    )
    .expect("effect-free PocketIC replay");
    assert!(replay.terminal);
    assert_eq!(replay.effects_applied, 0);
    assert_eq!(restarted.mutations, mutations_before_replay);
}

pub(super) struct Fixture {
    pub(super) desired: DesiredFleet,
    pub(super) platform: MockPlatform,
    pub(super) root: PathBuf,
}

#[test]
fn fresh_logical_controller_and_treasury_roles_create_and_replay_without_effect() {
    let mut fixture = fixture();
    let wasm = fixture.root.join("app.wasm");
    let mut coordinator = desired_canister("treasury", None, false, &wasm, None);
    coordinator.kind = DesiredCanisterKind::Coordinator;
    let mut child = desired_canister("created", None, false, &wasm, None);
    child.controller_canisters = vec!["treasury".to_string()];
    fixture.desired.canisters = vec![coordinator, child];
    fixture.desired.treasury = "treasury".to_string();
    fixture.desired.ledger_fee_cycles = "10B".to_string();
    fixture.desired.management_creation_fee_cycles = "50B".to_string();
    for canister in &mut fixture.desired.canisters {
        canister.initial_cycles = "20B".to_string();
        canister.minimum_cycles = "20B".to_string();
    }
    fixture.platform = MockPlatform::new(fixture.desired.clone(), Vec::new());
    fixture.platform.operator_cycles = 1_000_000_000_000_000;

    let desired_sha256 = "73".repeat(32);
    let planned = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        &fixture.desired.fleet,
        1_800_000_000_000_000_000,
        &mut fixture.platform,
    )
    .expect("plan fresh logical authority");
    assert_eq!(
        planned
            .plan
            .canisters
            .iter()
            .filter(|canister| canister.disposition == CanisterDisposition::Create)
            .count(),
        2
    );
    let applied = workflow::apply(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        &fixture.desired.fleet,
        &planned.plan.plan_sha256,
        &mut fixture.platform,
    )
    .expect("apply fresh logical authority");
    assert!(applied.terminal);

    let second = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        &fixture.desired.fleet,
        1_800_000_000_000_000_100,
        &mut fixture.platform,
    )
    .expect("replan converged logical authority");
    assert!(
        second
            .plan
            .canisters
            .iter()
            .all(|canister| canister.actions.is_empty())
    );

    fs::remove_dir_all(fixture.root).expect("remove test directory");
}

#[test]
fn protocol_planning_selects_the_largest_ordered_prefix_with_cycle_headroom() {
    let mut fixture = protocol_tranche_fixture(vec![499, 2]);

    let report = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &"12".repeat(32),
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut fixture.platform,
    )
    .expect("compile the largest affordable protocol prefix");

    assert_eq!(report.plan.protocol_actions.len(), 1);
    assert_eq!(
        report.plan.protocol_actions[0].name(),
        "fleet-component-provisioning-0"
    );
    assert_eq!(report.plan.conservation.maximum_execution_burn_cycles, 499);
    assert_eq!(report.plan.conservation.observed_controlled_cycles, 500);
    assert_eq!(report.plan.conservation.expected_post_operation_cycles, 1);
    assert!(fixture.platform.mutations.is_empty());

    let first_action_sha256 = crate::fleet_ensure::ops::action_sha256(
        report
            .plan
            .protocol_actions
            .first()
            .expect("first protocol tranche"),
    );
    let first = apply_fixture_plan(&mut fixture, &"12".repeat(32), &report.plan)
        .expect_err("the completed prefix requires one reviewed successor plan");
    assert!(matches!(
        first,
        workflow::EnsureWorkflowError::SuccessorReviewRequired {
            reason: crate::fleet_ensure::model::FleetEnsureSuccessorReviewReason::AdditionalEffect,
            review: Some(_)
        }
    ));
    assert_eq!(
        fixture.platform.mutations.get(&first_action_sha256),
        Some(&1)
    );

    let successor = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &"12".repeat(32),
        "test-fleet",
        1_800_000_000_000_000_100,
        &mut fixture.platform,
    )
    .expect("compile the remaining affordable protocol tranche");
    assert_eq!(successor.plan.protocol_actions.len(), 1);
    assert_eq!(
        successor.plan.protocol_actions[0].name(),
        "fleet-component-provisioning-1"
    );
    assert_eq!(successor.plan.conservation.maximum_execution_burn_cycles, 2);
    let successor_action_sha256 = crate::fleet_ensure::ops::action_sha256(
        successor
            .plan
            .protocol_actions
            .first()
            .expect("successor protocol tranche"),
    );
    let terminal = apply_fixture_plan(&mut fixture, &"12".repeat(32), &successor.plan)
        .expect("apply the terminal protocol tranche");
    assert!(terminal.terminal);
    assert_eq!(
        fixture.platform.mutations.get(&successor_action_sha256),
        Some(&1)
    );

    let mutations = fixture.platform.mutations.clone();
    let replay_plan = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &"12".repeat(32),
        "test-fleet",
        1_800_000_000_000_000_200,
        &mut fixture.platform,
    )
    .expect("compile effect-free protocol replay");
    assert!(workflow::ordered_actions(&replay_plan.plan).is_empty());
    let replay = apply_fixture_plan(&mut fixture, &"12".repeat(32), &replay_plan.plan)
        .expect("apply effect-free protocol replay");
    assert!(replay.terminal);
    assert_eq!(fixture.platform.mutations, mutations);

    fs::remove_dir_all(fixture.root).expect("remove test directory");
}

fn apply_fixture_plan(
    fixture: &mut Fixture,
    desired_sha256: &str,
    plan: &FleetEnsurePlan,
) -> Result<crate::fleet_ensure::model::FleetEnsureReport, workflow::EnsureWorkflowError<MockError>>
{
    workflow::apply(
        &fixture.root,
        &fixture.desired,
        desired_sha256,
        "test-fleet",
        &plan.plan_sha256,
        &mut fixture.platform,
    )
}

#[test]
fn first_unaffordable_protocol_action_rejects_with_exact_cycle_guidance() {
    let mut fixture = protocol_tranche_fixture(vec![501, 1]);

    let error = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &"13".repeat(32),
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut fixture.platform,
    )
    .expect_err("the first indivisible protocol action cannot fit");

    assert!(matches!(
        error,
        workflow::EnsureWorkflowError::Policy(
            crate::fleet_ensure::policy::EnsurePolicyError::InsufficientCycleConservation {
                action_count: 1,
                available: 500,
                required: 501,
                shortfall: 1,
            }
        )
    ));
    let paths = crate::fleet_ensure::ops::EnsurePaths::under(
        &fixture.root,
        &fixture.desired.environment,
        "test-fleet",
    );
    assert!(
        crate::fleet_ensure::ops::read_plan(&paths)
            .expect("read absent rejected plan")
            .is_none()
    );
    assert!(fixture.platform.mutations.is_empty());

    fs::remove_dir_all(fixture.root).expect("remove test directory");
}

pub(super) fn protocol_tranche_fixture(burns: Vec<u128>) -> Fixture {
    let mut fixture = fixture();
    fixture
        .desired
        .canisters
        .retain(|canister| canister.name == "treasury");
    fixture.desired.canisters[0].wasm = None;
    fixture.desired.protocol = Some(DesiredFleetProtocol {
        app_config: "canic.toml".to_string(),
        component_group_placements: Vec::new(),
        coordinator_candid: "coordinator.did".to_string(),
        root_candid: "root.did".to_string(),
        store_candid: "store.did".to_string(),
    });
    let current_hash = sha256_hex(b"current-wasm");
    fixture.platform = MockPlatform::new(
        fixture.desired.clone(),
        vec![live(
            TREASURY,
            500,
            Some(&current_hash),
            true,
            &[CONTROLLER],
        )],
    );
    fixture.platform.typed_protocol_burns = burns;
    fixture
}

fn fixture() -> Fixture {
    let root = temp_dir("canic-fleet-ensure");
    fs::create_dir_all(&root).expect("create fixture root");
    let wasm = root.join("app.wasm");
    let candid = root.join("drain.did");
    fs::write(&wasm, b"current-wasm").expect("write Wasm");
    fs::write(&candid, b"service : {};").expect("write Candid");
    let desired = DesiredFleet {
        bootstrap: None,
        canisters: vec![
            desired_canister("treasury", Some(TREASURY), false, &wasm, None),
            desired_canister("created", None, false, &wasm, None),
            desired_canister("app", Some(OLD_APP), false, &wasm, None),
            DesiredCanister {
                canic_init: None,
                controller_canisters: Vec::new(),
                controllers: vec![CONTROLLER.to_string()],
                drain: Some(drain(&candid)),
                initial_cycles: "0".to_string(),
                init_arg: None,
                init_candid: None,
                kind: DesiredCanisterKind::Auxiliary,
                minimum_cycles: "0".to_string(),
                name: "retired".to_string(),
                parent: None,
                presence: DesiredPresence::Absent,
                principal: Some(RETIRED.to_string()),
                protocol_binding: None,
                replace: false,
                subnet: SUBNET.to_string(),
                wasm: None,
            },
            desired_canister("replaced", Some(OLD_REPLACED), true, &wasm, Some(&candid)),
        ],
        cycles_ledger: LEDGER.to_string(),
        environment: "local".to_string(),
        fleet: "test-fleet".to_string(),
        ledger_fee_cycles: "10".to_string(),
        management_creation_fee_cycles: "50".to_string(),
        material_cycle_threshold: "5".to_string(),
        maximum_observation_burn_cycles: "0".to_string(),
        maximum_stalled_observations: 2,
        maximum_update_burn_cycles: "1".to_string(),
        operator: CONTROLLER.to_string(),
        protocol: None,
        protocol_steps: Vec::new(),
        schema_version: FLEET_ENSURE_SCHEMA_VERSION,
        treasury: "treasury".to_string(),
    };
    let current_hash = sha256_hex(b"current-wasm");
    let live = vec![
        live(TREASURY, 500, Some(&current_hash), true, &[CONTROLLER]),
        live(OLD_APP, 5, Some("old"), false, &[TREASURY]),
        live(RETIRED, 100, Some("old"), true, &[CONTROLLER]),
        live(OLD_REPLACED, 100, Some("old"), true, &[CONTROLLER]),
    ];
    let platform = MockPlatform::new(desired.clone(), live);
    Fixture {
        desired,
        platform,
        root,
    }
}

#[test]
fn retained_root_owned_balance_is_passively_reobserved_without_an_effect() {
    let mut fixture = fixture();
    fixture
        .desired
        .canisters
        .retain(|canister| canister.name == "treasury");
    fixture.platform = MockPlatform::new(
        fixture.desired.clone(),
        [live(
            TREASURY,
            500,
            Some(&sha256_hex(b"current-wasm")),
            true,
            &[CONTROLLER],
        )],
    );
    fixture
        .platform
        .live
        .get_mut(TREASURY)
        .expect("treasury live observation")
        .root_owned_lifecycle = Some(RootOwnedCanisterLifecycle::Idle);

    let desired_sha256 = "92".repeat(32);
    let planned = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &desired_sha256,
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut fixture.platform,
    )
    .expect("plan terminal retained estate");
    assert!(workflow::ordered_actions(&planned.plan).is_empty());

    fixture
        .platform
        .live
        .get_mut(TREASURY)
        .expect("treasury live observation")
        .root_owned_lifecycle = Some(RootOwnedCanisterLifecycle::Retained);
    fixture
        .platform
        .retain_root_owned_lifecycle_for_observations("treasury", 1);
    let terminal = apply_fixture_plan(&mut fixture, &desired_sha256, &planned.plan)
        .expect("passively reobserve retained Root-owned lifecycle");

    assert!(terminal.terminal);
    assert!(fixture.platform.mutations.is_empty());
    assert_eq!(
        fixture.platform.paced_root_owned_observations,
        [("treasury".to_string(), 1)]
    );
    let journal = crate::fleet_ensure::ops::read_journal(
        &crate::fleet_ensure::ops::EnsurePaths::under(&fixture.root, "local", "test-fleet"),
    )
    .expect("read terminal passive-observation journal")
    .expect("terminal passive-observation journal");
    assert_eq!(journal.completion, FleetEnsureCompletion::Converged);
    assert_eq!(journal.stalled_observations, 0);

    fs::remove_dir_all(fixture.root).expect("remove test directory");
}

#[test]
fn retained_root_owned_balance_stall_is_bounded_and_effect_free() {
    let mut fixture = fixture();
    fixture
        .desired
        .canisters
        .retain(|canister| canister.name == "treasury");
    let mut retained = live(
        TREASURY,
        500,
        Some(&sha256_hex(b"current-wasm")),
        true,
        &[CONTROLLER],
    );
    retained.root_owned_lifecycle = Some(RootOwnedCanisterLifecycle::Retained);
    fixture.platform = MockPlatform::new(fixture.desired.clone(), [retained]);
    fixture
        .platform
        .retain_root_owned_lifecycle_for_observations("treasury", 3);

    let error = workflow::plan(
        &fixture.root,
        &fixture.desired,
        &"93".repeat(32),
        "test-fleet",
        1_800_000_000_000_000_000,
        &mut fixture.platform,
    )
    .expect_err("retained Root-owned observation must remain bounded");

    assert!(matches!(
        error,
        workflow::EnsureWorkflowError::RootOwnedObservationStalled {
            ref target,
            observations: 2,
            ref last_lifecycle,
        } if target == "treasury" && last_lifecycle == "retained"
    ));
    assert!(fixture.platform.mutations.is_empty());
    assert_eq!(
        fixture.platform.paced_root_owned_observations,
        [("treasury".to_string(), 1)]
    );

    fs::remove_dir_all(fixture.root).expect("remove test directory");
}

fn desired_canister(
    name: &str,
    principal: Option<&str>,
    replace: bool,
    wasm: &Path,
    drain_candid: Option<&Path>,
) -> DesiredCanister {
    DesiredCanister {
        canic_init: None,
        controller_canisters: Vec::new(),
        controllers: vec![CONTROLLER.to_string()],
        drain: drain_candid.map(drain),
        initial_cycles: "20".to_string(),
        init_arg: None,
        init_candid: None,
        kind: if name == "treasury" {
            DesiredCanisterKind::Coordinator
        } else {
            DesiredCanisterKind::Auxiliary
        },
        minimum_cycles: "20".to_string(),
        name: name.to_string(),
        parent: None,
        presence: DesiredPresence::Present,
        principal: principal.map(str::to_string),
        protocol_binding: None,
        replace,
        subnet: SUBNET.to_string(),
        wasm: Some(wasm.display().to_string()),
    }
}

fn drain(candid: &Path) -> DrainAuthority {
    DrainAuthority {
        candid: candid.display().to_string(),
        destination: "treasury".to_string(),
        maximum_execution_burn_cycles: "2".to_string(),
        method: "canic_cycle_drain".to_string(),
    }
}

fn protocol_step(root: &Path) -> DesiredProtocolStep {
    protocol_step_for(root, "fleet-catalog-terminal", "app")
}

fn fleet_protocol_action(name: &str, action: CurrentFleetProtocolAction) -> EnsureAction {
    EnsureAction::FleetProtocol {
        action: Box::new(action),
        candid: "coordinator.did".to_string(),
        candid_sha256: "19".repeat(32),
        maximum_execution_burn_cycles: 1,
        name: name.to_string(),
        principal: TREASURY.to_string(),
    }
}

fn current_protocol_variants(plan: &FleetEnsurePlan) -> BTreeSet<&'static str> {
    plan.protocol_actions
        .iter()
        .filter_map(|action| match action {
            EnsureAction::SealAuthority { .. } => {
                panic!("authority sealing uses the production PocketIC adapter")
            }
            EnsureAction::FleetProtocol { action, .. } => Some(match action.as_ref() {
                CurrentFleetProtocolAction::ReconcilePoolAsset { .. } => "reconcile_pool_asset",
                CurrentFleetProtocolAction::ObservePoolReadiness { .. } => "observe_pool_readiness",
                CurrentFleetProtocolAction::MaintainPoolReadiness { .. } => {
                    "maintain_pool_readiness"
                }
                CurrentFleetProtocolAction::ActivateRegistry { .. } => "activate_registry",
                CurrentFleetProtocolAction::ActivateRegistryMirror { .. } => {
                    "activate_registry_mirror"
                }
                CurrentFleetProtocolAction::AdoptStore { .. } => "adopt_store",
                CurrentFleetProtocolAction::BootstrapStore { .. } => "bootstrap_store",
                CurrentFleetProtocolAction::JoinRoot { .. } => "join_root",
                CurrentFleetProtocolAction::PrepareComponentRegistry { .. } => {
                    "prepare_component_registry"
                }
                CurrentFleetProtocolAction::ProvisionComponents { .. } => "provision_components",
                CurrentFleetProtocolAction::PrepareStoreFixture { .. } => "prepare_store_fixture",
                CurrentFleetProtocolAction::PublishStoreFixtureChunk { .. } => {
                    "publish_store_fixture_chunk"
                }
                CurrentFleetProtocolAction::PublishStoreChunk { .. } => "publish_store_chunk",
                CurrentFleetProtocolAction::SynchronizeRegistry { .. } => "synchronize_registry",
            }),
            _ => None,
        })
        .collect()
}

fn reviewed_desired(plan: &FleetEnsurePlan) -> Option<&DesiredFleet> {
    plan.reviewed_desired
        .as_deref()
        .map(crate::fleet_ensure::model::ReviewedDesiredFleetRecord::desired)
}

pub(super) fn typed_protocol_action(operation_id: &str) -> EnsureAction {
    let operation_id = canic_core::cdk::utils::hash::decode_hex(operation_id)
        .ok()
        .and_then(|bytes| <[u8; 32]>::try_from(bytes).ok())
        .expect("Fleet ensure operation ID");
    let fleet = FleetBinding {
        fleet: FleetKey {
            canonical_network_id: CanonicalNetworkId::ic_mainnet(),
            fleet_id: FleetId::from_generated_bytes([7; 32]),
        },
        app: AppId::from("ensure_test"),
    };
    let authority = FleetRegistryAuthority {
        binding: FleetCoordinatorBinding {
            fleet: fleet.clone(),
            coordinator_subnet: SubnetId::from_principal(
                SUBNET.parse().expect("fixture Subnet Principal"),
            ),
            coordinator: TREASURY.parse().expect("fixture Coordinator Principal"),
        },
        epoch: 1,
    };
    EnsureAction::FleetProtocol {
        action: Box::new(CurrentFleetProtocolAction::ProvisionComponents {
            request: FleetComponentProvisioningPrepareRequest {
                operation_id,
                plan: FleetComponentProvisioningPlan {
                    fleet,
                    fleet_registry: FleetRegistryVersion {
                        authority,
                        revision: 1,
                        content_hash: [8; 32],
                    },
                    configuration_digest: ComponentDeploymentConfigurationDigest::from_bytes(
                        [9; 32],
                    ),
                    operation: FleetComponentProvisioningOperation::FreshInstall,
                    directory_confirmation_roots: Vec::new(),
                    batches: Vec::new(),
                },
            },
            plan_hash: [10; 32],
        }),
        candid: "coordinator.did".to_string(),
        candid_sha256: "11".repeat(32),
        maximum_execution_burn_cycles: 1,
        name: "fleet-component-provisioning".to_string(),
        principal: TREASURY.to_string(),
    }
}

fn empty_active_registry() -> FleetRegistry {
    let fleet = FleetBinding {
        fleet: FleetKey {
            canonical_network_id: CanonicalNetworkId::ic_mainnet(),
            fleet_id: FleetId::from_generated_bytes([7; 32]),
        },
        app: AppId::from("ensure_test"),
    };
    FleetRegistry {
        authority: FleetRegistryAuthority {
            binding: FleetCoordinatorBinding {
                fleet: fleet.clone(),
                coordinator_subnet: SubnetId::from_principal(
                    SUBNET.parse().expect("fixture Subnet Principal"),
                ),
                coordinator: TREASURY.parse().expect("fixture Coordinator Principal"),
            },
            epoch: 1,
        },
        revision: 1,
        admission: FleetAdmissionPolicy {
            schema_version: 1,
            fleet,
            generation: 1,
            fleet_principals: Vec::new(),
            rules: Vec::new(),
            policy_digest: [0; 32],
        },
        component_specs: Vec::new(),
        fleet_subnet_roots: Vec::new(),
        services: Vec::new(),
    }
}

fn protocol_step_for(root: &Path, name: &str, canister: &str) -> DesiredProtocolStep {
    let candid = root.join("protocol.did");
    let command_args = root.join("protocol-command.args");
    let status_args = root.join("protocol-status.args");
    let expected_status = root.join("protocol-expected.args");
    fs::write(
        &candid,
        b"service : { apply : (text) -> (); status : (text) -> (bool) query; };",
    )
    .expect("write protocol Candid");
    fs::write(&command_args, b"({{operation_id}})").expect("write protocol command");
    fs::write(&status_args, b"({{operation_id}})").expect("write protocol status");
    fs::write(&expected_status, b"(true)").expect("write expected protocol status");
    DesiredProtocolStep {
        canister: canister.to_string(),
        candid: candid.display().to_string(),
        command_args: command_args.display().to_string(),
        command_method: "apply".to_string(),
        expected_status: expected_status.display().to_string(),
        maximum_execution_burn_cycles: "1".to_string(),
        name: name.to_string(),
        status_args: status_args.display().to_string(),
        status_method: "status".to_string(),
    }
}

fn live(
    principal: &str,
    cycles: u128,
    module_sha256: Option<&str>,
    running: bool,
    controllers: &[&str],
) -> LiveCanister {
    LiveCanister {
        canister_version: Some(1),
        controllers: controllers
            .iter()
            .map(|value| (*value).to_string())
            .collect(),
        cycles,
        module_sha256: module_sha256.map(str::to_string),
        principal: principal.to_string(),
        reinstall_required: false,
        root_owned_lifecycle: None,
        status: if running {
            CanisterRuntimeStatus::Running
        } else {
            CanisterRuntimeStatus::Stopped
        },
    }
}

fn install_action(
    name: &str,
    canic_init: DesiredCanisterInit,
    mode: crate::fleet_ensure::model::InstallMode,
) -> EnsureAction {
    EnsureAction::Install {
        canic_init: Some(canic_init),
        reinstall_witness: None,
        init_arg: None,
        init_arg_sha256: None,
        init_candid: None,
        init_candid_sha256: None,
        mode,
        name: name.to_string(),
        principal: Principal::from_slice(&[99; 29]).to_text(),
        wasm: "artifact.wasm".to_string(),
        wasm_sha256: "ab".repeat(32),
    }
}

fn empty_outcome() -> EffectOutcome {
    EffectOutcome {
        created_principal: None,
        post_cycles: None,
        receipt: None,
    }
}

#[test]
#[expect(
    clippy::too_many_lines,
    reason = "one source authority table covers selected protocol admission and each independent live drift"
)]
fn reinstall_preparation_binds_exact_running_authority_before_sealing() {
    use crate::fleet_ensure::{
        model::{
            FleetReinstallSourceRecord, ReviewedDesiredFleetRecord,
            RootManagementCanisterObservation, RootManagementObservation,
        },
        policy::{
            EnsurePolicyError,
            reinstall::{PreparationInput, preparation},
        },
    };
    let fixture = protocol_tranche_fixture(Vec::new());
    let hash = sha256_hex(b"current-wasm");
    let source = FleetReinstallSourceRecord {
        terminal_retirement: None,
        reviewed_desired: ReviewedDesiredFleetRecord::capture(&fixture.desired),
        wasm_sha256_by_canister: BTreeMap::from([("treasury".to_string(), hash.clone())]),
        candid_sha256_by_path: BTreeMap::from([("coordinator.did".to_string(), "11".repeat(32))]),
    };
    let observed = RootManagementCanisterObservation {
        live: live(TREASURY, 500, Some(&hash), true, &[CONTROLLER]),
        name: "treasury".to_string(),
        subnet: SUBNET.to_string(),
    };
    let mut target = fixture.desired.clone();
    target.protocol.as_mut().unwrap().coordinator_candid = "selected-new-coordinator.did".into();
    let plan = |desired: &DesiredFleet, observed: RootManagementCanisterObservation| {
        preparation(PreparationInput {
            desired,
            source: &source,
            target_artifacts_sha256: &"42".repeat(32),
            observation: &RootManagementObservation {
                operator_cycles: 0,
                roots: BTreeMap::from([("treasury".to_string(), observed)]),
            },
            source_operation_id: &"21".repeat(32),
            desired_sha256: &"22".repeat(32),
            operation_id: &"23".repeat(32),
            time: 100,
        })
    };
    let accepted = plan(&target, observed.clone()).expect("exact current authority");
    assert!(matches!(
        accepted.canisters[0].actions.as_slice(),
        [EnsureAction::SealAuthority { candid, .. }] if candid == "coordinator.did"
    ));
    assert_eq!(
        accepted.reviewed_desired.as_ref().unwrap().desired(),
        &target
    );
    assert_eq!(
        accepted.reinstall.as_ref().unwrap().source.as_deref(),
        Some(&source)
    );
    assert!(
        plan(&fixture.desired, observed.clone()).is_ok(),
        "identical rebuild remains a wipe"
    );
    let mut changed = target.clone();
    changed.operator = OLD_APP.into();
    assert!(matches!(
        plan(&changed, observed.clone()),
        Err(EnsurePolicyError::RootManagementAuthorityMismatch { .. })
    ));
    let mut foreign = target.clone();
    foreign.cycles_ledger = OLD_APP.into();
    assert!(matches!(
        plan(&foreign, observed.clone()),
        Err(EnsurePolicyError::RootManagementAuthorityMismatch { .. })
    ));
    foreign = target.clone();
    foreign.canisters[0].subnet = OLD_APP.into();
    assert!(matches!(
        plan(&foreign, observed.clone()),
        Err(EnsurePolicyError::RootManagementAuthorityMismatch { .. })
    ));

    assert_eq!(
        accepted.reinstall.as_ref().unwrap().authorities[0].principal,
        TREASURY
    );
    let mut conflicts = Vec::new();
    let mut changed = observed.clone();
    changed.live.controllers = vec![OLD_APP.to_string()];
    conflicts.push(changed);
    let mut changed = observed.clone();
    changed.live.principal = OLD_APP.to_string();
    conflicts.push(changed);
    let mut changed = observed.clone();
    changed.subnet = OLD_APP.to_string();
    conflicts.push(changed);
    let mut changed = observed.clone();
    changed.live.module_sha256 = Some("ff".repeat(32));
    conflicts.push(changed);
    let mut changed = observed.clone();
    changed.live.status = CanisterRuntimeStatus::Stopped;
    conflicts.push(changed);
    let mut changed = observed;
    changed.name = "foreign".to_string();
    conflicts.push(changed);
    for changed in conflicts {
        assert!(matches!(
            plan(&target, changed),
            Err(EnsurePolicyError::RootManagementAuthorityMismatch { .. })
        ));
    }
    fs::remove_dir_all(fixture.root).unwrap();
}

#[test]
fn reinstall_preparation_requires_each_authority_to_cover_its_own_seal() {
    use crate::fleet_ensure::{
        model::{
            FleetReinstallSourceRecord, ReviewedDesiredFleetRecord,
            RootManagementCanisterObservation, RootManagementObservation,
        },
        policy::{
            EnsurePolicyError,
            reinstall::{PreparationInput, preparation},
        },
    };
    let mut fixture = protocol_tranche_fixture(Vec::new());
    fixture.desired.maximum_observation_burn_cycles = "2".into();
    fixture.desired.maximum_update_burn_cycles = "3".into();
    let mut root = fixture.desired.canisters[0].clone();
    root.name = "root".into();
    root.kind = DesiredCanisterKind::Root;
    root.parent = Some("treasury".into());
    root.principal = Some(OLD_APP.into());
    fixture.desired.canisters.push(root);
    let hash = sha256_hex(b"current-wasm");
    let source = FleetReinstallSourceRecord {
        terminal_retirement: None,
        reviewed_desired: ReviewedDesiredFleetRecord::capture(&fixture.desired),
        wasm_sha256_by_canister: BTreeMap::from([
            ("treasury".into(), hash.clone()),
            ("root".into(), hash.clone()),
        ]),
        candid_sha256_by_path: BTreeMap::from([
            ("coordinator.did".into(), "11".repeat(32)),
            ("root.did".into(), "22".repeat(32)),
        ]),
    };
    let plan = |treasury_cycles, root_cycles| {
        let roots = [
            ("treasury", TREASURY, treasury_cycles),
            ("root", OLD_APP, root_cycles),
        ]
        .into_iter()
        .map(|(name, principal, cycles)| {
            (
                name.into(),
                RootManagementCanisterObservation {
                    live: live(principal, cycles, Some(&hash), true, &[CONTROLLER]),
                    name: name.into(),
                    subnet: SUBNET.into(),
                },
            )
        })
        .collect();
        preparation(PreparationInput {
            desired: &fixture.desired,
            source: &source,
            target_artifacts_sha256: &"42".repeat(32),
            observation: &RootManagementObservation {
                operator_cycles: 0,
                roots,
            },
            source_operation_id: &"21".repeat(32),
            desired_sha256: &"22".repeat(32),
            operation_id: &"23".repeat(32),
            time: 100,
        })
    };
    for (treasury_cycles, root_cycles, name, principal) in [
        (1000, 18, "root", OLD_APP),
        (18, 1000, "treasury", TREASURY),
    ] {
        assert_eq!(
            plan(treasury_cycles, root_cycles),
            Err(EnsurePolicyError::AuthoritySealHeadroom {
                name: name.into(),
                principal: principal.into(),
                available: 18,
                required: 19,
                shortfall: 1,
            })
        );
    }
    let accepted = plan(19, 19).expect("each authority covers its own bound");
    assert_eq!(accepted.conservation.maximum_execution_burn_cycles, 38);
    assert_eq!(accepted.conservation.expected_post_operation_cycles, 0);
    assert_eq!(accepted.conservation.maximum_new_funding_cycles, 0);
    assert_eq!(accepted.conservation.maximum_operator_debit_cycles, 0);
    assert!(accepted.canisters.iter().all(|canister| matches!(
        canister.actions.as_slice(),
        [EnsureAction::SealAuthority { .. }]
    )));
    fs::remove_dir_all(fixture.root).unwrap();
}

#[test]
fn reinstall_seals_roots_before_coordinator_to_finish_inflight_funding() {
    let seal = |authority_kind| EnsureAction::SealAuthority {
        authority_kind,
        candid: "root.did".into(),
        candid_sha256: "11".repeat(32),
        name: "authority".into(),
        principal: Principal::anonymous().to_text(),
    };
    assert!(
        workflow::action_order(&seal(DesiredCanisterKind::Root))
            < workflow::action_order(&seal(DesiredCanisterKind::Coordinator))
    );
}

#[test]
fn reinstall_history_accepts_only_the_exact_intended_root_replacement() {
    use crate::fleet_ensure::model::{
        ReinstallHistoryWitness, ReinstallRootWitnessRecord, RootManagementBinding,
    };
    use crate::fleet_ensure::policy::reinstall::history_authority;
    let before = RootManagementBinding {
        controllers: vec![CONTROLLER.into()],
        module_sha256: "11".repeat(32),
        name: "root".into(),
        principal: TREASURY.into(),
        subnet: SUBNET.into(),
    };
    let mut after = before.clone();
    after.module_sha256 = "22".repeat(32);
    let witness = ReinstallHistoryWitness {
        prior_module_sha256: before.module_sha256.clone(),
        authority: before.clone(),
        candid: "source.did".into(),
        candid_sha256: "33".repeat(32),
        replacement: Some(Box::new(ReinstallRootWitnessRecord {
            authority: after.clone(),
            candid: "target.did".into(),
            candid_sha256: "44".repeat(32),
        })),
    };
    let select = |current: &RootManagementBinding, principal: &str, state: &EffectState| {
        history_authority(&witness, current, principal, &after.module_sha256, state)
    };
    assert_eq!(
        select(&before, TREASURY, &EffectState::Intent)
            .unwrap()
            .candid,
        "source.did"
    );
    assert_eq!(
        select(&after, TREASURY, &EffectState::Issued)
            .unwrap()
            .candid,
        "target.did"
    );
    assert_eq!(
        select(&after, TREASURY, &EffectState::Intent)
            .unwrap()
            .candid,
        "target.did"
    );
    assert!(select(&after, TREASURY, &EffectState::Applied).is_none());
    assert!(select(&after, OLD_APP, &EffectState::Issued).is_none());
    let mut foreign = after.clone();
    foreign.module_sha256 = "55".repeat(32);
    assert!(select(&foreign, TREASURY, &EffectState::Issued).is_none());
    foreign = after.clone();
    foreign.controllers = vec![OLD_APP.into()];
    assert!(select(&foreign, TREASURY, &EffectState::Issued).is_none());
    foreign = after.clone();
    foreign.subnet = OLD_APP.into();
    assert!(select(&foreign, TREASURY, &EffectState::Issued).is_none());
}

#[test]
fn reinstall_selected_artifact_identity_detects_in_place_changes() {
    let mut fixture = protocol_tranche_fixture(Vec::new());
    let wasm = fixture.root.join("selected.wasm");
    fs::write(&wasm, b"selected release bytes").unwrap();
    fixture.desired.canisters[0].wasm = Some(wasm.to_string_lossy().into_owned());
    let digest = || {
        crate::fleet_ensure::ops::reinstall::target_artifacts_sha256(
            &fixture.root,
            &fixture.desired,
        )
        .unwrap()
    };
    let selected = digest();
    fs::write(&wasm, b"different release bytes").unwrap();
    assert_ne!(digest(), selected);
    fs::write(&wasm, b"selected release bytes").unwrap();
    assert_eq!(digest(), selected);
    fs::remove_dir_all(fixture.root).unwrap();
}

#[test]
fn fixture_publication_rejects_a_journal_counter_beyond_its_reviewed_limit() {
    let mut fixture = publication_retry_fixture(1);
    let planned = workflow::plan(
        &fixture.root,
        &fixture.desired,
        "publication",
        "test-fleet",
        1,
        &mut fixture.platform,
    )
    .unwrap();
    let paths = crate::fleet_ensure::ops::EnsurePaths::under(&fixture.root, "local", "test-fleet");
    let hash = action_sha256(&planned.plan.protocol_actions[0]);
    fixture.platform.stall_before_mutation(hash, 1);
    assert!(matches!(
        workflow::apply(
            &fixture.root,
            &fixture.desired,
            "publication",
            "test-fleet",
            &planned.plan.plan_sha256,
            &mut fixture.platform
        ),
        Err(workflow::EnsureWorkflowError::Platform(_))
    ));
    let mut journal = crate::fleet_ensure::ops::read_journal(&paths)
        .unwrap()
        .unwrap();
    journal.effects[0].publication_attempts = 2;
    crate::fleet_ensure::ops::write_journal(&paths, &journal).unwrap();
    assert!(matches!(
        workflow::apply(
            &fixture.root,
            &fixture.desired,
            "publication",
            "test-fleet",
            &planned.plan.plan_sha256,
            &mut fixture.platform
        ),
        Err(workflow::EnsureWorkflowError::JournalIntegrity)
    ));
    assert!(fixture.platform.mutations.is_empty());
    fs::remove_dir_all(fixture.root).unwrap();
}

/// A deliberately non-executable completed source with a current immutable phase.
#[expect(
    clippy::too_many_lines,
    reason = "one completed-source fixture binds its plan, immutable phase and paid receipts"
)]
pub(super) fn terminal_retirement_fixture() -> (
    Fixture,
    crate::fleet_ensure::ops::EnsurePaths,
    FleetEnsurePlan,
) {
    use crate::fleet_ensure::{model::*, ops, policy::expected_plan_sha256};
    let mut fixture = protocol_tranche_fixture(Vec::new());
    fixture.desired.ledger_fee_cycles = "0.00000001B".into();
    fixture.platform.desired = fixture.desired.clone();
    for file in ["coordinator.did", "root.did", "store.did"] {
        fs::write(fixture.root.join(file), b"service : {};").unwrap();
    }
    let mut plan = workflow::plan(
        &fixture.root,
        &fixture.desired,
        "terminal-source",
        "test-fleet",
        1,
        &mut fixture.platform,
    )
    .unwrap()
    .plan;
    let mut source_desired = fixture.desired.clone();
    source_desired.canisters[0].wasm = Some("app.wasm".into());
    plan.reviewed_desired = Some(Box::new(ReviewedDesiredFleetRecord::capture(
        &source_desired,
    )));
    plan.conservation.maximum_new_funding_cycles = 110;
    plan.conservation.maximum_unavoidable_fee_cycles = 10;
    plan.conservation.maximum_operator_debit_cycles = 120;
    plan.conservation.maximum_execution_burn_cycles = 200;
    plan.continuation = Some(FleetEnsureContinuationAuthority {
        fixture_publication_retry_attempts: 0,
        app_config_sha256: sha256_hex(b"config"),
        application_artifact_union_sha256: sha256_hex(b"artifacts"),
        coordinator_candid_sha256: sha256_hex(b"service : {};"),
        maximum_successor_actions: 5,
        root_candid_sha256: sha256_hex(b"service : {};"),
        store_candid_sha256: sha256_hex(b"service : {};"),
    });
    let fund = EnsureAction::Fund {
        pool_funding: None,
        amount: 110,
        created_at_time: 1,
        expected_post_cycles: 610,
        funding_deficit_cycles: 100,
        funding_margin_cycles: 10,
        ledger: LEDGER.into(),
        name: "treasury".into(),
        principal: TREASURY.into(),
    };
    plan.canisters[0].actions = vec![fund.clone()];
    plan.protocol_actions.clear();
    plan.plan_sha256 = expected_plan_sha256(&plan);
    let mut phase = plan.clone();
    phase.continuation = None;
    phase.canisters[0].actions.clear();
    phase.protocol_actions = vec![typed_protocol_action(&plan.operation_id)];
    phase.plan_sha256 = expected_plan_sha256(&phase);
    let paths = ops::EnsurePaths::under(&fixture.root, "local", "test-fleet");
    let mut phase_paths = paths.clone();
    phase_paths.plan = paths
        .plan
        .with_file_name("phases")
        .join(format!("{}.json", phase.plan_sha256));
    ops::write_plan(&phase_paths, &phase).unwrap();
    ops::write_plan(&paths, &plan).unwrap();
    let mut raw: serde_json::Value =
        serde_json::from_slice(&fs::read(&paths.plan).unwrap()).unwrap();
    // Match the reported completed staging source: the informational forecast
    // lacks successor/retry/startup fields, while continuation authority retains
    // its exact action bound. No field may be filled in to execute this source.
    raw["recovery_review"] = serde_json::json!({
        "base_execution_burn_cycles": "0",
        "continuation_reserve_cycles": "0",
        "whole_continuation_ceiling_cycles": "0",
        "known_pool_funding": [],
        "discovery": "pending_current_protocol",
    });
    fs::write(&paths.plan, serde_json::to_vec(&raw).unwrap()).unwrap();
    let effects = [fund, phase.protocol_actions[0].clone()]
        .iter()
        .map(|action| EffectRecord {
            publication_attempts: 0,
            maintenance_attempts: 0,
            action_sha256: action_sha256(action),
            created_principal: None,
            destination_post_cycles: None,
            destination_pre_cycles: None,
            post_cycles: Some(610),
            pre_cycles: Some(500),
            pre_canister_version: None,
            progress_identity: None,
            receipt: Some("receipt-fixture".into()),
            state: EffectState::Applied,
        })
        .collect();
    let journal = FleetEnsureJournalRecord {
        funding_observations: BTreeMap::new(),
        funding_reviews: Vec::new(),
        successor_phases: vec![FleetEnsureSuccessorPhaseRecord {
            execution_burn_before_phase: 0,
            plan_sha256: phase.plan_sha256.clone(),
            plan: None,
        }],
        completion: FleetEnsureCompletion::Converged,
        estate_funding_required: None,
        effects,
        fleet: plan.fleet.clone(),
        initial_controlled_cycles: 500,
        initial_estate_funding_cycles_by_root: BTreeMap::new(),
        initial_operator_cycles: 1000,
        operation_id: plan.operation_id.clone(),
        plan_sha256: plan.plan_sha256.clone(),
        schema_version: 1,
        stalled_observations: 0,
    };
    ops::write_journal(&paths, &journal).unwrap();
    let mut state = ops::read_state(&paths, "test-fleet").unwrap();
    state.active_registry = Some(empty_active_registry());
    ops::write_state(&paths, &state).unwrap();
    (fixture, paths, phase)
}

#[test]
fn terminal_retirement_reads_receipts_without_making_source_executable() {
    use crate::fleet_ensure::ops::{self, reinstall::terminal};
    let (mut fixture, paths, _) = terminal_retirement_fixture();
    assert!(matches!(
        ops::read_plan(&paths),
        Err(ops::EnsureStateError::Decode { .. })
    ));
    let source = terminal::read(&paths, "local", "test-fleet").unwrap();
    assert_eq!(source.actions.len(), 2);
    assert_eq!(source.documents.phase_document_sha256.len(), 1);
    let before = fs::read(&paths.plan).unwrap();
    assert!(matches!(
        workflow::plan(
            &fixture.root,
            &fixture.desired,
            "changed",
            "test-fleet",
            2,
            &mut fixture.platform
        ),
        Err(workflow::EnsureWorkflowError::RetainedTerminalReviewRequired { .. })
    ));
    assert_eq!(fs::read(&paths.plan).unwrap(), before);
    assert!(fixture.platform.mutations.is_empty());
    fs::remove_dir_all(fixture.root).unwrap();
}

#[test]
fn terminal_retirement_requires_the_retained_continuation_action_bound() {
    use crate::fleet_ensure::ops::{self, reinstall::terminal};
    let (mut fixture, paths, _) = terminal_retirement_fixture();
    terminal::read(&paths, "local", "test-fleet").unwrap();
    let mut raw: serde_json::Value =
        serde_json::from_slice(&fs::read(&paths.plan).unwrap()).unwrap();
    raw["continuation"]
        .as_object_mut()
        .unwrap()
        .remove("maximum_successor_actions")
        .unwrap();
    fs::write(&paths.plan, serde_json::to_vec(&raw).unwrap()).unwrap();
    let before = [
        fs::read(&paths.plan).unwrap(),
        fs::read(&paths.journal).unwrap(),
        fs::read(&paths.state).unwrap(),
    ];
    assert!(matches!(
        terminal::read(&paths, "local", "test-fleet"),
        Err(ops::EnsureStateError::InvalidTerminalSource)
    ));
    assert!(matches!(
        workflow::plan_reinstall(
            &fixture.root,
            &fixture.desired,
            "target",
            "test-fleet",
            2,
            &mut fixture.platform,
        ),
        Err(workflow::EnsureWorkflowError::State(
            ops::EnsureStateError::InvalidTerminalSource
        ))
    ));
    assert!(fixture.platform.mutations.is_empty());
    for (index, path) in [&paths.plan, &paths.journal, &paths.state]
        .into_iter()
        .enumerate()
    {
        assert_eq!(fs::read(path).unwrap(), before[index]);
    }
    fs::remove_dir_all(fixture.root).unwrap();
}

#[test]
fn terminal_retirement_rejects_incomplete_or_changed_receipts_and_phases() {
    use crate::fleet_ensure::ops::{self, reinstall::terminal};
    let (fixture, paths, phase) = terminal_retirement_fixture();
    let original = ops::read_journal(&paths).unwrap().unwrap();
    for change in 0..8 {
        let mut journal = original.clone();
        match change {
            0 => journal.effects[0].state = EffectState::Issued,
            1 => journal.effects[0].receipt = None,
            2 => journal.effects[0].post_cycles = Some(500),
            3 => journal.effects[0].action_sha256 = sha256_hex(b"wrong effect"),
            4 => {
                journal.effects.pop();
            }
            5 => journal.effects[1].maintenance_attempts = 1,
            6 => journal.effects[1].publication_attempts = 1,
            7 => journal.completion = FleetEnsureCompletion::InProgress,
            _ => unreachable!(),
        }
        ops::write_journal(&paths, &journal).unwrap();
        assert!(
            matches!(
                terminal::read(&paths, "local", "test-fleet"),
                Err(ops::EnsureStateError::InvalidTerminalSource)
            ),
            "case {change}"
        );
    }
    ops::write_journal(&paths, &original).unwrap();
    let phase_path = paths
        .plan
        .with_file_name("phases")
        .join(format!("{}.json", phase.plan_sha256));
    let mut raw: serde_json::Value =
        serde_json::from_slice(&fs::read(&phase_path).unwrap()).unwrap();
    raw["operation_id"] = serde_json::json!(sha256_hex(b"different operation"));
    fs::write(phase_path, serde_json::to_vec(&raw).unwrap()).unwrap();
    assert!(terminal::read(&paths, "local", "test-fleet").is_err());
    fs::remove_dir_all(fixture.root).unwrap();
}

#[test]
fn terminal_retirement_requires_exact_paid_debit_and_bounded_live_conservation() {
    use crate::fleet_ensure::{
        ops::reinstall::terminal, policy::reinstall::terminal::conservation,
    };
    let (fixture, paths, _) = terminal_retirement_fixture();
    let source = terminal::read(&paths, "local", "test-fleet").unwrap();
    let observation = FleetObservation {
        additional_controlled_cycles: BTreeMap::new(),
        canisters: BTreeMap::new(),
        estate_funding_domains: BTreeMap::new(),
        ledger_fee_cycles: 10,
        operator_cycles: 880,
        protocol_ready: BTreeMap::new(),
    };
    let actual = conservation(&source, &observation, 600).unwrap();
    assert_eq!(actual.observed_net_cycle_debit_cycles, 10);
    assert_eq!(actual.received_new_funding_cycles, 110);
    assert_eq!(actual.operator_debit_cycles, 120);
    for balance in [879, 881, 1001] {
        let mut changed = observation.clone();
        changed.operator_cycles = balance;
        assert!(conservation(&source, &changed, 600).is_none());
    }
    assert!(conservation(&source, &observation, 409).is_none());
    let donated = conservation(&source, &observation, 611).unwrap();
    assert_eq!(donated.observed_net_cycle_credit_cycles, 1);
    assert_eq!(donated.observed_net_cycle_debit_cycles, 0);
    let mut changed = source;
    changed.conservation.scheduled_transfer_cycles = 1;
    assert!(conservation(&changed, &observation, 600).is_none());
    if let EnsureAction::Fund { ledger, .. } = &mut changed.actions[0] {
        *ledger = CONTROLLER.into();
    }
    assert!(conservation(&changed, &observation, 600).is_none());
    fs::remove_dir_all(fixture.root).unwrap();
}

#[test]
fn terminal_retirement_archives_phases_and_recovers_every_handoff_boundary() {
    use crate::fleet_ensure::{
        model::*,
        ops::{self, reinstall::terminal},
        policy::expected_plan_sha256,
    };
    let (fixture, paths, mut replacement) = terminal_retirement_fixture();
    let mut journal: serde_json::Value =
        serde_json::from_slice(&fs::read(&paths.journal).unwrap()).unwrap();
    journal
        .as_object_mut()
        .unwrap()
        .remove("funding_observations");
    fs::write(&paths.journal, serde_json::to_vec(&journal).unwrap()).unwrap();
    let source = terminal::read(&paths, "local", "test-fleet").unwrap();
    replacement.operation_id = sha256_hex(b"new reviewed operation");
    replacement.scope = FleetEnsurePlanScope::ReinstallPreparation;
    replacement.protocol_actions.clear();
    replacement.reinstall = Some(Box::new(FleetReinstallRecord {
        target_artifacts_sha256: None,
        source: Some(Box::new(
            terminal::capture(
                &fixture.root,
                &source,
                ActualCycleConservation {
                    estate_funding_cycles: 0,
                    exact_estate_creation_fee_cycles: 0,
                    exact_unavoidable_fee_cycles: 10,
                    final_controlled_cycles: 600,
                    observed_net_cycle_debit_cycles: 10,
                    observed_starting_cycles: 500,
                    observed_net_cycle_credit_cycles: 0,
                    operator_debit_cycles: 120,
                    received_new_funding_cycles: 110,
                },
            )
            .unwrap(),
        )),
        operation_id: replacement.operation_id.clone(),
        source_operation_id: source.documents.operation_id.clone(),
        authorities: Vec::new(),
        assets: Vec::new(),
        activation_reset: None,
    }));
    replacement.plan_sha256 = expected_plan_sha256(&replacement);
    ops::reinstall::adoption::tests::assert_terminal_handoff(&paths, replacement);
    fs::remove_dir_all(fixture.root).unwrap();
}

#[test]
#[expect(
    clippy::too_many_lines,
    reason = "one review journey checks inventory rejection, staged review and pre-apply drift without effects"
)]
fn terminal_retirement_review_binds_fresh_inventory_and_rechecks_before_apply() {
    use crate::fleet_ensure::{model::*, ops};
    let (mut fixture, paths, phase) = terminal_retirement_fixture();
    let mut journal: serde_json::Value =
        serde_json::from_slice(&fs::read(&paths.journal).unwrap()).unwrap();
    journal
        .as_object_mut()
        .unwrap()
        .remove("funding_observations");
    fs::write(&paths.journal, serde_json::to_vec(&journal).unwrap()).unwrap();
    let mut state = ops::read_state(&paths, "test-fleet").unwrap();
    state.principals.insert("treasury".into(), TREASURY.into());
    state.topology.insert(
        "treasury".into(),
        FleetEnsureTopologyRecord {
            kind: DesiredCanisterKind::Coordinator,
            module_hash: Some(sha256_hex(b"current-wasm")),
            parent: None,
            protocol_binding: None,
            role: Some("coordinator".into()),
        },
    );
    ops::write_state(&paths, &state).unwrap();
    fixture.platform.operator_cycles = 880;
    fixture.platform.live.get_mut(TREASURY).unwrap().cycles = 600;
    let inventory = TerminalFleetInventory {
        active_registry: state.active_registry.clone(),
        controlled_cycles_by_principal: BTreeMap::new(),
        entries: vec![RegistryEntry {
            pid: TREASURY.into(),
            role: Some("coordinator".into()),
            parent_pid: None,
            module_hash: Some(sha256_hex(b"current-wasm")),
            protocol_binding: None,
        }],
    };
    fixture.platform.terminal_inventory_expected_operation_id = Some(phase.operation_id.clone());
    fixture.platform.reinstall_authority = Some(BTreeMap::from([(
        "treasury".into(),
        RootManagementCanisterObservation {
            name: "treasury".into(),
            subnet: SUBNET.into(),
            live: fixture.platform.live[TREASURY].clone(),
        },
    )]));
    let original = [
        fs::read(&paths.plan).unwrap(),
        fs::read(&paths.journal).unwrap(),
        fs::read(&paths.state).unwrap(),
    ];
    // Missing or duplicate inventory cannot be masked by retained state or the burn allowance.
    for change in 0..5 {
        fixture.platform.terminal_inventory = inventory.clone();
        let entries = &mut fixture.platform.terminal_inventory.entries;
        match change {
            0 => entries.clear(),
            1 => entries.push(inventory.entries[0].clone()),
            2 => entries[0].parent_pid = Some(RETIRED.into()),
            3 => entries[0].module_hash = Some(sha256_hex(b"drifted module")),
            4 => entries[0].role = Some("drifted-role".into()),
            _ => unreachable!(),
        }
        assert!(matches!(
            workflow::plan_reinstall(
                &fixture.root,
                &fixture.desired,
                "target",
                "test-fleet",
                2,
                &mut fixture.platform
            ),
            Err(workflow::EnsureWorkflowError::ConvergenceDrift)
        ));
        assert_eq!(fixture.platform.desired, fixture.desired);
        assert!(fixture.platform.mutations.is_empty());
    }
    fixture.platform.terminal_inventory = inventory;
    let review = workflow::plan_reinstall(
        &fixture.root,
        &fixture.desired,
        "target",
        "test-fleet",
        2,
        &mut fixture.platform,
    )
    .unwrap();
    assert_eq!(
        review.plan.scope,
        FleetEnsurePlanScope::ReinstallPreparation
    );
    assert_ne!(review.plan.operation_id, phase.operation_id);
    assert!(
        review
            .plan
            .reinstall
            .as_ref()
            .unwrap()
            .source
            .as_ref()
            .unwrap()
            .terminal_retirement
            .is_some()
    );
    assert_eq!(fixture.platform.desired, fixture.desired);
    assert!(fixture.platform.mutations.is_empty());
    // Independent operator activity invalidates apply before the old operation is replaced.
    fixture.platform.operator_cycles -= 1;
    assert!(matches!(
        workflow::apply(
            &fixture.root,
            &fixture.desired,
            "target",
            "test-fleet",
            &review.plan.plan_sha256,
            &mut fixture.platform
        ),
        Err(workflow::EnsureWorkflowError::Conservation(_))
    ));
    assert!(fixture.platform.mutations.is_empty());
    assert_eq!(fixture.platform.desired, fixture.desired);
    for (index, path) in [&paths.plan, &paths.journal, &paths.state]
        .into_iter()
        .enumerate()
    {
        assert_eq!(fs::read(path).unwrap(), original[index]);
    }
    fs::remove_dir_all(fixture.root).unwrap();
}

#[test]
fn terminal_retirement_preserves_every_root_account_and_pool_identity() {
    use crate::fleet_ensure::{
        model::*, ops::reinstall::terminal, policy::reinstall::terminal::conservation,
    };
    let (fixture, paths, _) = terminal_retirement_fixture();
    let mut source = terminal::read(&paths, "local", "test-fleet").unwrap();
    source
        .conservation
        .estate_funding_domains
        .push(EstateFundingDomainPlan {
            allocated_workloads: 1,
            available_cycles: Some(30),
            available_pool_slots: 0,
            creation_amount_cycles: 50,
            cycles_ledger: LEDGER.into(),
            creation_execution_margin_cycles: 0,
            readiness_floor_cycles: 20,
            eligible_ready_pool_assets: 1,
            initial_pool_assets: vec![OLD_APP.into()],
            ledger_fee_cycles: 10,
            management_creation_fee_cycles: 50,
            maximum_creation_debit_cycles: 0,
            maximum_creation_fee_cycles: 0,
            maximum_funding_cycles: 0,
            occupied_pool_assets: 1,
            pending_creation_count: 0,
            pending_creation: None,
            pool_maximum_size: 1,
            planned_initial_workloads: 1,
            required_creation_count: 0,
            root: "root".into(),
            root_principal: Some(TREASURY.into()),
            shortfall_cycles: 0,
        });
    source
        .journal
        .initial_estate_funding_cycles_by_root
        .insert("root".into(), 30);
    let observation = FleetObservation {
        additional_controlled_cycles: BTreeMap::new(),
        canisters: BTreeMap::new(),
        estate_funding_domains: BTreeMap::from([(
            "root".into(),
            EstateFundingDomainObservation {
                balance_cycles: Some(30),
                cycles_ledger: LEDGER.into(),
                root_principal: Some(TREASURY.into()),
                pool: Some(EstatePoolInventoryObservation {
                    assets: vec![EstatePoolAssetObservation {
                        creation_receipt: None,
                        cycles: 20,
                        lifecycle: EstatePoolAssetLifecycle::Workload,
                        origin: EstatePoolAssetOrigin::Imported,
                        principal: OLD_APP.into(),
                    }],
                    maximum_size: 1,
                    minimum_size: 0,
                    pending_creation: None,
                    readiness_floor_cycles: 20,
                    creation_execution_margin_cycles: 0,
                }),
            },
        )]),
        ledger_fee_cycles: 10,
        operator_cycles: 880,
        protocol_ready: BTreeMap::new(),
    };
    assert!(conservation(&source, &observation, 600).is_some());
    for change in 0..6 {
        let mut changed = observation.clone();
        let domain = changed.estate_funding_domains.get_mut("root").unwrap();
        match change {
            0 => domain.balance_cycles = Some(29),
            1 => domain.balance_cycles = None,
            2 => domain.root_principal = Some(RETIRED.into()),
            3 => domain.cycles_ledger = CONTROLLER.into(),
            4 => domain.pool.as_mut().unwrap().assets.clear(),
            5 => {
                let assets = &mut domain.pool.as_mut().unwrap().assets;
                assets.push(assets[0].clone());
            }
            _ => unreachable!(),
        }
        assert!(
            conservation(&source, &changed, 600).is_none(),
            "case {change}"
        );
    }
    source.conservation.estate_funding_domains[0].required_creation_count = 1;
    assert!(conservation(&source, &observation, 600).is_none());
    fs::remove_dir_all(fixture.root).unwrap();
}

fn store_chunk_fixture() -> Fixture {
    let mut fixture = publication_retry_fixture(1);
    let chunks = (0..7_u8).map(|value| vec![value; 64]).collect::<Vec<_>>();
    let template_id = TemplateId::owned("concurrent-store".into());
    let version = TemplateVersion::owned("concurrent-release".into());
    let mut preparation = Some(TemplateChunkSetPrepareInput {
        manifest: None,
        template_id: template_id.clone(),
        version: version.clone(),
        payload_hash: canic_core::cdk::utils::hash::wasm_hash(&chunks.concat()),
        payload_size_bytes: 448,
        chunk_hashes: chunks
            .iter()
            .map(|bytes| canic_core::cdk::utils::hash::wasm_hash(bytes))
            .collect(),
    });
    fixture.platform.reviewed_protocol_actions = chunks
        .into_iter()
        .enumerate()
        .map(|(index, bytes)| {
            fleet_protocol_action(
                &format!("store-chunk-{index}"),
                CurrentFleetProtocolAction::PublishStoreChunk {
                    request: TemplateChunkInput {
                        preparation: preparation.take(),
                        template_id: template_id.clone(),
                        version: version.clone(),
                        chunk_index: u32::try_from(index).unwrap(),
                        bytes,
                    },
                },
            )
        })
        .collect();
    let paths = crate::fleet_ensure::ops::EnsurePaths::under(&fixture.root, "local", "test-fleet");
    fixture.platform.publication_journal = Some(paths.journal);
    for action in &mut fixture.platform.reviewed_protocol_actions {
        let EnsureAction::FleetProtocol {
            principal, candid, ..
        } = action
        else {
            unreachable!()
        };
        *principal = Principal::from_slice(&[72]).to_text();
        *candid = "store.did".into();
    }
    fixture
}

fn pool_batch_fixture() -> Fixture {
    let mut fixture = publication_retry_fixture(1);
    fixture.platform.reviewed_protocol_actions = (0..7_u8)
        .map(|index| {
            let mut action = fleet_protocol_action(
                &format!("pool-import-{index}"),
                CurrentFleetProtocolAction::ReconcilePoolAsset {
                    request: canic_core::dto::pool::PoolCanisterRequest {
                        canister_id: Principal::from_slice(&[100 + index]),
                    },
                    minimum_cycles: Cycles::new(100),
                },
            );
            let EnsureAction::FleetProtocol {
                principal, candid, ..
            } = &mut action
            else {
                unreachable!()
            };
            *principal = Principal::from_slice(&[71]).to_text();
            *candid = "root.did".into();
            action
        })
        .collect();
    let paths = crate::fleet_ensure::ops::EnsurePaths::under(&fixture.root, "local", "test-fleet");
    fixture.platform.publication_journal = Some(paths.journal);
    fixture
}

#[test]
fn pool_batches_stop_at_authority_and_duplicate_asset_boundaries() {
    use crate::fleet_ensure::policy::independent_effects::batch_len;
    let fixture = pool_batch_fixture();
    let actions = &fixture.platform.reviewed_protocol_actions;
    assert_eq!(batch_len(&actions.iter().collect::<Vec<_>>(), &[], 0), 4);
    for axis in 0..5 {
        let mut changed = actions.clone();
        let duplicate = changed[0].clone();
        let EnsureAction::FleetProtocol {
            principal,
            candid,
            candid_sha256,
            action,
            ..
        } = &mut changed[1]
        else {
            unreachable!()
        };
        match axis {
            0 => *principal = RETIRED.into(),
            1 => *candid = "other.did".into(),
            2 => *candid_sha256 = "ff".repeat(32),
            3 => changed[1] = duplicate,
            4 => {
                **action = CurrentFleetProtocolAction::MaintainPoolReadiness {
                    minimum_ready: 1,
                    readiness_floor: Cycles::new(100),
                    maximum_updates: 1,
                };
            }
            _ => unreachable!(),
        }
        assert_eq!(batch_len(&changed.iter().collect::<Vec<_>>(), &[], 0), 1);
    }
    fs::remove_dir_all(fixture.root).unwrap();
}

#[test]
fn store_chunks_batch_after_preparation_and_replay_without_effects() {
    let mut fixture = store_chunk_fixture();
    let planned = workflow::plan(
        &fixture.root,
        &fixture.desired,
        "chunks",
        "test-fleet",
        1,
        &mut fixture.platform,
    )
    .unwrap();
    let report = apply_fixture_plan(&mut fixture, "chunks", &planned.plan).unwrap();
    assert!(report.terminal);
    assert_eq!(
        fixture
            .platform
            .independent_batches
            .iter()
            .map(Vec::len)
            .collect::<Vec<_>>(),
        [4, 2]
    );
    let mutations = fixture.platform.mutations.clone();
    assert_eq!(mutations.len(), 7);
    assert!(mutations.values().all(|count| *count == 1));
    apply_fixture_plan(&mut fixture, "chunks", &planned.plan).unwrap();
    assert_eq!(fixture.platform.mutations, mutations);
    fs::remove_dir_all(fixture.root).unwrap();
}

#[test]
fn store_chunks_drain_failures_keep_sibling_receipts_and_resume_exactly() {
    // Exercise a rejected call and a committed call with a lost response.
    for lost_reply in [false, true] {
        let mut fixture = store_chunk_fixture();
        let planned = workflow::plan(
            &fixture.root,
            &fixture.desired,
            "chunks",
            "test-fleet",
            1,
            &mut fixture.platform,
        )
        .unwrap();
        let first = action_sha256(&planned.plan.protocol_actions[1]);
        if lost_reply {
            fixture.platform.fail_once(first.clone());
        } else {
            fixture.platform.stall_before_mutation(first.clone(), 1);
        }
        assert!(matches!(
            apply_fixture_plan(&mut fixture, "chunks", &planned.plan),
            Err(workflow::EnsureWorkflowError::Platform(_))
        ));
        assert_eq!(fixture.platform.independent_batches.len(), 1);
        let paths =
            crate::fleet_ensure::ops::EnsurePaths::under(&fixture.root, "local", "test-fleet");
        let journal = crate::fleet_ensure::ops::read_journal(&paths)
            .unwrap()
            .unwrap();
        assert_eq!(journal.effects.len(), 5, "no later batch issued");
        assert_eq!(
            journal.effects[1].state,
            if lost_reply {
                EffectState::Applied
            } else {
                EffectState::Intent
            }
        );
        for record in &journal.effects[2..] {
            assert_eq!(record.state, EffectState::Applied);
            assert_eq!(record.receipt.as_deref(), Some("protocol-receipt"));
        }
        // Also qualify a process death after submission but before any replies
        // or terminal observations reached disk. The Store evidence survives.
        let mut interrupted = journal;
        for record in &mut interrupted.effects[1..] {
            record.state = EffectState::Intent;
            record.receipt = None;
            record.progress_identity = None;
        }
        crate::fleet_ensure::ops::write_journal(&paths, &interrupted).unwrap();
        // Reconstruct the adapter without its process-local response cache.
        let mut resumed = MockPlatform::new(
            fixture.desired.clone(),
            fixture.platform.live.values().cloned(),
        );
        resumed.protocol_ready = fixture.platform.protocol_ready.clone();
        resumed.reviewed_protocol_actions = fixture.platform.reviewed_protocol_actions.clone();
        resumed.publication_journal = Some(paths.journal);
        fixture.platform = resumed;
        let report = apply_fixture_plan(&mut fixture, "chunks", &planned.plan).unwrap();
        assert!(report.terminal);
        assert_eq!(fixture.platform.mutations.contains_key(&first), !lost_reply);
        for action in &planned.plan.protocol_actions[2..5] {
            assert!(
                !fixture
                    .platform
                    .mutations
                    .contains_key(&action_sha256(action))
            );
        }
        fs::remove_dir_all(fixture.root).unwrap();
    }
}

#[test]
fn store_chunks_require_applied_preparation_and_exact_batch_authority() {
    use crate::fleet_ensure::policy::independent_effects::batch_len;
    let fixture = store_chunk_fixture();
    let actions = &fixture.platform.reviewed_protocol_actions;
    let mut record = EffectRecord {
        action_sha256: action_sha256(&actions[0]),
        state: EffectState::Applied,
        publication_attempts: 0,
        maintenance_attempts: 0,
        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,
    };
    let refs = actions.iter().collect::<Vec<_>>();
    assert_eq!(batch_len(&refs, &[record.clone()], 1), 4);
    record.state = EffectState::Issued;
    assert_eq!(batch_len(&refs, &[record.clone()], 1), 0);
    record.state = EffectState::Applied;
    assert_eq!(batch_len(&refs, &[], 0), 0);
    for axis in 0..6 {
        let mut changed = actions.clone();
        let EnsureAction::FleetProtocol {
            action,
            candid,
            candid_sha256,
            principal,
            ..
        } = &mut changed[2]
        else {
            unreachable!()
        };
        let CurrentFleetProtocolAction::PublishStoreChunk { request } = action.as_mut() else {
            unreachable!()
        };
        match axis {
            0 => *principal = RETIRED.to_string(),
            1 => *candid = "other.did".into(),
            2 => *candid_sha256 = "ff".repeat(32),
            3 => request.version = TemplateVersion::owned("different".into()),
            4 => request.template_id = TemplateId::owned("different".into()),
            5 => request.chunk_index = 1,
            _ => unreachable!(),
        }
        assert_eq!(
            batch_len(&changed.iter().collect::<Vec<_>>(), &[record.clone()], 1),
            1
        );
    }
    fs::remove_dir_all(fixture.root).unwrap();
}

#[test]
fn store_chunks_stall_bound_reports_the_failed_chunk_after_draining() {
    let mut fixture = store_chunk_fixture();
    fixture.desired.maximum_stalled_observations = 1;
    fixture.platform.desired = fixture.desired.clone();
    let planned = workflow::plan(
        &fixture.root,
        &fixture.desired,
        "chunks",
        "test-fleet",
        1,
        &mut fixture.platform,
    )
    .unwrap();
    fixture
        .platform
        .stall_before_mutation(action_sha256(&planned.plan.protocol_actions[2]), 1);
    assert!(
        matches!(apply_fixture_plan(&mut fixture, "chunks", &planned.plan),
        Err(workflow::EnsureWorkflowError::Stalled { action, observations: 1, .. }) if action == "store-chunk-2")
    );
    assert_eq!(fixture.platform.independent_batches.len(), 1);
    for action in [
        &planned.plan.protocol_actions[1],
        &planned.plan.protocol_actions[3],
        &planned.plan.protocol_actions[4],
    ] {
        assert_eq!(
            fixture.platform.mutations.get(&action_sha256(action)),
            Some(&1)
        );
    }
    fs::remove_dir_all(fixture.root).unwrap();
}

#[test]
fn terminal_retirement_missing_observations_cannot_add_allowance_or_hide_invalid_evidence() {
    use crate::fleet_ensure::ops::{self, reinstall::terminal};
    let (mut fixture, paths, _) = terminal_retirement_fixture();
    let mut raw: serde_json::Value =
        serde_json::from_slice(&fs::read(&paths.journal).unwrap()).unwrap();
    raw.as_object_mut().unwrap().remove("funding_observations");
    let original = serde_json::to_vec(&raw).unwrap();
    fs::write(&paths.journal, &original).unwrap();
    let source = terminal::read(&paths, "local", "test-fleet").unwrap();
    assert_eq!(source.conservation.maximum_execution_burn_cycles, 200);
    assert_eq!(
        source.documents.journal_document_sha256,
        sha256_hex(&original)
    );
    assert!(matches!(
        ops::read_journal(&paths),
        Err(ops::EnsureStateError::Decode { .. })
    ));
    assert!(matches!(
        workflow::plan(
            &fixture.root,
            &fixture.desired,
            "target",
            "test-fleet",
            2,
            &mut fixture.platform
        ),
        Err(workflow::EnsureWorkflowError::RetainedTerminalReviewRequired { .. })
    ));
    assert_eq!(fs::read(&paths.journal).unwrap(), original);
    assert!(fixture.platform.mutations.is_empty());
    for (field, value) in [
        ("funding_observations", serde_json::Value::Null),
        ("funding_observations", serde_json::json!({"root": {}})),
        ("unreviewed_payment", serde_json::json!({"cycles": "1"})),
        ("schema_version", serde_json::json!(2)),
        ("completion", serde_json::json!("in_progress")),
        ("initial_controlled_cycles", serde_json::Value::Null),
    ] {
        let mut changed = raw.clone();
        changed[field] = value;
        fs::write(&paths.journal, serde_json::to_vec(&changed).unwrap()).unwrap();
        assert!(matches!(
            terminal::read(&paths, "local", "test-fleet"),
            Err(ops::EnsureStateError::InvalidTerminalSource)
        ));
    }
    fs::remove_dir_all(fixture.root).unwrap();
}

#[test]
#[ignore = "read-only inspection of an explicitly selected retained workspace"]
fn terminal_retirement_inspects_selected_workspace_without_mutation() {
    use crate::fleet_ensure::ops::{self, reinstall::terminal};
    let root = PathBuf::from(
        std::env::var_os("CANIC_RETAINED_REVIEW_WORKSPACE").expect("selected workspace"),
    );
    let environment =
        std::env::var("CANIC_RETAINED_REVIEW_ENVIRONMENT").expect("selected environment");
    let fleet = std::env::var("CANIC_RETAINED_REVIEW_FLEET").expect("selected Fleet");
    let paths = ops::EnsurePaths::under(&root, &environment, &fleet);
    let original = [&paths.plan, &paths.journal, &paths.state].map(|path| fs::read(path).unwrap());
    let source = terminal::read(&paths, &environment, &fleet).unwrap();
    println!(
        "source={:?}; effects={}; phases={}",
        source.documents,
        source.actions.len(),
        source.journal.successor_phases.len()
    );
    for (path, before) in [&paths.plan, &paths.journal, &paths.state]
        .into_iter()
        .zip(original)
    {
        assert_eq!(fs::read(path).unwrap(), before);
    }
}