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::future::Future;
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};

use aws_lc_rs::digest;
use lodestone_wire::{
    Digest, ErrorCode, GetIndexRequest, Identity, IndexResourceSpec, Locator, MutationOutcome,
    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 = 2;
pub const DEFAULT_EXACT_WORKFLOW_STORE_CONFLICT_LIMIT: u32 = 8;
const MAX_EFFECT_CALL_MILLIS: u64 = 599_000;
const MAX_SAFE_REVISION: u64 = 9_007_199_254_740_991;

/// 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,
}

/// Durable progress of one exact caller-side mutation workflow.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ExactWorkflowPhaseV2 {
    Initialized,
    Prepared,
    SubmissionStarted,
    SubmissionObserved,
    Terminal,
}

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,
        }
    }
}

fn exact_submission_disposition(outcome: &MutationOutcome) -> ExactSubmissionDispositionV1 {
    match outcome {
        MutationOutcome::Terminal => ExactSubmissionDispositionV1::Resolved,
        MutationOutcome::Pending { .. } => ExactSubmissionDispositionV1::Submitted,
        MutationOutcome::Indeterminate { .. } | MutationOutcome::ResolutionRequired { .. } => {
            ExactSubmissionDispositionV1::Indeterminate
        }
    }
}

fn resolution_disposition(outcome: ExactMutationOutcomeV1) -> ExactSubmissionDispositionV1 {
    match outcome {
        ExactMutationOutcomeV1::NotSubmitted | ExactMutationOutcomeV1::EffectDeadlineExpired => {
            ExactSubmissionDispositionV1::NotSubmitted
        }
        ExactMutationOutcomeV1::KnownRetired => ExactSubmissionDispositionV1::KnownRetired,
        ExactMutationOutcomeV1::Indeterminate | ExactMutationOutcomeV1::ClockRegression => {
            ExactSubmissionDispositionV1::Indeterminate
        }
        ExactMutationOutcomeV1::AlreadySatisfied
        | ExactMutationOutcomeV1::Applied
        | ExactMutationOutcomeV1::Resolved
        | ExactMutationOutcomeV1::Conflict => ExactSubmissionDispositionV1::Resolved,
    }
}

fn append_resolution_disposition(outcome: &RetainedAppendBulkOutcome) -> ExactSubmissionDispositionV1 {
    if outcome
        .items
        .iter()
        .any(|item| item.state == RetainedAppendState::SubmittedIndeterminate)
    {
        ExactSubmissionDispositionV1::Indeterminate
    } else if outcome
        .items
        .iter()
        .any(|item| item.state == RetainedAppendState::Retired)
    {
        ExactSubmissionDispositionV1::KnownRetired
    } else if outcome.items.iter().any(|item| {
        matches!(
            item.state,
            RetainedAppendState::Prepared
                | RetainedAppendState::NotSubmittedRetryable
                | RetainedAppendState::NotSubmittedTerminal
        )
    }) {
        ExactSubmissionDispositionV1::NotSubmitted
    } else {
        ExactSubmissionDispositionV1::Resolved
    }
}

/// 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 ExactWorkflowRecoveryV2 {
    pub format_version: u16,
    pub revision: u64,
    pub phase: ExactWorkflowPhaseV2,
    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 ExactExecutionV2<T> {
    pub result: T,
    pub recovery: ExactWorkflowRecoveryV2,
}

/// Boxed future used by the object-safe caller-owned recovery-store boundary.
pub type ExactWorkflowStoreFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;

/// Closed failures returned by a caller-owned exact-workflow store.
#[derive(Clone, Debug, PartialEq)]
pub enum ExactWorkflowStoreError {
    Unavailable,
    Conflict {
        observed: Option<Box<ExactWorkflowRecoveryV2>>,
    },
    Malformed,
    UnsupportedFormat {
        observed: u16,
    },
    IntentMismatch,
    Internal,
}

impl Display for ExactWorkflowStoreError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter.write_str(match self {
            Self::Unavailable => "the exact workflow recovery store is unavailable",
            Self::Conflict { .. } => "the exact workflow recovery store compare-and-exchange conflicted",
            Self::Malformed => "the exact workflow recovery store returned malformed current state",
            Self::UnsupportedFormat { .. } => "the exact workflow recovery store returned an unsupported format",
            Self::IntentMismatch => "the exact workflow recovery store key belongs to another intent",
            Self::Internal => "the exact workflow recovery store failed internally",
        })
    }
}

impl Error for ExactWorkflowStoreError {}

/// Linearizable durable CAS scoped by the caller to one exact workflow key.
///
/// The implementation owns storage mechanics only. Lodestone validates every
/// loaded value and computes every legal successor before invoking the store.
pub trait ExactWorkflowRecoveryStore: Send + Sync {
    fn load(&self) -> ExactWorkflowStoreFuture<'_, Result<Option<ExactWorkflowRecoveryV2>, ExactWorkflowStoreError>>;

    fn compare_exchange(
        &self,
        expected_revision: Option<u64>,
        next: ExactWorkflowRecoveryV2,
    ) -> ExactWorkflowStoreFuture<'_, Result<(), ExactWorkflowStoreError>>;
}

/// Filesystem-backed adapter over Lodestone's sealed create-new workflow
/// journal. The directory is exclusive to one exact request intent.
#[derive(Clone)]
pub struct FilesystemExactWorkflowRecoveryStore {
    path: PathBuf,
    session: Arc<tokio::sync::Mutex<Option<FilesystemExactWorkflowSession>>>,
}

struct FilesystemExactWorkflowSession {
    journal: WorkflowJournal<ExactMutationIntent>,
    recovery: ExactWorkflowRecoveryV2,
}

impl fmt::Debug for FilesystemExactWorkflowRecoveryStore {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("FilesystemExactWorkflowRecoveryStore")
            .field("path", &self.path)
            .finish()
    }
}

impl FilesystemExactWorkflowRecoveryStore {
    #[must_use]
    pub fn new(path: impl Into<PathBuf>) -> Self {
        Self {
            path: path.into(),
            session: Arc::new(tokio::sync::Mutex::new(None)),
        }
    }

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

impl ExactWorkflowRecoveryStore for FilesystemExactWorkflowRecoveryStore {
    fn load(&self) -> ExactWorkflowStoreFuture<'_, Result<Option<ExactWorkflowRecoveryV2>, ExactWorkflowStoreError>> {
        Box::pin(async move {
            let mut session = self.session.lock().await;
            if let Some(current) = session.as_ref() {
                return Ok(Some(current.recovery.clone()));
            }
            let Some(opened) = open_filesystem_recovery(&self.path).await? else {
                return Ok(None);
            };
            let recovery = opened.recovery.clone();
            *session = Some(opened);
            Ok(Some(recovery))
        })
    }

    fn compare_exchange(
        &self,
        expected_revision: Option<u64>,
        next: ExactWorkflowRecoveryV2,
    ) -> ExactWorkflowStoreFuture<'_, Result<(), ExactWorkflowStoreError>> {
        Box::pin(async move {
            let mut session = self.session.lock().await;
            if session.is_none() {
                *session = open_filesystem_recovery(&self.path).await?;
            }
            match (session.as_mut(), expected_revision) {
                (Some(current), Some(expected)) if current.recovery.revision == expected => {
                    let transition = transition_between(&current.recovery, &next)?;
                    current
                        .journal
                        .append(transition)
                        .await
                        .map_err(map_journal_store_error)?;
                    current.recovery = next;
                    Ok(())
                }
                (Some(current), _) => Err(ExactWorkflowStoreError::Conflict {
                    observed: Some(Box::new(current.recovery.clone())),
                }),
                (None, Some(_)) => Err(ExactWorkflowStoreError::Conflict { observed: None }),
                (None, None) => {
                    validate_initial_recovery(&next)?;
                    let intent = intent_from_recovery(&next);
                    let replay = WorkflowJournal::open_or_create(&self.path, intent)
                        .await
                        .map_err(map_journal_store_error)?;
                    let created = replay.was_created();
                    let opened = filesystem_recovery_from_replay(replay)?;
                    if !created {
                        let observed = opened.recovery.clone();
                        *session = Some(opened);
                        return Err(ExactWorkflowStoreError::Conflict {
                            observed: Some(Box::new(observed)),
                        });
                    }
                    if opened.recovery != next {
                        return Err(ExactWorkflowStoreError::Malformed);
                    }
                    *session = Some(opened);
                    Ok(())
                }
            }
        })
    }
}

async fn open_filesystem_recovery(
    path: &Path,
) -> Result<Option<FilesystemExactWorkflowSession>, ExactWorkflowStoreError> {
    let replay = WorkflowJournal::<ExactMutationIntent>::open_existing(path)
        .await
        .map_err(map_journal_store_error)?;
    replay.map(filesystem_recovery_from_replay).transpose()
}

