lodestone_client/

development_workflows.rs

//! Closed recovery models for supported local development workflows.

use std::collections::{BTreeMap, BTreeSet};
use std::path::{Component, Path, PathBuf};

use lodestone_wire::{
    Digest, Identity, IndexResourceSpec, LODESTONE_API_INDEX_RESOURCE_SPEC_FORMAT_VERSION, Locator, Resource,
    ServiceCredentialReservation, U64,
};
use serde::{Deserialize, Serialize};

use crate::{
    LocalOperatorWorkflowKind, PublishedFileIdentity, WorkflowJournalError, WorkflowJournalIntent,
    WorkflowJournalTransition,
};

pub const DEVELOPMENT_MATERIALIZE_FORMAT_VERSION: u16 = 1;
pub const DEVELOPMENT_PKI_SUCCESSOR_FORMAT_VERSION: u16 = 1;
pub const DEVELOPMENT_RECONCILE_FORMAT_VERSION: u16 = 1;
const MAX_ENDPOINT_BYTES: usize = 2_048;
const MAX_NAME_BYTES: usize = 255;

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct DevelopmentMaterializeIntent {
    pub format_version: u16,
    pub output_root: String,
    pub supply_root: String,
    pub supply_package_sha256: [u8; 32],
    pub supply_source_revision: String,
    pub native_api_sha256: [u8; 32],
    pub native_endpoint: String,
    pub subscription_endpoint: String,
    pub internal_endpoint: String,
    pub bootstrap_endpoint: String,
}

impl crate::workflow_journal::sealed::Sealed for DevelopmentMaterializeIntent {}

impl WorkflowJournalIntent for DevelopmentMaterializeIntent {
    type Transition = DevelopmentMaterializeTransition;

    const KIND: LocalOperatorWorkflowKind = LocalOperatorWorkflowKind::DevelopmentMaterialize;

