lodestone_types/

state_outcomes.rs

//! Transport-neutral state commands and exact write-outcome identities.
//!
//! RESP is deliberately absent from this module. Public native requests and
//! the RESP leaf both translate into these semantics, while only native
//! requests carry caller-known recovery identities.

use std::collections::BTreeSet;
use std::error::Error;
use std::fmt::{self, Display, Formatter};

use serde::{Deserialize, Deserializer, Serialize, de};

use crate::{
    CollectionIndexRef, DurabilityBoundary, DurabilityClass, DurabilityLoss, FenceToken, MutationLifecycleAuthority,
    MutationLifecycleError, MutationLifecycleObservation, MutationLifecycleResolver, MutationReplayIdentity,
    OperationFingerprint, PreparedMutationDisposition, RoutingRevision, SeqNo, ShardId, StateIncarnation,
};

pub const STATE_COMMAND_FORMAT_VERSION: u16 = 1;
pub const STATE_WRITE_OUTCOME_FORMAT_VERSION: u16 = 1;
pub const REGISTERED_STATE_RECOVERY_LOCATOR_FORMAT_VERSION: u16 = 1;
pub const EPHEMERAL_STATE_ACTION_REFERENCE_FORMAT_VERSION: u16 = 1;
pub const MAX_STATE_COMMAND_BYTES: usize = 2 * 1024 * 1024;
pub const MAX_STATE_COMMAND_KEYS: usize = 128;
pub const MAX_STATE_COMMAND_KEY_BYTES: usize = 254;
pub const MAX_STATE_COMMAND_VALUE_BYTES: usize = 1024 * 1024;
pub const MAX_STATE_COMMAND_HASH_FIELDS: usize = 1024;
// JSON byte vectors use at most four bytes per source byte (three decimal
// digits plus a comma). The remaining envelope, scalar fields, and at most
// 1,024 hash fields or 128 key/entry containers fit under this deliberately
// generous structural allowance. Near the limit validation measures the
// actual encoding instead of relying on this upper bound.
const STATE_COMMAND_JSON_STRUCTURAL_ALLOWANCE: usize =
    256 + 128 * (MAX_STATE_COMMAND_HASH_FIELDS + MAX_STATE_COMMAND_KEYS);
pub const MAX_REGISTERED_STATE_OUTCOME_ENTRIES: usize = 4_096;
pub const REGISTERED_STATE_OUTCOME_SEQUENCE_WINDOW: u64 = 65_536;
pub const MAX_EPHEMERAL_STATE_OUTCOME_ENTRIES: usize = 1_024;
pub const EPHEMERAL_STATE_OUTCOME_SEQUENCE_WINDOW: u64 = 8_192;

#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct StateActionId([u8; 32]);

impl StateActionId {
    pub fn new(bytes: [u8; 32]) -> Result<Self, StateContractError> {
        if bytes == [0; 32] {
            return Err(StateContractError::ZeroActionId);
        }
        Ok(Self(bytes))
    }

    #[must_use]
    pub const fn get(self) -> [u8; 32] {
        self.0
    }

    #[must_use]
    pub const fn as_bytes(&self) -> &[u8; 32] {
        &self.0
    }
}