fn filesystem_recovery_from_replay(
    replay: crate::WorkflowJournalReplay<ExactMutationIntent>,
) -> Result<FilesystemExactWorkflowSession, ExactWorkflowStoreError> {
    let mut recovery = recovery_from_intent(replay.intent());
    let (journal, transitions) = replay.into_parts();
    for transition in transitions {
        reduce_transition(&mut recovery, &transition).map_err(|_| ExactWorkflowStoreError::Malformed)?;
    }
    validate_recovery_shape(&recovery).map_err(|_| ExactWorkflowStoreError::Malformed)?;
    Ok(FilesystemExactWorkflowSession { journal, recovery })
}

const fn map_journal_store_error(error: WorkflowJournalError) -> ExactWorkflowStoreError {
    match error {
        WorkflowJournalError::UnsupportedFormat(observed) => ExactWorkflowStoreError::UnsupportedFormat { observed },
        WorkflowJournalError::IntentMismatch => ExactWorkflowStoreError::IntentMismatch,
        WorkflowJournalError::InvalidIntent
        | WorkflowJournalError::InvalidTransition
        | WorkflowJournalError::Corrupt
        | WorkflowJournalError::Codec
        | WorkflowJournalError::LimitExceeded => ExactWorkflowStoreError::Malformed,
        WorkflowJournalError::UnsupportedPlatform
        | WorkflowJournalError::InvalidPath
        | WorkflowJournalError::InsecurePath
        | WorkflowJournalError::Busy
        | WorkflowJournalError::Io
        | WorkflowJournalError::RandomUnavailable
        | WorkflowJournalError::ReopenRequired
        | WorkflowJournalError::WorkerUnavailable => ExactWorkflowStoreError::Unavailable,
    }
}

/// 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<Arc<dyn ExactWorkflowRecoveryStore>>,
    recovery: Option<ExactWorkflowRecoveryV2>,
    clock: Arc<dyn ExactWorkflowClock>,
    cancellation: CallCancellation,
    store_conflict_limit: u32,
}

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(),
            store_conflict_limit: DEFAULT_EXACT_WORKFLOW_STORE_CONFLICT_LIMIT,
        }
    }

    #[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_recovery_store(mut self, store: Arc<dyn ExactWorkflowRecoveryStore>) -> Self {
        self.store = Some(store);
        self
    }

    #[must_use]
    pub fn with_store_conflict_limit(mut self, value: u32) -> Self {
        self.store_conflict_limit = value;
        self
    }

    #[must_use]
    pub fn with_recovery(mut self, recovery: ExactWorkflowRecoveryV2) -> 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),
    Store {
        source: ExactWorkflowStoreError,
        last_durable_recovery: Option<Box<ExactWorkflowRecoveryV2>>,
    },
    StoreConflictBudgetExhausted {
        last_durable_recovery: Box<ExactWorkflowRecoveryV2>,
    },
    Cancelled {
        last_durable_recovery: Box<ExactWorkflowRecoveryV2>,
    },
    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::Store { .. } => "the exact mutation recovery store failed",
            Self::StoreConflictBudgetExhausted { .. } => {
                "the exact mutation recovery store conflict budget was exhausted"
            }
            Self::Cancelled { .. } => "the exact mutation workflow was cancelled",
            Self::Client(_) => "the exact mutation client call failed",
        })
    }
}

impl Error for ExactWorkflowError {}

impl ExactWorkflowError {
    #[must_use]
    pub fn last_durable_recovery(&self) -> Option<&ExactWorkflowRecoveryV2> {
        match self {
            Self::Store {
                last_durable_recovery, ..
            } => last_durable_recovery.as_deref(),
            Self::StoreConflictBudgetExhausted { last_durable_recovery }
            | Self::Cancelled { last_durable_recovery } => Some(last_durable_recovery),
            _ => None,
        }
    }
}

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,
    effect_deadline_unix_millis: U64,
}

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()
            || parse_u64(&self.effect_deadline_unix_millis).is_none_or(|value| value == 0)
            || 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 {
    ClockObserved { unix_millis: U64 },
    Prepared { preparation: ExactPreparationV1 },
    SubmissionStarted,
    SubmissionObserved { disposition: ExactSubmissionDispositionV1 },
    Terminal { result: ExactTerminalResultV1 },
}

impl sealed::Sealed for ExactMutationTransition {}

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

enum ExactSession {
    Durable {
        store: Arc<dyn ExactWorkflowRecoveryStore>,
        conflict_limit: u32,
        deadline_unix_millis: u64,
        clock: Arc<dyn ExactWorkflowClock>,
        cancellation: CallCancellation,
    },
    Ephemeral,
}

