lodestone_client/

exact_workflows.rs

//! Recovery-correct native mutation workflows.
//!
//! The server remains the effect authority. This module owns only the
//! caller-side identity, deadline, and resolver state needed to avoid minting
//! a replacement effect after response loss.

use std::error::Error;
use std::fmt::{self, Display, Formatter};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};

use aws_lc_rs::digest;
use lodestone_wire::{
    Digest, ErrorCode, GetIndexRequest, Identity, IndexResourceSpec, Locator, ObserveAppendCommittedRequest,
    PrepareAppendBulkRequest, PrepareCreateIndexRequest, PrepareDocumentMutationRequest, ResolveAppendBulkRequest,
    ResolveCatalogActionRequest, ResolveDocumentMutationRequest, Resource, RetainedAppendBulkOutcome,
    RetainedAppendItem, RetainedAppendItemOutcome, RetainedAppendObservationOutcome, RetainedAppendObservationState,
    RetainedAppendRetrySafety, RetainedAppendState, SubmitAppendBulkRequest, SubmitCatalogActionRequest,
    SubmitDocumentMutationRequest, U64,
};
use serde::{Deserialize, Serialize};

use crate::workflow_journal::sealed;
use crate::{
    CallCancellation, CallOptions, Client, ClientError, LocalOperatorWorkflowKind, SubmissionDisposition,
    WorkflowJournal, WorkflowJournalError, WorkflowJournalIntent, WorkflowJournalTransition,
};

pub const EXACT_MUTATION_FORMAT_VERSION: u16 = 1;
const MAX_EFFECT_CALL_MILLIS: u64 = 599_000;

/// Returns the catalog submit discriminator for a prepared index creation.
///
/// The preparation authority selects retained-append creation whenever the
/// caller explicitly requests retained event identity, so recovery must
/// preserve that same discriminator at submit time.
#[must_use]
pub const fn create_index_action_kind(retained_event_identity: Option<bool>) -> &'static str {
    if matches!(retained_event_identity, Some(true)) {
        "create_retained_append_log_index"
    } else {
        "create_index"
    }
}

/// Semantic owner of one recovery record.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ExactMutationKindV1 {
    Append,
    Document,
    Index,
}

/// Closed result vocabulary shared by all exact mutation helpers.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ExactMutationOutcomeV1 {
    AlreadySatisfied,
    Applied,
    Resolved,
    Conflict,
    NotSubmitted,
    Indeterminate,
    KnownRetired,
    EffectDeadlineExpired,
    ClockRegression,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
enum ExactSubmissionDispositionV1 {
    NotSubmitted,
    Submitted,
    Resolved,
    KnownRetired,
    Indeterminate,
}

impl From<SubmissionDisposition> for ExactSubmissionDispositionV1 {
    fn from(value: SubmissionDisposition) -> Self {
        match value {
            SubmissionDisposition::NotSubmitted => Self::NotSubmitted,
            SubmissionDisposition::Submitted => Self::Submitted,
            SubmissionDisposition::Resolved => Self::Resolved,
            SubmissionDisposition::KnownRetired => Self::KnownRetired,
            SubmissionDisposition::Indeterminate => Self::Indeterminate,
        }
    }
}

/// Redacted retained-append item observation. Server error messages are never
/// persisted in a recovery record.
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct AppendExactItemResultV1 {
    pub item_action_id: Digest,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub locator: Option<Locator>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub semantic_digest: Option<Digest>,
    pub state: RetainedAppendState,
    pub retry_safety: RetainedAppendRetrySafety,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub sequence: Option<U64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub primary_term: Option<U64>,
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct AppendExactResultV1 {
    pub outcome: ExactMutationOutcomeV1,
    pub batch_id: Digest,
    pub index: Resource,
    pub items: Vec<AppendExactItemResultV1>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub visibility: Vec<RetainedAppendObservationOutcome>,
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct DocumentMutationExactResultV1 {
    pub outcome: ExactMutationOutcomeV1,
    pub index: Resource,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub locator: Option<Locator>,
    pub state: String,
    pub submitted: bool,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub sequence: Option<U64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub mapping_generation: Option<U64>,
    pub mapping_adopted: bool,
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ReconcileIndexExactResultV1 {
    pub outcome: ExactMutationOutcomeV1,
    pub name: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub locator: Option<Locator>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub resource: Option<Resource>,
    pub state: String,
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(tag = "kind", content = "result", rename_all = "snake_case", deny_unknown_fields)]
enum ExactTerminalResultV1 {
    Append(AppendExactResultV1),
    Document(DocumentMutationExactResultV1),
    Index(ReconcileIndexExactResultV1),
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(tag = "kind", content = "value", rename_all = "snake_case", deny_unknown_fields)]
enum ExactPreparationV1 {
    Append(RetainedAppendBulkOutcome),
    Document {
        locator: Locator,
        semantic_digest: Digest,
    },
    Index {
        locator: Locator,
        expires_at: U64,
        semantic_digest: Digest,
        resource_spec: Box<IndexResourceSpec>,
    },
}

/// Non-secret state sufficient to resume the same effect. Callers that do not
/// use the filesystem adapter must persist this value after an unresolved
/// return.
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ExactWorkflowRecoveryV1 {
    pub format_version: u16,
    pub mutation_kind: ExactMutationKindV1,
    pub request_sha256: Digest,
    pub cluster_id: Identity,
    pub collection_id: Identity,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub index: Option<Resource>,
    pub caller_action_id: Digest,
    pub effect_deadline_unix_millis: U64,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub last_observed_unix_millis: Option<U64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    preparation: Option<ExactPreparationV1>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    submission_disposition: Option<ExactSubmissionDispositionV1>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    terminal: Option<ExactTerminalResultV1>,
}

/// One result plus the exact non-secret recovery state observed at return.
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ExactExecutionV1<T> {
    pub result: T,
    pub recovery: ExactWorkflowRecoveryV1,
}

/// Filesystem-backed adapter over Lodestone's sealed create-new workflow
/// journal. The directory is exclusive to one exact request intent.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ExactWorkflowStore {
    path: PathBuf,
}

impl ExactWorkflowStore {
    #[must_use]
    pub fn new(path: impl Into<PathBuf>) -> Self {
        Self { path: path.into() }
    }

    #[must_use]
    pub fn path(&self) -> &Path {
        &self.path
    }
}

/// Injectable wall clock. Time is a liveness budget, never ordering or commit
/// evidence.
pub trait ExactWorkflowClock: Send + Sync {
    fn unix_millis(&self) -> Result<u64, ExactWorkflowError>;
}

#[derive(Debug)]
struct SystemExactWorkflowClock;

impl ExactWorkflowClock for SystemExactWorkflowClock {
    fn unix_millis(&self) -> Result<u64, ExactWorkflowError> {
        let duration = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map_err(|_| ExactWorkflowError::ClockUnavailable)?;
        u64::try_from(duration.as_millis()).map_err(|_| ExactWorkflowError::ClockUnavailable)
    }
}

/// Immutable budgets and optional recovery storage for one exact invocation.
#[derive(Clone)]
pub struct ExactWorkflowOptions {
    effect_deadline_unix_millis: u64,
    observation_deadline_unix_millis: Option<u64>,
    store: Option<ExactWorkflowStore>,
    recovery: Option<ExactWorkflowRecoveryV1>,
    clock: Arc<dyn ExactWorkflowClock>,
    cancellation: CallCancellation,
}

impl ExactWorkflowOptions {
    #[must_use]
    pub fn new(effect_deadline_unix_millis: u64) -> Self {
        Self {
            effect_deadline_unix_millis,
            observation_deadline_unix_millis: None,
            store: None,
            recovery: None,
            clock: Arc::new(SystemExactWorkflowClock),
            cancellation: CallCancellation::new(),
        }
    }

    #[must_use]
    pub fn with_observation_deadline_unix_millis(mut self, value: u64) -> Self {
        self.observation_deadline_unix_millis = Some(value);
        self
    }

    #[must_use]
    pub fn with_store(mut self, store: ExactWorkflowStore) -> Self {
        self.store = Some(store);
        self
    }

    #[must_use]
    pub fn with_recovery(mut self, recovery: ExactWorkflowRecoveryV1) -> Self {
        self.recovery = Some(recovery);
        self
    }

    #[must_use]
    pub fn with_clock(mut self, clock: Arc<dyn ExactWorkflowClock>) -> Self {
        self.clock = clock;
        self
    }

    #[must_use]
    pub fn with_cancellation(mut self, cancellation: CallCancellation) -> Self {
        self.cancellation = cancellation;
        self
    }
}

#[derive(Debug)]
pub enum ExactWorkflowError {
    InvalidInput,
    InvalidRecovery,
    RequestCodec,
    CoreShape,
    ClockUnavailable,
    Journal(WorkflowJournalError),
    Client(ClientError),
}

impl Display for ExactWorkflowError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter.write_str(match self {
            Self::InvalidInput => "the exact mutation input is invalid",
            Self::InvalidRecovery => "the exact mutation recovery state does not match this request",
            Self::RequestCodec => "the exact mutation request cannot be canonically encoded",
            Self::CoreShape => "Core changed an exact mutation identity or outcome shape",
            Self::ClockUnavailable => "the exact mutation wall clock is unavailable",
            Self::Journal(_) => "the exact mutation journal could not be verified or updated",
            Self::Client(_) => "the exact mutation client call failed",
        })
    }
}