impl<'de> Deserialize<'de> for StateActionId {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let bytes = <[u8; 32]>::deserialize(deserializer)?;
        Self::new(bytes).map_err(de::Error::custom)
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct StateHashFieldV1 {
    pub field: Vec<u8>,
    pub value: Vec<u8>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", content = "value", rename_all = "snake_case", deny_unknown_fields)]
pub enum StateDataValueV1 {
    Bytes(Vec<u8>),
    Hash(Vec<StateHashFieldV1>),
    /// Byte-exact canonical JSON object data. This is user data rather than an
    /// open control-plane shape; adapters choose whether their dialect can
    /// represent it.
    Json(Vec<u8>),
}

impl StateDataValueV1 {
    pub fn validate(&self) -> Result<(), StateContractError> {
        match self {
            Self::Bytes(bytes) => validate_value_bytes(bytes.len()),
            Self::Json(bytes) => {
                validate_value_bytes(bytes.len())?;
                let parsed: serde_json::Value =
                    serde_json::from_slice(bytes).map_err(|_| StateContractError::InvalidJsonValue)?;
                if !parsed.is_object()
                    || serde_json::to_vec(&parsed).map_err(|_| StateContractError::InvalidJsonValue)? != *bytes
                {
                    return Err(StateContractError::InvalidJsonValue);
                }
                Ok(())
            }
            Self::Hash(fields) => validate_hash_fields(fields),
        }
    }
}

fn validate_hash_fields(fields: &[StateHashFieldV1]) -> Result<(), StateContractError> {
    if fields.len() > MAX_STATE_COMMAND_HASH_FIELDS {
        return Err(StateContractError::TooManyHashFields(fields.len()));
    }
    let mut prior: Option<&[u8]> = None;
    let mut total = 0_usize;
    for field in fields {
        if field.field.is_empty() || prior.is_some_and(|value| value >= field.field.as_slice()) {
            return Err(StateContractError::NonCanonicalHashFields);
        }
        prior = Some(&field.field);
        total = total
            .checked_add(field.field.len())
            .and_then(|value| value.checked_add(field.value.len()))
            .ok_or(StateContractError::ValueTooLarge(usize::MAX))?;
    }
    validate_value_bytes(total)
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum StateSetConditionV1 {
    Always,
    IfAbsent,
    IfPresent,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct StateSetEntryV1 {
    pub key: Vec<u8>,
    pub value: Vec<u8>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum StateOperationV1 {
    Get {
        key: Vec<u8>,
    },
    Set {
        key: Vec<u8>,
        value: Vec<u8>,
        condition: StateSetConditionV1,
        expires_at_millis: Option<u64>,
    },
    Delete {
        keys: Vec<Vec<u8>>,
    },
    Exists {
        keys: Vec<Vec<u8>>,
    },
    ExpireAt {
        key: Vec<u8>,
        absolute_millis: u64,
    },
    TtlStatus {
        key: Vec<u8>,
    },
    Persist {
        key: Vec<u8>,
    },
    MultiGet {
        keys: Vec<Vec<u8>>,
    },
    MultiSet {
        entries: Vec<StateSetEntryV1>,
    },
    Increment {
        key: Vec<u8>,
        delta: i64,
    },
    HashGet {
        key: Vec<u8>,
        field: Vec<u8>,
    },
    HashSet {
        key: Vec<u8>,
        fields: Vec<StateHashFieldV1>,
    },
    HashGetAll {
        key: Vec<u8>,
    },
}

/// Allocation-free view over the routing keys carried by one state command.
pub struct StateRoutingKeys<'a> {
    inner: StateRoutingKeysInner<'a>,
}

enum StateRoutingKeysInner<'a> {
    One(Option<&'a [u8]>),
    Keys(std::slice::Iter<'a, Vec<u8>>),
    Entries(std::slice::Iter<'a, StateSetEntryV1>),
}

impl<'a> Iterator for StateRoutingKeys<'a> {
    type Item = &'a [u8];

    fn next(&mut self) -> Option<Self::Item> {
        match &mut self.inner {
            StateRoutingKeysInner::One(key) => key.take(),
            StateRoutingKeysInner::Keys(keys) => keys.next().map(Vec::as_slice),
            StateRoutingKeysInner::Entries(entries) => entries.next().map(|entry| entry.key.as_slice()),
        }
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        let remaining = self.len();
        (remaining, Some(remaining))
    }
}

impl ExactSizeIterator for StateRoutingKeys<'_> {
    fn len(&self) -> usize {
        match &self.inner {
            StateRoutingKeysInner::One(key) => usize::from(key.is_some()),
            StateRoutingKeysInner::Keys(keys) => keys.len(),
            StateRoutingKeysInner::Entries(entries) => entries.len(),
        }
    }
}

impl StateOperationV1 {
    #[must_use]
    pub const fn is_write(&self) -> bool {
        matches!(
            self,
            Self::Set { .. }
                | Self::Delete { .. }
                | Self::ExpireAt { .. }
                | Self::Persist { .. }
                | Self::MultiSet { .. }
                | Self::Increment { .. }
                | Self::HashSet { .. }
        )
    }

    #[must_use]
    pub fn routing_keys(&self) -> Vec<&[u8]> {
        self.routing_keys_iter().collect()
    }

    /// Iterates routing keys without allocating an intermediate collection.
    #[must_use]
    pub fn routing_keys_iter(&self) -> StateRoutingKeys<'_> {
        match self {
            Self::Get { key }
            | Self::Set { key, .. }
            | Self::ExpireAt { key, .. }
            | Self::TtlStatus { key }
            | Self::Persist { key }
            | Self::Increment { key, .. }
            | Self::HashGet { key, .. }
            | Self::HashSet { key, .. }
            | Self::HashGetAll { key } => StateRoutingKeys {
                inner: StateRoutingKeysInner::One(Some(key)),
            },
            Self::Delete { keys } | Self::Exists { keys } | Self::MultiGet { keys } => StateRoutingKeys {
                inner: StateRoutingKeysInner::Keys(keys.iter()),
            },
            Self::MultiSet { entries } => StateRoutingKeys {
                inner: StateRoutingKeysInner::Entries(entries.iter()),
            },
        }
    }

    pub fn validate(&self) -> Result<(), StateContractError> {
        let keys = self.routing_keys_iter();
        let key_count = keys.len();
        if key_count == 0 || key_count > MAX_STATE_COMMAND_KEYS {
            return Err(StateContractError::InvalidKeyCount(key_count));
        }
        for key in keys {
            validate_key(key)?;
        }
        match self {
            Self::Set { value, .. } => validate_value_bytes(value.len())?,
            Self::MultiSet { entries } => {
                let mut unique = BTreeSet::new();
                for entry in entries {
                    validate_value_bytes(entry.value.len())?;
                    if !unique.insert(entry.key.as_slice()) {
                        return Err(StateContractError::DuplicateWriteKey);
                    }
                }
                if entries.windows(2).any(|pair| pair[0].key >= pair[1].key) {
                    return Err(StateContractError::NonCanonicalWriteKeys);
                }
            }
            Self::Delete { keys } => {
                let mut unique = BTreeSet::new();
                if keys.iter().any(|key| !unique.insert(key.as_slice())) {
                    return Err(StateContractError::DuplicateWriteKey);
                }
                if keys.windows(2).any(|pair| pair[0] >= pair[1]) {
                    return Err(StateContractError::NonCanonicalWriteKeys);
                }
            }
            Self::HashGet { field, .. } => {
                if field.is_empty() || field.len() > MAX_STATE_COMMAND_VALUE_BYTES {
                    return Err(StateContractError::InvalidHashField);
                }
            }
            Self::HashSet { fields, .. } => validate_hash_fields(fields)?,
            Self::Get { .. }
            | Self::Exists { .. }
            | Self::ExpireAt { .. }
            | Self::TtlStatus { .. }
            | Self::Persist { .. }
            | Self::MultiGet { .. }
            | Self::Increment { .. }
            | Self::HashGetAll { .. } => {}
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct StateCommandV1 {
    pub format_version: u16,
    pub operation: StateOperationV1,
}

impl StateCommandV1 {
    #[must_use]
    pub const fn new(operation: StateOperationV1) -> Self {
        Self {
            format_version: STATE_COMMAND_FORMAT_VERSION,
            operation,
        }
    }

    pub fn validate(&self) -> Result<(), StateContractError> {
        if self.format_version != STATE_COMMAND_FORMAT_VERSION {
            return Err(StateContractError::UnsupportedVersion(self.format_version));
        }
        self.operation.validate()?;
        if self.serialized_upper_bound() <= MAX_STATE_COMMAND_BYTES {
            return Ok(());
        }
        let bytes = serde_json::to_vec(self).map_err(|error| StateContractError::Codec(error.to_string()))?;
        if bytes.len() > MAX_STATE_COMMAND_BYTES {
            return Err(StateContractError::CommandTooLarge(bytes.len()));
        }
        Ok(())
    }

    fn serialized_upper_bound(&self) -> usize {
        // INVARIANT: every byte vector is included by this exhaustive match.
        // Keeping this bound conservative leaves the exact JSON length check
        // authoritative for commands close to the public 2 MiB ceiling.
        let key_bytes = self
            .operation
            .routing_keys_iter()
            .fold(0_usize, |total, key| total.saturating_add(key.len()));
        let other_bytes = match &self.operation {
            StateOperationV1::Set { value, .. } => value.len(),
            StateOperationV1::MultiSet { entries } => entries
                .iter()
                .fold(0_usize, |total, entry| total.saturating_add(entry.value.len())),
            StateOperationV1::HashGet { field, .. } => field.len(),
            StateOperationV1::HashSet { fields, .. } => fields.iter().fold(0_usize, |total, field| {
                total
                    .saturating_add(field.field.len())
                    .saturating_add(field.value.len())
            }),
            StateOperationV1::Get { .. }
            | StateOperationV1::Delete { .. }
            | StateOperationV1::Exists { .. }
            | StateOperationV1::ExpireAt { .. }
            | StateOperationV1::TtlStatus { .. }
            | StateOperationV1::Persist { .. }
            | StateOperationV1::MultiGet { .. }
            | StateOperationV1::Increment { .. }
            | StateOperationV1::HashGetAll { .. } => 0,
        };
        key_bytes
            .saturating_add(other_bytes)
            .saturating_mul(4)
            .saturating_add(STATE_COMMAND_JSON_STRUCTURAL_ALLOWANCE)
    }

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

    #[must_use]
    pub fn routing_keys(&self) -> Vec<&[u8]> {
        self.operation.routing_keys()
    }

    /// Iterates routing keys without allocating an intermediate collection.
    #[must_use]
    pub fn routing_keys_iter(&self) -> StateRoutingKeys<'_> {
        self.operation.routing_keys_iter()
    }

    /// Captures only the result-shape and bounded-count contract needed after
    /// execution has consumed the command payload.
    #[must_use]
    pub fn result_expectation(&self) -> StateCommandResultExpectation {
        match &self.operation {
            StateOperationV1::Get { .. } => StateCommandResultExpectation::Get,
            StateOperationV1::Set { .. } => StateCommandResultExpectation::Set,
            StateOperationV1::Delete { keys } => StateCommandResultExpectation::Delete {
                maximum_removed: keys.len(),
            },
            StateOperationV1::Exists { keys } => StateCommandResultExpectation::Exists {
                maximum_count: keys.len(),
            },
            StateOperationV1::ExpireAt { .. } => StateCommandResultExpectation::Expire,
            StateOperationV1::TtlStatus { .. } => StateCommandResultExpectation::TtlStatus,
            StateOperationV1::Persist { .. } => StateCommandResultExpectation::Persist,
            StateOperationV1::MultiGet { keys } => StateCommandResultExpectation::MultiGet {
                expected_values: keys.len(),
            },
            StateOperationV1::MultiSet { entries } => StateCommandResultExpectation::MultiSet {
                expected_written: entries.len(),
            },
            StateOperationV1::Increment { .. } => StateCommandResultExpectation::Increment,
            StateOperationV1::HashGet { .. } => StateCommandResultExpectation::HashGet,
            StateOperationV1::HashSet { fields, .. } => StateCommandResultExpectation::HashSet {
                maximum_fields_added: fields.len(),
            },
            StateOperationV1::HashGetAll { .. } => StateCommandResultExpectation::HashGetAll,
        }
    }

    pub fn semantic_digest(
        &self,
        collection_index_ref: CollectionIndexRef,
        state_incarnation: StateIncarnation,
    ) -> Result<OperationFingerprint, StateContractError> {
        self.validate()?;
        if state_incarnation.get() == 0 {
            return Err(StateContractError::ZeroStateIncarnation);
        }
        let body = serde_json::to_vec(self).map_err(|error| StateContractError::Codec(error.to_string()))?;
        let mut hasher = blake3::Hasher::new();
        hash_part(&mut hasher, b"lodestone/state-command-semantic/v1");
        hash_part(&mut hasher, &collection_index_ref.collection_id().get().to_be_bytes());
        hash_part(&mut hasher, &collection_index_ref.index_id().get().to_be_bytes());
        hash_part(&mut hasher, &state_incarnation.get().to_be_bytes());
        hash_part(&mut hasher, &body);
        Ok(OperationFingerprint::new(*hasher.finalize().as_bytes()))
    }
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct StateWriteIntentV1 {
    pub action_id: StateActionId,
    pub submitted_state_incarnation: StateIncarnation,
    pub semantic_digest: OperationFingerprint,
}

impl StateWriteIntentV1 {
    pub fn validate_for(
        self,
        command: &StateCommandV1,
        collection_index_ref: CollectionIndexRef,
    ) -> Result<(), StateContractError> {
        if !command.is_write() || self.submitted_state_incarnation.get() == 0 {
            return Err(StateContractError::InvalidWriteIntent);
        }
        if command.semantic_digest(collection_index_ref, self.submitted_state_incarnation)? != self.semantic_digest {
            return Err(StateContractError::SemanticDigestMismatch);
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum StateWriteResultV1 {
    Set { applied: bool },
    Delete { removed: u32 },
    Expire { changed: bool },
    Persist { changed: bool },
    MultiSet { written: u32 },
    Increment { value: i64 },
    HashSet { fields_added: u32 },
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum StateTtlStatusResultV1 {
    Missing,
    Persistent,
    ExpiresInMillis { remaining_millis: u64 },
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum StateCommandResultV1 {
    Get { value: Option<StateDataValueV1> },
    Set { applied: bool },
    Delete { removed: u32 },
    Exists { count: u32 },
    Expire { changed: bool },
    TtlStatus { status: StateTtlStatusResultV1 },
    Persist { changed: bool },
    MultiGet { values: Vec<Option<StateDataValueV1>> },
    MultiSet { written: u32 },
    Increment { value: i64 },
    HashGet { value: Option<Vec<u8>> },
    HashSet { fields_added: u32 },
    HashGetAll { fields: Vec<StateHashFieldV1> },
}

/// Non-serialized validation descriptor retained when execution consumes a
/// state command's owned key/value payload.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum StateCommandResultExpectation {
    Get,
    Set,
    Delete { maximum_removed: usize },
    Exists { maximum_count: usize },
    Expire,
    TtlStatus,
    Persist,
    MultiGet { expected_values: usize },
    MultiSet { expected_written: usize },
    Increment,
    HashGet,
    HashSet { maximum_fields_added: usize },
    HashGetAll,
}

impl StateCommandResultV1 {
    pub fn validate_for(&self, command: &StateCommandV1) -> Result<(), StateContractError> {
        self.validate_for_expectation(command.result_expectation())
    }

    pub fn validate_for_expectation(
        &self,
        expectation: StateCommandResultExpectation,
    ) -> Result<(), StateContractError> {
        let shape_matches = matches!(
            (expectation, self),
            (StateCommandResultExpectation::Get, Self::Get { .. })
                | (StateCommandResultExpectation::Set, Self::Set { .. })
                | (StateCommandResultExpectation::Delete { .. }, Self::Delete { .. })
                | (StateCommandResultExpectation::Exists { .. }, Self::Exists { .. })
                | (StateCommandResultExpectation::Expire, Self::Expire { .. })
                | (StateCommandResultExpectation::TtlStatus, Self::TtlStatus { .. })
                | (StateCommandResultExpectation::Persist, Self::Persist { .. })
                | (StateCommandResultExpectation::MultiGet { .. }, Self::MultiGet { .. })
                | (StateCommandResultExpectation::MultiSet { .. }, Self::MultiSet { .. })
                | (StateCommandResultExpectation::Increment, Self::Increment { .. })
                | (StateCommandResultExpectation::HashGet, Self::HashGet { .. })
                | (StateCommandResultExpectation::HashSet { .. }, Self::HashSet { .. })
                | (StateCommandResultExpectation::HashGetAll, Self::HashGetAll { .. })
        );
        if !shape_matches {
            return Err(StateContractError::InvalidCommandResult);
        }
        match (expectation, self) {
            (StateCommandResultExpectation::Delete { maximum_removed }, Self::Delete { removed })
                if usize::try_from(*removed).map_or(true, |count| count > maximum_removed) =>
            {
                return Err(StateContractError::InvalidCommandResult);
            }
            (StateCommandResultExpectation::Exists { maximum_count }, Self::Exists { count })
                if usize::try_from(*count).map_or(true, |count| count > maximum_count) =>
            {
                return Err(StateContractError::InvalidCommandResult);
            }
            (StateCommandResultExpectation::MultiGet { expected_values }, Self::MultiGet { values })
                if values.len() != expected_values =>
            {
                return Err(StateContractError::InvalidCommandResult);
            }
            (StateCommandResultExpectation::MultiSet { expected_written }, Self::MultiSet { written })
                if usize::try_from(*written) != Ok(expected_written) =>
            {
                return Err(StateContractError::InvalidCommandResult);
            }
            (StateCommandResultExpectation::HashSet { maximum_fields_added }, Self::HashSet { fields_added })
                if usize::try_from(*fields_added).map_or(true, |count| count > maximum_fields_added) =>
            {
                return Err(StateContractError::InvalidCommandResult);
            }
            _ => {}
        }
        match self {
            Self::Get { value: Some(value) } => value.validate()?,
            Self::MultiGet { values } => {
                if values.len() > MAX_STATE_COMMAND_KEYS {
                    return Err(StateContractError::InvalidCommandResult);
                }
                for value in values.iter().flatten() {
                    value.validate()?;
                }
            }
            Self::Delete { removed } | Self::Exists { count: removed } | Self::MultiSet { written: removed }
                if usize::try_from(*removed).map_or(true, |count| count > MAX_STATE_COMMAND_KEYS) =>
            {
                return Err(StateContractError::InvalidCommandResult);
            }
            Self::HashSet { fields_added }
                if usize::try_from(*fields_added).map_or(true, |count| count > MAX_STATE_COMMAND_HASH_FIELDS) =>
            {
                return Err(StateContractError::InvalidCommandResult);
            }
            Self::HashGet { value: Some(value) } => validate_value_bytes(value.len())?,
            Self::HashGetAll { fields } => StateDataValueV1::Hash(fields.clone()).validate()?,
            _ => {}
        }
        Ok(())
    }

    pub fn as_write_result(&self) -> Option<StateWriteResultV1> {
        match self {
            Self::Set { applied } => Some(StateWriteResultV1::Set { applied: *applied }),
            Self::Delete { removed } => Some(StateWriteResultV1::Delete { removed: *removed }),
            Self::Expire { changed } => Some(StateWriteResultV1::Expire { changed: *changed }),
            Self::Persist { changed } => Some(StateWriteResultV1::Persist { changed: *changed }),
            Self::MultiSet { written } => Some(StateWriteResultV1::MultiSet { written: *written }),
            Self::Increment { value } => Some(StateWriteResultV1::Increment { value: *value }),
            Self::HashSet { fields_added } => Some(StateWriteResultV1::HashSet {
                fields_added: *fields_added,
            }),
            Self::Get { .. }
            | Self::Exists { .. }
            | Self::TtlStatus { .. }
            | Self::MultiGet { .. }
            | Self::HashGet { .. }
            | Self::HashGetAll { .. } => None,
        }
    }
}

impl StateWriteResultV1 {
    pub fn validate(&self) -> Result<(), StateContractError> {
        match self {
            Self::Delete { removed } | Self::MultiSet { written: removed }
                if usize::try_from(*removed).map_or(true, |count| count > MAX_STATE_COMMAND_KEYS) =>
            {
                Err(StateContractError::InvalidWriteResult)
            }
            Self::HashSet { fields_added }
                if usize::try_from(*fields_added).map_or(true, |count| count > MAX_STATE_COMMAND_HASH_FIELDS) =>
            {
                Err(StateContractError::InvalidWriteResult)
            }
            _ => Ok(()),
        }
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct StateWriteOutcomeRecordV1 {
    pub format_version: u16,
    pub action_id: StateActionId,
    pub semantic_digest: OperationFingerprint,
    pub submitted_state_incarnation: StateIncarnation,
    pub durability_class: DurabilityClass,
    pub first_sequence: SeqNo,
    pub last_sequence: SeqNo,
    pub result: StateWriteResultV1,
}

impl StateWriteOutcomeRecordV1 {
    pub fn validate(&self) -> Result<(), StateContractError> {
        if self.format_version != STATE_WRITE_OUTCOME_FORMAT_VERSION {
            return Err(StateContractError::UnsupportedVersion(self.format_version));
        }
        if self.submitted_state_incarnation.get() == 0
            || self.first_sequence.get() == 0
            || self.last_sequence < self.first_sequence
            || !matches!(
                self.durability_class,
                DurabilityClass::RegionalObject | DurabilityClass::ZonalLowLatency | DurabilityClass::Ephemeral
            )
        {
            return Err(StateContractError::InvalidOutcomeSequence);
        }
        self.result.validate()?;
        let expected_operations = match &self.result {
            StateWriteResultV1::Delete { removed: 0 } => 1,
            StateWriteResultV1::Delete { removed } | StateWriteResultV1::MultiSet { written: removed } => {
                u64::from(*removed)
            }
            StateWriteResultV1::Set { .. }
            | StateWriteResultV1::Expire { .. }
            | StateWriteResultV1::Persist { .. }
            | StateWriteResultV1::Increment { .. }
            | StateWriteResultV1::HashSet { .. } => 1,
        };
        let actual_operations = self
            .last_sequence
            .get()
            .checked_sub(self.first_sequence.get())
            .and_then(|span| span.checked_add(1))
            .ok_or(StateContractError::InvalidOutcomeSequence)?;
        if actual_operations != expected_operations {
            return Err(StateContractError::InvalidOutcomeSequence);
        }
        Ok(())
    }
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum StateSearchVisibilityV1 {
    Pending,
    Visible,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct StateWriteCompletionV1 {
    pub outcome: StateWriteOutcomeRecordV1,
    pub durability_class: DurabilityClass,
    pub durability_boundary: DurabilityBoundary,
    pub maximum_loss: DurabilityLoss,
    pub search_visibility: StateSearchVisibilityV1,
    pub deduplicated: bool,
}

impl StateWriteCompletionV1 {
    pub fn validate(&self) -> Result<(), StateContractError> {
        self.outcome.validate()?;
        if self.durability_class != self.outcome.durability_class {
            return Err(StateContractError::DurabilityMismatch);
        }
        match self.durability_class {
            DurabilityClass::RegionalObject | DurabilityClass::ZonalLowLatency
                if self.durability_boundary == DurabilityBoundary::Registered
                    && self.maximum_loss == DurabilityLoss::Zero =>
            {
                Ok(())
            }
            DurabilityClass::Ephemeral
                if self.durability_boundary == DurabilityBoundary::IncarnationMemory
                    && self.maximum_loss == DurabilityLoss::WholeIncarnation =>
            {
                Ok(())
            }
            _ => Err(StateContractError::DurabilityMismatch),
        }
    }
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct StateOutcomeRetentionV1 {
    pub maximum_entries: u32,
    pub sequence_window: u64,
    pub retired_through: Option<SeqNo>,
    pub current_high_water: Option<SeqNo>,
}

impl StateOutcomeRetentionV1 {
    pub fn validate(self) -> Result<(), StateContractError> {
        if self.maximum_entries == 0
            || usize::try_from(self.maximum_entries)
                .map_or(true, |maximum| maximum > MAX_REGISTERED_STATE_OUTCOME_ENTRIES)
            || self.sequence_window == 0
            || self.sequence_window > REGISTERED_STATE_OUTCOME_SEQUENCE_WINDOW
            || self.retired_through.is_some_and(|sequence| sequence.get() == 0)
            || self.current_high_water.is_some_and(|sequence| sequence.get() == 0)
            || (self.retired_through.is_some() && self.current_high_water.is_none())
            || self
                .retired_through
                .zip(self.current_high_water)
                .is_some_and(|(retired, high)| retired > high)
        {
            return Err(StateContractError::InvalidRetention);
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct RegisteredStateRecoveryLocatorV1 {
    pub format_version: u16,
    pub collection_index_ref: CollectionIndexRef,
    pub shard_id: ShardId,
    pub routing_revision: RoutingRevision,
    pub submitted_fence: FenceToken,
    pub action_id: StateActionId,
    pub submitted_state_incarnation: StateIncarnation,
    pub semantic_digest: OperationFingerprint,
    pub observed_registered_high_water: Option<SeqNo>,
    pub retention_maximum_entries: u32,
    pub retention_sequence_window: u64,
}

impl RegisteredStateRecoveryLocatorV1 {
    pub fn validate(&self) -> Result<(), StateContractError> {
        if self.format_version != REGISTERED_STATE_RECOVERY_LOCATOR_FORMAT_VERSION {
            return Err(StateContractError::UnsupportedVersion(self.format_version));
        }
        if self.submitted_state_incarnation.get() == 0 || self.submitted_fence.epoch().get() == 0 {
            return Err(StateContractError::InvalidLocator);
        }
        StateOutcomeRetentionV1 {
            maximum_entries: self.retention_maximum_entries,
            sequence_window: self.retention_sequence_window,
            retired_through: None,
            current_high_water: self.observed_registered_high_water,
        }
        .validate()
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct EphemeralStateActionReferenceV1 {
    pub format_version: u16,
    pub collection_index_ref: CollectionIndexRef,
    pub shard_id: ShardId,
    pub routing_revision: RoutingRevision,
    pub action_id: StateActionId,
    pub reset_epoch: StateIncarnation,
    pub semantic_digest: OperationFingerprint,
    pub observed_sequence_high_water: Option<SeqNo>,
}

impl EphemeralStateActionReferenceV1 {
    pub fn validate(&self) -> Result<(), StateContractError> {
        if self.format_version != EPHEMERAL_STATE_ACTION_REFERENCE_FORMAT_VERSION {
            return Err(StateContractError::UnsupportedVersion(self.format_version));
        }
        if self.reset_epoch.get() == 0
            || self
                .observed_sequence_high_water
                .is_some_and(|sequence| sequence.get() == 0)
        {
            return Err(StateContractError::InvalidLocator);
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(
    tag = "durability",
    content = "locator",
    rename_all = "snake_case",
    deny_unknown_fields
)]
pub enum PreparedStateWriteV1 {
    Registered(RegisteredStateRecoveryLocatorV1),
    Ephemeral(EphemeralStateActionReferenceV1),
}

impl PreparedStateWriteV1 {
    // INVARIANT: construction binds the complete pre-submit route, fence,
    // action, incarnation, command digest, and retention cut atomically.
    #[allow(clippy::too_many_arguments)]
    pub fn registered(
        collection_index_ref: CollectionIndexRef,
        shard_id: ShardId,
        routing_revision: RoutingRevision,
        submitted_fence: FenceToken,
        action_id: StateActionId,
        state_incarnation: StateIncarnation,
        command: &StateCommandV1,
        retention: StateOutcomeRetentionV1,
    ) -> Result<Self, StateContractError> {
        retention.validate()?;
        let locator = RegisteredStateRecoveryLocatorV1 {
            format_version: REGISTERED_STATE_RECOVERY_LOCATOR_FORMAT_VERSION,
            collection_index_ref,
            shard_id,
            routing_revision,
            submitted_fence,
            action_id,
            submitted_state_incarnation: state_incarnation,
            semantic_digest: command.semantic_digest(collection_index_ref, state_incarnation)?,
            observed_registered_high_water: retention.current_high_water,
            retention_maximum_entries: retention.maximum_entries,
            retention_sequence_window: retention.sequence_window,
        };
        locator.validate()?;
        Ok(Self::Registered(locator))
    }

    pub fn ephemeral(
        collection_index_ref: CollectionIndexRef,
        shard_id: ShardId,
        routing_revision: RoutingRevision,
        action_id: StateActionId,
        reset_epoch: StateIncarnation,
        command: &StateCommandV1,
        retention: StateOutcomeRetentionV1,
    ) -> Result<Self, StateContractError> {
        retention.validate()?;
        if usize::try_from(retention.maximum_entries)
            .map_or(true, |maximum| maximum > MAX_EPHEMERAL_STATE_OUTCOME_ENTRIES)
            || retention.sequence_window > EPHEMERAL_STATE_OUTCOME_SEQUENCE_WINDOW
        {
            return Err(StateContractError::InvalidRetention);
        }
        let reference = EphemeralStateActionReferenceV1 {
            format_version: EPHEMERAL_STATE_ACTION_REFERENCE_FORMAT_VERSION,
            collection_index_ref,
            shard_id,
            routing_revision,
            action_id,
            reset_epoch,
            semantic_digest: command.semantic_digest(collection_index_ref, reset_epoch)?,
            observed_sequence_high_water: retention.current_high_water,
        };
        reference.validate()?;
        Ok(Self::Ephemeral(reference))
    }

    pub fn intent_for(
        &self,
        command: &StateCommandV1,
        expected_collection_index_ref: CollectionIndexRef,
    ) -> Result<StateWriteIntentV1, StateContractError> {
        let (collection_index_ref, action_id, submitted_state_incarnation, semantic_digest) = match self {
            Self::Registered(locator) => {
                locator.validate()?;
                (
                    locator.collection_index_ref,
                    locator.action_id,
                    locator.submitted_state_incarnation,
                    locator.semantic_digest,
                )
            }
            Self::Ephemeral(reference) => {
                reference.validate()?;
                (
                    reference.collection_index_ref,
                    reference.action_id,
                    reference.reset_epoch,
                    reference.semantic_digest,
                )
            }
        };
        if collection_index_ref != expected_collection_index_ref {
            return Err(StateContractError::InvalidLocator);
        }
        let intent = StateWriteIntentV1 {
            action_id,
            submitted_state_incarnation,
            semantic_digest,
        };
        intent.validate_for(command, expected_collection_index_ref)?;
        Ok(intent)
    }

    #[must_use]
    pub const fn shard_id(&self) -> ShardId {
        match self {
            Self::Registered(locator) => locator.shard_id,
            Self::Ephemeral(reference) => reference.shard_id,
        }
    }

    #[must_use]
    pub const fn routing_revision(&self) -> RoutingRevision {
        match self {
            Self::Registered(locator) => locator.routing_revision,
            Self::Ephemeral(reference) => reference.routing_revision,
        }
    }

    #[must_use]
    pub const fn observed_high_water(&self) -> Option<SeqNo> {
        match self {
            Self::Registered(locator) => locator.observed_registered_high_water,
            Self::Ephemeral(reference) => reference.observed_sequence_high_water,
        }
    }

    pub fn mutation_replay_identity(&self) -> Result<MutationReplayIdentity, MutationLifecycleError> {
        let mut authority = blake3::Hasher::new();
        authority.update(b"lodestone/state-write-mutation-authority/v1\0");
        let (scope, shard_id, routing_revision, incarnation, action_id, semantic_digest) = match self {
            Self::Registered(locator) => {
                authority.update(&[1]);
                authority.update(&locator.submitted_fence.to_binary());
                (
                    locator.collection_index_ref,
                    locator.shard_id,
                    locator.routing_revision,
                    locator.submitted_state_incarnation,
                    locator.action_id,
                    locator.semantic_digest,
                )
            }
            Self::Ephemeral(reference) => {
                authority.update(&[2]);
                (
                    reference.collection_index_ref,
                    reference.shard_id,
                    reference.routing_revision,
                    reference.reset_epoch,
                    reference.action_id,
                    reference.semantic_digest,
                )
            }
        };
        authority.update(&scope.collection_id().to_be_bytes());
        authority.update(&scope.index_id().to_be_bytes());
        authority.update(&shard_id.get().to_be_bytes());
        authority.update(&routing_revision.get().to_be_bytes());
        authority.update(&incarnation.get().to_be_bytes());
        MutationReplayIdentity::new(
            *authority.finalize().as_bytes(),
            action_id.get(),
            *semantic_digest.as_bytes(),
        )
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum StateWriteResolutionV1 {
    NotSubmitted,
    Resolved {
        completion: StateWriteCompletionV1,
    },
    KnownRetired {
        retired_through: SeqNo,
    },
    StateReset {
        submitted_reset_epoch: StateIncarnation,
        current_reset_epoch: StateIncarnation,
    },
    Indeterminate {
        observed_high_water: Option<SeqNo>,
        current_high_water: Option<SeqNo>,
        retired_through: Option<SeqNo>,
    },
}

impl MutationLifecycleResolver<PreparedStateWriteV1> for StateWriteResolutionV1 {
    fn mutation_lifecycle_authority(&self, _locator: &PreparedStateWriteV1) -> MutationLifecycleAuthority {
        MutationLifecycleAuthority::State
    }

    fn resolve_mutation_lifecycle(
        &self,
        locator: &PreparedStateWriteV1,
        visible: bool,
    ) -> Result<MutationLifecycleObservation, MutationLifecycleError> {
        let identity = locator.mutation_replay_identity()?;
        match self {
            Self::NotSubmitted => Ok(MutationLifecycleObservation::proven_not_submitted(identity)),
            Self::Resolved { completion }
                if visible || matches!(completion.search_visibility, StateSearchVisibilityV1::Visible) =>
            {
                Ok(MutationLifecycleObservation::visible(identity))
            }
            Self::Resolved { .. } => Ok(MutationLifecycleObservation::committed(identity)),
            Self::Indeterminate { .. } => Ok(MutationLifecycleObservation::prepared(
                identity,
                PreparedMutationDisposition::StillPending,
            )),
            Self::KnownRetired { .. } | Self::StateReset { .. } => Err(MutationLifecycleError::OutcomeEvidenceRetired),
        }
    }
}

pub fn resolve_absent_state_action_v1(
    observed_high_water: Option<SeqNo>,
    retention: StateOutcomeRetentionV1,
) -> Result<StateWriteResolutionV1, StateContractError> {
    retention.validate()?;
    if observed_high_water.is_some_and(|observed| retention.current_high_water.is_none_or(|current| observed > current))
    {
        return Ok(StateWriteResolutionV1::Indeterminate {
            observed_high_water,
            current_high_water: retention.current_high_water,
            retired_through: retention.retired_through,
        });
    }
    if retention
        .retired_through
        .is_some_and(|floor| observed_high_water.is_none_or(|observed| floor > observed))
    {
        return Ok(StateWriteResolutionV1::KnownRetired {
            retired_through: retention.retired_through.ok_or(StateContractError::InvalidRetention)?,
        });
    }
    // INVARIANT: if retirement has not crossed the preparation cut, every
    // action submitted after that cut would still be in the outcome map.
    // Absence therefore proves non-submission even if unrelated state
    // sequences advanced the high-water mark.
    Ok(StateWriteResolutionV1::NotSubmitted)
}

pub fn encode_state_command_v1(command: &StateCommandV1) -> Result<Vec<u8>, StateContractError> {
    command.validate()?;
    serde_json::to_vec(command).map_err(|error| StateContractError::Codec(error.to_string()))
}

pub fn decode_state_command_v1(bytes: &[u8]) -> Result<StateCommandV1, StateContractError> {
    if bytes.is_empty() || bytes.len() > MAX_STATE_COMMAND_BYTES {
        return Err(StateContractError::CommandTooLarge(bytes.len()));
    }
    let command = serde_json::from_slice::<StateCommandV1>(bytes)
        .map_err(|error| StateContractError::Codec(error.to_string()))?;
    command.validate()?;
    let canonical = serde_json::to_vec(&command).map_err(|error| StateContractError::Codec(error.to_string()))?;
    if canonical != bytes {
        return Err(StateContractError::NonCanonicalEncoding);
    }
    Ok(command)
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum StateContractError {
    Codec(String),
    CommandTooLarge(usize),
    DuplicateWriteKey,
    DurabilityMismatch,
    InvalidHashField,
    InvalidJsonValue,
    InvalidCommandResult,
    InvalidKeyCount(usize),
    InvalidLocator,
    InvalidOutcomeSequence,
    InvalidRetention,
    InvalidWriteIntent,
    InvalidWriteResult,
    KeyTooLarge(usize),
    NonCanonicalEncoding,
    NonCanonicalHashFields,
    NonCanonicalWriteKeys,
    SemanticDigestMismatch,
    TooManyHashFields(usize),
    UnsupportedVersion(u16),
    ValueTooLarge(usize),
    ZeroActionId,
    ZeroStateIncarnation,
}

impl Display for StateContractError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        match self {
            Self::Codec(error) => write!(formatter, "state contract codec refusal: {error}"),
            Self::CommandTooLarge(bytes) => write!(formatter, "state command has {bytes} bytes"),
            Self::DuplicateWriteKey => formatter.write_str("state write batch contains a duplicate key"),
            Self::DurabilityMismatch => formatter.write_str("state completion durability fields disagree"),
            Self::InvalidHashField => formatter.write_str("state hash field is empty or oversized"),
            Self::InvalidJsonValue => formatter.write_str("state JSON value is not a canonical object"),
            Self::InvalidCommandResult => formatter.write_str("state command result differs from its request"),
            Self::InvalidKeyCount(count) => write!(formatter, "state command has invalid key count {count}"),
            Self::InvalidLocator => formatter.write_str("state recovery locator identity is invalid"),
            Self::InvalidOutcomeSequence => formatter.write_str("state outcome sequence range is invalid"),
            Self::InvalidRetention => formatter.write_str("state outcome retention bounds are invalid"),
            Self::InvalidWriteIntent => formatter.write_str("state write intent does not bind a write"),
            Self::InvalidWriteResult => formatter.write_str("state write result exceeds its command bound"),
            Self::KeyTooLarge(bytes) => write!(formatter, "state key has invalid length {bytes}"),
            Self::NonCanonicalEncoding => formatter.write_str("state command encoding is not canonical"),
            Self::NonCanonicalHashFields => formatter.write_str("state hash fields are not sorted and unique"),
            Self::NonCanonicalWriteKeys => formatter.write_str("state write keys are not sorted and unique"),
            Self::SemanticDigestMismatch => formatter.write_str("state write semantic digest differs"),
            Self::TooManyHashFields(count) => write!(formatter, "state hash has {count} fields"),
            Self::UnsupportedVersion(version) => write!(formatter, "unsupported state contract version {version}"),
            Self::ValueTooLarge(bytes) => write!(formatter, "state value has {bytes} bytes"),
            Self::ZeroActionId => formatter.write_str("state action identity must be nonzero"),
            Self::ZeroStateIncarnation => formatter.write_str("state incarnation must be nonzero"),
        }
    }
}

impl Error for StateContractError {}

fn validate_key(key: &[u8]) -> Result<(), StateContractError> {
    if key.is_empty() || key.len() > MAX_STATE_COMMAND_KEY_BYTES {
        return Err(StateContractError::KeyTooLarge(key.len()));
    }
    Ok(())
}

fn validate_value_bytes(bytes: usize) -> Result<(), StateContractError> {
    if bytes > MAX_STATE_COMMAND_VALUE_BYTES {
        return Err(StateContractError::ValueTooLarge(bytes));
    }
    Ok(())
}

fn hash_part(hasher: &mut blake3::Hasher, bytes: &[u8]) {
    hasher.update(&(bytes.len() as u64).to_be_bytes());
    hasher.update(bytes);
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{CollectionId, Epoch, IndexId, RecoveryEra};

    fn scope() -> Result<CollectionIndexRef, Box<dyn Error>> {
        Ok(CollectionIndexRef::new(CollectionId::new(7)?, IndexId::new(9)?))
    }

    fn command() -> StateCommandV1 {
        StateCommandV1::new(StateOperationV1::Increment {
            key: b"counter".to_vec(),
            delta: 1,
        })
    }

    #[test]
    fn compact_result_expectation_preserves_shape_and_count_validation() -> Result<(), Box<dyn Error>> {
        let command = StateCommandV1::new(StateOperationV1::MultiGet {
            keys: vec![b"{state}a".to_vec(), b"{state}b".to_vec()],
        });
        let expectation = command.result_expectation();
        StateCommandResultV1::MultiGet {
            values: vec![None, Some(StateDataValueV1::Bytes(b"value".to_vec()))],
        }
        .validate_for_expectation(expectation)?;
        assert_eq!(
            StateCommandResultV1::MultiGet { values: vec![None] }.validate_for_expectation(expectation),
            Err(StateContractError::InvalidCommandResult)
        );
        assert_eq!(
            StateCommandResultV1::Get { value: None }.validate_for_expectation(expectation),
            Err(StateContractError::InvalidCommandResult)
        );

        assert_eq!(
            StateCommandResultV1::HashSet { fields_added: 3 }.validate_for_expectation(
                StateCommandResultExpectation::HashSet {
                    maximum_fields_added: 2,
                },
            ),
            Err(StateContractError::InvalidCommandResult)
        );
        Ok(())
    }

    #[test]
    fn transport_command_is_canonical_bounded_and_resp_free() -> Result<(), Box<dyn Error>> {
        let command = command();
        let bytes = encode_state_command_v1(&command)?;
        assert_eq!(decode_state_command_v1(&bytes)?, command);
        assert!(!String::from_utf8(bytes)?.to_ascii_lowercase().contains("resp"));

        let duplicate = StateCommandV1::new(StateOperationV1::MultiSet {
            entries: vec![
                StateSetEntryV1 {
                    key: b"same".to_vec(),
                    value: b"one".to_vec(),
                },
                StateSetEntryV1 {
                    key: b"same".to_vec(),
                    value: b"two".to_vec(),
                },
            ],
        });
        assert_eq!(duplicate.validate(), Err(StateContractError::DuplicateWriteKey));
        let reordered = StateCommandV1::new(StateOperationV1::Delete {
            keys: vec![b"z".to_vec(), b"a".to_vec()],
        });
        assert_eq!(reordered.validate(), Err(StateContractError::NonCanonicalWriteKeys));
        StateDataValueV1::Json(br#"{"value":1}"#.to_vec()).validate()?;
        assert_eq!(
            StateDataValueV1::Json(br#"{ "value": 1 }"#.to_vec()).validate(),
            Err(StateContractError::InvalidJsonValue)
        );
        Ok(())
    }

    #[test]
    fn fast_command_size_bound_covers_every_variant_and_exact_fallback() -> Result<(), Box<dyn Error>> {
        let key = b"{state}key".to_vec();
        let value = vec![0, 9, 10, 99, 100, 255];
        let operations = vec![
            StateOperationV1::Get { key: key.clone() },
            StateOperationV1::Set {
                key: key.clone(),
                value: value.clone(),
                condition: StateSetConditionV1::IfAbsent,
                expires_at_millis: Some(u64::MAX),
            },
            StateOperationV1::Delete {
                keys: vec![key.clone()],
            },
            StateOperationV1::Exists {
                keys: vec![key.clone()],
            },
            StateOperationV1::ExpireAt {
                key: key.clone(),
                absolute_millis: u64::MAX,
            },
            StateOperationV1::TtlStatus { key: key.clone() },
            StateOperationV1::Persist { key: key.clone() },
            StateOperationV1::MultiGet {
                keys: vec![key.clone(), key.clone()],
            },
            StateOperationV1::MultiSet {
                entries: (0_u8..128)
                    .map(|ordinal| StateSetEntryV1 {
                        key: vec![ordinal],
                        value: value.clone(),
                    })
                    .collect(),
            },
            StateOperationV1::Increment {
                key: key.clone(),
                delta: i64::MIN,
            },
            StateOperationV1::HashGet {
                key: key.clone(),
                field: value.clone(),
            },
            StateOperationV1::HashSet {
                key: key.clone(),
                fields: (0_u16..1024)
                    .map(|ordinal| StateHashFieldV1 {
                        field: ordinal.to_be_bytes().to_vec(),
                        value: value.clone(),
                    })
                    .collect(),
            },
            StateOperationV1::HashGetAll { key },
        ];
        for operation in operations {
            let command = StateCommandV1::new(operation);
            command.operation.validate()?;
            let actual_bytes = serde_json::to_vec(&command)?.len();
            assert!(command.serialized_upper_bound() >= actual_bytes);
            assert_eq!(command.validate().is_ok(), actual_bytes <= MAX_STATE_COMMAND_BYTES);
        }

        let accepted_fallback = StateCommandV1::new(StateOperationV1::Set {
            key: b"key".to_vec(),
            value: vec![0; 500_000],
            condition: StateSetConditionV1::Always,
            expires_at_millis: None,
        });
        assert!(accepted_fallback.serialized_upper_bound() > MAX_STATE_COMMAND_BYTES);
        accepted_fallback.validate()?;
        assert!(serde_json::to_vec(&accepted_fallback)?.len() <= MAX_STATE_COMMAND_BYTES);

        let refused_fallback = StateCommandV1::new(StateOperationV1::Set {
            key: b"key".to_vec(),
            value: vec![255; 600_000],
            condition: StateSetConditionV1::Always,
            expires_at_millis: None,
        });
        assert!(refused_fallback.serialized_upper_bound() > MAX_STATE_COMMAND_BYTES);
        assert!(matches!(
            refused_fallback.validate(),
            Err(StateContractError::CommandTooLarge(bytes)) if bytes > MAX_STATE_COMMAND_BYTES
        ));
        Ok(())
    }

    #[test]
    fn semantic_digest_binds_scope_incarnation_and_complete_command() -> Result<(), Box<dyn Error>> {
        let command = command();
        let digest = command.semantic_digest(scope()?, StateIncarnation::new(3))?;
        let intent = StateWriteIntentV1 {
            action_id: StateActionId::new([1; 32])?,
            submitted_state_incarnation: StateIncarnation::new(3),
            semantic_digest: digest,
        };
        intent.validate_for(&command, scope()?)?;
        assert_eq!(
            intent.validate_for(
                &command,
                CollectionIndexRef::new(CollectionId::new(8)?, IndexId::new(9)?)
            ),
            Err(StateContractError::SemanticDigestMismatch)
        );
        Ok(())
    }

    #[test]
    fn absent_resolution_distinguishes_not_submitted_retired_and_indeterminate() -> Result<(), Box<dyn Error>> {
        let retention = |current: Option<u64>, retired: Option<u64>| StateOutcomeRetentionV1 {
            maximum_entries: 4,
            sequence_window: 16,
            current_high_water: current.map(SeqNo::new),
            retired_through: retired.map(SeqNo::new),
        };
        assert_eq!(
            resolve_absent_state_action_v1(Some(SeqNo::new(5)), retention(Some(5), None))?,
            StateWriteResolutionV1::NotSubmitted
        );
        assert_eq!(
            resolve_absent_state_action_v1(Some(SeqNo::new(5)), retention(Some(12), Some(8)))?,
            StateWriteResolutionV1::KnownRetired {
                retired_through: SeqNo::new(8)
            }
        );
        assert_eq!(
            resolve_absent_state_action_v1(Some(SeqNo::new(5)), retention(Some(12), Some(4)))?,
            StateWriteResolutionV1::NotSubmitted
        );
        assert!(matches!(
            resolve_absent_state_action_v1(Some(SeqNo::new(12)), retention(Some(5), None))?,
            StateWriteResolutionV1::Indeterminate { .. }
        ));
        assert_eq!(
            retention(None, Some(1)).validate(),
            Err(StateContractError::InvalidRetention)
        );
        assert_eq!(
            retention(Some(0), None).validate(),
            Err(StateContractError::InvalidRetention)
        );
        Ok(())
    }

    #[test]
    fn locator_and_completion_reject_wrong_durability_or_identity() -> Result<(), Box<dyn Error>> {
        let locator = RegisteredStateRecoveryLocatorV1 {
            format_version: REGISTERED_STATE_RECOVERY_LOCATOR_FORMAT_VERSION,
            collection_index_ref: scope()?,
            shard_id: ShardId::new(3),
            routing_revision: RoutingRevision::new(4),
            submitted_fence: FenceToken::new(RecoveryEra::new(1)?, Epoch::new(2))?,
            action_id: StateActionId::new([3; 32])?,
            submitted_state_incarnation: StateIncarnation::new(5),
            semantic_digest: OperationFingerprint::new([4; 32]),
            observed_registered_high_water: Some(SeqNo::new(8)),
            retention_maximum_entries: 16,
            retention_sequence_window: 64,
        };
        locator.validate()?;
        let outcome = StateWriteOutcomeRecordV1 {
            format_version: STATE_WRITE_OUTCOME_FORMAT_VERSION,
            action_id: locator.action_id,
            semantic_digest: locator.semantic_digest,
            submitted_state_incarnation: locator.submitted_state_incarnation,
            durability_class: DurabilityClass::RegionalObject,
            first_sequence: SeqNo::new(9),
            last_sequence: SeqNo::new(9),
            result: StateWriteResultV1::Increment { value: 42 },
        };
        let completion = StateWriteCompletionV1 {
            outcome: outcome.clone(),
            durability_class: DurabilityClass::RegionalObject,
            durability_boundary: DurabilityBoundary::IncarnationMemory,
            maximum_loss: DurabilityLoss::Zero,
            search_visibility: StateSearchVisibilityV1::Pending,
            deduplicated: false,
        };
        assert_eq!(completion.validate(), Err(StateContractError::DurabilityMismatch));
        let rebound = StateWriteCompletionV1 {
            outcome,
            durability_class: DurabilityClass::ZonalLowLatency,
            durability_boundary: DurabilityBoundary::Registered,
            maximum_loss: DurabilityLoss::Zero,
            search_visibility: StateSearchVisibilityV1::Pending,
            deduplicated: true,
        };
        assert_eq!(rebound.validate(), Err(StateContractError::DurabilityMismatch));
        Ok(())
    }

    #[test]
    fn state_resolver_projects_closed_indeterminate_outcomes() -> Result<(), Box<dyn Error>> {
        let prepared = PreparedStateWriteV1::registered(
            scope()?,
            ShardId::new(3),
            RoutingRevision::new(4),
            FenceToken::new(RecoveryEra::new(1)?, Epoch::new(2))?,
            StateActionId::new([0x31; 32])?,
            StateIncarnation::new(5),
            &command(),
            StateOutcomeRetentionV1 {
                maximum_entries: 16,
                sequence_window: 64,
                retired_through: None,
                current_high_water: Some(SeqNo::new(8)),
            },
        )?;
        assert_eq!(
            StateWriteResolutionV1::NotSubmitted
                .resolve_mutation_lifecycle(&prepared, false)?
                .resolve_submitted_indeterminate(),
            crate::SubmittedIndeterminateResolution::ProvenNotSubmitted
        );
        assert_eq!(
            StateWriteResolutionV1::Indeterminate {
                observed_high_water: Some(SeqNo::new(8)),
                current_high_water: Some(SeqNo::new(8)),
                retired_through: None,
            }
            .resolve_mutation_lifecycle(&prepared, false)?
            .resolve_submitted_indeterminate(),
            crate::SubmittedIndeterminateResolution::StillPending
        );
        Ok(())
    }

    #[test]
    fn outcome_sequence_span_is_exact_for_single_batch_and_no_effect_results() -> Result<(), Box<dyn Error>> {
        let action_id = StateActionId::new([5; 32])?;
        let outcome = |first, last, result| StateWriteOutcomeRecordV1 {
            format_version: STATE_WRITE_OUTCOME_FORMAT_VERSION,
            action_id,
            semantic_digest: OperationFingerprint::new([6; 32]),
            submitted_state_incarnation: StateIncarnation::new(7),
            durability_class: DurabilityClass::RegionalObject,
            first_sequence: SeqNo::new(first),
            last_sequence: SeqNo::new(last),
            result,
        };
        outcome(4, 4, StateWriteResultV1::Delete { removed: 0 }).validate()?;
        outcome(4, 6, StateWriteResultV1::MultiSet { written: 3 }).validate()?;
        assert_eq!(
            outcome(4, 5, StateWriteResultV1::Increment { value: 1 }).validate(),
            Err(StateContractError::InvalidOutcomeSequence)
        );
        assert_eq!(
            outcome(4, 6, StateWriteResultV1::MultiSet { written: 2 }).validate(),
            Err(StateContractError::InvalidOutcomeSequence)
        );
        Ok(())
    }

    #[test]
    fn prepared_write_binds_route_scope_fence_reset_and_retention_before_submission() -> Result<(), Box<dyn Error>> {
        let command = command();
        let retention = StateOutcomeRetentionV1 {
            maximum_entries: 16,
            sequence_window: 64,
            retired_through: None,
            current_high_water: Some(SeqNo::new(8)),
        };
        let registered = PreparedStateWriteV1::registered(
            scope()?,
            ShardId::new(3),
            RoutingRevision::new(4),
            FenceToken::new(RecoveryEra::new(1)?, Epoch::new(2))?,
            StateActionId::new([8; 32])?,
            StateIncarnation::new(5),
            &command,
            retention,
        )?;
        assert_eq!(registered.observed_high_water(), Some(SeqNo::new(8)));
        registered.intent_for(&command, scope()?)?;
        assert_eq!(
            registered.intent_for(
                &command,
                CollectionIndexRef::new(CollectionId::new(9)?, IndexId::new(9)?)
            ),
            Err(StateContractError::InvalidLocator)
        );

        let oversized_ephemeral = StateOutcomeRetentionV1 {
            maximum_entries: u32::try_from(MAX_EPHEMERAL_STATE_OUTCOME_ENTRIES + 1)?,
            sequence_window: EPHEMERAL_STATE_OUTCOME_SEQUENCE_WINDOW,
            retired_through: None,
            current_high_water: None,
        };
        assert_eq!(
            PreparedStateWriteV1::ephemeral(
                scope()?,
                ShardId::new(3),
                RoutingRevision::new(4),
                StateActionId::new([9; 32])?,
                StateIncarnation::new(5),
                &command,
                oversized_ephemeral,
            ),
            Err(StateContractError::InvalidRetention)
        );
        Ok(())
    }
}