impl ExactSession {
    async fn record(
        &self,
        state: &mut ExactWorkflowRecoveryV2,
        transition: ExactMutationTransition,
    ) -> Result<bool, ExactWorkflowError> {
        let Self::Durable {
            store,
            conflict_limit,
            deadline_unix_millis,
            clock,
            cancellation,
        } = self
        else {
            reduce_transition(state, &transition)?;
            return Ok(true);
        };
        let mut conflicts = 0_u32;
        loop {
            if cancellation.is_cancelled() {
                return Err(ExactWorkflowError::Cancelled {
                    last_durable_recovery: Box::new(state.clone()),
                });
            }
            if transition_is_reflected(state, &transition) {
                return Ok(false);
            }
            if conflicts > 0 && clock.unix_millis()? >= *deadline_unix_millis {
                return Err(ExactWorkflowError::StoreConflictBudgetExhausted {
                    last_durable_recovery: Box::new(state.clone()),
                });
            }
            let mut next = state.clone();
            reduce_transition(&mut next, &transition)?;
            match store.compare_exchange(Some(state.revision), next.clone()).await {
                Ok(()) => {
                    *state = next;
                    return Ok(true);
                }
                Err(ExactWorkflowStoreError::Conflict {
                    observed: Some(observed),
                }) => {
                    let observed = *observed;
                    validate_recovery_descendant(state, &observed)?;
                    *state = observed;
                    if transition_is_reflected(state, &transition) {
                        return Ok(false);
                    }
                    conflicts = conflicts.saturating_add(1);
                    if conflicts > *conflict_limit {
                        return Err(ExactWorkflowError::StoreConflictBudgetExhausted {
                            last_durable_recovery: Box::new(state.clone()),
                        });
                    }
                }
                Err(ExactWorkflowStoreError::Conflict { observed: None }) => {
                    return Err(store_failure(ExactWorkflowStoreError::Malformed, Some(state.clone())));
                }
                Err(error) => return Err(store_failure(error, Some(state.clone()))),
            }
        }
    }
}

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<ExactExecutionV2<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(),
            options.effect_deadline_unix_millis,
        );
        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 let Some(ExactTerminalResultV1::Append(result)) = &recovery.terminal {
            return Ok(execution(result.clone(), recovery));
        }
        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?;
            if let Some(ExactTerminalResultV1::Append(result)) = &recovery.terminal {
                return Ok(execution(result.clone(), recovery));
            }
        }

        let preparations = append_preparation(&recovery)?.items.clone();
        let may_submit = recovery.phase == ExactWorkflowPhaseV2::Prepared
            || recovery.submission_disposition == Some(ExactSubmissionDispositionV1::NotSubmitted);
        if may_submit && now < options.effect_deadline_unix_millis {
            let owns_submission = session
                .record(&mut recovery, ExactMutationTransition::SubmissionStarted)
                .await?;
            if let Some(ExactTerminalResultV1::Append(result)) = &recovery.terminal {
                return Ok(execution(result.clone(), recovery));
            }
            if owns_submission {
                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
                        };
                        let published = session
                            .record(
                                &mut recovery,
                                ExactMutationTransition::SubmissionObserved { disposition },
                            )
                            .await?;
                        if let Some(ExactTerminalResultV1::Append(result)) = &recovery.terminal {
                            return Ok(execution(result.clone(), recovery));
                        }
                        if published
                            && 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 let Some(ExactTerminalResultV1::Append(result)) = &recovery.terminal {
                            return Ok(execution(result.clone(), recovery));
                        }
                        if recovery.submission_disposition == Some(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)?;
        record_resolution_observation(&session, &mut recovery, append_resolution_disposition(&outcome)).await?;
        if let Some(ExactTerminalResultV1::Append(result)) = &recovery.terminal {
            return Ok(execution(result.clone(), recovery));
        }
        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<ExactExecutionV2<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,
            options.effect_deadline_unix_millis,
        );
        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 let Some(ExactTerminalResultV1::Document(result)) = &recovery.terminal {
            return Ok(execution(result.clone(), recovery));
        }
        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?;
            if let Some(ExactTerminalResultV1::Document(result)) = &recovery.terminal {
                return Ok(execution(result.clone(), recovery));
            }
        }
        let (locator, semantic_digest) = document_preparation(&recovery)?;
        let may_submit = recovery.phase == ExactWorkflowPhaseV2::Prepared
            || recovery.submission_disposition == Some(ExactSubmissionDispositionV1::NotSubmitted);
        if may_submit && now < options.effect_deadline_unix_millis {
            let owns_submission = session
                .record(&mut recovery, ExactMutationTransition::SubmissionStarted)
                .await?;
            if let Some(ExactTerminalResultV1::Document(result)) = &recovery.terminal {
                return Ok(execution(result.clone(), recovery));
            }
            if owns_submission {
                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(),
                            record: request.record.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);
                        }
                        let published = session
                            .record(
                                &mut recovery,
                                ExactMutationTransition::SubmissionObserved {
                                    disposition: exact_submission_disposition(&value.mutation_outcome),
                                },
                            )
                            .await?;
                        if let Some(ExactTerminalResultV1::Document(result)) = &recovery.terminal {
                            return Ok(execution(result.clone(), recovery));
                        }
                        if published
                            && matches!(&value.mutation_outcome, MutationOutcome::Terminal)
                            && matches!(value.state.as_str(), "applied" | "dropped")
                        {
                            let submitted = value.state == "applied";
                            return finish_document(
                                &session,
                                &mut recovery,
                                DocumentMutationExactResultV1 {
                                    outcome: ExactMutationOutcomeV1::Applied,
                                    index: request.index,
                                    locator: Some(value.locator),
                                    state: value.state,
                                    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 let Some(ExactTerminalResultV1::Document(result)) = &recovery.terminal {
                            return Ok(execution(result.clone(), recovery));
                        }
                        if recovery.submission_disposition == Some(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),
        };
        record_resolution_observation(&session, &mut recovery, resolution_disposition(outcome)).await?;
        if let Some(ExactTerminalResultV1::Document(result)) = &recovery.terminal {
            return Ok(execution(result.clone(), recovery));
        }
        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<ExactExecutionV2<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,
            options.effect_deadline_unix_millis,
        );
        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 let Some(ExactTerminalResultV1::Index(result)) = &recovery.terminal {
            return Ok(execution(result.clone(), recovery));
        }
        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?;
            if let Some(ExactTerminalResultV1::Index(result)) = &recovery.terminal {
                return Ok(execution(result.clone(), recovery));
            }
        }
        let (locator, _expires_at, semantic_digest, resource_spec) = index_preparation(&recovery)?;
        // INVARIANT: preparation expiry belongs to the catalog authority's
        // logical clock. The caller clock owns only its immutable effect
        // deadline; the authority alone decides whether this preparation is
        // still live when submission reaches it.
        let submit_deadline = options.effect_deadline_unix_millis;
        let submission = 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,
        };
        if may_start_catalog_submission(&recovery, now, submit_deadline) {
            let owns_submission = session
                .record(&mut recovery, ExactMutationTransition::SubmissionStarted)
                .await?;
            if let Some(ExactTerminalResultV1::Index(result)) = &recovery.terminal {
                return Ok(execution(result.clone(), recovery));
            }
            if owns_submission {
                match self
                    .submit_catalog_action(submission.clone(), effect_options(now, submit_deadline, &options)?)
                    .await
                {
                    Ok(value) => {
                        if value.locator != locator {
                            return Err(ExactWorkflowError::CoreShape);
                        }
                        session
                            .record(
                                &mut recovery,
                                ExactMutationTransition::SubmissionObserved {
                                    disposition: exact_submission_disposition(&value.mutation_outcome),
                                },
                            )
                            .await?;
                        if let Some(ExactTerminalResultV1::Index(result)) = &recovery.terminal {
                            return Ok(execution(result.clone(), recovery));
                        }
                    }
                    Err(error) => {
                        let disposition = ExactSubmissionDispositionV1::from(error.submission_disposition());
                        session
                            .record(
                                &mut recovery,
                                ExactMutationTransition::SubmissionObserved { disposition },
                            )
                            .await?;
                        if let Some(ExactTerminalResultV1::Index(result)) = &recovery.terminal {
                            return Ok(execution(result.clone(), recovery));
                        }
                        if recovery.submission_disposition == Some(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 mut value = match self
            .resolve_catalog_action(
                ResolveCatalogActionRequest {
                    locator: locator.clone(),
                },
                resolve_options.clone(),
            )
            .await
        {
            Ok(value) => value,
            Err(_) => {
                return Ok(execution(
                    index_status(&request, &recovery, ExactMutationOutcomeV1::Indeterminate),
                    recovery,
                ));
            }
        };
        if value.locator != locator {
            return Err(ExactWorkflowError::CoreShape);
        }
        let exact_replay =
            catalog_resolution_requires_exact_replay(value.state.as_str(), recovery.submission_disposition);
        if exact_replay {
            // The owner has proved that this exact effect is already durably
            // admitted. Calling the same endpoint resumes resolution; it does
            // not authorize a first submission or extend the effect identity.
            match self.submit_catalog_action(submission, resolve_options.clone()).await {
                Ok(replayed) if replayed.locator == locator => {}
                Ok(_) => return Err(ExactWorkflowError::CoreShape),
                Err(error) if error.submission_disposition() == SubmissionDisposition::NotSubmitted => {
                    return Ok(execution(
                        index_status(&request, &recovery, ExactMutationOutcomeV1::NotSubmitted),
                        recovery,
                    ));
                }
                Err(_) => {
                    return Ok(execution(
                        index_status(&request, &recovery, ExactMutationOutcomeV1::Indeterminate),
                        recovery,
                    ));
                }
            }
            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),
        };
        record_resolution_observation(&session, &mut recovery, resolution_disposition(outcome)).await?;
        if let Some(ExactTerminalResultV1::Index(result)) = &recovery.terminal {
            return Ok(execution(result.clone(), recovery));
        }
        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(),
        )
    }
}

fn catalog_resolution_requires_exact_replay(state: &str, submission: Option<ExactSubmissionDispositionV1>) -> bool {
    state == "pending"
        || state == "prepared"
            && matches!(
                submission,
                Some(
                    ExactSubmissionDispositionV1::Submitted
                        | ExactSubmissionDispositionV1::Resolved
                        | ExactSubmissionDispositionV1::Indeterminate
                )
            )
}

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,
    effect_deadline_unix_millis: u64,
) -> ExactMutationIntent {
    ExactMutationIntent {
        format_version: EXACT_MUTATION_FORMAT_VERSION,
        mutation_kind,
        request_sha256,
        cluster_id,
        collection_id,
        index,
        caller_action_id,
        effect_deadline_unix_millis: wire_u64(effect_deadline_unix_millis),
    }
}

async fn open_session(
    intent: &ExactMutationIntent,
    options: &ExactWorkflowOptions,
) -> Result<(ExactSession, ExactWorkflowRecoveryV2), ExactWorkflowError> {
    intent.validate()?;
    if options.effect_deadline_unix_millis == 0 || options.store.is_some() && options.recovery.is_some() {
        return Err(ExactWorkflowError::InvalidRecovery);
    }
    let (session, recovery) = if let Some(store) = &options.store {
        let initial = recovery_from_intent(intent);
        let loaded = store.load().await.map_err(|error| store_failure(error, None))?;
        let recovery = match loaded {
            Some(value) => {
                validate_store_recovery(&value, intent)?;
                value
            }
            None => match store.compare_exchange(None, initial.clone()).await {
                Ok(()) => initial,
                Err(ExactWorkflowStoreError::Conflict {
                    observed: Some(observed),
                }) => {
                    let observed = *observed;
                    validate_store_recovery(&observed, intent)?;
                    observed
                }
                Err(ExactWorkflowStoreError::Conflict { observed: None }) => {
                    return Err(store_failure(ExactWorkflowStoreError::Malformed, None));
                }
                Err(error) => return Err(store_failure(error, None)),
            },
        };
        let conflict_deadline = options
            .observation_deadline_unix_millis
            .unwrap_or(options.effect_deadline_unix_millis);
        (
            ExactSession::Durable {
                store: Arc::clone(store),
                conflict_limit: options.store_conflict_limit,
                deadline_unix_millis: conflict_deadline,
                clock: Arc::clone(&options.clock),
                cancellation: options.cancellation.clone(),
            },
            recovery,
        )
    } else {
        let recovery = match &options.recovery {
            Some(value) => {
                validate_recovery_matches(value, intent)?;
                value.clone()
            }
            None => recovery_from_intent(intent),
        };
        (ExactSession::Ephemeral, recovery)
    };
    Ok((session, recovery))
}