    fn validate(&self) -> Result<(), WorkflowJournalError> {
        if self.format_version != DEVELOPMENT_MATERIALIZE_FORMAT_VERSION
            || !valid_absolute_path(&self.output_root)
            || !valid_absolute_path(&self.supply_root)
            || self.output_root == self.supply_root
            || !valid_sha256(&self.supply_package_sha256)
            || !valid_git_object(&self.supply_source_revision)
            || !valid_sha256(&self.native_api_sha256)
            || !valid_development_endpoints(
                &self.native_endpoint,
                &self.subscription_endpoint,
                &self.internal_endpoint,
                &self.bootstrap_endpoint,
            )
        {
            return Err(WorkflowJournalError::InvalidIntent);
        }
        Ok(())
    }
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DevelopmentMaterializeArtifact {
    CaCertificate,
    CaPrivateKey,
    ProcessCertificate,
    ProcessPrivateKey,
    GatewayCertificate,
    GatewayPrivateKey,
    BootstrapCredential,
    BootstrapIntent,
    ServerConfiguration,
    DevelopmentIndexDefinition,
    InstanceManifest,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DevelopmentMaterializePrivatePlan {
    Pki,
}

impl DevelopmentMaterializeArtifact {
    fn is_manifest(self) -> bool {
        self == Self::InstanceManifest
    }

    fn required_before_manifest() -> BTreeSet<Self> {
        [
            Self::CaCertificate,
            Self::CaPrivateKey,
            Self::ProcessCertificate,
            Self::ProcessPrivateKey,
            Self::GatewayCertificate,
            Self::GatewayPrivateKey,
            Self::BootstrapCredential,
            Self::BootstrapIntent,
            Self::ServerConfiguration,
            Self::DevelopmentIndexDefinition,
        ]
        .into_iter()
        .collect()
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "state", rename_all = "snake_case", deny_unknown_fields)]
pub enum DevelopmentMaterializeTransition {
    InstanceIdentityRecorded {
        instance_identity: String,
    },
    ClockRecorded {
        unix_millis: U64,
    },
    PrivatePlanStaged {
        plan: DevelopmentMaterializePrivatePlan,
        file: PublishedFileIdentity,
    },
    PrivatePlanPublished {
        plan: DevelopmentMaterializePrivatePlan,
        file: PublishedFileIdentity,
    },
    ArtifactStaged {
        artifact: DevelopmentMaterializeArtifact,
        file: PublishedFileIdentity,
    },
    ArtifactPublished {
        artifact: DevelopmentMaterializeArtifact,
        file: PublishedFileIdentity,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        public_sha256: Option<[u8; 32]>,
    },
    Verified {
        manifest_sha256: [u8; 32],
    },
    Completed,
}

impl crate::workflow_journal::sealed::Sealed for DevelopmentMaterializeTransition {}

impl WorkflowJournalTransition for DevelopmentMaterializeTransition {
    fn validate(&self) -> Result<(), WorkflowJournalError> {
        let valid = match self {
            Self::InstanceIdentityRecorded { instance_identity } => valid_instance_identity(instance_identity),
            Self::ClockRecorded { unix_millis } => valid_u64(unix_millis),
            Self::PrivatePlanStaged { file, .. }
            | Self::PrivatePlanPublished { file, .. }
            | Self::ArtifactStaged { file, .. } => file.is_valid(),
            Self::ArtifactPublished {
                artifact,
                file,
                public_sha256,
            } => {
                file.is_valid()
                    && public_sha256.as_ref().is_none_or(valid_sha256)
                    && (!artifact.is_manifest() || public_sha256.as_ref().is_some_and(valid_sha256))
            }
            Self::Verified { manifest_sha256 } => valid_sha256(manifest_sha256),
            Self::Completed => true,
        };
        if valid {
            Ok(())
        } else {
            Err(WorkflowJournalError::InvalidTransition)
        }
    }
}

#[derive(Clone, Debug, Default, PartialEq)]
pub struct DevelopmentMaterializeState {
    instance_identity: Option<String>,
    unix_millis: Option<U64>,
    private_plans: BTreeMap<DevelopmentMaterializePrivatePlan, PublishedFileIdentity>,
    published_private_plans: BTreeMap<DevelopmentMaterializePrivatePlan, PublishedFileIdentity>,
    staged: BTreeMap<DevelopmentMaterializeArtifact, PublishedFileIdentity>,
    published: BTreeMap<DevelopmentMaterializeArtifact, PublishedFileIdentity>,
    published_manifest_sha256: Option<[u8; 32]>,
    manifest_sha256: Option<[u8; 32]>,
    completed: bool,
}

impl DevelopmentMaterializeState {
    pub fn replay(transitions: &[DevelopmentMaterializeTransition]) -> Result<Self, WorkflowJournalError> {
        let mut state = Self::default();
        for transition in transitions {
            transition.validate()?;
            state.apply(transition)?;
        }
        Ok(state)
    }

    fn apply(&mut self, transition: &DevelopmentMaterializeTransition) -> Result<(), WorkflowJournalError> {
        if self.completed {
            return Err(WorkflowJournalError::Corrupt);
        }
        match transition {
            DevelopmentMaterializeTransition::InstanceIdentityRecorded { instance_identity }
                if self.instance_identity.is_none() && self.unix_millis.is_none() =>
            {
                self.instance_identity = Some(instance_identity.clone());
            }
            DevelopmentMaterializeTransition::ClockRecorded { unix_millis }
                if self.instance_identity.is_some() && self.unix_millis.is_none() =>
            {
                self.unix_millis = Some(unix_millis.clone());
            }
            DevelopmentMaterializeTransition::PrivatePlanStaged { plan, file }
                if self.unix_millis.is_some()
                    && self.staged.is_empty()
                    && self.published.is_empty()
                    && !self.private_plans.contains_key(plan) =>
            {
                self.private_plans.insert(*plan, file.clone());
            }
            DevelopmentMaterializeTransition::PrivatePlanPublished { plan, file }
                if !self.published_private_plans.contains_key(plan) && self.private_plans.get(plan) == Some(file) =>
            {
                self.published_private_plans.insert(*plan, file.clone());
            }
            DevelopmentMaterializeTransition::ArtifactStaged { artifact, file }
                if self
                    .published_private_plans
                    .contains_key(&DevelopmentMaterializePrivatePlan::Pki)
                    && !self.staged.contains_key(artifact)
                    && !self.published.contains_key(artifact)
                    && (!artifact.is_manifest()
                        || self.published.keys().copied().collect::<BTreeSet<_>>()
                            == DevelopmentMaterializeArtifact::required_before_manifest()) =>
            {
                self.staged.insert(*artifact, file.clone());
            }
            DevelopmentMaterializeTransition::ArtifactPublished {
                artifact,
                file,
                public_sha256,
            } if !self.published.contains_key(artifact) && self.staged.get(artifact) == Some(file) => {
                self.published.insert(*artifact, file.clone());
                if artifact.is_manifest() {
                    self.published_manifest_sha256 = *public_sha256;
                }
            }
            DevelopmentMaterializeTransition::Verified { manifest_sha256 }
                if self
                    .published
                    .contains_key(&DevelopmentMaterializeArtifact::InstanceManifest)
                    && self.published_manifest_sha256.as_ref() == Some(manifest_sha256)
                    && self.manifest_sha256.is_none() =>
            {
                self.manifest_sha256 = Some(*manifest_sha256);
            }
            DevelopmentMaterializeTransition::Completed if self.manifest_sha256.is_some() => {
                self.completed = true;
            }
            _ => return Err(WorkflowJournalError::Corrupt),
        }
        Ok(())
    }

    #[must_use]
    pub fn instance_identity(&self) -> Option<&str> {
        self.instance_identity.as_deref()
    }

    #[must_use]
    pub const fn clock(&self) -> Option<&U64> {
        self.unix_millis.as_ref()
    }

    #[must_use]
    pub fn staged(&self, artifact: DevelopmentMaterializeArtifact) -> Option<&PublishedFileIdentity> {
        self.staged.get(&artifact)
    }

    #[must_use]
    pub fn private_plan(&self, plan: DevelopmentMaterializePrivatePlan) -> Option<&PublishedFileIdentity> {
        self.private_plans.get(&plan)
    }

    #[must_use]
    pub fn published_private_plan(&self, plan: DevelopmentMaterializePrivatePlan) -> Option<&PublishedFileIdentity> {
        self.published_private_plans.get(&plan)
    }

    #[must_use]
    pub fn published(&self, artifact: DevelopmentMaterializeArtifact) -> Option<&PublishedFileIdentity> {
        self.published.get(&artifact)
    }

    #[must_use]
    pub const fn completed(&self) -> bool {
        self.completed
    }
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DevelopmentPkiSuccessorOperation {
    RenewLeaves,
    RotateCa,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct DevelopmentPkiSuccessorIntent {
    pub format_version: u16,
    pub operation: DevelopmentPkiSuccessorOperation,
    pub instance_identity: String,
    pub parent_manifest_path: String,
    pub parent_manifest_sha256: [u8; 32],
    pub successor_root: String,
    pub parent_instance_generation: U64,
    pub successor_instance_generation: U64,
    pub parent_ca_generation: U64,
    pub successor_ca_generation: U64,
    pub parent_process_leaf_generation: U64,
    pub successor_process_leaf_generation: U64,
    pub parent_gateway_leaf_generation: U64,
    pub successor_gateway_leaf_generation: U64,
}

impl crate::workflow_journal::sealed::Sealed for DevelopmentPkiSuccessorIntent {}

impl WorkflowJournalIntent for DevelopmentPkiSuccessorIntent {
    type Transition = DevelopmentPkiSuccessorTransition;

    const KIND: LocalOperatorWorkflowKind = LocalOperatorWorkflowKind::DevelopmentPkiSuccessor;

    fn validate(&self) -> Result<(), WorkflowJournalError> {
        let instance_generations =
            ordered_successor(&self.parent_instance_generation, &self.successor_instance_generation);
        let process_generations = ordered_successor(
            &self.parent_process_leaf_generation,
            &self.successor_process_leaf_generation,
        );
        let gateway_generations = ordered_successor(
            &self.parent_gateway_leaf_generation,
            &self.successor_gateway_leaf_generation,
        );
        let ca_generations = match self.operation {
            DevelopmentPkiSuccessorOperation::RenewLeaves => {
                canonical_u64(&self.parent_ca_generation).filter(|value| *value > 0)
                    == canonical_u64(&self.successor_ca_generation).filter(|value| *value > 0)
            }
            DevelopmentPkiSuccessorOperation::RotateCa => {
                ordered_successor(&self.parent_ca_generation, &self.successor_ca_generation)
            }
        };
        let parent_manifest = Path::new(&self.parent_manifest_path);
        let parent_root = parent_manifest.parent();
        if self.format_version != DEVELOPMENT_PKI_SUCCESSOR_FORMAT_VERSION
            || !valid_instance_identity(&self.instance_identity)
            || !valid_absolute_path(&self.parent_manifest_path)
            || !valid_sha256(&self.parent_manifest_sha256)
            || !valid_absolute_path(&self.successor_root)
            || parent_manifest.file_name().and_then(|name| name.to_str()) != Some("instance.json")
            || parent_root.is_none_or(|root| paths_overlap(root, Path::new(&self.successor_root)))
            || !instance_generations
            || !process_generations
            || !gateway_generations
            || !ca_generations
        {
            return Err(WorkflowJournalError::InvalidIntent);
        }
        Ok(())
    }
}

fn paths_overlap(left: &Path, right: &Path) -> bool {
    left.starts_with(right) || right.starts_with(left)
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DevelopmentPkiSuccessorArtifact {
    CaCertificate,
    CaPrivateKey,
    ProcessCertificate,
    ProcessPrivateKey,
    GatewayCertificate,
    GatewayPrivateKey,
    ServerConfiguration,
    InstanceManifest,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DevelopmentPkiSuccessorPrivatePlan {
    Pki,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "state", rename_all = "snake_case", deny_unknown_fields)]
pub enum DevelopmentPkiSuccessorTransition {
    ClockRecorded {
        unix_millis: U64,
    },
    PrivatePlanStaged {
        plan: DevelopmentPkiSuccessorPrivatePlan,
        file: PublishedFileIdentity,
    },
    PrivatePlanPublished {
        plan: DevelopmentPkiSuccessorPrivatePlan,
        file: PublishedFileIdentity,
    },
    ArtifactStaged {
        artifact: DevelopmentPkiSuccessorArtifact,
        file: PublishedFileIdentity,
    },
    ArtifactPublished {
        artifact: DevelopmentPkiSuccessorArtifact,
        file: PublishedFileIdentity,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        public_sha256: Option<[u8; 32]>,
    },
    Verified {
        manifest_sha256: [u8; 32],
    },
    Completed,
}

impl crate::workflow_journal::sealed::Sealed for DevelopmentPkiSuccessorTransition {}

impl WorkflowJournalTransition for DevelopmentPkiSuccessorTransition {
    fn validate(&self) -> Result<(), WorkflowJournalError> {
        let valid = match self {
            Self::ClockRecorded { unix_millis } => valid_u64(unix_millis),
            Self::PrivatePlanStaged { file, .. }
            | Self::PrivatePlanPublished { file, .. }
            | Self::ArtifactStaged { file, .. } => file.is_valid(),
            Self::ArtifactPublished {
                artifact,
                file,
                public_sha256,
            } => {
                file.is_valid()
                    && public_sha256.as_ref().is_none_or(valid_sha256)
                    && (*artifact != DevelopmentPkiSuccessorArtifact::InstanceManifest
                        || public_sha256.as_ref().is_some_and(valid_sha256))
            }
            Self::Verified { manifest_sha256 } => valid_sha256(manifest_sha256),
            Self::Completed => true,
        };
        if valid {
            Ok(())
        } else {
            Err(WorkflowJournalError::InvalidTransition)
        }
    }
}

#[derive(Clone, Debug, Default, PartialEq)]
pub struct DevelopmentPkiSuccessorState {
    clock: Option<U64>,
    private_plans: BTreeMap<DevelopmentPkiSuccessorPrivatePlan, PublishedFileIdentity>,
    published_private_plans: BTreeMap<DevelopmentPkiSuccessorPrivatePlan, PublishedFileIdentity>,
    staged: BTreeMap<DevelopmentPkiSuccessorArtifact, PublishedFileIdentity>,
    published: BTreeMap<DevelopmentPkiSuccessorArtifact, PublishedFileIdentity>,
    published_manifest_sha256: Option<[u8; 32]>,
    manifest_sha256: Option<[u8; 32]>,
    completed: bool,
}

impl DevelopmentPkiSuccessorState {
    pub fn replay(
        intent: &DevelopmentPkiSuccessorIntent,
        transitions: &[DevelopmentPkiSuccessorTransition],
    ) -> Result<Self, WorkflowJournalError> {
        intent.validate()?;
        let mut state = Self::default();
        for transition in transitions {
            transition.validate()?;
            state.apply(intent.operation, transition)?;
        }
        Ok(state)
    }

    fn apply(
        &mut self,
        operation: DevelopmentPkiSuccessorOperation,
        transition: &DevelopmentPkiSuccessorTransition,
    ) -> Result<(), WorkflowJournalError> {
        if self.completed {
            return Err(WorkflowJournalError::Corrupt);
        }
        match transition {
            DevelopmentPkiSuccessorTransition::ClockRecorded { unix_millis } if self.clock.is_none() => {
                self.clock = Some(unix_millis.clone());
            }
            DevelopmentPkiSuccessorTransition::PrivatePlanStaged { plan, file }
                if self.clock.is_some()
                    && self.staged.is_empty()
                    && self.published.is_empty()
                    && !self.private_plans.contains_key(plan) =>
            {
                self.private_plans.insert(*plan, file.clone());
            }
            DevelopmentPkiSuccessorTransition::PrivatePlanPublished { plan, file }
                if !self.published_private_plans.contains_key(plan) && self.private_plans.get(plan) == Some(file) =>
            {
                self.published_private_plans.insert(*plan, file.clone());
            }
            DevelopmentPkiSuccessorTransition::ArtifactStaged { artifact, file }
                if self.clock.is_some()
                    && self
                        .published_private_plans
                        .contains_key(&DevelopmentPkiSuccessorPrivatePlan::Pki)
                    && !self.staged.contains_key(artifact)
                    && !self.published.contains_key(artifact)
                    && (artifact_admitted_for_pki_operation(operation, *artifact)
                        || (*artifact == DevelopmentPkiSuccessorArtifact::InstanceManifest
                            && self.published.keys().copied().collect::<BTreeSet<_>>()
                                == required_pki_artifacts(operation))) =>
            {
                self.staged.insert(*artifact, file.clone());
            }
            DevelopmentPkiSuccessorTransition::ArtifactPublished {
                artifact,
                file,
                public_sha256,
            } if !self.published.contains_key(artifact) && self.staged.get(artifact) == Some(file) => {
                self.published.insert(*artifact, file.clone());
                if *artifact == DevelopmentPkiSuccessorArtifact::InstanceManifest {
                    self.published_manifest_sha256 = *public_sha256;
                }
            }
            DevelopmentPkiSuccessorTransition::Verified { manifest_sha256 }
                if self
                    .published
                    .contains_key(&DevelopmentPkiSuccessorArtifact::InstanceManifest)
                    && self.published_manifest_sha256.as_ref() == Some(manifest_sha256)
                    && self.manifest_sha256.is_none() =>
            {
                self.manifest_sha256 = Some(*manifest_sha256);
            }
            DevelopmentPkiSuccessorTransition::Completed if self.manifest_sha256.is_some() => {
                self.completed = true;
            }
            _ => return Err(WorkflowJournalError::Corrupt),
        }
        Ok(())
    }

    #[must_use]
    pub const fn clock(&self) -> Option<&U64> {
        self.clock.as_ref()
    }

    #[must_use]
    pub fn private_plan(&self, plan: DevelopmentPkiSuccessorPrivatePlan) -> Option<&PublishedFileIdentity> {
        self.private_plans.get(&plan)
    }

    #[must_use]
    pub fn published_private_plan(&self, plan: DevelopmentPkiSuccessorPrivatePlan) -> Option<&PublishedFileIdentity> {
        self.published_private_plans.get(&plan)
    }

    #[must_use]
    pub fn staged(&self, artifact: DevelopmentPkiSuccessorArtifact) -> Option<&PublishedFileIdentity> {
        self.staged.get(&artifact)
    }

    #[must_use]
    pub fn published(&self, artifact: DevelopmentPkiSuccessorArtifact) -> Option<&PublishedFileIdentity> {
        self.published.get(&artifact)
    }

    #[must_use]
    pub const fn completed(&self) -> bool {
        self.completed
    }
}

fn artifact_admitted_for_pki_operation(
    operation: DevelopmentPkiSuccessorOperation,
    artifact: DevelopmentPkiSuccessorArtifact,
) -> bool {
    artifact != DevelopmentPkiSuccessorArtifact::InstanceManifest
        && (operation == DevelopmentPkiSuccessorOperation::RotateCa
            || !matches!(
                artifact,
                DevelopmentPkiSuccessorArtifact::CaCertificate | DevelopmentPkiSuccessorArtifact::CaPrivateKey
            ))
}

fn required_pki_artifacts(operation: DevelopmentPkiSuccessorOperation) -> BTreeSet<DevelopmentPkiSuccessorArtifact> {
    let mut required = [
        DevelopmentPkiSuccessorArtifact::ProcessCertificate,
        DevelopmentPkiSuccessorArtifact::ProcessPrivateKey,
        DevelopmentPkiSuccessorArtifact::GatewayCertificate,
        DevelopmentPkiSuccessorArtifact::GatewayPrivateKey,
        DevelopmentPkiSuccessorArtifact::ServerConfiguration,
    ]
    .into_iter()
    .collect::<BTreeSet<_>>();
    if operation == DevelopmentPkiSuccessorOperation::RotateCa {
        required.insert(DevelopmentPkiSuccessorArtifact::CaCertificate);
        required.insert(DevelopmentPkiSuccessorArtifact::CaPrivateKey);
    }
    required
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct DevelopmentReconcileIntent {
    pub format_version: u16,
    pub instance_root: String,
    pub instance_manifest_sha256: [u8; 32],
    pub index_definition_path: String,
    pub index_definition_sha256: [u8; 32],
    pub credential_output: String,
    pub profile_output: String,
    pub material_output: String,
    pub timeout_millis: U64,
    pub reservation_ttl_millis: U64,
    pub credential_ttl_millis: U64,
}

impl crate::workflow_journal::sealed::Sealed for DevelopmentReconcileIntent {}

impl WorkflowJournalIntent for DevelopmentReconcileIntent {
    type Transition = DevelopmentReconcileTransition;

    const KIND: LocalOperatorWorkflowKind = LocalOperatorWorkflowKind::DevelopmentReconcile;

    fn validate(&self) -> Result<(), WorkflowJournalError> {
        let paths = [
            &self.instance_root,
            &self.index_definition_path,
            &self.credential_output,
            &self.profile_output,
            &self.material_output,
        ];
        let distinct_outputs = [&self.credential_output, &self.profile_output, &self.material_output]
            .into_iter()
            .collect::<BTreeSet<_>>()
            .len()
            == 3;
        if self.format_version != DEVELOPMENT_RECONCILE_FORMAT_VERSION
            || paths.into_iter().any(|path| !valid_absolute_path(path))
            || !distinct_outputs
            || !valid_sha256(&self.instance_manifest_sha256)
            || !valid_sha256(&self.index_definition_sha256)
            || !valid_positive_bounded_duration(&self.timeout_millis)
            || !valid_positive_bounded_duration(&self.reservation_ttl_millis)
            || !valid_positive_bounded_duration(&self.credential_ttl_millis)
            || !ordered_u64(&self.reservation_ttl_millis, &self.credential_ttl_millis)
        {
            return Err(WorkflowJournalError::InvalidIntent);
        }
        Ok(())
    }
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DevelopmentReconcileRoleTarget {
    Administrator,
    Consumer,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DevelopmentReconcileTerminalOutcome {
    EffectDeadlineExpired,
    Conflict,
    ReservationExpiredNoEffect,
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(tag = "state", rename_all = "snake_case", deny_unknown_fields)]
pub enum DevelopmentReconcileTransition {
    EffectDeadlineRecorded {
        unix_millis: U64,
    },
    AuthorityObserved {
        cluster_id: Identity,
        collection_id: Identity,
        schema_sha256: Digest,
        logical_time: U64,
        reservation_expires_at: U64,
        credential_expires_at: U64,
    },
    CatalogPrepared {
        locator: Locator,
        semantic_digest: Digest,
        expires_at: U64,
        resource_spec: IndexResourceSpec,
    },
    CatalogApplied {
        locator: Locator,
        resource: Resource,
    },
    IndexActive {
        resource: Resource,
        index_revision: U64,
    },
    RolePrepared {
        target: DevelopmentReconcileRoleTarget,
        locator: Locator,
        semantic_digest: Digest,
        expires_at: U64,
        predecessor_role_version: U64,
        successor_role_version: U64,
    },
    RoleApplied {
        target: DevelopmentReconcileRoleTarget,
        locator: Locator,
        target_policy_revision: U64,
    },
    RoleVerified {
        target: DevelopmentReconcileRoleTarget,
        role_version: U64,
        observed_policy_revision: U64,
    },
    CredentialReserved {
        reservation: ServiceCredentialReservation,
    },
    CredentialStaged {
        file: PublishedFileIdentity,
    },
    CredentialActivated {
        principal_id: Identity,
        api_key_id: Identity,
        generation: U64,
    },
    CredentialPublished {
        file: PublishedFileIdentity,
    },
    ProfilePublished {
        file: PublishedFileIdentity,
        profile_sha256: [u8; 32],
    },
    ConsumerVerified {
        observed_policy_revision: U64,
    },
    MaterialPublished {
        file: PublishedFileIdentity,
        material_sha256: [u8; 32],
    },
    Terminal {
        outcome: DevelopmentReconcileTerminalOutcome,
    },
    Completed,
}

impl crate::workflow_journal::sealed::Sealed for DevelopmentReconcileTransition {}

impl WorkflowJournalTransition for DevelopmentReconcileTransition {
    fn validate(&self) -> Result<(), WorkflowJournalError> {
        let valid = match self {
            Self::EffectDeadlineRecorded { unix_millis } => valid_u64(unix_millis),
            Self::AuthorityObserved {
                cluster_id,
                collection_id,
                schema_sha256,
                logical_time,
                reservation_expires_at,
                credential_expires_at,
            } => {
                valid_identity(cluster_id)
                    && valid_identity(collection_id)
                    && valid_digest(schema_sha256)
                    && valid_u64(logical_time)
                    && ordered_u64(logical_time, reservation_expires_at)
                    && ordered_u64(reservation_expires_at, credential_expires_at)
            }
            Self::CatalogPrepared {
                locator,
                semantic_digest,
                expires_at,
                resource_spec,
            } => {
                valid_locator(locator)
                    && valid_digest(semantic_digest)
                    && valid_u64(expires_at)
                    && resource_spec.format_version == u32::from(LODESTONE_API_INDEX_RESOURCE_SPEC_FORMAT_VERSION)
            }
            Self::CatalogApplied { locator, resource } => valid_locator(locator) && valid_resource(resource),
            Self::IndexActive {
                resource,
                index_revision,
            } => valid_resource(resource) && valid_u64(index_revision),
            Self::RolePrepared {
                locator,
                semantic_digest,
                expires_at,
                predecessor_role_version,
                successor_role_version,
                ..
            } => {
                valid_locator(locator)
                    && valid_digest(semantic_digest)
                    && valid_u64(expires_at)
                    && ordered_successor(predecessor_role_version, successor_role_version)
            }
            Self::RoleApplied {
                locator,
                target_policy_revision,
                ..
            } => valid_locator(locator) && valid_u64(target_policy_revision),
            Self::RoleVerified {
                role_version,
                observed_policy_revision,
                ..
            } => valid_u64(role_version) && valid_u64(observed_policy_revision),
            Self::CredentialReserved { reservation } => valid_reservation(reservation),
            Self::CredentialStaged { file } | Self::CredentialPublished { file } => file.is_valid(),
            Self::CredentialActivated {
                principal_id,
                api_key_id,
                generation,
            } => valid_identity(principal_id) && valid_identity(api_key_id) && valid_u64(generation),
            Self::ProfilePublished { file, profile_sha256 } => file.is_valid() && valid_sha256(profile_sha256),
            Self::ConsumerVerified {
                observed_policy_revision,
            } => valid_u64(observed_policy_revision),
            Self::MaterialPublished { file, material_sha256 } => file.is_valid() && valid_sha256(material_sha256),
            Self::Terminal { .. } => true,
            Self::Completed => true,
        };
        if valid {
            Ok(())
        } else {
            Err(WorkflowJournalError::InvalidTransition)
        }
    }
}

#[derive(Clone, Debug, Default, PartialEq)]
pub struct DevelopmentReconcileState {
    effect_deadline: Option<U64>,
    binding: Option<(Identity, Identity, Digest, U64, U64, U64)>,
    catalog_preparation: Option<(Locator, Digest, U64, IndexResourceSpec)>,
    resource: Option<Resource>,
    index_active: bool,
    role_preparations: BTreeMap<DevelopmentReconcileRoleTarget, (Locator, Digest, U64, U64, U64)>,
    role_applied: BTreeMap<DevelopmentReconcileRoleTarget, U64>,
    role_verified: BTreeMap<DevelopmentReconcileRoleTarget, (U64, U64)>,
    reservation: Option<ServiceCredentialReservation>,
    credential_staged: Option<PublishedFileIdentity>,
    credential_activated: bool,
    credential_published: Option<PublishedFileIdentity>,
    profile_published: Option<PublishedFileIdentity>,
    consumer_verified: Option<U64>,
    material_published: Option<PublishedFileIdentity>,
    terminal: Option<DevelopmentReconcileTerminalOutcome>,
    completed: bool,
}

impl DevelopmentReconcileState {
    pub fn replay(
        intent: &DevelopmentReconcileIntent,
        transitions: &[DevelopmentReconcileTransition],
    ) -> Result<Self, WorkflowJournalError> {
        intent.validate()?;
        let mut state = Self::default();
        for transition in transitions {
            transition.validate()?;
            state.apply(intent, transition)?;
        }
        Ok(state)
    }

    fn apply(
        &mut self,
        intent: &DevelopmentReconcileIntent,
        transition: &DevelopmentReconcileTransition,
    ) -> Result<(), WorkflowJournalError> {
        if self.completed || self.terminal.is_some() {
            return Err(WorkflowJournalError::Corrupt);
        }
        match transition {
            DevelopmentReconcileTransition::EffectDeadlineRecorded { unix_millis }
                if self.effect_deadline.is_none() =>
            {
                self.effect_deadline = Some(unix_millis.clone());
            }
            DevelopmentReconcileTransition::AuthorityObserved {
                cluster_id,
                collection_id,
                schema_sha256,
                logical_time,
                reservation_expires_at,
                credential_expires_at,
            } if self.effect_deadline.is_some()
                && self.binding.is_none()
                && checked_add_u64(logical_time, &intent.reservation_ttl_millis)
                    == canonical_u64(reservation_expires_at)
                && checked_add_u64(logical_time, &intent.credential_ttl_millis)
                    == canonical_u64(credential_expires_at) =>
            {
                self.binding = Some((
                    cluster_id.clone(),
                    collection_id.clone(),
                    schema_sha256.clone(),
                    logical_time.clone(),
                    reservation_expires_at.clone(),
                    credential_expires_at.clone(),
                ));
            }
            DevelopmentReconcileTransition::CatalogPrepared {
                locator,
                semantic_digest,
                expires_at,
                resource_spec,
            } if self.binding.is_some() && self.resource.is_none() => {
                // A preparation is effect-free. If its exact catalog root is
                // stale before submission, reconciliation may durably replace
                // it while preserving the same desired index definition.
                self.catalog_preparation = Some((
                    locator.clone(),
                    semantic_digest.clone(),
                    expires_at.clone(),
                    resource_spec.clone(),
                ));
            }
            DevelopmentReconcileTransition::CatalogApplied { locator, resource }
                if self
                    .catalog_preparation
                    .as_ref()
                    .is_some_and(|(expected, _, _, _)| expected == locator)
                    && self.resource.is_none() =>
            {
                self.resource = Some(resource.clone());
            }
            DevelopmentReconcileTransition::IndexActive { resource, .. }
                if self.binding.is_some()
                    && self.resource.as_ref().is_none_or(|current| current == resource)
                    && !self.index_active =>
            {
                self.resource = Some(resource.clone());
                self.index_active = true;
            }
            DevelopmentReconcileTransition::RolePrepared {
                target,
                locator,
                semantic_digest,
                expires_at,
                predecessor_role_version,
                successor_role_version,
            } if self.role_target_is_next(*target) && !self.role_applied.contains_key(target) => {
                // Security policy can advance between prepare and submit. A
                // not-submitted stale preparation is safely replaceable.
                self.role_preparations.insert(
                    *target,
                    (
                        locator.clone(),
                        semantic_digest.clone(),
                        expires_at.clone(),
                        predecessor_role_version.clone(),
                        successor_role_version.clone(),
                    ),
                );
            }
            DevelopmentReconcileTransition::RoleApplied {
                target,
                locator,
                target_policy_revision,
            } if self
                .role_preparations
                .get(target)
                .is_some_and(|(expected, _, _, _, _)| expected == locator)
                && !self.role_applied.contains_key(target) =>
            {
                self.role_applied.insert(*target, target_policy_revision.clone());
            }
            DevelopmentReconcileTransition::RoleVerified {
                target,
                role_version,
                observed_policy_revision,
            } if self.role_target_is_next(*target)
                && self.role_verification_matches(*target, role_version, observed_policy_revision) =>
            {
                self.role_verified
                    .insert(*target, (role_version.clone(), observed_policy_revision.clone()));
            }
            DevelopmentReconcileTransition::CredentialReserved { reservation }
                if self
                    .role_verified
                    .contains_key(&DevelopmentReconcileRoleTarget::Consumer)
                    && self.reservation.is_none() =>
            {
                let Some((_, _, _, _, reservation_expires_at, credential_expires_at)) = self.binding.as_ref() else {
                    return Err(WorkflowJournalError::Corrupt);
                };
                if &reservation.reservation_expires_at != reservation_expires_at
                    || &reservation.credential_expires_at != credential_expires_at
                {
                    return Err(WorkflowJournalError::Corrupt);
                }
                self.reservation = Some(reservation.clone());
            }
            DevelopmentReconcileTransition::CredentialStaged { file }
                if self.reservation.is_some() && self.credential_staged.is_none() =>
            {
                self.credential_staged = Some(file.clone());
            }
            DevelopmentReconcileTransition::CredentialActivated {
                principal_id,
                api_key_id,
                generation,
            } if self.reservation.as_ref().is_some_and(|reservation| {
                &reservation.principal_id == principal_id
                    && &reservation.api_key_id == api_key_id
                    && &reservation.generation == generation
            }) && self.credential_staged.is_some()
                && !self.credential_activated =>
            {
                self.credential_activated = true;
            }
            DevelopmentReconcileTransition::CredentialPublished { file }
                if self.credential_activated
                    && self.credential_staged.as_ref() == Some(file)
                    && self.credential_published.is_none() =>
            {
                self.credential_published = Some(file.clone());
            }
            DevelopmentReconcileTransition::ProfilePublished { file, .. }
                if self.credential_published.is_some() && self.profile_published.is_none() =>
            {
                self.profile_published = Some(file.clone());
            }
            DevelopmentReconcileTransition::ConsumerVerified {
                observed_policy_revision,
            } if self.profile_published.is_some() && self.consumer_verified.is_none() => {
                self.consumer_verified = Some(observed_policy_revision.clone());
            }
            DevelopmentReconcileTransition::MaterialPublished { file, .. }
                if self.consumer_verified.is_some() && self.material_published.is_none() =>
            {
                self.material_published = Some(file.clone());
            }
            DevelopmentReconcileTransition::Terminal { outcome, .. } if self.effect_deadline.is_some() => {
                self.terminal = Some(*outcome);
            }
            DevelopmentReconcileTransition::Completed if self.material_published.is_some() => {
                self.completed = true;
            }
            _ => return Err(WorkflowJournalError::Corrupt),
        }
        Ok(())
    }

    fn role_target_is_next(&self, target: DevelopmentReconcileRoleTarget) -> bool {
        self.index_active
            && match target {
                DevelopmentReconcileRoleTarget::Administrator => !self
                    .role_verified
                    .contains_key(&DevelopmentReconcileRoleTarget::Administrator),
                DevelopmentReconcileRoleTarget::Consumer => {
                    self.role_verified
                        .contains_key(&DevelopmentReconcileRoleTarget::Administrator)
                        && !self
                            .role_verified
                            .contains_key(&DevelopmentReconcileRoleTarget::Consumer)
                }
            }
    }

    fn role_verification_matches(
        &self,
        target: DevelopmentReconcileRoleTarget,
        role_version: &U64,
        observed_policy_revision: &U64,
    ) -> bool {
        match self.role_preparations.get(&target) {
            Some((_, _, _, _, successor_role_version)) => {
                successor_role_version == role_version
                    && self.role_applied.get(&target).is_some_and(|target_policy_revision| {
                        ordered_or_equal_u64(target_policy_revision, observed_policy_revision)
                    })
            }
            None => !self.role_applied.contains_key(&target),
        }
    }

    #[must_use]
    pub const fn effect_deadline(&self) -> Option<&U64> {
        self.effect_deadline.as_ref()
    }

    #[must_use]
    pub const fn resource(&self) -> Option<&Resource> {
        self.resource.as_ref()
    }

    #[must_use]
    pub const fn authority_binding(&self) -> Option<&(Identity, Identity, Digest, U64, U64, U64)> {
        self.binding.as_ref()
    }

    #[must_use]
    pub const fn catalog_preparation(&self) -> Option<&(Locator, Digest, U64, IndexResourceSpec)> {
        self.catalog_preparation.as_ref()
    }

    #[must_use]
    pub const fn index_active(&self) -> bool {
        self.index_active
    }

    #[must_use]
    pub fn role_preparation(
        &self,
        target: DevelopmentReconcileRoleTarget,
    ) -> Option<&(Locator, Digest, U64, U64, U64)> {
        self.role_preparations.get(&target)
    }

    #[must_use]
    pub fn role_applied(&self, target: DevelopmentReconcileRoleTarget) -> Option<&U64> {
        self.role_applied.get(&target)
    }

    #[must_use]
    pub fn role_verified(&self, target: DevelopmentReconcileRoleTarget) -> Option<&(U64, U64)> {
        self.role_verified.get(&target)
    }

    #[must_use]
    pub const fn reservation(&self) -> Option<&ServiceCredentialReservation> {
        self.reservation.as_ref()
    }

    #[must_use]
    pub const fn credential_staged(&self) -> Option<&PublishedFileIdentity> {
        self.credential_staged.as_ref()
    }

    #[must_use]
    pub const fn credential_activated(&self) -> bool {
        self.credential_activated
    }

    #[must_use]
    pub const fn credential_published(&self) -> Option<&PublishedFileIdentity> {
        self.credential_published.as_ref()
    }

    #[must_use]
    pub const fn profile_published(&self) -> Option<&PublishedFileIdentity> {
        self.profile_published.as_ref()
    }

    #[must_use]
    pub const fn consumer_policy_revision(&self) -> Option<&U64> {
        self.consumer_verified.as_ref()
    }

    #[must_use]
    pub const fn material_published(&self) -> Option<&PublishedFileIdentity> {
        self.material_published.as_ref()
    }

    #[must_use]
    pub const fn terminal(&self) -> Option<DevelopmentReconcileTerminalOutcome> {
        self.terminal
    }

    #[must_use]
    pub const fn completed(&self) -> bool {
        self.completed
    }
}

fn valid_absolute_path(value: &str) -> bool {
    if value.is_empty() || value.len() > 4_096 || value.bytes().any(|byte| byte.is_ascii_control()) {
        return false;
    }
    let path = Path::new(value);
    path.is_absolute()
        && path.parent().is_some()
        && path
            .components()
            .all(|component| matches!(component, Component::RootDir | Component::Normal(_)))
        && path.components().collect::<PathBuf>().as_os_str() == path.as_os_str()
}

fn valid_sha256(value: &[u8; 32]) -> bool {
    value.iter().any(|byte| *byte != 0)
}

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

fn valid_git_object(value: &str) -> bool {
    matches!(value.len(), 40 | 64)
        && value
            .bytes()
            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
        && value.bytes().any(|byte| byte != b'0')
}

fn valid_development_endpoints(native: &str, subscription: &str, internal: &str, bootstrap: &str) -> bool {
    let values = [native, subscription, internal, bootstrap];
    if values.iter().any(|value| {
        value.is_empty() || value.len() > MAX_ENDPOINT_BYTES || !value.bytes().all(|byte| byte.is_ascii_graphic())
    }) {
        return false;
    }
    let Some(authority) = native.strip_prefix("lodestone+tls://") else {
        return false;
    };
    let (advertised_host, port_text) = if let Some(port) = authority.strip_prefix("[::1]:") {
        ("[::1]", port)
    } else if let Some(port) = authority.strip_prefix("localhost:") {
        ("localhost", port)
    } else if let Some(port) = authority.strip_prefix("127.0.0.1:") {
        ("127.0.0.1", port)
    } else {
        return false;
    };
    let Some(base_port) = port_text
        .parse::<u16>()
        .ok()
        .filter(|port| port.to_string() == port_text && (1024..=65532).contains(port))
    else {
        return false;
    };
    let bind_host = match advertised_host {
        "[::1]" => "[::1]",
        _ => "127.0.0.1",
    };
    subscription == format!("https://{advertised_host}:{}", base_port + 1)
        && internal == format!("{bind_host}:{}", base_port + 2)
        && bootstrap == format!("{bind_host}:{}", base_port + 3)
}

fn valid_identity(value: &Identity) -> bool {
    value.value.len() == 32
        && value
            .value
            .bytes()
            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
        && value.value.bytes().any(|byte| byte != b'0')
}

fn valid_u64(value: &U64) -> bool {
    value
        .value
        .parse::<u64>()
        .ok()
        .is_some_and(|parsed| parsed.to_string() == value.value)
}

fn canonical_u64(value: &U64) -> Option<u64> {
    value
        .value
        .parse::<u64>()
        .ok()
        .filter(|parsed| parsed.to_string() == value.value)
}

fn ordered_successor(predecessor: &U64, successor: &U64) -> bool {
    canonical_u64(predecessor)
        .filter(|value| *value > 0)
        .and_then(|value| value.checked_add(1))
        .is_some_and(|expected| canonical_u64(successor) == Some(expected))
}

fn ordered_u64(predecessor: &U64, successor: &U64) -> bool {
    canonical_u64(predecessor)
        .zip(canonical_u64(successor))
        .is_some_and(|(predecessor, successor)| predecessor < successor)
}

fn ordered_or_equal_u64(predecessor: &U64, successor: &U64) -> bool {
    canonical_u64(predecessor)
        .zip(canonical_u64(successor))
        .is_some_and(|(predecessor, successor)| predecessor <= successor)
}

fn checked_add_u64(left: &U64, right: &U64) -> Option<u64> {
    canonical_u64(left)?.checked_add(canonical_u64(right)?)
}

fn valid_positive_bounded_duration(value: &U64) -> bool {
    canonical_u64(value).is_some_and(|value| (1..=86_400_000).contains(&value))
}

fn valid_digest(value: &Digest) -> bool {
    value.value.len() == 64
        && value
            .value
            .bytes()
            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
        && value.value.bytes().any(|byte| byte != b'0')
}

fn valid_locator(value: &Locator) -> bool {
    !value.authority.is_empty()
        && value.authority.len() <= MAX_NAME_BYTES
        && value.authority.bytes().all(|byte| byte.is_ascii_graphic())
        && !value.value.is_empty()
        && value.value.len() <= 4_096
        && value.value.bytes().all(|byte| byte.is_ascii_graphic())
}

fn valid_resource(value: &Resource) -> bool {
    value.kind == "index" && valid_index_name(&value.name) && valid_identity(&value.incarnation)
}

fn valid_index_name(value: &str) -> bool {
    !value.is_empty()
        && value.len() <= MAX_NAME_BYTES
        && value
            .bytes()
            .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || matches!(byte, b'-' | b'_'))
}

fn valid_reservation(value: &ServiceCredentialReservation) -> bool {
    valid_locator(&value.locator)
        && value.locator.authority == "service_credential_reservation"
        && valid_digest(&value.reservation_action_id)
        && valid_digest(&value.activation_action_id)
        && valid_identity(&value.principal_id)
        && valid_identity(&value.api_key_id)
        && valid_u64(&value.generation)
        && valid_u64(&value.reservation_expires_at)
        && valid_u64(&value.credential_expires_at)
        && value.verifier_algorithm == "sha256_identity_bound_v1"
        && value.verifier_algorithm_version == 1
        && value.required_secret_bytes == 32
}

#[cfg(test)]
#[allow(clippy::expect_used)] // Fixed nonzero byte fixtures are valid by construction.
mod tests {
    use super::*;

    fn identity(value: u128) -> Identity {
        Identity {
            value: format!("{value:032x}"),
        }
    }

    fn instance_identity(value: u8) -> String {
        format!("{value:02x}").repeat(32)
    }

    fn digest(value: u8) -> Digest {
        Digest {
            value: format!("{value:02x}").repeat(32),
        }
    }

    fn u64_value(value: u64) -> U64 {
        U64 {
            value: value.to_string(),
        }
    }

    fn file(value: u64) -> PublishedFileIdentity {
        PublishedFileIdentity {
            device: 1,
            inode: value,
            size: 1,
            sha256: [u8::try_from(value).unwrap_or(1); 32],
        }
    }

    fn reconcile_intent() -> DevelopmentReconcileIntent {
        DevelopmentReconcileIntent {
            format_version: DEVELOPMENT_RECONCILE_FORMAT_VERSION,
            instance_root: "/tmp/lodestone".to_owned(),
            instance_manifest_sha256: [1; 32],
            index_definition_path: "/tmp/lodestone/operator/development-index.json".to_owned(),
            index_definition_sha256: [2; 32],
            credential_output: "/tmp/output/credential.json".to_owned(),
            profile_output: "/tmp/output/profile.json".to_owned(),
            material_output: "/tmp/output/material.json".to_owned(),
            timeout_millis: u64_value(90),
            reservation_ttl_millis: u64_value(10),
            credential_ttl_millis: u64_value(20),
        }
    }

    #[test]
    fn materialization_manifest_is_last_and_completion_requires_verification() -> Result<(), WorkflowJournalError> {
        let mut transitions = vec![
            DevelopmentMaterializeTransition::InstanceIdentityRecorded {
                instance_identity: instance_identity(1),
            },
            DevelopmentMaterializeTransition::ClockRecorded {
                unix_millis: u64_value(1),
            },
            DevelopmentMaterializeTransition::PrivatePlanStaged {
                plan: DevelopmentMaterializePrivatePlan::Pki,
                file: file(200),
            },
            DevelopmentMaterializeTransition::PrivatePlanPublished {
                plan: DevelopmentMaterializePrivatePlan::Pki,
                file: file(200),
            },
        ];
        for (position, artifact) in DevelopmentMaterializeArtifact::required_before_manifest()
            .into_iter()
            .enumerate()
        {
            let file = file(u64::try_from(position).unwrap_or(0) + 1);
            transitions.push(DevelopmentMaterializeTransition::ArtifactStaged {
                artifact,
                file: file.clone(),
            });
            transitions.push(DevelopmentMaterializeTransition::ArtifactPublished {
                artifact,
                file,
                public_sha256: None,
            });
        }
        let manifest = file(100);
        transitions.push(DevelopmentMaterializeTransition::ArtifactStaged {
            artifact: DevelopmentMaterializeArtifact::InstanceManifest,
            file: manifest.clone(),
        });
        transitions.push(DevelopmentMaterializeTransition::ArtifactPublished {
            artifact: DevelopmentMaterializeArtifact::InstanceManifest,
            file: manifest,
            public_sha256: Some([3; 32]),
        });
        assert!(DevelopmentMaterializeState::replay(&transitions).is_ok());
        transitions.push(DevelopmentMaterializeTransition::Verified {
            manifest_sha256: [3; 32],
        });
        transitions.push(DevelopmentMaterializeTransition::Completed);
        assert!(DevelopmentMaterializeState::replay(&transitions)?.completed());

        let out_of_order = [
            DevelopmentMaterializeTransition::InstanceIdentityRecorded {
                instance_identity: instance_identity(1),
            },
            DevelopmentMaterializeTransition::ClockRecorded {
                unix_millis: u64_value(1),
            },
            DevelopmentMaterializeTransition::ArtifactPublished {
                artifact: DevelopmentMaterializeArtifact::InstanceManifest,
                file: file(1),
                public_sha256: Some([3; 32]),
            },
        ];
        assert_eq!(
            DevelopmentMaterializeState::replay(&out_of_order),
            Err(WorkflowJournalError::Corrupt)
        );
        Ok(())
    }

    #[test]
    fn pki_successor_requires_ca_artifacts_only_for_ca_rotation() -> Result<(), WorkflowJournalError> {
        let intent = DevelopmentPkiSuccessorIntent {
            format_version: DEVELOPMENT_PKI_SUCCESSOR_FORMAT_VERSION,
            operation: DevelopmentPkiSuccessorOperation::RenewLeaves,
            instance_identity: instance_identity(1),
            parent_manifest_path: "/tmp/parent/instance.json".to_owned(),
            parent_manifest_sha256: [1; 32],
            successor_root: "/tmp/successor".to_owned(),
            parent_instance_generation: u64_value(1),
            successor_instance_generation: u64_value(2),
            parent_ca_generation: u64_value(1),
            successor_ca_generation: u64_value(1),
            parent_process_leaf_generation: u64_value(1),
            successor_process_leaf_generation: u64_value(2),
            parent_gateway_leaf_generation: u64_value(1),
            successor_gateway_leaf_generation: u64_value(2),
        };
        intent.validate()?;
        let illegal = [DevelopmentPkiSuccessorTransition::ArtifactStaged {
            artifact: DevelopmentPkiSuccessorArtifact::CaPrivateKey,
            file: file(1),
        }];
        assert_eq!(
            DevelopmentPkiSuccessorState::replay(&intent, &illegal),
            Err(WorkflowJournalError::Corrupt)
        );

        let mut transitions = vec![
            DevelopmentPkiSuccessorTransition::ClockRecorded {
                unix_millis: u64_value(1),
            },
            DevelopmentPkiSuccessorTransition::PrivatePlanStaged {
                plan: DevelopmentPkiSuccessorPrivatePlan::Pki,
                file: file(90),
            },
            DevelopmentPkiSuccessorTransition::PrivatePlanPublished {
                plan: DevelopmentPkiSuccessorPrivatePlan::Pki,
                file: file(90),
            },
        ];
        for (position, artifact) in required_pki_artifacts(intent.operation).into_iter().enumerate() {
            let identity = file(u64::try_from(position).unwrap_or(0) + 1);
            transitions.push(DevelopmentPkiSuccessorTransition::ArtifactStaged {
                artifact,
                file: identity.clone(),
            });
            transitions.push(DevelopmentPkiSuccessorTransition::ArtifactPublished {
                artifact,
                file: identity,
                public_sha256: None,
            });
        }
        let manifest = file(100);
        transitions.push(DevelopmentPkiSuccessorTransition::ArtifactStaged {
            artifact: DevelopmentPkiSuccessorArtifact::InstanceManifest,
            file: manifest.clone(),
        });
        transitions.push(DevelopmentPkiSuccessorTransition::ArtifactPublished {
            artifact: DevelopmentPkiSuccessorArtifact::InstanceManifest,
            file: manifest,
            public_sha256: Some([9; 32]),
        });
        transitions.push(DevelopmentPkiSuccessorTransition::Verified {
            manifest_sha256: [9; 32],
        });
        transitions.push(DevelopmentPkiSuccessorTransition::Completed);
        assert!(DevelopmentPkiSuccessorState::replay(&intent, &transitions)?.completed());
        Ok(())
    }

    #[test]
    fn reconciliation_enforces_administrator_scope_before_consumer_scope() -> Result<(), WorkflowJournalError> {
        let resource = Resource {
            kind: "index".to_owned(),
            name: "events".to_owned(),
            incarnation: identity(4),
        };
        let prefix = [
            DevelopmentReconcileTransition::EffectDeadlineRecorded {
                unix_millis: u64_value(100),
            },
            DevelopmentReconcileTransition::AuthorityObserved {
                cluster_id: identity(1),
                collection_id: identity(2),
                schema_sha256: digest(3),
                logical_time: u64_value(10),
                reservation_expires_at: u64_value(20),
                credential_expires_at: u64_value(30),
            },
            DevelopmentReconcileTransition::IndexActive {
                resource,
                index_revision: u64_value(1),
            },
            DevelopmentReconcileTransition::RoleVerified {
                target: DevelopmentReconcileRoleTarget::Consumer,
                role_version: u64_value(1),
                observed_policy_revision: u64_value(1),
            },
        ];
        assert_eq!(
            DevelopmentReconcileState::replay(&reconcile_intent(), &prefix),
            Err(WorkflowJournalError::Corrupt)
        );
        Ok(())
    }

    #[test]
    fn reconciliation_terminal_outcome_is_final_but_unresolved_work_remains_resumable() {
        let deadline = DevelopmentReconcileTransition::EffectDeadlineRecorded {
            unix_millis: u64_value(100),
        };
        let terminal = DevelopmentReconcileTransition::Terminal {
            outcome: DevelopmentReconcileTerminalOutcome::EffectDeadlineExpired,
        };
        let state = DevelopmentReconcileState::replay(&reconcile_intent(), &[deadline.clone(), terminal]);
        assert_eq!(
            state.map(|value| value.terminal()),
            Ok(Some(DevelopmentReconcileTerminalOutcome::EffectDeadlineExpired))
        );
        assert_eq!(
            DevelopmentReconcileState::replay(
                &reconcile_intent(),
                &[
                    deadline,
                    DevelopmentReconcileTransition::Terminal {
                        outcome: DevelopmentReconcileTerminalOutcome::Conflict,
                    },
                    DevelopmentReconcileTransition::Completed,
                ],
            ),
            Err(WorkflowJournalError::Corrupt)
        );

        // An unresolved remote locator is already retained by its preparation
        // transition. No terminal record is appended, so a later invocation can
        // spend a fresh observation budget resolving the same locator.
        let mut resumable = vec![
            DevelopmentReconcileTransition::EffectDeadlineRecorded {
                unix_millis: u64_value(100),
            },
            DevelopmentReconcileTransition::AuthorityObserved {
                cluster_id: identity(1),
                collection_id: identity(2),
                schema_sha256: digest(3),
                logical_time: u64_value(10),
                reservation_expires_at: u64_value(20),
                credential_expires_at: u64_value(30),
            },
            DevelopmentReconcileTransition::CatalogPrepared {
                locator: Locator {
                    authority: "catalog".to_owned(),
                    value: "create-events".to_owned(),
                },
                semantic_digest: digest(4),
                expires_at: u64_value(40),
                resource_spec: IndexResourceSpec {
                    format_version: u32::from(LODESTONE_API_INDEX_RESOURCE_SPEC_FORMAT_VERSION),
                    revision: u64_value(1),
                    initial_shards: 1,
                    target_shard_bytes: u64_value(1),
                    durability: crate::ResourceDurabilityClass::RegionalObject,
                    storage_class: crate::ResourceStorageClass::Standard,
                    ram_admission_weight: 1,
                    ram_hard_max_bytes: u64_value(1),
                    nvme_admission_weight: 1,
                    nvme_hard_max_bytes: u64_value(1),
                    refresh_objective_millis: 1,
                    memtable_hard_max_bytes: u64_value(1),
                    seal_parallelism: 1,
                    merge_parallelism: 1,
                    vector: lodestone_wire::VectorResourcePolicy {
                        quality: lodestone_wire::VectorQualityObjective::Balanced,
                        indexing: lodestone_wire::VectorIndexingObjective::Balanced,
                        residency: None,
                        calibration: None,
                    },
                    lifecycle_binding: None,
                    encryption_required: false,
                    resolver_capability: None,
                },
            },
        ];
        let mut replacement = resumable[2].clone();
        if let DevelopmentReconcileTransition::CatalogPrepared {
            locator,
            semantic_digest,
            ..
        } = &mut replacement
        {
            locator.value = "create-events-after-root-movement".to_owned();
            *semantic_digest = digest(5);
        }
        resumable.push(replacement);
        let state = DevelopmentReconcileState::replay(&reconcile_intent(), &resumable);
        assert!(state.is_ok_and(|value| {
            value.terminal().is_none()
                && value
                    .catalog_preparation()
                    .is_some_and(|(locator, semantic_digest, _, _)| {
                        locator.value == "create-events-after-root-movement" && semantic_digest == &digest(5)
                    })
        }));
    }
}