impl Error for ExactWorkflowError {}

impl From<WorkflowJournalError> for ExactWorkflowError {
    fn from(value: WorkflowJournalError) -> Self {
        Self::Journal(value)
    }
}

/// Append request plus the optional visibility observation required by the
/// caller. Its canonical encoding is bound into recovery state.
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct AppendExactRequestV1 {
    pub batch_id: Digest,
    pub index: Resource,
    pub items: Vec<RetainedAppendItem>,
    #[serde(default)]
    pub observe_committed: bool,
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
struct ExactMutationIntent {
    format_version: u16,
    mutation_kind: ExactMutationKindV1,
    request_sha256: Digest,
    cluster_id: Identity,
    collection_id: Identity,
    index: Option<Resource>,
    caller_action_id: Digest,
}

impl sealed::Sealed for ExactMutationIntent {}

impl WorkflowJournalIntent for ExactMutationIntent {
    type Transition = ExactMutationTransition;

    const KIND: LocalOperatorWorkflowKind = LocalOperatorWorkflowKind::ExactMutation;

    fn validate(&self) -> Result<(), WorkflowJournalError> {
        if self.format_version != EXACT_MUTATION_FORMAT_VERSION
            || self.request_sha256.validate().is_err()
            || self.cluster_id.validate().is_err()
            || self.collection_id.validate().is_err()
            || self.caller_action_id.validate().is_err()
            || self.index.as_ref().is_some_and(|value| !valid_index(value))
        {
            return Err(WorkflowJournalError::InvalidIntent);
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(tag = "transition", rename_all = "snake_case", deny_unknown_fields)]
enum ExactMutationTransition {
    DeadlineEstablished { unix_millis: U64 },
    ClockObserved { unix_millis: U64 },
    Prepared { preparation: ExactPreparationV1 },
    SubmissionObserved { disposition: ExactSubmissionDispositionV1 },
    Terminal { result: ExactTerminalResultV1 },
}

impl sealed::Sealed for ExactMutationTransition {}

impl WorkflowJournalTransition for ExactMutationTransition {
    fn validate(&self) -> Result<(), WorkflowJournalError> {
        match self {
            Self::DeadlineEstablished { unix_millis } if parse_u64(unix_millis).is_some_and(|value| value > 0) => {
                Ok(())
            }
            Self::ClockObserved { unix_millis } if parse_u64(unix_millis).is_some() => Ok(()),
            Self::Prepared { preparation } if valid_preparation(preparation) => Ok(()),
            Self::SubmissionObserved { .. } => Ok(()),
            Self::Terminal { result } if valid_terminal(result) => Ok(()),
            _ => Err(WorkflowJournalError::InvalidTransition),
        }
    }
}

enum ExactSession {
    Durable(WorkflowJournal<ExactMutationIntent>),
    Ephemeral,
}

impl ExactSession {
    async fn record(
        &self,
        state: &mut ExactWorkflowRecoveryV1,
        transition: ExactMutationTransition,
    ) -> Result<(), ExactWorkflowError> {
        let mut next = state.clone();
        reduce_transition(&mut next, &transition)?;
        if let Self::Durable(journal) = self {
            journal.append(transition).await?;
        }
        *state = next;
        Ok(())
    }
}

impl Client {
    /// Runs retained append through prepare, submit, and resolver without ever
    /// replacing the caller's batch or item action identities.
    pub async fn append_exact(
        &self,
        request: AppendExactRequestV1,
        options: ExactWorkflowOptions,
    ) -> Result<ExactExecutionV1<AppendExactResultV1>, ExactWorkflowError> {
        validate_append_request(&request)?;
        let request_sha256 = canonical_sha256(&request)?;
        let (cluster_id, collection_id) = self.exact_profile_scope();
        let intent = exact_intent(
            ExactMutationKindV1::Append,
            request_sha256,
            cluster_id,
            collection_id,
            Some(request.index.clone()),
            request.batch_id.clone(),
        );
        let (session, mut recovery) = open_session(&intent, &options).await?;
        let mut options = options;
        options.effect_deadline_unix_millis = recovery_effect_deadline(&recovery)?;
        if let Some(ExactTerminalResultV1::Append(result)) = &recovery.terminal {
            return Ok(execution(result.clone(), recovery));
        }
        let now = observe_clock(&session, &mut recovery, &options).await?;
        if clock_regressed(&recovery, now) {
            return Ok(execution(
                append_status(&request, &recovery, ExactMutationOutcomeV1::ClockRegression),
                recovery,
            ));
        }

        if recovery.preparation.is_none() {
            if now >= options.effect_deadline_unix_millis {
                let result = append_status(&request, &recovery, ExactMutationOutcomeV1::EffectDeadlineExpired);
                return finish_append(&session, &mut recovery, result).await;
            }
            let prepared = self
                .prepare_append_bulk(
                    PrepareAppendBulkRequest {
                        batch_id: request.batch_id.clone(),
                        index: request.index.clone(),
                        items: request.items.clone(),
                    },
                    effect_options(now, options.effect_deadline_unix_millis, &options)?,
                )
                .await
                .map_err(ExactWorkflowError::Client)?
                .outcome;
            validate_append_outcome(&request, &prepared, true)?;
            session
                .record(
                    &mut recovery,
                    ExactMutationTransition::Prepared {
                        preparation: ExactPreparationV1::Append(redact_append_outcome(prepared)),
                    },
                )
                .await?;
        }

        let preparations = append_preparation(&recovery)?.items.clone();
        let may_submit = !matches!(
            recovery.submission_disposition,
            Some(
                ExactSubmissionDispositionV1::Submitted
                    | ExactSubmissionDispositionV1::Resolved
                    | ExactSubmissionDispositionV1::KnownRetired
                    | ExactSubmissionDispositionV1::Indeterminate
            )
        );
        if may_submit && now < options.effect_deadline_unix_millis {
            match self
                .submit_append_bulk(
                    SubmitAppendBulkRequest {
                        batch_id: request.batch_id.clone(),
                        index: request.index.clone(),
                        items: request.items.clone(),
                        preparations,
                    },
                    effect_options(now, options.effect_deadline_unix_millis, &options)?,
                )
                .await
            {
                Ok(submitted) => {
                    validate_append_outcome(&request, &submitted.outcome, false)?;
                    let disposition = if submitted.outcome.items.iter().any(|item| {
                        matches!(
                            item.state,
                            RetainedAppendState::Committed
                                | RetainedAppendState::Replayed
                                | RetainedAppendState::SubmittedIndeterminate
                        )
                    }) {
                        ExactSubmissionDispositionV1::Submitted
                    } else {
                        ExactSubmissionDispositionV1::NotSubmitted
                    };
                    session
                        .record(
                            &mut recovery,
                            ExactMutationTransition::SubmissionObserved { disposition },
                        )
                        .await?;
                    if let Some(mut result) = append_terminal_result(&request, &submitted.outcome, false) {
                        if !terminal_outcome(result.outcome) {
                            return Ok(execution(result, recovery));
                        }
                        if request.observe_committed
                            && matches!(
                                result.outcome,
                                ExactMutationOutcomeV1::Applied | ExactMutationOutcomeV1::Resolved
                            )
                        {
                            let Some(observe_options) = resolution_options(now, &options)? else {
                                result.outcome = ExactMutationOutcomeV1::Indeterminate;
                                return Ok(execution(result, recovery));
                            };
                            result = observe_append_visibility(self, &request, result, observe_options).await?;
                            if result.outcome == ExactMutationOutcomeV1::Indeterminate {
                                return Ok(execution(result, recovery));
                            }
                        }
                        return finish_append(&session, &mut recovery, result).await;
                    }
                }
                Err(error) => {
                    let disposition = ExactSubmissionDispositionV1::from(error.submission_disposition());
                    session
                        .record(
                            &mut recovery,
                            ExactMutationTransition::SubmissionObserved { disposition },
                        )
                        .await?;
                    if disposition == ExactSubmissionDispositionV1::NotSubmitted {
                        return Ok(execution(
                            append_status(&request, &recovery, ExactMutationOutcomeV1::NotSubmitted),
                            recovery,
                        ));
                    }
                }
            }
        }

        let Some(resolve_options) = resolution_options(now, &options)? else {
            return Ok(execution(
                append_status(&request, &recovery, ExactMutationOutcomeV1::Indeterminate),
                recovery,
            ));
        };
        let resolved = self
            .resolve_append_bulk(
                ResolveAppendBulkRequest {
                    batch_id: request.batch_id.clone(),
                    index: request.index.clone(),
                    items: append_preparation(&recovery)?.items.clone(),
                },
                resolve_options,
            )
            .await;
        let outcome = match resolved {
            Ok(value) => value.outcome,
            Err(_) => {
                return Ok(execution(
                    append_status(&request, &recovery, ExactMutationOutcomeV1::Indeterminate),
                    recovery,
                ));
            }
        };
        validate_append_outcome(&request, &outcome, false)?;
        let Some(mut result) = append_terminal_result(&request, &outcome, true) else {
            return Ok(execution(
                append_result(&request, &outcome, ExactMutationOutcomeV1::Indeterminate),
                recovery,
            ));
        };
        if request.observe_committed
            && matches!(
                result.outcome,
                ExactMutationOutcomeV1::Applied | ExactMutationOutcomeV1::Resolved
            )
        {
            let Some(observe_options) = resolution_options(now, &options)? else {
                result.outcome = ExactMutationOutcomeV1::Indeterminate;
                return Ok(execution(result, recovery));
            };
            result = observe_append_visibility(self, &request, result, observe_options).await?;
            if result.outcome == ExactMutationOutcomeV1::Indeterminate {
                return Ok(execution(result, recovery));
            }
        }
        finish_append(&session, &mut recovery, result).await
    }

    /// Runs one document mutation with a stable prepared locator and resolves
    /// any possibly submitted result before returning a terminal claim.
    pub async fn mutate_document_exact(
        &self,
        request: PrepareDocumentMutationRequest,
        options: ExactWorkflowOptions,
    ) -> Result<ExactExecutionV1<DocumentMutationExactResultV1>, ExactWorkflowError> {
        if !valid_index(&request.index) || request.document_id.is_empty() {
            return Err(ExactWorkflowError::InvalidInput);
        }
        let request_sha256 = canonical_sha256(&request)?;
        let (cluster_id, collection_id) = self.exact_profile_scope();
        let intent = exact_intent(
            ExactMutationKindV1::Document,
            request_sha256.clone(),
            cluster_id,
            collection_id,
            Some(request.index.clone()),
            request_sha256,
        );
        let (session, mut recovery) = open_session(&intent, &options).await?;
        let mut options = options;
        options.effect_deadline_unix_millis = recovery_effect_deadline(&recovery)?;
        if let Some(ExactTerminalResultV1::Document(result)) = &recovery.terminal {
            return Ok(execution(result.clone(), recovery));
        }
        let now = observe_clock(&session, &mut recovery, &options).await?;
        if clock_regressed(&recovery, now) {
            return Ok(execution(
                document_status(&request, &recovery, ExactMutationOutcomeV1::ClockRegression),
                recovery,
            ));
        }
        if recovery.preparation.is_none() {
            if now >= options.effect_deadline_unix_millis {
                let result = document_status(&request, &recovery, ExactMutationOutcomeV1::EffectDeadlineExpired);
                return finish_document(&session, &mut recovery, result).await;
            }
            let prepared = self
                .prepare_document_mutation(
                    request.clone(),
                    effect_options(now, options.effect_deadline_unix_millis, &options)?,
                )
                .await
                .map_err(ExactWorkflowError::Client)?;
            if prepared.locator.authority.is_empty() || prepared.semantic_digest.validate().is_err() {
                return Err(ExactWorkflowError::CoreShape);
            }
            session
                .record(
                    &mut recovery,
                    ExactMutationTransition::Prepared {
                        preparation: ExactPreparationV1::Document {
                            locator: prepared.locator,
                            semantic_digest: prepared.semantic_digest,
                        },
                    },
                )
                .await?;
        }
        let (locator, semantic_digest) = document_preparation(&recovery)?;
        let may_submit = !matches!(
            recovery.submission_disposition,
            Some(
                ExactSubmissionDispositionV1::Submitted
                    | ExactSubmissionDispositionV1::Resolved
                    | ExactSubmissionDispositionV1::KnownRetired
                    | ExactSubmissionDispositionV1::Indeterminate
            )
        );
        if may_submit && now < options.effect_deadline_unix_millis {
            match self
                .submit_document_mutation(
                    SubmitDocumentMutationRequest {
                        locator: locator.clone(),
                        semantic_digest,
                        index: request.index.clone(),
                        document_id: request.document_id.clone(),
                        operation: request.operation.clone(),
                        source: request.source.clone(),
                        routing: request.routing.clone(),
                        condition: request.condition.clone(),
                        pipeline: request.pipeline.clone(),
                    },
                    effect_options(now, options.effect_deadline_unix_millis, &options)?,
                )
                .await
            {
                Ok(value) => {
                    if value.locator != locator {
                        return Err(ExactWorkflowError::CoreShape);
                    }
                    session
                        .record(
                            &mut recovery,
                            ExactMutationTransition::SubmissionObserved {
                                disposition: if value.submitted {
                                    ExactSubmissionDispositionV1::Submitted
                                } else {
                                    ExactSubmissionDispositionV1::NotSubmitted
                                },
                            },
                        )
                        .await?;
                    if value.state == "applied" {
                        return finish_document(
                            &session,
                            &mut recovery,
                            DocumentMutationExactResultV1 {
                                outcome: ExactMutationOutcomeV1::Applied,
                                index: request.index,
                                locator: Some(value.locator),
                                state: value.state,
                                submitted: value.submitted,
                                sequence: value.sequence,
                                mapping_generation: value.mapping_generation,
                                mapping_adopted: value.mapping_adopted,
                            },
                        )
                        .await;
                    }
                }
                Err(error) => {
                    let disposition = ExactSubmissionDispositionV1::from(error.submission_disposition());
                    session
                        .record(
                            &mut recovery,
                            ExactMutationTransition::SubmissionObserved { disposition },
                        )
                        .await?;
                    if disposition == ExactSubmissionDispositionV1::NotSubmitted {
                        return Ok(execution(
                            document_status(&request, &recovery, ExactMutationOutcomeV1::NotSubmitted),
                            recovery,
                        ));
                    }
                }
            }
        }
        let Some(resolve_options) = resolution_options(now, &options)? else {
            return Ok(execution(
                document_status(&request, &recovery, ExactMutationOutcomeV1::Indeterminate),
                recovery,
            ));
        };
        let resolve_options = resolve_options
            .with_index_name(request.index.name.clone())
            .map_err(ExactWorkflowError::Client)?;
        let value = match self
            .resolve_document_mutation(
                ResolveDocumentMutationRequest {
                    locator: locator.clone(),
                },
                resolve_options,
            )
            .await
        {
            Ok(value) => value,
            Err(_) => {
                return Ok(execution(
                    document_status(&request, &recovery, ExactMutationOutcomeV1::Indeterminate),
                    recovery,
                ));
            }
        };
        if value.locator != locator || value.source.is_some() {
            return Err(ExactWorkflowError::CoreShape);
        }
        let outcome = match value.state.as_str() {
            "applied" => ExactMutationOutcomeV1::Resolved,
            "failed" => ExactMutationOutcomeV1::Resolved,
            "expired_unknown" => ExactMutationOutcomeV1::KnownRetired,
            "prepared" => ExactMutationOutcomeV1::NotSubmitted,
            "pending" => ExactMutationOutcomeV1::Indeterminate,
            _ => return Err(ExactWorkflowError::CoreShape),
        };
        let result = DocumentMutationExactResultV1 {
            outcome,
            index: request.index,
            locator: Some(value.locator),
            state: value.state,
            submitted: !matches!(outcome, ExactMutationOutcomeV1::NotSubmitted),
            sequence: value.sequence,
            mapping_generation: value.mapping_generation,
            mapping_adopted: value.mapping_adopted,
        };
        if matches!(
            outcome,
            ExactMutationOutcomeV1::Indeterminate | ExactMutationOutcomeV1::NotSubmitted
        ) {
            Ok(execution(result, recovery))
        } else {
            finish_document(&session, &mut recovery, result).await
        }
    }

    /// Reconciles only an absent index to one exact desired definition. An
    /// existing different definition is a conflict; this helper never updates.
    pub async fn reconcile_index_exact(
        &self,
        request: PrepareCreateIndexRequest,
        options: ExactWorkflowOptions,
    ) -> Result<ExactExecutionV1<ReconcileIndexExactResultV1>, ExactWorkflowError> {
        if request.name.is_empty() || request.resource_spec.is_none() {
            return Err(ExactWorkflowError::InvalidInput);
        }
        let request_sha256 = canonical_sha256(&request)?;
        let (cluster_id, collection_id) = self.exact_profile_scope();
        let intent = exact_intent(
            ExactMutationKindV1::Index,
            request_sha256.clone(),
            cluster_id,
            collection_id,
            None,
            request_sha256,
        );
        let (session, mut recovery) = open_session(&intent, &options).await?;
        let mut options = options;
        options.effect_deadline_unix_millis = recovery_effect_deadline(&recovery)?;
        if let Some(ExactTerminalResultV1::Index(result)) = &recovery.terminal {
            return Ok(execution(result.clone(), recovery));
        }
        let now = observe_clock(&session, &mut recovery, &options).await?;
        if clock_regressed(&recovery, now) {
            return Ok(execution(
                index_status(&request, &recovery, ExactMutationOutcomeV1::ClockRegression),
                recovery,
            ));
        }

        if recovery.preparation.is_none() {
            let Some(lookup_options) = resolution_options(now, &options)? else {
                let result = index_status(&request, &recovery, ExactMutationOutcomeV1::EffectDeadlineExpired);
                return finish_index(&session, &mut recovery, result).await;
            };
            let lookup = self
                .get_index(
                    GetIndexRequest {
                        name: request.name.clone(),
                        expected_incarnation: None,
                    },
                    lookup_options,
                )
                .await;
            match lookup {
                Ok(existing) => {
                    let exact = existing.resource.current_mapping == request.mapping
                        && request.resource_spec.as_ref() == Some(&existing.resource.resource_spec)
                        && existing.resource.summary.retained_event_identity
                            == request.retained_event_identity.unwrap_or(false);
                    let outcome = if exact && existing.resource.summary.state == "active" {
                        ExactMutationOutcomeV1::AlreadySatisfied
                    } else if exact {
                        ExactMutationOutcomeV1::Indeterminate
                    } else {
                        ExactMutationOutcomeV1::Conflict
                    };
                    let result = ReconcileIndexExactResultV1 {
                        outcome,
                        name: request.name,
                        locator: None,
                        resource: Some(Resource {
                            kind: "index".to_owned(),
                            name: existing.resource.summary.name,
                            incarnation: existing.resource.summary.incarnation,
                        }),
                        state: existing.resource.summary.state,
                    };
                    if outcome == ExactMutationOutcomeV1::Indeterminate {
                        return Ok(execution(result, recovery));
                    }
                    return finish_index(&session, &mut recovery, result).await;
                }
                Err(ClientError::Server(error)) if error.code == ErrorCode::NotFound => {}
                Err(error) => return Err(ExactWorkflowError::Client(error)),
            }
            if now >= options.effect_deadline_unix_millis {
                let result = index_status(&request, &recovery, ExactMutationOutcomeV1::EffectDeadlineExpired);
                return finish_index(&session, &mut recovery, result).await;
            }
            let prepared = self
                .prepare_create_index(
                    request.clone(),
                    effect_options(now, options.effect_deadline_unix_millis, &options)?,
                )
                .await
                .map_err(ExactWorkflowError::Client)?;
            if prepared.locator.authority.is_empty()
                || prepared.semantic_digest.validate().is_err()
                || prepared.resource_spec != *request.resource_spec.as_ref().ok_or(ExactWorkflowError::InvalidInput)?
            {
                return Err(ExactWorkflowError::CoreShape);
            }
            session
                .record(
                    &mut recovery,
                    ExactMutationTransition::Prepared {
                        preparation: ExactPreparationV1::Index {
                            locator: prepared.locator,
                            expires_at: prepared.expires_at,
                            semantic_digest: prepared.semantic_digest,
                            resource_spec: Box::new(prepared.resource_spec),
                        },
                    },
                )
                .await?;
        }
        let (locator, expires_at, semantic_digest, resource_spec) = index_preparation(&recovery)?;
        let expiry = parse_u64(&expires_at).ok_or(ExactWorkflowError::InvalidRecovery)?;
        let submit_deadline = options.effect_deadline_unix_millis.min(expiry);
        let may_submit = !matches!(
            recovery.submission_disposition,
            Some(
                ExactSubmissionDispositionV1::Submitted
                    | ExactSubmissionDispositionV1::Resolved
                    | ExactSubmissionDispositionV1::KnownRetired
                    | ExactSubmissionDispositionV1::Indeterminate
            )
        );
        if may_submit && now < submit_deadline {
            match self
                .submit_catalog_action(
                    SubmitCatalogActionRequest {
                        locator: locator.clone(),
                        semantic_digest,
                        action_kind: create_index_action_kind(request.retained_event_identity).to_owned(),
                        name: request.name.clone(),
                        mapping: Some(request.mapping.clone()),
                        resource_spec: Some(resource_spec),
                        expected_incarnation: None,
                    },
                    effect_options(now, submit_deadline, &options)?,
                )
                .await
            {
                Ok(value) => {
                    if value.locator != locator {
                        return Err(ExactWorkflowError::CoreShape);
                    }
                    session
                        .record(
                            &mut recovery,
                            ExactMutationTransition::SubmissionObserved {
                                disposition: if value.submitted {
                                    ExactSubmissionDispositionV1::Submitted
                                } else {
                                    ExactSubmissionDispositionV1::NotSubmitted
                                },
                            },
                        )
                        .await?;
                }
                Err(error) => {
                    let disposition = ExactSubmissionDispositionV1::from(error.submission_disposition());
                    session
                        .record(
                            &mut recovery,
                            ExactMutationTransition::SubmissionObserved { disposition },
                        )
                        .await?;
                    if disposition == ExactSubmissionDispositionV1::NotSubmitted {
                        return Ok(execution(
                            index_status(&request, &recovery, ExactMutationOutcomeV1::NotSubmitted),
                            recovery,
                        ));
                    }
                }
            }
        }
        let Some(resolve_options) = resolution_options(now, &options)? else {
            return Ok(execution(
                index_status(&request, &recovery, ExactMutationOutcomeV1::Indeterminate),
                recovery,
            ));
        };
        let resolve_options = resolve_options
            .with_index_name(request.name.clone())
            .map_err(ExactWorkflowError::Client)?;
        let value = match self
            .resolve_catalog_action(
                ResolveCatalogActionRequest {
                    locator: locator.clone(),
                },
                resolve_options,
            )
            .await
        {
            Ok(value) => value,
            Err(_) => {
                return Ok(execution(
                    index_status(&request, &recovery, ExactMutationOutcomeV1::Indeterminate),
                    recovery,
                ));
            }
        };
        if value.locator != locator {
            return Err(ExactWorkflowError::CoreShape);
        }
        let outcome = match value.state.as_str() {
            "applied" => ExactMutationOutcomeV1::Resolved,
            "failed" => ExactMutationOutcomeV1::Conflict,
            "expired_unknown" => ExactMutationOutcomeV1::KnownRetired,
            "prepared" => ExactMutationOutcomeV1::NotSubmitted,
            "pending" => ExactMutationOutcomeV1::Indeterminate,
            _ => return Err(ExactWorkflowError::CoreShape),
        };
        let mut result = ReconcileIndexExactResultV1 {
            outcome,
            name: request.name,
            locator: Some(value.locator),
            resource: value.resource,
            state: value.state,
        };
        if outcome == ExactMutationOutcomeV1::Resolved {
            let Some(verify_options) = resolution_options(now, &options)? else {
                result.outcome = ExactMutationOutcomeV1::Indeterminate;
                return Ok(execution(result, recovery));
            };
            let existing = match self
                .get_index(
                    GetIndexRequest {
                        name: result.name.clone(),
                        expected_incarnation: result.resource.as_ref().map(|value| value.incarnation.clone()),
                    },
                    verify_options,
                )
                .await
            {
                Ok(value) => value.resource,
                Err(_) => {
                    result.outcome = ExactMutationOutcomeV1::Indeterminate;
                    return Ok(execution(result, recovery));
                }
            };
            if existing.current_mapping != request.mapping
                || request.resource_spec.as_ref() != Some(&existing.resource_spec)
                || existing.summary.retained_event_identity != request.retained_event_identity.unwrap_or(false)
            {
                result.outcome = ExactMutationOutcomeV1::Conflict;
                result.state = "definition_conflict".to_owned();
            } else if existing.summary.state != "active" {
                result.outcome = ExactMutationOutcomeV1::Indeterminate;
                result.state = existing.summary.state;
                return Ok(execution(result, recovery));
            } else {
                result.resource = Some(Resource {
                    kind: "index".to_owned(),
                    name: existing.summary.name,
                    incarnation: existing.summary.incarnation,
                });
                result.state = "active".to_owned();
            }
        }
        if matches!(
            outcome,
            ExactMutationOutcomeV1::Indeterminate | ExactMutationOutcomeV1::NotSubmitted
        ) {
            Ok(execution(result, recovery))
        } else {
            finish_index(&session, &mut recovery, result).await
        }
    }

    fn exact_profile_scope(&self) -> (Identity, Identity) {
        let profile = self.connection_profile().as_wire();
        (
            profile.expected_cluster_id.clone(),
            profile.expected_collection_id.clone(),
        )
    }
}

async fn observe_append_visibility(
    client: &Client,
    request: &AppendExactRequestV1,
    mut result: AppendExactResultV1,
    options: CallOptions,
) -> Result<AppendExactResultV1, ExactWorkflowError> {
    let mut visibility = Vec::with_capacity(result.items.len());
    for item in &result.items {
        let Some(locator) = item.locator.clone() else {
            return Err(ExactWorkflowError::CoreShape);
        };
        match client
            .observe_append_committed(
                ObserveAppendCommittedRequest {
                    index: request.index.clone(),
                    locator,
                },
                options.clone(),
            )
            .await
        {
            Ok(value) => visibility.push(value.outcome),
            Err(_) => {
                result.outcome = ExactMutationOutcomeV1::Indeterminate;
                result.visibility = visibility;
                return Ok(result);
            }
        }
    }
    if visibility
        .iter()
        .any(|value| value.state != RetainedAppendObservationState::Visible)
    {
        result.outcome = ExactMutationOutcomeV1::Indeterminate;
    }
    result.visibility = visibility;
    Ok(result)
}

fn exact_intent(
    mutation_kind: ExactMutationKindV1,
    request_sha256: Digest,
    cluster_id: Identity,
    collection_id: Identity,
    index: Option<Resource>,
    caller_action_id: Digest,
) -> ExactMutationIntent {
    ExactMutationIntent {
        format_version: EXACT_MUTATION_FORMAT_VERSION,
        mutation_kind,
        request_sha256,
        cluster_id,
        collection_id,
        index,
        caller_action_id,
    }
}

async fn open_session(
    intent: &ExactMutationIntent,
    options: &ExactWorkflowOptions,
) -> Result<(ExactSession, ExactWorkflowRecoveryV1), ExactWorkflowError> {
    intent.validate()?;
    if options.effect_deadline_unix_millis == 0 || options.store.is_some() && options.recovery.is_some() {
        return Err(ExactWorkflowError::InvalidRecovery);
    }
    let (session, mut recovery) = if let Some(store) = &options.store {
        let replay = WorkflowJournal::open_or_create(store.path(), intent.clone()).await?;
        let (journal, transitions) = replay.into_parts();
        let mut recovery = recovery_from_intent(intent);
        for transition in transitions {
            reduce_transition(&mut recovery, &transition)?;
        }
        (ExactSession::Durable(journal), recovery)
    } else {
        let recovery = match &options.recovery {
            Some(value) => {
                validate_recovery_matches(value, intent)?;
                value.clone()
            }
            None => recovery_from_intent(intent),
        };
        (ExactSession::Ephemeral, recovery)
    };
    if parse_u64(&recovery.effect_deadline_unix_millis) == Some(0) {
        session
            .record(
                &mut recovery,
                ExactMutationTransition::DeadlineEstablished {
                    unix_millis: wire_u64(options.effect_deadline_unix_millis),
                },
            )
            .await?;
    }
    Ok((session, recovery))
}

fn recovery_from_intent(intent: &ExactMutationIntent) -> ExactWorkflowRecoveryV1 {
    ExactWorkflowRecoveryV1 {
        format_version: intent.format_version,
        mutation_kind: intent.mutation_kind,
        request_sha256: intent.request_sha256.clone(),
        cluster_id: intent.cluster_id.clone(),
        collection_id: intent.collection_id.clone(),
        index: intent.index.clone(),
        caller_action_id: intent.caller_action_id.clone(),
        effect_deadline_unix_millis: wire_u64(0),
        last_observed_unix_millis: None,
        preparation: None,
        submission_disposition: None,
        terminal: None,
    }
}

fn validate_recovery_matches(
    recovery: &ExactWorkflowRecoveryV1,
    intent: &ExactMutationIntent,
) -> Result<(), ExactWorkflowError> {
    let expected = recovery_from_intent(intent);
    if recovery.format_version != expected.format_version
        || recovery.mutation_kind != expected.mutation_kind
        || recovery.request_sha256 != expected.request_sha256
        || recovery.cluster_id != expected.cluster_id
        || recovery.collection_id != expected.collection_id
        || recovery.index != expected.index
        || recovery.caller_action_id != expected.caller_action_id
        || parse_u64(&recovery.effect_deadline_unix_millis).is_none_or(|value| value == 0)
        || recovery
            .last_observed_unix_millis
            .as_ref()
            .is_some_and(|value| parse_u64(value).is_none())
        || recovery
            .preparation
            .as_ref()
            .is_some_and(|value| !valid_preparation(value))
        || recovery
            .preparation
            .as_ref()
            .is_some_and(|value| preparation_kind(value) != recovery.mutation_kind)
        || recovery.terminal.as_ref().is_some_and(|value| !valid_terminal(value))
        || recovery
            .terminal
            .as_ref()
            .is_some_and(|value| terminal_kind(value) != recovery.mutation_kind)
        || recovery.submission_disposition.is_some() && recovery.preparation.is_none()
    {
        return Err(ExactWorkflowError::InvalidRecovery);
    }
    Ok(())
}

fn reduce_transition(
    state: &mut ExactWorkflowRecoveryV1,
    transition: &ExactMutationTransition,
) -> Result<(), ExactWorkflowError> {
    match transition {
        ExactMutationTransition::DeadlineEstablished { unix_millis }
            if parse_u64(&state.effect_deadline_unix_millis) == Some(0)
                && parse_u64(unix_millis).is_some_and(|value| value > 0) =>
        {
            state.effect_deadline_unix_millis = unix_millis.clone();
        }
        ExactMutationTransition::ClockObserved { unix_millis } => {
            let next = parse_u64(unix_millis).ok_or(ExactWorkflowError::InvalidRecovery)?;
            if state
                .last_observed_unix_millis
                .as_ref()
                .and_then(parse_u64)
                .is_some_and(|prior| next < prior)
            {
                return Err(ExactWorkflowError::InvalidRecovery);
            }
            state.last_observed_unix_millis = Some(unix_millis.clone());
        }
        ExactMutationTransition::Prepared { preparation }
            if state.preparation.is_none()
                && state.terminal.is_none()
                && preparation_kind(preparation) == state.mutation_kind =>
        {
            state.preparation = Some(preparation.clone());
        }
        ExactMutationTransition::SubmissionObserved { disposition }
            if state.preparation.is_some()
                && state.terminal.is_none()
                && !matches!(
                    state.submission_disposition,
                    Some(
                        ExactSubmissionDispositionV1::Submitted
                            | ExactSubmissionDispositionV1::Resolved
                            | ExactSubmissionDispositionV1::KnownRetired
                            | ExactSubmissionDispositionV1::Indeterminate
                    )
                ) =>
        {
            state.submission_disposition = Some(*disposition);
        }
        ExactMutationTransition::Terminal { result }
            if state.terminal.is_none() && terminal_kind(result) == state.mutation_kind =>
        {
            state.terminal = Some(result.clone());
        }
        _ => return Err(ExactWorkflowError::InvalidRecovery),
    }
    Ok(())
}

async fn observe_clock(
    session: &ExactSession,
    recovery: &mut ExactWorkflowRecoveryV1,
    options: &ExactWorkflowOptions,
) -> Result<u64, ExactWorkflowError> {
    let now = options.clock.unix_millis()?;
    if recovery
        .last_observed_unix_millis
        .as_ref()
        .and_then(parse_u64)
        .is_some_and(|prior| now < prior)
    {
        return Ok(now);
    }
    session
        .record(
            recovery,
            ExactMutationTransition::ClockObserved {
                unix_millis: wire_u64(now),
            },
        )
        .await?;
    Ok(now)
}

fn clock_regressed(recovery: &ExactWorkflowRecoveryV1, now: u64) -> bool {
    recovery
        .last_observed_unix_millis
        .as_ref()
        .and_then(parse_u64)
        .is_some_and(|observed| now < observed)
}

fn recovery_effect_deadline(recovery: &ExactWorkflowRecoveryV1) -> Result<u64, ExactWorkflowError> {
    parse_u64(&recovery.effect_deadline_unix_millis)
        .filter(|value| *value > 0)
        .ok_or(ExactWorkflowError::InvalidRecovery)
}

fn effect_options(now: u64, deadline: u64, options: &ExactWorkflowOptions) -> Result<CallOptions, ExactWorkflowError> {
    bounded_options(now, deadline, options)
}

fn resolution_options(now: u64, options: &ExactWorkflowOptions) -> Result<Option<CallOptions>, ExactWorkflowError> {
    let deadline = if now < options.effect_deadline_unix_millis {
        options.effect_deadline_unix_millis
    } else {
        match options.observation_deadline_unix_millis {
            Some(value) if value > now => value,
            _ => return Ok(None),
        }
    };
    bounded_options(now, deadline, options).map(Some)
}

fn bounded_options(now: u64, deadline: u64, options: &ExactWorkflowOptions) -> Result<CallOptions, ExactWorkflowError> {
    if deadline <= now {
        return Err(ExactWorkflowError::InvalidInput);
    }
    let bounded = deadline.min(now.saturating_add(MAX_EFFECT_CALL_MILLIS));
    CallOptions::from_absolute_deadline_unix_millis(bounded)
        .map(|value| value.with_cancellation(options.cancellation.clone()))
        .map_err(ExactWorkflowError::Client)
}

fn validate_append_request(request: &AppendExactRequestV1) -> Result<(), ExactWorkflowError> {
    if request.batch_id.validate().is_err()
        || !valid_index(&request.index)
        || request.items.is_empty()
        || request.items.len() > 1_000
        || request.items.iter().any(|item| {
            item.item_action_id.validate().is_err() || item.document_id.is_empty() || item.document_id.len() > 512
        })
    {
        return Err(ExactWorkflowError::InvalidInput);
    }
    let mut identities = std::collections::BTreeSet::new();
    if request
        .items
        .iter()
        .any(|item| !identities.insert(item.item_action_id.value.as_str()))
    {
        return Err(ExactWorkflowError::InvalidInput);
    }
    Ok(())
}

fn validate_append_outcome(
    request: &AppendExactRequestV1,
    outcome: &RetainedAppendBulkOutcome,
    prepared: bool,
) -> Result<(), ExactWorkflowError> {
    if outcome.batch_id != request.batch_id
        || outcome.index != request.index
        || outcome.items.len() != request.items.len()
    {
        return Err(ExactWorkflowError::CoreShape);
    }
    for (expected, observed) in request.items.iter().zip(&outcome.items) {
        if observed.item_action_id != expected.item_action_id
            || observed
                .locator
                .as_ref()
                .is_some_and(|value| value.authority.is_empty() || value.value.is_empty())
            || observed
                .semantic_digest
                .as_ref()
                .is_some_and(|value| value.validate().is_err())
            || prepared
                && (observed.state != RetainedAppendState::Prepared
                    || observed.retry_safety != RetainedAppendRetrySafety::Safe
                    || observed.locator.is_none()
                    || observed.semantic_digest.is_none()
                    || observed.sequence.is_some()
                    || observed.primary_term.is_some()
                    || observed.error.is_some())
        {
            return Err(ExactWorkflowError::CoreShape);
        }
    }
    Ok(())
}

fn redact_append_outcome(mut outcome: RetainedAppendBulkOutcome) -> RetainedAppendBulkOutcome {
    for item in &mut outcome.items {
        item.error = None;
    }
    outcome
}

fn append_preparation(recovery: &ExactWorkflowRecoveryV1) -> Result<&RetainedAppendBulkOutcome, ExactWorkflowError> {
    match &recovery.preparation {
        Some(ExactPreparationV1::Append(value)) => Ok(value),
        _ => Err(ExactWorkflowError::InvalidRecovery),
    }
}

fn document_preparation(recovery: &ExactWorkflowRecoveryV1) -> Result<(Locator, Digest), ExactWorkflowError> {
    match &recovery.preparation {
        Some(ExactPreparationV1::Document {
            locator,
            semantic_digest,
        }) => Ok((locator.clone(), semantic_digest.clone())),
        _ => Err(ExactWorkflowError::InvalidRecovery),
    }
}

fn index_preparation(
    recovery: &ExactWorkflowRecoveryV1,
) -> Result<(Locator, U64, Digest, IndexResourceSpec), ExactWorkflowError> {
    match &recovery.preparation {
        Some(ExactPreparationV1::Index {
            locator,
            expires_at,
            semantic_digest,
            resource_spec,
        }) => Ok((
            locator.clone(),
            expires_at.clone(),
            semantic_digest.clone(),
            resource_spec.as_ref().clone(),
        )),
        _ => Err(ExactWorkflowError::InvalidRecovery),
    }
}

fn append_terminal_result(
    request: &AppendExactRequestV1,
    outcome: &RetainedAppendBulkOutcome,
    recovered: bool,
) -> Option<AppendExactResultV1> {
    let state = if outcome
        .items
        .iter()
        .any(|item| item.state == RetainedAppendState::SubmittedIndeterminate)
    {
        return None;
    } else if outcome
        .items
        .iter()
        .any(|item| item.state == RetainedAppendState::Conflict)
    {
        ExactMutationOutcomeV1::Conflict
    } else if outcome
        .items
        .iter()
        .any(|item| item.state == RetainedAppendState::Retired)
    {
        ExactMutationOutcomeV1::KnownRetired
    } else if outcome.items.iter().any(|item| {
        matches!(
            item.state,
            RetainedAppendState::NotSubmittedRetryable | RetainedAppendState::NotSubmittedTerminal
        )
    }) {
        ExactMutationOutcomeV1::NotSubmitted
    } else if outcome.items.iter().all(|item| {
        matches!(
            item.state,
            RetainedAppendState::Committed | RetainedAppendState::Replayed
        )
    }) {
        if recovered {
            ExactMutationOutcomeV1::Resolved
        } else {
            ExactMutationOutcomeV1::Applied
        }
    } else {
        return None;
    };
    Some(append_result(request, outcome, state))
}

fn append_result(
    request: &AppendExactRequestV1,
    outcome: &RetainedAppendBulkOutcome,
    state: ExactMutationOutcomeV1,
) -> AppendExactResultV1 {
    AppendExactResultV1 {
        outcome: state,
        batch_id: request.batch_id.clone(),
        index: request.index.clone(),
        items: outcome.items.iter().map(redact_append_item).collect(),
        visibility: Vec::new(),
    }
}

fn append_status(
    request: &AppendExactRequestV1,
    recovery: &ExactWorkflowRecoveryV1,
    state: ExactMutationOutcomeV1,
) -> AppendExactResultV1 {
    let items = append_preparation(recovery)
        .map(|value| value.items.iter().map(redact_append_item).collect())
        .unwrap_or_default();
    AppendExactResultV1 {
        outcome: state,
        batch_id: request.batch_id.clone(),
        index: request.index.clone(),
        items,
        visibility: Vec::new(),
    }
}

fn redact_append_item(item: &RetainedAppendItemOutcome) -> AppendExactItemResultV1 {
    AppendExactItemResultV1 {
        item_action_id: item.item_action_id.clone(),
        locator: item.locator.clone(),
        semantic_digest: item.semantic_digest.clone(),
        state: item.state,
        retry_safety: item.retry_safety,
        sequence: item.sequence.clone(),
        primary_term: item.primary_term.clone(),
    }
}

fn document_status(
    request: &PrepareDocumentMutationRequest,
    recovery: &ExactWorkflowRecoveryV1,
    outcome: ExactMutationOutcomeV1,
) -> DocumentMutationExactResultV1 {
    let locator = document_preparation(recovery).map(|value| value.0).ok();
    DocumentMutationExactResultV1 {
        outcome,
        index: request.index.clone(),
        locator,
        state: outcome_name(outcome).to_owned(),
        submitted: recovery
            .submission_disposition
            .is_some_and(|value| !matches!(value, ExactSubmissionDispositionV1::NotSubmitted)),
        sequence: None,
        mapping_generation: None,
        mapping_adopted: false,
    }
}

fn index_status(
    request: &PrepareCreateIndexRequest,
    recovery: &ExactWorkflowRecoveryV1,
    outcome: ExactMutationOutcomeV1,
) -> ReconcileIndexExactResultV1 {
    let locator = match &recovery.preparation {
        Some(ExactPreparationV1::Index { locator, .. }) => Some(locator.clone()),
        _ => None,
    };
    ReconcileIndexExactResultV1 {
        outcome,
        name: request.name.clone(),
        locator,
        resource: None,
        state: outcome_name(outcome).to_owned(),
    }
}

const fn outcome_name(outcome: ExactMutationOutcomeV1) -> &'static str {
    match outcome {
        ExactMutationOutcomeV1::AlreadySatisfied => "already_satisfied",
        ExactMutationOutcomeV1::Applied => "applied",
        ExactMutationOutcomeV1::Resolved => "resolved",
        ExactMutationOutcomeV1::Conflict => "conflict",
        ExactMutationOutcomeV1::NotSubmitted => "not_submitted",
        ExactMutationOutcomeV1::Indeterminate => "indeterminate",
        ExactMutationOutcomeV1::KnownRetired => "known_retired",
        ExactMutationOutcomeV1::EffectDeadlineExpired => "effect_deadline_expired",
        ExactMutationOutcomeV1::ClockRegression => "clock_regression",
    }
}

async fn finish_append(
    session: &ExactSession,
    recovery: &mut ExactWorkflowRecoveryV1,
    result: AppendExactResultV1,
) -> Result<ExactExecutionV1<AppendExactResultV1>, ExactWorkflowError> {
    session
        .record(
            recovery,
            ExactMutationTransition::Terminal {
                result: ExactTerminalResultV1::Append(result.clone()),
            },
        )
        .await?;
    Ok(execution(result, recovery.clone()))
}

async fn finish_document(
    session: &ExactSession,
    recovery: &mut ExactWorkflowRecoveryV1,
    result: DocumentMutationExactResultV1,
) -> Result<ExactExecutionV1<DocumentMutationExactResultV1>, ExactWorkflowError> {
    session
        .record(
            recovery,
            ExactMutationTransition::Terminal {
                result: ExactTerminalResultV1::Document(result.clone()),
            },
        )
        .await?;
    Ok(execution(result, recovery.clone()))
}

async fn finish_index(
    session: &ExactSession,
    recovery: &mut ExactWorkflowRecoveryV1,
    result: ReconcileIndexExactResultV1,
) -> Result<ExactExecutionV1<ReconcileIndexExactResultV1>, ExactWorkflowError> {
    session
        .record(
            recovery,
            ExactMutationTransition::Terminal {
                result: ExactTerminalResultV1::Index(result.clone()),
            },
        )
        .await?;
    Ok(execution(result, recovery.clone()))
}

fn execution<T>(result: T, recovery: ExactWorkflowRecoveryV1) -> ExactExecutionV1<T> {
    ExactExecutionV1 { result, recovery }
}

fn valid_preparation(value: &ExactPreparationV1) -> bool {
    match value {
        ExactPreparationV1::Append(outcome) => {
            !outcome.items.is_empty()
                && outcome.items.len() <= 1_000
                && outcome.items.iter().all(|item| {
                    item.item_action_id.validate().is_ok()
                        && item
                            .locator
                            .as_ref()
                            .is_some_and(|locator| !locator.authority.is_empty() && !locator.value.is_empty())
                        && item
                            .semantic_digest
                            .as_ref()
                            .is_some_and(|digest| digest.validate().is_ok())
                        && item.state == RetainedAppendState::Prepared
                        && item.retry_safety == RetainedAppendRetrySafety::Safe
                        && item.error.is_none()
                })
        }
        ExactPreparationV1::Document {
            locator,
            semantic_digest,
        } => !locator.authority.is_empty() && !locator.value.is_empty() && semantic_digest.validate().is_ok(),
        ExactPreparationV1::Index {
            locator,
            expires_at,
            semantic_digest,
            ..
        } => {
            !locator.authority.is_empty()
                && !locator.value.is_empty()
                && parse_u64(expires_at).is_some_and(|value| value > 0)
                && semantic_digest.validate().is_ok()
        }
    }
}

fn valid_terminal(value: &ExactTerminalResultV1) -> bool {
    match value {
        ExactTerminalResultV1::Append(result) => {
            result.batch_id.validate().is_ok()
                && valid_index(&result.index)
                && terminal_outcome(result.outcome)
                && result.items.iter().all(|item| item.item_action_id.validate().is_ok())
        }
        ExactTerminalResultV1::Document(result) => {
            valid_index(&result.index)
                && result
                    .locator
                    .as_ref()
                    .is_none_or(|locator| !locator.authority.is_empty() && !locator.value.is_empty())
                && terminal_outcome(result.outcome)
        }
        ExactTerminalResultV1::Index(result) => !result.name.is_empty() && terminal_outcome(result.outcome),
    }
}

const fn terminal_outcome(value: ExactMutationOutcomeV1) -> bool {
    matches!(
        value,
        ExactMutationOutcomeV1::AlreadySatisfied
            | ExactMutationOutcomeV1::Applied
            | ExactMutationOutcomeV1::Resolved
            | ExactMutationOutcomeV1::Conflict
            | ExactMutationOutcomeV1::KnownRetired
            | ExactMutationOutcomeV1::EffectDeadlineExpired
    )
}

const fn preparation_kind(value: &ExactPreparationV1) -> ExactMutationKindV1 {
    match value {
        ExactPreparationV1::Append(_) => ExactMutationKindV1::Append,
        ExactPreparationV1::Document { .. } => ExactMutationKindV1::Document,
        ExactPreparationV1::Index { .. } => ExactMutationKindV1::Index,
    }
}

const fn terminal_kind(value: &ExactTerminalResultV1) -> ExactMutationKindV1 {
    match value {
        ExactTerminalResultV1::Append(_) => ExactMutationKindV1::Append,
        ExactTerminalResultV1::Document(_) => ExactMutationKindV1::Document,
        ExactTerminalResultV1::Index(_) => ExactMutationKindV1::Index,
    }
}

fn valid_index(value: &Resource) -> bool {
    value.kind == "index" && !value.name.is_empty() && value.name.len() <= 255 && value.incarnation.validate().is_ok()
}

fn canonical_sha256<T: Serialize>(value: &T) -> Result<Digest, ExactWorkflowError> {
    let bytes = serde_json::to_vec(value).map_err(|_| ExactWorkflowError::RequestCodec)?;
    let observed = digest::digest(&digest::SHA256, &bytes);
    Ok(Digest {
        value: lower_hex(observed.as_ref()),
    })
}

fn parse_u64(value: &U64) -> Option<u64> {
    if value.value.is_empty() || value.value.len() > 20 || value.value.starts_with('0') && value.value != "0" {
        return None;
    }
    value.value.parse().ok()
}

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

fn lower_hex(bytes: &[u8]) -> String {
    const HEX: &[u8; 16] = b"0123456789abcdef";
    let mut output = String::with_capacity(bytes.len().saturating_mul(2));
    for byte in bytes {
        output.push(char::from(HEX[usize::from(byte >> 4)]));
        output.push(char::from(HEX[usize::from(byte & 0x0f)]));
    }
    output
}

#[cfg(test)]
mod tests {
    use std::os::unix::fs::PermissionsExt as _;

    use super::*;

    fn identity(character: char) -> Identity {
        Identity {
            value: character.to_string().repeat(32),
        }
    }

    fn digest(character: char) -> Digest {
        Digest {
            value: character.to_string().repeat(64),
        }
    }

    fn resource() -> Resource {
        Resource {
            kind: "index".to_owned(),
            name: "events".to_owned(),
            incarnation: identity('a'),
        }
    }

    #[test]
    fn retained_event_identity_preserves_the_catalog_submit_kind() {
        assert_eq!(create_index_action_kind(None), "create_index");
        assert_eq!(create_index_action_kind(Some(false)), "create_index");
        assert_eq!(create_index_action_kind(Some(true)), "create_retained_append_log_index");
    }

    #[test]
    fn recovery_refuses_changed_request_or_scope_and_retains_first_deadline() -> Result<(), Box<dyn Error>> {
        let intent = exact_intent(
            ExactMutationKindV1::Append,
            digest('b'),
            identity('c'),
            identity('d'),
            Some(resource()),
            digest('e'),
        );
        let mut recovery = recovery_from_intent(&intent);
        reduce_transition(
            &mut recovery,
            &ExactMutationTransition::DeadlineEstablished {
                unix_millis: wire_u64(100),
            },
        )?;
        validate_recovery_matches(&recovery, &intent)?;
        assert_eq!(recovery_effect_deadline(&recovery)?, 100);
        let mut changed = intent.clone();
        changed.collection_id = identity('f');
        assert!(matches!(
            validate_recovery_matches(&recovery, &changed),
            Err(ExactWorkflowError::InvalidRecovery)
        ));
        changed = intent.clone();
        changed.request_sha256 = digest('f');
        assert!(matches!(
            validate_recovery_matches(&recovery, &changed),
            Err(ExactWorkflowError::InvalidRecovery)
        ));
        Ok(())
    }

    #[test]
    fn reducer_never_regresses_clock_or_replaces_possible_submission() -> Result<(), Box<dyn Error>> {
        let intent = exact_intent(
            ExactMutationKindV1::Append,
            digest('b'),
            identity('c'),
            identity('d'),
            Some(resource()),
            digest('e'),
        );
        let mut state = recovery_from_intent(&intent);
        reduce_transition(
            &mut state,
            &ExactMutationTransition::DeadlineEstablished {
                unix_millis: wire_u64(100),
            },
        )?;
        reduce_transition(
            &mut state,
            &ExactMutationTransition::ClockObserved {
                unix_millis: wire_u64(10),
            },
        )?;
        assert!(
            reduce_transition(
                &mut state,
                &ExactMutationTransition::ClockObserved {
                    unix_millis: wire_u64(9),
                },
            )
            .is_err()
        );
        state.preparation = Some(ExactPreparationV1::Append(RetainedAppendBulkOutcome {
            batch_id: digest('e'),
            index: resource(),
            items: vec![RetainedAppendItemOutcome {
                item_action_id: digest('f'),
                locator: Some(Locator {
                    authority: "retained_append".to_owned(),
                    value: "01".to_owned(),
                }),
                semantic_digest: Some(digest('1')),
                state: RetainedAppendState::Prepared,
                retry_safety: RetainedAppendRetrySafety::Safe,
                sequence: None,
                primary_term: None,
                error: None,
            }],
            complete: true,
        }));
        reduce_transition(
            &mut state,
            &ExactMutationTransition::SubmissionObserved {
                disposition: ExactSubmissionDispositionV1::Indeterminate,
            },
        )?;
        assert!(
            reduce_transition(
                &mut state,
                &ExactMutationTransition::SubmissionObserved {
                    disposition: ExactSubmissionDispositionV1::NotSubmitted,
                },
            )
            .is_err()
        );
        Ok(())
    }

    #[test]
    fn not_submitted_append_outcome_remains_recovery_resumable() -> Result<(), Box<dyn Error>> {
        let request = AppendExactRequestV1 {
            batch_id: digest('a'),
            index: resource(),
            items: vec![RetainedAppendItem {
                item_action_id: digest('b'),
                document_id: "event-1".to_owned(),
                source: lodestone_wire::RawJson {
                    bytes_base64: "e30=".to_owned(),
                },
            }],
            observe_committed: true,
        };
        let outcome = RetainedAppendBulkOutcome {
            batch_id: request.batch_id.clone(),
            index: request.index.clone(),
            items: vec![RetainedAppendItemOutcome {
                item_action_id: request.items[0].item_action_id.clone(),
                locator: Some(Locator {
                    authority: "retained_append".to_owned(),
                    value: "01".to_owned(),
                }),
                semantic_digest: Some(digest('c')),
                state: RetainedAppendState::NotSubmittedRetryable,
                retry_safety: RetainedAppendRetrySafety::Safe,
                sequence: None,
                primary_term: None,
                error: None,
            }],
            complete: false,
        };

        let result = append_terminal_result(&request, &outcome, false).ok_or("not-submitted observation absent")?;
        assert_eq!(result.outcome, ExactMutationOutcomeV1::NotSubmitted);
        assert!(!terminal_outcome(result.outcome));
        Ok(())
    }

    #[test]
    fn append_recovery_redacts_server_error_messages() {
        let outcome = RetainedAppendBulkOutcome {
            batch_id: digest('a'),
            index: resource(),
            items: vec![RetainedAppendItemOutcome {
                item_action_id: digest('b'),
                locator: Some(Locator {
                    authority: "retained_append".to_owned(),
                    value: "01".to_owned(),
                }),
                semantic_digest: Some(digest('c')),
                state: RetainedAppendState::Prepared,
                retry_safety: RetainedAppendRetrySafety::Safe,
                sequence: None,
                primary_term: None,
                error: None,
            }],
            complete: true,
        };
        assert!(redact_append_outcome(outcome).items[0].error.is_none());
    }

    #[tokio::test]
    async fn durable_resume_retains_the_first_effect_deadline() -> Result<(), Box<dyn Error>> {
        let mut suffix = [0_u8; 8];
        getrandom::fill(&mut suffix)?;
        let parent =
            std::fs::canonicalize(std::env::temp_dir())?.join(format!("lodestone-exact-{}", lower_hex(&suffix)));
        std::fs::create_dir(&parent)?;
        std::fs::set_permissions(&parent, std::fs::Permissions::from_mode(0o700))?;
        let store = ExactWorkflowStore::new(parent.join("journal"));
        let intent = exact_intent(
            ExactMutationKindV1::Append,
            digest('b'),
            identity('c'),
            identity('d'),
            Some(resource()),
            digest('e'),
        );
        let first = ExactWorkflowOptions::new(100).with_store(store.clone());
        let (session, recovery) = open_session(&intent, &first).await?;
        assert_eq!(recovery_effect_deadline(&recovery)?, 100);
        drop(session);

        let resumed = ExactWorkflowOptions::new(999).with_store(store);
        let (session, recovery) = open_session(&intent, &resumed).await?;
        assert_eq!(recovery_effect_deadline(&recovery)?, 100);
        drop(session);
        std::fs::remove_dir_all(&parent)?;
        Ok(())
    }
}