fn recovery_from_intent(intent: &ExactMutationIntent) -> ExactWorkflowRecoveryV2 {
    ExactWorkflowRecoveryV2 {
        format_version: intent.format_version,
        revision: 0,
        phase: ExactWorkflowPhaseV2::Initialized,
        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: intent.effect_deadline_unix_millis.clone(),
        last_observed_unix_millis: None,
        preparation: None,
        submission_disposition: None,
        terminal: None,
    }
}

fn validate_recovery_matches(
    recovery: &ExactWorkflowRecoveryV2,
    intent: &ExactMutationIntent,
) -> Result<(), ExactWorkflowError> {
    let expected = recovery_from_intent(intent);
    validate_recovery_shape(recovery)?;
    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
    {
        return Err(ExactWorkflowError::InvalidRecovery);
    }
    Ok(())
}

fn reduce_transition(
    state: &mut ExactWorkflowRecoveryV2,
    transition: &ExactMutationTransition,
) -> Result<(), ExactWorkflowError> {
    if state.revision >= MAX_SAFE_REVISION || state.phase == ExactWorkflowPhaseV2::Terminal {
        return Err(ExactWorkflowError::InvalidRecovery);
    }
    match transition {
        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.phase == ExactWorkflowPhaseV2::Initialized
                && preparation_kind(preparation) == state.mutation_kind =>
        {
            state.preparation = Some(preparation.clone());
            state.phase = ExactWorkflowPhaseV2::Prepared;
        }
        ExactMutationTransition::SubmissionStarted if state.preparation.is_some() && state.terminal.is_none() => {
            if state.phase < ExactWorkflowPhaseV2::SubmissionStarted {
                state.phase = ExactWorkflowPhaseV2::SubmissionStarted;
            } else if state.submission_disposition == Some(ExactSubmissionDispositionV1::NotSubmitted) {
                // INVARIANT: before a retry that follows proof of no effect,
                // clear that proof to the conservative may-have-submitted
                // state before the transport can observe the request.
                state.submission_disposition = Some(ExactSubmissionDispositionV1::Indeterminate);
            } else {
                return Err(ExactWorkflowError::InvalidRecovery);
            }
        }
        ExactMutationTransition::SubmissionObserved { disposition }
            if state.preparation.is_some()
                && state.terminal.is_none()
                && (state.phase >= ExactWorkflowPhaseV2::SubmissionStarted
                    || state.phase == ExactWorkflowPhaseV2::Prepared
                        && *disposition == ExactSubmissionDispositionV1::NotSubmitted)
                && valid_disposition_successor(state.submission_disposition, *disposition) =>
        {
            state.submission_disposition = Some(*disposition);
            state.phase = ExactWorkflowPhaseV2::SubmissionObserved;
        }
        ExactMutationTransition::Terminal { result }
            if state.terminal.is_none() && terminal_kind(result) == state.mutation_kind =>
        {
            state.terminal = Some(result.clone());
            state.phase = ExactWorkflowPhaseV2::Terminal;
        }
        _ => return Err(ExactWorkflowError::InvalidRecovery),
    }
    state.revision = state
        .revision
        .checked_add(1)
        .filter(|value| *value <= MAX_SAFE_REVISION)
        .ok_or(ExactWorkflowError::InvalidRecovery)?;
    validate_recovery_shape(state)?;
    Ok(())
}

fn validate_recovery_shape(recovery: &ExactWorkflowRecoveryV2) -> Result<(), ExactWorkflowError> {
    let preparation_valid = recovery
        .preparation
        .as_ref()
        .is_none_or(|value| valid_preparation(value) && preparation_kind(value) == recovery.mutation_kind);
    let terminal_valid = recovery
        .terminal
        .as_ref()
        .is_none_or(|value| valid_terminal(value) && terminal_kind(value) == recovery.mutation_kind);
    let phase_valid = match recovery.phase {
        ExactWorkflowPhaseV2::Initialized => {
            recovery.preparation.is_none() && recovery.submission_disposition.is_none() && recovery.terminal.is_none()
        }
        ExactWorkflowPhaseV2::Prepared | ExactWorkflowPhaseV2::SubmissionStarted => {
            recovery.preparation.is_some() && recovery.submission_disposition.is_none() && recovery.terminal.is_none()
        }
        ExactWorkflowPhaseV2::SubmissionObserved => {
            recovery.preparation.is_some() && recovery.submission_disposition.is_some() && recovery.terminal.is_none()
        }
        ExactWorkflowPhaseV2::Terminal => recovery.terminal.is_some(),
    };
    if recovery.format_version != EXACT_MUTATION_FORMAT_VERSION
        || recovery.revision > MAX_SAFE_REVISION
        || recovery.request_sha256.validate().is_err()
        || recovery.cluster_id.validate().is_err()
        || recovery.collection_id.validate().is_err()
        || recovery.caller_action_id.validate().is_err()
        || recovery.index.as_ref().is_some_and(|value| !valid_index(value))
        || 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())
        || !preparation_valid
        || !terminal_valid
        || !phase_valid
        || recovery.revision == 0
            && (recovery.phase != ExactWorkflowPhaseV2::Initialized || recovery.last_observed_unix_millis.is_some())
    {
        return Err(ExactWorkflowError::InvalidRecovery);
    }
    Ok(())
}

const fn valid_disposition_successor(
    current: Option<ExactSubmissionDispositionV1>,
    next: ExactSubmissionDispositionV1,
) -> bool {
    match current {
        None | Some(ExactSubmissionDispositionV1::NotSubmitted) => true,
        Some(ExactSubmissionDispositionV1::Submitted) => !matches!(next, ExactSubmissionDispositionV1::NotSubmitted),
        Some(ExactSubmissionDispositionV1::Indeterminate) => true,
        Some(ExactSubmissionDispositionV1::Resolved) => matches!(next, ExactSubmissionDispositionV1::Resolved),
        Some(ExactSubmissionDispositionV1::KnownRetired) => {
            matches!(next, ExactSubmissionDispositionV1::KnownRetired)
        }
    }
}

fn validate_initial_recovery(recovery: &ExactWorkflowRecoveryV2) -> Result<(), ExactWorkflowStoreError> {
    validate_recovery_shape(recovery).map_err(|_| ExactWorkflowStoreError::Malformed)?;
    if recovery.revision != 0 || recovery.phase != ExactWorkflowPhaseV2::Initialized {
        return Err(ExactWorkflowStoreError::Malformed);
    }
    Ok(())
}

fn intent_from_recovery(recovery: &ExactWorkflowRecoveryV2) -> ExactMutationIntent {
    ExactMutationIntent {
        format_version: recovery.format_version,
        mutation_kind: recovery.mutation_kind,
        request_sha256: recovery.request_sha256.clone(),
        cluster_id: recovery.cluster_id.clone(),
        collection_id: recovery.collection_id.clone(),
        index: recovery.index.clone(),
        caller_action_id: recovery.caller_action_id.clone(),
        effect_deadline_unix_millis: recovery.effect_deadline_unix_millis.clone(),
    }
}

fn transition_between(
    current: &ExactWorkflowRecoveryV2,
    next: &ExactWorkflowRecoveryV2,
) -> Result<ExactMutationTransition, ExactWorkflowStoreError> {
    validate_recovery_shape(current).map_err(|_| ExactWorkflowStoreError::Malformed)?;
    validate_recovery_shape(next).map_err(|_| ExactWorkflowStoreError::Malformed)?;
    if next.revision
        != current
            .revision
            .checked_add(1)
            .ok_or(ExactWorkflowStoreError::Malformed)?
        || !same_recovery_intent(current, next)
    {
        return Err(ExactWorkflowStoreError::Malformed);
    }
    let retry_started = current.submission_disposition == Some(ExactSubmissionDispositionV1::NotSubmitted)
        && next.submission_disposition == Some(ExactSubmissionDispositionV1::Indeterminate)
        && current.phase == next.phase;
    let candidates = [
        next.last_observed_unix_millis
            .as_ref()
            .filter(|value| current.last_observed_unix_millis.as_ref() != Some(*value))
            .map(|value| ExactMutationTransition::ClockObserved {
                unix_millis: value.clone(),
            }),
        next.preparation
            .as_ref()
            .filter(|value| current.preparation.as_ref() != Some(*value))
            .map(|value| ExactMutationTransition::Prepared {
                preparation: value.clone(),
            }),
        (retry_started
            || current.phase < ExactWorkflowPhaseV2::SubmissionStarted
                && next.phase >= ExactWorkflowPhaseV2::SubmissionStarted)
            .then_some(ExactMutationTransition::SubmissionStarted),
        next.submission_disposition
            .filter(|_| !retry_started)
            .map(|disposition| ExactMutationTransition::SubmissionObserved { disposition }),
        next.terminal
            .as_ref()
            .filter(|value| current.terminal.as_ref() != Some(*value))
            .map(|value| ExactMutationTransition::Terminal { result: value.clone() }),
    ];
    let mut matched = None;
    for candidate in candidates.into_iter().flatten() {
        let mut reduced = current.clone();
        if reduce_transition(&mut reduced, &candidate).is_ok() && reduced == *next {
            if matched.is_some() {
                return Err(ExactWorkflowStoreError::Malformed);
            }
            matched = Some(candidate);
        }
    }
    matched.ok_or(ExactWorkflowStoreError::Malformed)
}

fn validate_store_recovery(
    recovery: &ExactWorkflowRecoveryV2,
    intent: &ExactMutationIntent,
) -> Result<(), ExactWorkflowError> {
    if recovery.format_version != EXACT_MUTATION_FORMAT_VERSION {
        return Err(store_failure(
            ExactWorkflowStoreError::UnsupportedFormat {
                observed: recovery.format_version,
            },
            Some(recovery.clone()),
        ));
    }
    validate_recovery_shape(recovery)
        .map_err(|_| store_failure(ExactWorkflowStoreError::Malformed, Some(recovery.clone())))?;
    if !recovery_matches_intent(recovery, intent) {
        return Err(store_failure(
            ExactWorkflowStoreError::IntentMismatch,
            Some(recovery.clone()),
        ));
    }
    Ok(())
}

fn validate_recovery_descendant(
    current: &ExactWorkflowRecoveryV2,
    observed: &ExactWorkflowRecoveryV2,
) -> Result<(), ExactWorkflowError> {
    if observed.format_version != EXACT_MUTATION_FORMAT_VERSION {
        return Err(store_failure(
            ExactWorkflowStoreError::UnsupportedFormat {
                observed: observed.format_version,
            },
            Some(current.clone()),
        ));
    }
    validate_recovery_shape(observed)
        .map_err(|_| store_failure(ExactWorkflowStoreError::Malformed, Some(current.clone())))?;
    let clock_monotonic = current.last_observed_unix_millis.as_ref().is_none_or(|prior| {
        observed
            .last_observed_unix_millis
            .as_ref()
            .and_then(parse_u64)
            .zip(parse_u64(prior))
            .is_some_and(|(next, prior)| next >= prior)
    });
    let disposition_monotonic = current.submission_disposition.is_none_or(|prior| {
        observed
            .submission_disposition
            .is_some_and(|next| valid_disposition_successor(Some(prior), next))
    });
    if !same_recovery_intent(current, observed) {
        return Err(store_failure(
            ExactWorkflowStoreError::IntentMismatch,
            Some(current.clone()),
        ));
    }
    if observed.revision <= current.revision
        || observed.phase < current.phase
        || !clock_monotonic
        || current
            .preparation
            .as_ref()
            .is_some_and(|value| observed.preparation.as_ref() != Some(value))
        || !disposition_monotonic
        || current
            .terminal
            .as_ref()
            .is_some_and(|value| observed.terminal.as_ref() != Some(value))
    {
        return Err(store_failure(ExactWorkflowStoreError::Malformed, Some(current.clone())));
    }
    Ok(())
}

fn transition_is_reflected(state: &ExactWorkflowRecoveryV2, transition: &ExactMutationTransition) -> bool {
    if state.terminal.is_some() {
        // INVARIANT: the first CAS-committed terminal result is authoritative.
        // A losing replica must adopt it instead of attempting to overwrite it
        // or returning a result derived from stale local observations.
        return true;
    }
    match transition {
        ExactMutationTransition::ClockObserved { unix_millis } => parse_u64(unix_millis).is_some_and(|requested| {
            state
                .last_observed_unix_millis
                .as_ref()
                .and_then(parse_u64)
                .is_some_and(|observed| observed >= requested)
        }),
        ExactMutationTransition::Prepared { preparation } => state.preparation.as_ref() == Some(preparation),
        ExactMutationTransition::SubmissionStarted => {
            state.preparation.is_some()
                && (state.phase == ExactWorkflowPhaseV2::SubmissionStarted
                    || state.phase >= ExactWorkflowPhaseV2::SubmissionObserved
                        && state.submission_disposition != Some(ExactSubmissionDispositionV1::NotSubmitted))
        }
        ExactMutationTransition::SubmissionObserved { disposition } => state
            .submission_disposition
            .is_some_and(|observed| valid_disposition_successor(Some(*disposition), observed)),
        ExactMutationTransition::Terminal { .. } => false,
    }
}

fn same_recovery_intent(left: &ExactWorkflowRecoveryV2, right: &ExactWorkflowRecoveryV2) -> bool {
    left.format_version == right.format_version
        && left.mutation_kind == right.mutation_kind
        && left.request_sha256 == right.request_sha256
        && left.cluster_id == right.cluster_id
        && left.collection_id == right.collection_id
        && left.index == right.index
        && left.caller_action_id == right.caller_action_id
        && left.effect_deadline_unix_millis == right.effect_deadline_unix_millis
}

fn recovery_matches_intent(recovery: &ExactWorkflowRecoveryV2, intent: &ExactMutationIntent) -> bool {
    recovery.format_version == intent.format_version
        && recovery.mutation_kind == intent.mutation_kind
        && recovery.request_sha256 == intent.request_sha256
        && recovery.cluster_id == intent.cluster_id
        && recovery.collection_id == intent.collection_id
        && recovery.index == intent.index
        && recovery.caller_action_id == intent.caller_action_id
}

fn store_failure(source: ExactWorkflowStoreError, recovery: Option<ExactWorkflowRecoveryV2>) -> ExactWorkflowError {
    ExactWorkflowError::Store {
        source,
        last_durable_recovery: recovery.map(Box::new),
    }
}

async fn record_resolution_observation(
    session: &ExactSession,
    recovery: &mut ExactWorkflowRecoveryV2,
    disposition: ExactSubmissionDispositionV1,
) -> Result<(), ExactWorkflowError> {
    // INVARIANT: a reflected disposition is as authoritative as one committed
    // by this caller. Only SubmissionStarted ownership controls whether this
    // caller may perform an effect; resolution and visibility remain reads.
    session
        .record(recovery, ExactMutationTransition::SubmissionObserved { disposition })
        .await
        .map(|_| ())
}

async fn observe_clock(
    session: &ExactSession,
    recovery: &mut ExactWorkflowRecoveryV2,
    options: &ExactWorkflowOptions,
) -> Result<u64, ExactWorkflowError> {
    let now = options.clock.unix_millis()?;
    // INVARIANT: once submission may have happened, later wall-clock samples
    // cannot fence another effect. Only an authoritative not-submitted result
    // reopens submission, so visibility polling must not consume the bounded
    // durable journal with liveness-only clock observations.
    if recovery.phase >= ExactWorkflowPhaseV2::SubmissionStarted
        && recovery.submission_disposition != Some(ExactSubmissionDispositionV1::NotSubmitted)
    {
        return Ok(now);
    }
    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: &ExactWorkflowRecoveryV2, 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: &ExactWorkflowRecoveryV2) -> 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: &ExactWorkflowRecoveryV2) -> Result<&RetainedAppendBulkOutcome, ExactWorkflowError> {
    match &recovery.preparation {
        Some(ExactPreparationV1::Append(value)) => Ok(value),
        _ => Err(ExactWorkflowError::InvalidRecovery),
    }
}

fn document_preparation(recovery: &ExactWorkflowRecoveryV2) -> 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: &ExactWorkflowRecoveryV2,
) -> 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 may_start_catalog_submission(
    recovery: &ExactWorkflowRecoveryV2,
    now_unix_millis: u64,
    effect_deadline_unix_millis: u64,
) -> bool {
    // The preparation's authority-time expiry is deliberately absent here.
    // It cannot be compared with the caller's Unix clock.
    (recovery.phase == ExactWorkflowPhaseV2::Prepared
        || recovery.submission_disposition == Some(ExactSubmissionDispositionV1::NotSubmitted))
        && now_unix_millis < effect_deadline_unix_millis
}

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: &ExactWorkflowRecoveryV2,
    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: &ExactWorkflowRecoveryV2,
    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: &ExactWorkflowRecoveryV2,
    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 ExactWorkflowRecoveryV2,
    result: AppendExactResultV1,
) -> Result<ExactExecutionV2<AppendExactResultV1>, ExactWorkflowError> {
    session
        .record(
            recovery,
            ExactMutationTransition::Terminal {
                result: ExactTerminalResultV1::Append(result.clone()),
            },
        )
        .await?;
    let durable = match &recovery.terminal {
        Some(ExactTerminalResultV1::Append(durable)) => durable.clone(),
        _ => return Err(ExactWorkflowError::InvalidRecovery),
    };
    Ok(execution(durable, recovery.clone()))
}

async fn finish_document(
    session: &ExactSession,
    recovery: &mut ExactWorkflowRecoveryV2,
    result: DocumentMutationExactResultV1,
) -> Result<ExactExecutionV2<DocumentMutationExactResultV1>, ExactWorkflowError> {
    session
        .record(
            recovery,
            ExactMutationTransition::Terminal {
                result: ExactTerminalResultV1::Document(result.clone()),
            },
        )
        .await?;
    let durable = match &recovery.terminal {
        Some(ExactTerminalResultV1::Document(durable)) => durable.clone(),
        _ => return Err(ExactWorkflowError::InvalidRecovery),
    };
    Ok(execution(durable, recovery.clone()))
}

async fn finish_index(
    session: &ExactSession,
    recovery: &mut ExactWorkflowRecoveryV2,
    result: ReconcileIndexExactResultV1,
) -> Result<ExactExecutionV2<ReconcileIndexExactResultV1>, ExactWorkflowError> {
    session
        .record(
            recovery,
            ExactMutationTransition::Terminal {
                result: ExactTerminalResultV1::Index(result.clone()),
            },
        )
        .await?;
    let durable = match &recovery.terminal {
        Some(ExactTerminalResultV1::Index(durable)) => durable.clone(),
        _ => return Err(ExactWorkflowError::InvalidRecovery),
    };
    Ok(execution(durable, recovery.clone()))
}

fn execution<T>(result: T, recovery: ExactWorkflowRecoveryV2) -> ExactExecutionV2<T> {
    ExactExecutionV2 { 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 std::sync::Mutex;

    use super::*;

    #[derive(Default)]
    struct PostgresShapedRecoveryStore {
        row: Mutex<Option<ExactWorkflowRecoveryV2>>,
    }

    impl ExactWorkflowRecoveryStore for PostgresShapedRecoveryStore {
        fn load(
            &self,
        ) -> ExactWorkflowStoreFuture<'_, Result<Option<ExactWorkflowRecoveryV2>, ExactWorkflowStoreError>> {
            Box::pin(async move {
                self.row
                    .lock()
                    .map(|row| row.clone())
                    .map_err(|_| ExactWorkflowStoreError::Internal)
            })
        }

        fn compare_exchange(
            &self,
            expected_revision: Option<u64>,
            next: ExactWorkflowRecoveryV2,
        ) -> ExactWorkflowStoreFuture<'_, Result<(), ExactWorkflowStoreError>> {
            Box::pin(async move {
                let mut row = self.row.lock().map_err(|_| ExactWorkflowStoreError::Internal)?;
                let observed_revision = row.as_ref().map(|value| value.revision);
                if observed_revision != expected_revision {
                    return Err(ExactWorkflowStoreError::Conflict {
                        observed: row.clone().map(Box::new),
                    });
                }
                *row = Some(next);
                Ok(())
            })
        }
    }

    struct AdvancingConflictStore {
        row: Mutex<ExactWorkflowRecoveryV2>,
    }

    struct UnavailableRecoveryStore;

    struct AdvancingClock {
        next: Mutex<u64>,
    }

    impl ExactWorkflowClock for AdvancingClock {
        fn unix_millis(&self) -> Result<u64, ExactWorkflowError> {
            let mut next = self.next.lock().map_err(|_| ExactWorkflowError::ClockUnavailable)?;
            let observed = *next;
            *next = next.saturating_add(1);
            Ok(observed)
        }
    }

    impl ExactWorkflowRecoveryStore for UnavailableRecoveryStore {
        fn load(
            &self,
        ) -> ExactWorkflowStoreFuture<'_, Result<Option<ExactWorkflowRecoveryV2>, ExactWorkflowStoreError>> {
            Box::pin(async { Err(ExactWorkflowStoreError::Unavailable) })
        }

        fn compare_exchange(
            &self,
            _expected_revision: Option<u64>,
            _next: ExactWorkflowRecoveryV2,
        ) -> ExactWorkflowStoreFuture<'_, Result<(), ExactWorkflowStoreError>> {
            Box::pin(async { Err(ExactWorkflowStoreError::Unavailable) })
        }
    }

    impl ExactWorkflowRecoveryStore for AdvancingConflictStore {
        fn load(
            &self,
        ) -> ExactWorkflowStoreFuture<'_, Result<Option<ExactWorkflowRecoveryV2>, ExactWorkflowStoreError>> {
            Box::pin(async move {
                self.row
                    .lock()
                    .map(|row| Some(row.clone()))
                    .map_err(|_| ExactWorkflowStoreError::Internal)
            })
        }

        fn compare_exchange(
            &self,
            _expected_revision: Option<u64>,
            _next: ExactWorkflowRecoveryV2,
        ) -> ExactWorkflowStoreFuture<'_, Result<(), ExactWorkflowStoreError>> {
            Box::pin(async move {
                let mut row = self.row.lock().map_err(|_| ExactWorkflowStoreError::Internal)?;
                let next_clock = row
                    .last_observed_unix_millis
                    .as_ref()
                    .and_then(parse_u64)
                    .unwrap_or(0)
                    .saturating_add(1);
                reduce_transition(
                    &mut row,
                    &ExactMutationTransition::ClockObserved {
                        unix_millis: wire_u64(next_clock),
                    },
                )
                .map_err(|_| ExactWorkflowStoreError::Internal)?;
                Err(ExactWorkflowStoreError::Conflict {
                    observed: Some(Box::new(row.clone())),
                })
            })
        }
    }

    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'),
        }
    }

    fn append_preparation_value() -> ExactPreparationV1 {
        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,
            achievement: None,
        })
    }

    fn index_preparation_value(expires_at: u64) -> ExactPreparationV1 {
        ExactPreparationV1::Index {
            locator: Locator {
                authority: "catalog_confirmation".to_owned(),
                value: "logical-clock-preparation".to_owned(),
            },
            expires_at: wire_u64(expires_at),
            semantic_digest: digest('8'),
            resource_spec: Box::new(IndexResourceSpec {
                format_version: u32::from(lodestone_wire::LODESTONE_API_INDEX_RESOURCE_SPEC_FORMAT_VERSION),
                revision: wire_u64(1),
                initial_shards: 1,
                target_shard_bytes: wire_u64(1),
                durability: lodestone_wire::ResourceDurabilityClass::RegionalObject,
                storage_class: lodestone_wire::ResourceStorageClass::Standard,
                ram_admission_weight: 1,
                ram_hard_max_bytes: wire_u64(1),
                nvme_admission_weight: 1,
                nvme_hard_max_bytes: wire_u64(1),
                refresh_objective_millis: 1,
                memtable_hard_max_bytes: wire_u64(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: true,
                resolver_capability: None,
            }),
        }
    }

    #[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 catalog_recovery_replays_only_an_exact_prepared_or_pending_identity() {
        assert!(catalog_resolution_requires_exact_replay("pending", None));
        assert!(catalog_resolution_requires_exact_replay(
            "prepared",
            Some(ExactSubmissionDispositionV1::Indeterminate),
        ));
        assert!(catalog_resolution_requires_exact_replay(
            "prepared",
            Some(ExactSubmissionDispositionV1::Submitted),
        ));
        assert!(!catalog_resolution_requires_exact_replay("prepared", None));
        assert!(!catalog_resolution_requires_exact_replay(
            "prepared",
            Some(ExactSubmissionDispositionV1::NotSubmitted),
        ));
        assert!(!catalog_resolution_requires_exact_replay(
            "expired_unknown",
            Some(ExactSubmissionDispositionV1::Indeterminate),
        ));
    }

    #[test]
    fn catalog_logical_expiry_never_clamps_the_client_unix_deadline() -> Result<(), Box<dyn Error>> {
        const CLIENT_NOW: u64 = 1_793_100_501_940;
        const EFFECT_DEADLINE: u64 = CLIENT_NOW + 10_000;
        const AUTHORITY_LOGICAL_EXPIRY: u64 = 8_390_001;

        let intent = exact_intent(
            ExactMutationKindV1::Index,
            digest('b'),
            identity('c'),
            identity('d'),
            None,
            digest('e'),
            EFFECT_DEADLINE,
        );
        let mut prepared = recovery_from_intent(&intent);
        reduce_transition(
            &mut prepared,
            &ExactMutationTransition::ClockObserved {
                unix_millis: wire_u64(CLIENT_NOW),
            },
        )?;
        reduce_transition(
            &mut prepared,
            &ExactMutationTransition::Prepared {
                preparation: index_preparation_value(AUTHORITY_LOGICAL_EXPIRY),
            },
        )?;

        let restarted: ExactWorkflowRecoveryV2 = serde_json::from_slice(&serde_json::to_vec(&prepared)?)?;
        assert_eq!(index_preparation(&restarted)?.1, wire_u64(AUTHORITY_LOGICAL_EXPIRY));
        assert!(may_start_catalog_submission(
            &restarted,
            CLIENT_NOW + 5_000,
            EFFECT_DEADLINE,
        ));
        assert!(!may_start_catalog_submission(
            &restarted,
            EFFECT_DEADLINE,
            EFFECT_DEADLINE,
        ));

        let mut uncertain = restarted.clone();
        reduce_transition(&mut uncertain, &ExactMutationTransition::SubmissionStarted)?;
        reduce_transition(
            &mut uncertain,
            &ExactMutationTransition::SubmissionObserved {
                disposition: ExactSubmissionDispositionV1::Indeterminate,
            },
        )?;
        assert!(!may_start_catalog_submission(
            &uncertain,
            CLIENT_NOW + 5_000,
            EFFECT_DEADLINE,
        ));

        let mut proven_not_submitted = restarted;
        reduce_transition(&mut proven_not_submitted, &ExactMutationTransition::SubmissionStarted)?;
        reduce_transition(
            &mut proven_not_submitted,
            &ExactMutationTransition::SubmissionObserved {
                disposition: ExactSubmissionDispositionV1::NotSubmitted,
            },
        )?;
        assert!(may_start_catalog_submission(
            &proven_not_submitted,
            CLIENT_NOW + 5_000,
            EFFECT_DEADLINE,
        ));
        Ok(())
    }

    #[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'),
            100,
        );
        let recovery = recovery_from_intent(&intent);
        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'),
            100,
        );
        let mut state = recovery_from_intent(&intent);
        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()
        );
        let preparation = 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,
            achievement: None,
        });
        reduce_transition(
            &mut state,
            &ExactMutationTransition::Prepared {
                preparation: preparation.clone(),
            },
        )?;
        reduce_transition(&mut state, &ExactMutationTransition::SubmissionStarted)?;
        reduce_transition(
            &mut state,
            &ExactMutationTransition::SubmissionObserved {
                disposition: ExactSubmissionDispositionV1::Indeterminate,
            },
        )?;
        assert!(reduce_transition(&mut state, &ExactMutationTransition::Prepared { preparation },).is_err());
        Ok(())
    }

    #[test]
    fn reducer_refuses_revision_overflow_without_changing_state() {
        let intent = exact_intent(
            ExactMutationKindV1::Append,
            digest('b'),
            identity('c'),
            identity('d'),
            Some(resource()),
            digest('e'),
            100,
        );
        let mut state = recovery_from_intent(&intent);
        state.revision = MAX_SAFE_REVISION;
        let unchanged = state.clone();
        assert!(
            reduce_transition(
                &mut state,
                &ExactMutationTransition::ClockObserved {
                    unix_millis: wire_u64(1),
                },
            )
            .is_err()
        );
        assert_eq!(state, unchanged);
    }

    #[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(),
                record: lodestone_wire::NativeRecord::Document {
                    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,
            achievement: None,
        };

        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,
            achievement: None,
        };
        assert!(redact_append_outcome(outcome).items[0].error.is_none());
    }

    #[tokio::test]
    async fn caller_owned_cas_converges_two_replicas_across_every_phase() -> Result<(), Box<dyn Error>> {
        let store: Arc<dyn ExactWorkflowRecoveryStore> = Arc::new(PostgresShapedRecoveryStore::default());
        let intent = exact_intent(
            ExactMutationKindV1::Append,
            digest('b'),
            identity('c'),
            identity('d'),
            Some(resource()),
            digest('e'),
            10_000,
        );
        let options = ExactWorkflowOptions::new(10_000)
            .with_observation_deadline_unix_millis(20_000)
            .with_recovery_store(Arc::clone(&store));
        let (left_session, mut left) = open_session(&intent, &options).await?;
        let (right_session, mut right) = open_session(&intent, &options).await?;
        assert_eq!(left.revision, 0);
        assert_eq!(right.revision, 0);

        assert!(
            left_session
                .record(
                    &mut left,
                    ExactMutationTransition::ClockObserved {
                        unix_millis: wire_u64(1),
                    },
                )
                .await?
        );
        assert!(
            !right_session
                .record(
                    &mut right,
                    ExactMutationTransition::ClockObserved {
                        unix_millis: wire_u64(1),
                    },
                )
                .await?
        );

        let preparation = append_preparation_value();
        assert!(
            left_session
                .record(
                    &mut left,
                    ExactMutationTransition::Prepared {
                        preparation: preparation.clone(),
                    },
                )
                .await?
        );
        assert!(
            !right_session
                .record(&mut right, ExactMutationTransition::Prepared { preparation },)
                .await?
        );

        assert!(
            left_session
                .record(&mut left, ExactMutationTransition::SubmissionStarted)
                .await?
        );
        assert!(
            !right_session
                .record(&mut right, ExactMutationTransition::SubmissionStarted)
                .await?
        );

        let observation = ExactMutationTransition::SubmissionObserved {
            disposition: ExactSubmissionDispositionV1::Resolved,
        };
        assert!(left_session.record(&mut left, observation.clone()).await?);
        assert!(!right_session.record(&mut right, observation).await?);

        let result = ExactTerminalResultV1::Append(AppendExactResultV1 {
            outcome: ExactMutationOutcomeV1::Resolved,
            batch_id: digest('e'),
            index: resource(),
            items: Vec::new(),
            visibility: Vec::new(),
        });
        assert!(
            left_session
                .record(&mut left, ExactMutationTransition::Terminal { result: result.clone() },)
                .await?
        );
        let losing = AppendExactResultV1 {
            outcome: ExactMutationOutcomeV1::Conflict,
            batch_id: digest('e'),
            index: resource(),
            items: Vec::new(),
            visibility: Vec::new(),
        };
        let adopted = finish_append(&right_session, &mut right, losing).await?;
        assert_eq!(adopted.result.outcome, ExactMutationOutcomeV1::Resolved);
        assert_eq!(left, right);
        assert_eq!(left.phase, ExactWorkflowPhaseV2::Terminal);
        assert_eq!(left.revision, 5);
        assert_eq!(store.load().await?, Some(left));
        Ok(())
    }

    #[test]
    fn rust_recovery_v2_matches_the_cross_language_fixture() -> Result<(), Box<dyn Error>> {
        let fixture: serde_json::Value = serde_json::from_str(include_str!(
            "../../../clients/native/fixtures/lodestone-exact-workflow-recovery-v2.json"
        ))?;
        if fixture.get("format_version").and_then(serde_json::Value::as_u64) != Some(2) {
            return Err("exact recovery fixture format differs".into());
        }
        let values = fixture
            .get("states")
            .and_then(serde_json::Value::as_array)
            .ok_or("exact recovery fixture states are absent")?;
        let states: Vec<ExactWorkflowRecoveryV2> = values
            .iter()
            .cloned()
            .map(serde_json::from_value)
            .collect::<Result<_, _>>()?;
        assert_eq!(states.len(), 6);
        for state in &states {
            validate_recovery_shape(state)?;
        }
        for pair in states.windows(2) {
            let [current, next] = pair else {
                return Err("exact recovery fixture window differs".into());
            };
            let transition = transition_between(current, next)?;
            let mut reduced = current.clone();
            reduce_transition(&mut reduced, &transition)?;
            assert_eq!(&reduced, next);
        }
        assert_eq!(
            states.last().map(|value| value.phase),
            Some(ExactWorkflowPhaseV2::Terminal)
        );
        Ok(())
    }

    #[tokio::test]
    async fn store_values_fail_closed_by_format_shape_and_intent() -> Result<(), Box<dyn Error>> {
        let intent = exact_intent(
            ExactMutationKindV1::Append,
            digest('b'),
            identity('c'),
            identity('d'),
            Some(resource()),
            digest('e'),
            10_000,
        );
        let initial = recovery_from_intent(&intent);

        let mut future = initial.clone();
        future.format_version = EXACT_MUTATION_FORMAT_VERSION.saturating_add(1);
        let future_store: Arc<dyn ExactWorkflowRecoveryStore> = Arc::new(PostgresShapedRecoveryStore {
            row: Mutex::new(Some(future)),
        });
        let future_result = open_session(
            &intent,
            &ExactWorkflowOptions::new(10_000).with_recovery_store(future_store),
        )
        .await;
        assert!(matches!(
            future_result,
            Err(ExactWorkflowError::Store {
                source: ExactWorkflowStoreError::UnsupportedFormat { .. },
                ..
            })
        ));

        let mut malformed = initial.clone();
        malformed.phase = ExactWorkflowPhaseV2::Prepared;
        let malformed_store: Arc<dyn ExactWorkflowRecoveryStore> = Arc::new(PostgresShapedRecoveryStore {
            row: Mutex::new(Some(malformed)),
        });
        let malformed_result = open_session(
            &intent,
            &ExactWorkflowOptions::new(10_000).with_recovery_store(malformed_store),
        )
        .await;
        assert!(matches!(
            malformed_result,
            Err(ExactWorkflowError::Store {
                source: ExactWorkflowStoreError::Malformed,
                ..
            })
        ));

        let mut mismatched = initial;
        mismatched.request_sha256 = digest('9');
        let mismatch_store: Arc<dyn ExactWorkflowRecoveryStore> = Arc::new(PostgresShapedRecoveryStore {
            row: Mutex::new(Some(mismatched)),
        });
        let mismatch_result = open_session(
            &intent,
            &ExactWorkflowOptions::new(10_000).with_recovery_store(mismatch_store),
        )
        .await;
        assert!(matches!(
            mismatch_result,
            Err(ExactWorkflowError::Store {
                source: ExactWorkflowStoreError::IntentMismatch,
                ..
            })
        ));

        let unavailable: Arc<dyn ExactWorkflowRecoveryStore> = Arc::new(UnavailableRecoveryStore);
        let unavailable_result = open_session(
            &intent,
            &ExactWorkflowOptions::new(10_000).with_recovery_store(unavailable),
        )
        .await;
        assert!(matches!(
            unavailable_result,
            Err(ExactWorkflowError::Store {
                source: ExactWorkflowStoreError::Unavailable,
                last_durable_recovery: None,
            })
        ));
        Ok(())
    }

    #[tokio::test]
    async fn store_conflicts_and_cancellation_are_bounded_with_recovery() -> Result<(), Box<dyn Error>> {
        let intent = exact_intent(
            ExactMutationKindV1::Append,
            digest('b'),
            identity('c'),
            identity('d'),
            Some(resource()),
            digest('e'),
            u64::MAX,
        );
        let store: Arc<dyn ExactWorkflowRecoveryStore> = Arc::new(AdvancingConflictStore {
            row: Mutex::new(recovery_from_intent(&intent)),
        });
        let options = ExactWorkflowOptions::new(u64::MAX)
            .with_observation_deadline_unix_millis(u64::MAX)
            .with_store_conflict_limit(0)
            .with_recovery_store(store);
        let (session, mut recovery) = open_session(&intent, &options).await?;
        let conflict = session
            .record(
                &mut recovery,
                ExactMutationTransition::Prepared {
                    preparation: append_preparation_value(),
                },
            )
            .await;
        assert!(matches!(
            conflict,
            Err(ExactWorkflowError::StoreConflictBudgetExhausted {
                last_durable_recovery
            }) if last_durable_recovery.revision == 1
        ));

        let cancellation = CallCancellation::new();
        cancellation.cancel();
        let store: Arc<dyn ExactWorkflowRecoveryStore> = Arc::new(PostgresShapedRecoveryStore::default());
        let options = ExactWorkflowOptions::new(u64::MAX)
            .with_cancellation(cancellation)
            .with_recovery_store(store);
        let (session, mut recovery) = open_session(&intent, &options).await?;
        let cancelled = session
            .record(
                &mut recovery,
                ExactMutationTransition::ClockObserved {
                    unix_millis: wire_u64(1),
                },
            )
            .await;
        assert!(matches!(
            cancelled,
            Err(ExactWorkflowError::Cancelled {
                last_durable_recovery
            }) if last_durable_recovery.revision == 0
        ));
        Ok(())
    }

    #[tokio::test]
    async fn durable_visibility_polling_does_not_exhaust_recovery_records() -> Result<(), Box<dyn Error>> {
        let intent = exact_intent(
            ExactMutationKindV1::Append,
            digest('b'),
            identity('c'),
            identity('d'),
            Some(resource()),
            digest('e'),
            u64::MAX,
        );
        let store: Arc<dyn ExactWorkflowRecoveryStore> = Arc::new(PostgresShapedRecoveryStore::default());
        let clock: Arc<dyn ExactWorkflowClock> = Arc::new(AdvancingClock { next: Mutex::new(1) });
        let options = ExactWorkflowOptions::new(u64::MAX)
            .with_clock(clock)
            .with_recovery_store(Arc::clone(&store));
        let (session, mut recovery) = open_session(&intent, &options).await?;

        assert_eq!(observe_clock(&session, &mut recovery, &options).await?, 1);
        assert!(
            session
                .record(
                    &mut recovery,
                    ExactMutationTransition::Prepared {
                        preparation: append_preparation_value(),
                    },
                )
                .await?
        );
        assert!(
            session
                .record(&mut recovery, ExactMutationTransition::SubmissionStarted)
                .await?
        );
        assert!(
            session
                .record(
                    &mut recovery,
                    ExactMutationTransition::SubmissionObserved {
                        disposition: ExactSubmissionDispositionV1::Resolved,
                    },
                )
                .await?
        );
        assert_eq!(recovery.revision, 4);

        for expected_now in 2..=301 {
            assert_eq!(observe_clock(&session, &mut recovery, &options).await?, expected_now);
            record_resolution_observation(&session, &mut recovery, ExactSubmissionDispositionV1::Resolved).await?;
        }
        assert_eq!(recovery.revision, 4);
        assert_eq!(store.load().await?.map(|stored| stored.revision), Some(4));
        Ok(())
    }

    #[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 path = parent.join("journal");
        let first_intent = exact_intent(
            ExactMutationKindV1::Append,
            digest('b'),
            identity('c'),
            identity('d'),
            Some(resource()),
            digest('e'),
            100,
        );
        {
            let store = Arc::new(FilesystemExactWorkflowRecoveryStore::new(path.clone()));
            let first = ExactWorkflowOptions::new(100).with_recovery_store(store);
            let (session, recovery) = open_session(&first_intent, &first).await?;
            assert_eq!(recovery_effect_deadline(&recovery)?, 100);
            drop(session);
        }

        let resumed_intent = ExactMutationIntent {
            effect_deadline_unix_millis: wire_u64(999),
            ..first_intent
        };
        let store = Arc::new(FilesystemExactWorkflowRecoveryStore::new(path));
        let resumed = ExactWorkflowOptions::new(999).with_recovery_store(store);
        let (session, recovery) = open_session(&resumed_intent, &resumed).await?;
        assert_eq!(recovery_effect_deadline(&recovery)?, 100);
        drop(session);
        std::fs::remove_dir_all(&parent)?;
        Ok(())
    }

    #[tokio::test]
    async fn filesystem_store_restarts_at_each_v2_durability_boundary() -> 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-v2-{}", lower_hex(&suffix)));
        std::fs::create_dir(&parent)?;
        std::fs::set_permissions(&parent, std::fs::Permissions::from_mode(0o700))?;
        let path = parent.join("journal");
        let intent = exact_intent(
            ExactMutationKindV1::Append,
            digest('b'),
            identity('c'),
            identity('d'),
            Some(resource()),
            digest('e'),
            10_000,
        );
        let mut state = recovery_from_intent(&intent);
        let store = FilesystemExactWorkflowRecoveryStore::new(path.clone());
        store.compare_exchange(None, state.clone()).await?;
        drop(store);

        let transitions = [
            ExactMutationTransition::Prepared {
                preparation: append_preparation_value(),
            },
            ExactMutationTransition::SubmissionStarted,
            ExactMutationTransition::SubmissionObserved {
                disposition: ExactSubmissionDispositionV1::Indeterminate,
            },
            ExactMutationTransition::SubmissionObserved {
                disposition: ExactSubmissionDispositionV1::Resolved,
            },
            ExactMutationTransition::Terminal {
                result: ExactTerminalResultV1::Append(AppendExactResultV1 {
                    outcome: ExactMutationOutcomeV1::Resolved,
                    batch_id: digest('e'),
                    index: resource(),
                    items: Vec::new(),
                    visibility: Vec::new(),
                }),
            },
        ];
        for transition in transitions {
            let store = FilesystemExactWorkflowRecoveryStore::new(path.clone());
            assert_eq!(store.load().await?, Some(state.clone()));
            let mut next = state.clone();
            reduce_transition(&mut next, &transition)?;
            store.compare_exchange(Some(state.revision), next.clone()).await?;
            state = next;
            drop(store);
        }
        let store = FilesystemExactWorkflowRecoveryStore::new(path);
        let loaded = store.load().await?.ok_or("filesystem recovery disappeared")?;
        assert_eq!(loaded.phase, ExactWorkflowPhaseV2::Terminal);
        assert_eq!(loaded.caller_action_id, digest('e'));
        assert_eq!(
            append_preparation(&loaded)?.items[0]
                .locator
                .as_ref()
                .map(|value| value.value.as_str()),
            Some("01")
        );
        drop(store);
        std::fs::remove_dir_all(&parent)?;
        Ok(())
    }
}