loonfs_api/v0/

operations.rs

//! Request/response shapes for the v0 HTTP API's operation endpoints:
//! namespace lifecycle (create/fork/status/delete), path-oriented filesystem
//! operations, file revisions, maintenance (checkpoint/retention), and the
//! shared [`ApiError`] body. Explicit commits and the change feed live in
//! [`super::commits`]; read-result shapes live in [`super::reads`].

use super::ContentToken;
use crate::{
    AbsolutePath, AttributeKey, AttributeRevisionNo, AttributeValue, ChangeSeq, CheckpointId,
    CommitId, ContentRef, InodeId, ManifestId, NamespaceId, RevisionNo, WriterEpoch,
};
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;

/// HTTP error body used by LoonFS APIs.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
pub struct ApiError {
    /// Stable machine-readable reason from the [`ErrorCode`](crate::ErrorCode)
    /// registry.
    ///
    /// Carried as a string so clients keep working when a newer server
    /// introduces a code they do not know; use
    /// [`ErrorCode::parse`](crate::ErrorCode::parse) for typed access.
    pub code: String,
    /// For `not_supported` errors, the capability-document feature key the
    /// client should reconcile against.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub feature: Option<String>,
    /// Human-readable error message.
    pub message: String,
    /// Correlation id the server assigned to the failed request; the same
    /// value is sent as the `x-request-id` response header.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub request_id: Option<String>,
    /// Structured context for the code, present when the failure carries
    /// machine-usable identity (API spec, "Standard error contract"). Boxed
    /// so the rare detailed error does not widen every error-carrying result.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub details: Option<Box<ErrorDetails>>,
}

/// Optional machine-readable details for an [`ApiError`].
///
/// Clients make retry decisions from the error code and use these fields for
/// relevant identifiers such as commit ids, writer epochs, and revisions.
/// Fields may be absent and clients must ignore fields they do not use.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
pub struct ErrorDetails {
    /// Idempotency key of the commit the error concerns.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub commit_id: Option<CommitId>,
    /// Sequence at which that commit id already landed. Present when the
    /// failure was decided against a durable commit receipt, which is what
    /// holds the sequence; absent when nothing has committed under the id
    /// yet and two live requests are simply claiming it at once.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub committed_seq: Option<ChangeSeq>,
    /// Semantic identity of the mutation that already landed under that
    /// commit id, from the same receipt as `committed_seq` and present
    /// exactly when it is. A retry recomputes this value from the request it
    /// just made — see
    /// [`put_retry_fingerprint`](crate::put_retry_fingerprint) — and equality
    /// is what proves the two are the same request.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub committed_fingerprint: Option<String>,
    /// Position, in the request's operation list, of the operation that
    /// failed. A commit applies all of its operations or none of them, so
    /// this names the one that stopped the whole request.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub operation_index: Option<u32>,
    /// Epoch the failing writer session held when it was displaced.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub fenced_epoch: Option<WriterEpoch>,
    /// Epoch that currently owns the namespace.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub active_writer_epoch: Option<WriterEpoch>,
    /// Writer id recorded by the current epoch's acquirer, when the head
    /// recorded one.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub active_writer: Option<String>,
    /// Unix milliseconds at which the current epoch's acquirer took it, when
    /// the head recorded one. Writer ids are process labels, so two runs on
    /// one machine can share one; the stamp is what tells them apart.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub active_acquired_at_ms: Option<u64>,
    /// Inode the failed precondition or operation targeted.
    #[serde(
        default,
        skip_serializing_if = "Option::is_none",
        with = "crate::public_inode_id::option"
    )]
    #[cfg_attr(
        feature = "openapi",
        schema(schema_with = crate::public_inode_id::optional_schema)
    )]
    pub inode_id: Option<InodeId>,
    /// Revision the request expected to be current.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub expected_revision_no: Option<RevisionNo>,
    /// Revision that is actually current; absent when the inode has none.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub actual_revision_no: Option<RevisionNo>,
    /// Attribute revision the request expected to be current.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub expected_attributes_revision_no: Option<AttributeRevisionNo>,
    /// Attribute revision that is actually current for the inode.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub actual_attributes_revision_no: Option<AttributeRevisionNo>,
    /// Change-feed cursor the request asked to resume after.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub after_seq: Option<ChangeSeq>,
    /// Oldest sequence still promised for incremental replay.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub retention_floor_seq: Option<ChangeSeq>,
    /// Deletion generation an undelete asked to recover.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub requested_deletion_seq: Option<ChangeSeq>,
    /// Deletion generation actually active for the inode.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub active_deletion_seq: Option<ChangeSeq>,
    /// Head sequence a namespace delete required the namespace to still be
    /// at.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub expected_head_seq: Option<ChangeSeq>,
    /// Head sequence the namespace was actually at, which is what a caller
    /// that still means to delete it retries against.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub actual_head_seq: Option<ChangeSeq>,
}

/// Request to create a namespace.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(deny_unknown_fields)]
pub struct CreateNamespaceRequest {
    /// Durable namespace id to create.
    pub namespace_id: NamespaceId,
}

/// Request to fork a namespace.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(deny_unknown_fields)]
pub struct ForkNamespaceRequest {
    /// Durable namespace id for the fork target.
    pub new_namespace_id: NamespaceId,
}

/// Status summary for one namespace.
///
/// This is the point-lookup answer to "does this namespace exist, and where
/// is its head?" — cheaper than listing all namespaces when only one matters.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
pub struct NamespaceStatusResponse {
    /// Namespace being inspected.
    pub namespace_id: NamespaceId,
    /// Current visible namespace sequence.
    pub head_seq: ChangeSeq,
    /// Current manifest pointer recorded by the head.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub current_manifest_id: Option<ManifestId>,
    /// Number of visible WAL segments after the current manifest.
    pub wal_tail_segments: u64,
    /// Oldest sequence still promised for incremental replay.
    pub retention_floor_seq: ChangeSeq,
}

/// Result of deleting a namespace.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
pub struct DeleteNamespaceResponse {
    /// Namespace whose history ended.
    pub namespace_id: NamespaceId,
    /// The head's last committed sequence; the delete linearized
    /// immediately after it, so this is where history ended.
    pub head_seq: ChangeSeq,
}

/// Destination-conflict behavior for path-oriented puts, moves, and copies.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "snake_case")]
pub enum DestinationBehavior {
    /// Fail if the destination path already exists.
    #[default]
    NoReplace,
    /// Replace the current file at the destination; only a file
    /// destination can be replaced.
    Replace,
}

/// Directory delete behavior for path-oriented deletes.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "snake_case")]
pub enum DeleteDirectoryBehavior {
    /// Fail if the target is a non-empty directory.
    #[default]
    NonRecursive,
    /// Delete a directory subtree.
    Recursive,
}

/// One path-oriented filesystem operation.
///
/// Unknown fields are rejected because concurrency guards are optional. A
/// misspelled guard must fail decoding instead of silently becoming `None`
/// and allowing an unguarded write. Any future fieldless variant must use
/// empty braces so serde also rejects unexpected fields for that variant.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum FilesystemOperation {
    /// Create one directory.
    #[cfg_attr(feature = "openapi", schema(title = "FsOpCreateDirectory"))]
    CreateDirectory {
        /// Absolute destination path, rejected when invalid or already bound.
        path: AbsolutePath,
        /// Also create missing ancestor directories (the same auto-create
        /// `put_file` performs). The final component must still be new.
        #[serde(default, skip_serializing_if = "std::ops::Not::not")]
        parents: bool,
    },
    /// Create or replace one file with an already-durable content ref.
    #[cfg_attr(feature = "openapi", schema(title = "FsOpPutFile"))]
    PutFile {
        /// Absolute destination path; missing ancestors are created automatically.
        path: AbsolutePath,
        /// Immutable bytes that must be covered by a valid preparation proof.
        content_ref: ContentRef,
        /// Whether an existing file may receive a new revision instead of causing a conflict.
        #[serde(default)]
        behavior: DestinationBehavior,
        /// When set (with `replace` behavior), the put applies only while
        /// the file's current revision is still this one; a raced write
        /// fails the request instead of silently stacking on it, and a
        /// missing file answers `path_not_found`.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        expected_revision_no: Option<RevisionNo>,
    },
    /// Delete one path.
    #[cfg_attr(feature = "openapi", schema(title = "FsOpDeletePath"))]
    DeletePath {
        /// Absolute path that must resolve to a visible inode.
        path: AbsolutePath,
        /// Whether a non-empty directory may be tombstoned recursively.
        #[serde(default)]
        behavior: DeleteDirectoryBehavior,
        /// When set, the delete applies only if the path still resolves to
        /// this inode; a raced rebinding fails the request instead of
        /// deleting (and reporting a recovery handle for) the wrong inode.
        #[serde(
            default,
            skip_serializing_if = "Option::is_none",
            with = "crate::public_inode_id::option"
        )]
        #[cfg_attr(
            feature = "openapi",
            schema(schema_with = crate::public_inode_id::optional_schema)
        )]
        expected_inode_id: Option<InodeId>,
    },
    /// Move one path to another path.
    #[cfg_attr(feature = "openapi", schema(title = "FsOpMovePath"))]
    MovePath {
        /// Absolute source path that must resolve to a visible inode.
        from_path: AbsolutePath,
        /// Absolute destination whose parent must be visible and writable.
        to_path: AbsolutePath,
        /// Whether an existing destination file may be replaced.
        #[serde(default)]
        behavior: DestinationBehavior,
    },
    /// Copy one file path to another path.
    #[cfg_attr(feature = "openapi", schema(title = "FsOpCopyPath"))]
    CopyPath {
        /// Absolute source path that must resolve to a visible file.
        from_path: AbsolutePath,
        /// Absolute destination whose parent must be visible and writable.
        to_path: AbsolutePath,
        /// Whether an existing destination file may receive a copied revision.
        #[serde(default)]
        behavior: DestinationBehavior,
    },
    /// Restore a deleted file or subtree.
    ///
    /// `inode_id` and `deletion_seq` identify one exact deletion. A stale
    /// sequence returns `not_deleted` and cannot undo a later deletion.
    #[cfg_attr(feature = "openapi", schema(title = "FsOpUndelete"))]
    Undelete {
        /// Deleted inode to make reachable again.
        #[serde(with = "crate::public_inode_id")]
        #[cfg_attr(
            feature = "openapi",
            schema(schema_with = crate::public_inode_id::schema)
        )]
        inode_id: InodeId,
        /// Observed deletion sequence, which prevents cancelling a newer tombstone generation.
        deletion_seq: ChangeSeq,
        /// Optional destination for the restored inode.
        ///
        /// When absent, the inode is rebound to the parent and name recorded by the
        /// deletion. Parent identity, rather than an old path string, keeps this
        /// correct after ancestor renames. An explicit path is required when the
        /// deletion recorded no binding.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        path: Option<AbsolutePath>,
    },
    /// Restore an older revision as the current revision for a path.
    #[cfg_attr(feature = "openapi", schema(title = "FsOpRestoreRevision"))]
    RestoreRevision {
        /// Absolute path that must resolve to a visible file.
        path: AbsolutePath,
        /// Existing historical revision whose content will be copied into a new current revision.
        source_revision_no: RevisionNo,
    },
    /// Write and remove attributes on the inode one path resolves to.
    #[cfg_attr(feature = "openapi", schema(title = "FsOpUpdateAttributes"))]
    UpdateAttributes {
        /// Absolute path that must resolve to a visible file or directory.
        path: AbsolutePath,
        /// Attributes to write. Each key replaces whatever the inode
        /// currently holds under it; keys the inode holds and this map does
        /// not name are left alone.
        #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
        set: BTreeMap<AttributeKey, AttributeValue>,
        /// Attribute keys to remove.
        ///
        /// A list preserves duplicate entries so validation can report them instead
        /// of silently deduplicating the request.
        #[serde(default, skip_serializing_if = "Vec::is_empty")]
        remove: Vec<AttributeKey>,
        /// When set, the update applies only if the path still resolves to
        /// this inode; a raced rebinding fails the request instead of
        /// writing attributes onto the wrong inode.
        #[serde(
            default,
            skip_serializing_if = "Option::is_none",
            with = "crate::public_inode_id::option"
        )]
        #[cfg_attr(
            feature = "openapi",
            schema(schema_with = crate::public_inode_id::optional_schema)
        )]
        expected_inode_id: Option<InodeId>,
        /// When set, the update applies only while the inode's attribute
        /// revision is still this one. Absent means the update is applied
        /// over whatever revision is current; either way the write carries
        /// its own revision guard, so a concurrent update never merges
        /// silently.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        expected_attributes_revision_no: Option<AttributeRevisionNo>,
    },
}

/// A request to commit one or more filesystem operations.
///
/// Operations run in order and either all succeed or none are committed. A
/// request with one operation uses the same fingerprint rules as a batch.
///
/// Unknown fields are rejected here for the same reason they are on
/// [`FilesystemOperation`]: the fields a typo can hide are the ones that
/// decide whether the commit is guarded at all.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(deny_unknown_fields)]
pub struct CommitRequest {
    /// Caller-supplied idempotency key for the whole request.
    pub commit_id: CommitId,
    /// Actor responsible for the commit, as supplied by the application.
    pub actor: crate::ActorRef,
    /// Caller annotation recorded on the commit and reported by the change
    /// feed. Part of the commit's identity: reusing `commit_id` with a
    /// different message is a `commit_id_reuse_conflict`, exactly as it is
    /// for an explicit commit.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub message: Option<String>,
    /// Proofs for any new external content refs introduced by this request.
    /// One proof covers every operation that names its content ref.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub content_tokens: Vec<ContentToken>,
    /// Ordered operations to apply. Must be non-empty; they commit all
    /// together or not at all.
    pub operations: Vec<FilesystemOperation>,
}

impl CommitRequest {
    /// A request carrying exactly one operation.
    pub fn single(
        commit_id: CommitId,
        actor: crate::ActorRef,
        message: Option<String>,
        operation: FilesystemOperation,
    ) -> Self {
        Self {
            commit_id,
            actor,
            message,
            content_tokens: Vec::new(),
            operations: vec![operation],
        }
    }
}

/// One immutable file revision.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
pub struct FileRevision {
    /// File inode that owns this revision.
    #[serde(with = "crate::public_inode_id")]
    #[cfg_attr(
        feature = "openapi",
        schema(schema_with = crate::public_inode_id::schema)
    )]
    pub inode_id: InodeId,
    /// Revision number within the file inode.
    pub revision_no: RevisionNo,
    /// Namespace sequence that created this revision.
    pub committed_seq: ChangeSeq,
    /// Wall-clock stamp of the commit that created this revision, in Unix
    /// milliseconds. Observational: `committed_seq` is the order.
    pub committed_at_ms: u64,
    /// Actor responsible for this revision, as supplied by the application.
    pub actor: crate::ActorRef,
    /// Content stored for this revision.
    pub content_ref: ContentRef,
}

/// Response for listing file revisions.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
pub struct ListFileRevisionsResponse {
    /// Namespace that was read.
    pub namespace_id: NamespaceId,
    /// File inode whose revisions were returned.
    #[serde(with = "crate::public_inode_id")]
    #[cfg_attr(
        feature = "openapi",
        schema(schema_with = crate::public_inode_id::schema)
    )]
    pub inode_id: InodeId,
    /// Namespace head sequence used for the read.
    pub head_seq: ChangeSeq,
    /// Retained revisions in order.
    pub revisions: Vec<FileRevision>,
    /// Opaque cursor for the next page, if more revisions are available.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub next_cursor: Option<String>,
}

/// Request to create a durable checkpoint pin.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(deny_unknown_fields)]
pub struct CreateCheckpointRequest {
    /// Label recorded on the checkpoint record. A label, not a key: several
    /// records may carry the same name over different bases.
    pub name: String,
    /// Optional lifetime; the server computes the record's expiry from its
    /// own clock. Absent means the pin holds until explicitly released.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub ttl_ms: Option<u64>,
}

/// Result of creating a checkpoint.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
pub struct CreateCheckpointResponse {
    /// Namespace that was checkpointed.
    pub namespace_id: NamespaceId,
    /// Checkpoint that was created.
    #[serde(flatten)]
    pub checkpoint: Checkpoint,
}

/// Result of releasing a checkpoint pin.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
pub struct ReleaseCheckpointResponse {
    /// Namespace the checkpoint belonged to.
    pub namespace_id: NamespaceId,
    /// Checkpoint the release targeted.
    pub checkpoint_id: CheckpointId,
}

/// Who a checkpoint record answers to, as the record durably records it.
///
/// The two owners have different releases, so a listing that names the
/// owner also says which records the release endpoint will act on: a user
/// pin is released by id, and a fork lease is released by deleting the
/// target namespace it protects.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum CheckpointOwnerSummary {
    /// An operator-created pin, released by id or by its own expiry.
    #[cfg_attr(feature = "openapi", schema(title = "CheckpointOwnerUser"))]
    User {
        /// The label the creator recorded. Not a key: several records may
        /// carry one label over different bases.
        name: String,
    },
    /// A fork target keeping its source basis alive for the length of one
    /// fork attempt.
    #[cfg_attr(feature = "openapi", schema(title = "CheckpointOwnerFork"))]
    Fork {
        /// Namespace whose continued existence keeps this pin standing.
        target_namespace_id: NamespaceId,
    },
}

/// One checkpoint resource, reported from what its durable record carries.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
pub struct Checkpoint {
    /// Durable checkpoint id used to address the checkpoint for release.
    pub checkpoint_id: CheckpointId,
    /// Who owns the checkpoint, including the label carried by a user pin.
    pub owner: CheckpointOwnerSummary,
    /// Time the checkpoint record was created, in Unix milliseconds.
    pub created_at_ms: u64,
    /// When garbage collection may release the record without being asked,
    /// in Unix milliseconds. Absent means the pin holds until it is
    /// released. An instant already in the past is a record whose expiry
    /// has passed and which no collection pass has reached yet: it is still
    /// a root, so it is still listed.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub expires_at_ms: Option<u64>,
    /// Sequence covered by the checkpoint's pinned basis.
    pub checkpoint_seq: ChangeSeq,
    /// Manifest pinned by the checkpoint.
    pub manifest_id: ManifestId,
}

/// One page of active checkpoint records.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
pub struct ListCheckpointsResponse {
    /// Namespace the records belong to.
    pub namespace_id: NamespaceId,
    /// Active records in ascending checkpoint-id order. Released records are
    /// omitted even if garbage collection has not deleted them yet.
    pub checkpoints: Vec<Checkpoint>,
    /// Opaque cursor for the next page.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub next_cursor: Option<String>,
}

/// How one WAL flush satisfied its goal.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "snake_case")]
pub enum FlushWalOutcome {
    /// The root already covered the head; nothing was published.
    AlreadyCurrent,
    /// This call published a new manifest and advanced the root to it.
    Published,
    /// This call published a manifest, but a newer root already covered
    /// the attempted sequence.
    Superseded,
}

/// Result of one WAL flush: how the metadata root covers the head.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
pub struct FlushWalResponse {
    /// Namespace whose WAL tail was flushed.
    pub namespace_id: NamespaceId,
    /// Head sequence the flush attempted to cover.
    pub target_head_seq: ChangeSeq,
    /// Manifest `metadata/root.json` references after the operation.
    pub manifest_id: ManifestId,
    /// Sequence covered by that manifest.
    pub manifest_head_seq: ChangeSeq,
    /// How the root came to cover the head.
    pub outcome: FlushWalOutcome,
}

/// Optional overrides for one garbage-collection pass. Absent fields use
/// the server's conservative defaults.
///
/// Every field is optional, so a typo would take the default instead of the
/// override the caller asked for. Unknown fields are rejected so a misspelled
/// `max_objects` fails loudly rather than running an unbounded pass.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(deny_unknown_fields)]
pub struct GcRequest {
    /// Objects younger than this are never deleted, reachable or not. The
    /// window has a derived safety floor (publication budgets plus provider
    /// deadlines); a smaller value is rejected as `invalid_request`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub grace_window_ms: Option<u64>,
    /// Maximum objects this invocation may read or decide. Omit to retain
    /// the run-to-completion behavior.
    ///
    /// A completed upload session past its reclamation grace makes the pass
    /// read every live manifest and retained WAL segment to find out
    /// whether anything still references its content, and that read is
    /// charged here like any other. A budget too small to finish it does
    /// not stall the pass: the session is retained, the response sets
    /// `content_reclamation_deferred`, and the sweep carries on through
    /// everything else. What a chronically small budget costs is content
    /// left unreclaimed, not progress. Give a pass at least as many objects
    /// as the namespace has live manifests and retained segments for that
    /// content to come back.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub max_objects: Option<u64>,
    /// Opaque resume token returned as `next_cursor` by an earlier pass
    /// against the same namespace.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub cursor: Option<String>,
}

/// Why a pass kept what it kept: `retained_candidates` split by the
/// decision that spared each candidate.
///
/// The reasons are a closed set — one per place the sweep decides against
/// deleting — so every field is always reported, and a zero is the answer
/// that nothing was kept for that reason. The counts sum to
/// `retained_candidates`.
///
/// Retention is a decision per candidate examined, not per object in the
/// namespace: one object examined by two passes is counted by each.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
pub struct RetainedCandidates {
    /// Selected as unreachable, then found reachable by the re-verification
    /// that runs immediately before every deletion. A candidate the pass
    /// already knew was reachable is never examined at all, so this counts
    /// the namespace moving underneath the pass rather than the size of its
    /// live set.
    pub referenced: u64,
    /// Unreachable, but younger than the grace window by the object's own
    /// provider timestamp. A later pass deletes it.
    pub grace_window: u64,
    /// Unreachable, and the provider reported no last-modified time at all,
    /// so the object's age is unknown and it is treated as young.
    pub no_provider_timestamp: u64,
    /// Unreachable, but this namespace published no manifest old enough to
    /// say what it referenced when the grace window opened, so nothing
    /// proves the object was already unreferenced then. A reader that pinned
    /// its anchor inside the window may still be reading it, and the pass
    /// keeps it until a manifest ages past the window.
    pub no_reference_manifest: u64,
    /// Root resolution failed somewhere in this pass, so manifest and table
    /// deletion was suppressed wholesale (`degraded_retention` is set too).
    pub degraded_roots: u64,
    /// A key under a swept family that this collector does not recognize as
    /// one of its own. Never deleted, whatever its age.
    pub unrecognized_key: u64,
    /// A checkpoint record this pass could have advanced but could not
    /// prove ready: a lost compare-and-swap, an unreadable record, a fork
    /// target not provably gone, a released record still inside its grace
    /// window, or an active pin that is simply doing its job. The pins
    /// themselves are listed by
    /// `GET /v0/admin/namespaces/{ns}/checkpoints`.
    pub checkpoint_not_releasable: u64,
    /// An upload session waiting out a window a clock resolves: an open
    /// session's lease plus the grace, an aborted session's grace, or a
    /// completed session's derived content-reclamation grace.
    /// `next_reclamation_at_ms` reports the soonest of these.
    pub upload_session_window: u64,
    /// An upload session held over for a reason no clock resolves: a lost
    /// compare-and-swap, a record that vanished mid-pass, or a reference
    /// set this pass could not establish. Only a later pass answers it.
    pub upload_session_undecided: u64,
    /// A completed session whose content reclamation was skipped because
    /// the reference scan did not fit in `max_objects`
    /// (`content_reclamation_deferred` is set too).
    pub content_scan_deferred: u64,
}

/// Result of one mark-and-sweep garbage-collection pass.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
pub struct GcResponse {
    /// Namespace the pass ran against.
    pub namespace_id: NamespaceId,
    /// Unreferenced WAL segments deleted.
    pub deleted_wal_segments: u64,
    /// Unreferenced metadata tables deleted.
    pub deleted_metadata_tables: u64,
    /// Unreferenced manifests deleted.
    pub deleted_manifests: u64,
    /// Released checkpoint records deleted after their grace window.
    pub deleted_checkpoint_records: u64,
    /// Fork-owned checkpoint records released because their target namespace
    /// is provably gone.
    pub released_fork_checkpoints: u64,
    /// Checkpoint records released because their expiry passed, or because
    /// they sit on a terminally deleted namespace.
    #[serde(default)]
    pub released_expired_checkpoints: u64,
    /// Upload-session control objects deleted after the reap window.
    #[serde(default)]
    pub deleted_upload_sessions: u64,
    /// Content objects reclaimed because their upload session completed,
    /// aged past the derived reclamation grace, and nothing the namespace
    /// can reach references them. The upload half's cleanup of abandoned
    /// sessions is not counted here: it deletes unconditionally, whether or
    /// not the session ever wrote anything.
    #[serde(default)]
    pub deleted_content_objects: u64,
    /// Active checkpoint records released because their basis manifest is
    /// verifiably gone.
    #[serde(default)]
    pub released_missing_basis_checkpoints: u64,
    /// Candidates retained at delete time (grace window, missing
    /// timestamps, or reachable from the fresh root set).
    pub retained_candidates: u64,
    /// The same total, split by the decision that spared each candidate.
    /// The total above stays because it is what every existing consumer
    /// reads; this says why.
    #[serde(default)]
    pub retained: RetainedCandidates,
    /// True when ambiguous roots suppressed manifest/table deletion.
    pub degraded_retention: bool,
    /// True when the pass skipped completed-content reclamation because
    /// what it needs — the namespace's live roots, then the reference
    /// collection over them — did not fit in `max_objects`. Nothing was
    /// ever decided from a partial collection; a later pass with room for
    /// the whole scan reclaims what this one left behind. A pass that had
    /// room for the roots swept every other candidate normally around the
    /// skip, and one that did not also reports `budget_exhausted`.
    #[serde(default)]
    pub content_reclamation_deferred: bool,
    /// True when the pass stopped because `max_objects` ran out before it
    /// finished. Whatever it did before that is reported here and stands;
    /// rerun with the returned cursor, or with a larger budget, to
    /// continue. A budget too small for the namespace's own roots stops a
    /// pass before it decides anything at all, which is what this says and
    /// an empty report on its own does not.
    #[serde(default)]
    pub budget_exhausted: bool,
    /// Opaque resume token when more candidates remain. Resuming rebuilds
    /// every safety proof; the token carries enumeration position only and
    /// is valid only against the same namespace.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub next_cursor: Option<String>,
    /// The soonest instant still ahead of this pass at which something it
    /// retained becomes reclaimable: an open session's lease plus the grace
    /// window, an aborted session's grace, or a completed session's derived
    /// content-reclamation grace. A scheduler reads this to decide when to
    /// come back, so a namespace needs no other side channel to have its
    /// reclamation happen.
    ///
    /// It reports what this pass saw and nothing more. A pass that stopped
    /// on `next_cursor` examined only part of the keyspace, and candidates
    /// that age out under a plain grace window on their object timestamps
    /// carry no deadline here at all, so absence is never a claim that
    /// nothing is owed.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub next_reclamation_at_ms: Option<u64>,
}

impl GcResponse {
    /// An empty report for `namespace_id`, before any candidate is examined.
    pub fn empty(namespace_id: NamespaceId) -> Self {
        Self {
            namespace_id,
            deleted_wal_segments: 0,
            deleted_metadata_tables: 0,
            deleted_manifests: 0,
            deleted_checkpoint_records: 0,
            released_fork_checkpoints: 0,
            released_expired_checkpoints: 0,
            deleted_upload_sessions: 0,
            deleted_content_objects: 0,
            released_missing_basis_checkpoints: 0,
            retained_candidates: 0,
            retained: RetainedCandidates::default(),
            degraded_retention: false,
            content_reclamation_deferred: false,
            budget_exhausted: false,
            next_cursor: None,
            next_reclamation_at_ms: None,
        }
    }

    /// Records one retained candidate under the reason that spared it.
    ///
    /// The total and the breakdown move together here so they cannot drift:
    /// every sweep site names a reason, and no site can count a retention
    /// without naming one.
    pub fn retain(&mut self, reason: RetainedReason) {
        self.retained_candidates += 1;
        *reason.counter(&mut self.retained) += 1;
    }
}

/// The reason one candidate was retained, as the sweep site knows it. Each
/// variant is the field of [`RetainedCandidates`] it counts into, where the
/// reason itself is described.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RetainedReason {
    /// Counts into [`RetainedCandidates::referenced`].
    Referenced,
    /// Counts into [`RetainedCandidates::grace_window`].
    GraceWindow,
    /// Counts into [`RetainedCandidates::no_provider_timestamp`].
    NoProviderTimestamp,
    /// Counts into [`RetainedCandidates::no_reference_manifest`].
    NoReferenceManifest,
    /// Counts into [`RetainedCandidates::degraded_roots`].
    DegradedRoots,
    /// Counts into [`RetainedCandidates::unrecognized_key`].
    UnrecognizedKey,
    /// Counts into [`RetainedCandidates::checkpoint_not_releasable`].
    CheckpointNotReleasable,
    /// Counts into [`RetainedCandidates::upload_session_window`].
    UploadSessionWindow,
    /// Counts into [`RetainedCandidates::upload_session_undecided`].
    UploadSessionUndecided,
    /// Counts into [`RetainedCandidates::content_scan_deferred`].
    ContentScanDeferred,
}

impl RetainedReason {
    fn counter(self, retained: &mut RetainedCandidates) -> &mut u64 {
        match self {
            Self::Referenced => &mut retained.referenced,
            Self::GraceWindow => &mut retained.grace_window,
            Self::NoProviderTimestamp => &mut retained.no_provider_timestamp,
            Self::NoReferenceManifest => &mut retained.no_reference_manifest,
            Self::DegradedRoots => &mut retained.degraded_roots,
            Self::UnrecognizedKey => &mut retained.unrecognized_key,
            Self::CheckpointNotReleasable => &mut retained.checkpoint_not_releasable,
            Self::UploadSessionWindow => &mut retained.upload_session_window,
            Self::UploadSessionUndecided => &mut retained.upload_session_undecided,
            Self::ContentScanDeferred => &mut retained.content_scan_deferred,
        }
    }
}

impl RetainedCandidates {
    /// Every reason and its count, in a fixed order, for callers that
    /// report the breakdown rather than read one field of it.
    pub fn by_reason(&self) -> [(&'static str, u64); 10] {
        [
            ("referenced", self.referenced),
            ("grace_window", self.grace_window),
            ("no_provider_timestamp", self.no_provider_timestamp),
            ("no_reference_manifest", self.no_reference_manifest),
            ("degraded_roots", self.degraded_roots),
            ("unrecognized_key", self.unrecognized_key),
            ("checkpoint_not_releasable", self.checkpoint_not_releasable),
            ("upload_session_window", self.upload_session_window),
            ("upload_session_undecided", self.upload_session_undecided),
            ("content_scan_deferred", self.content_scan_deferred),
        ]
    }

    /// Folds another pass's breakdown into this one.
    pub fn add(&mut self, other: &Self) {
        self.referenced += other.referenced;
        self.grace_window += other.grace_window;
        self.no_provider_timestamp += other.no_provider_timestamp;
        self.no_reference_manifest += other.no_reference_manifest;
        self.degraded_roots += other.degraded_roots;
        self.unrecognized_key += other.unrecognized_key;
        self.checkpoint_not_releasable += other.checkpoint_not_releasable;
        self.upload_session_window += other.upload_session_window;
        self.upload_session_undecided += other.upload_session_undecided;
        self.content_scan_deferred += other.content_scan_deferred;
    }

    /// The reason with the highest count, and that count. `None` when
    /// nothing was retained. Ties go to the first in [`Self::by_reason`]
    /// order, so one pass's report is stable.
    pub fn top_reason(&self) -> Option<(&'static str, u64)> {
        self.by_reason()
            .into_iter()
            .filter(|(_, count)| *count > 0)
            // `max_by_key` keeps the last of equal maxima, so the reversal
            // is what makes a tie report the earlier reason.
            .rev()
            .max_by_key(|(_, count)| *count)
    }
}

/// Result of advancing the retention floor.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
pub struct AdvanceRetentionResponse {
    /// New minimum sequence for incremental replay.
    pub retention_floor_seq: ChangeSeq,
}

/// One explicit maintenance step: the actions it selects, and nothing more.
///
/// Selection is presence. Each field names one independent action, and a
/// step runs exactly the ones the body carries — a request that selects
/// nothing is rejected rather than quietly doing nothing. Unknown fields are
/// rejected for the same reason: a misspelled selector would leave its action
/// unrun, and the caller would read the empty report as "there was nothing to
/// do".
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(deny_unknown_fields)]
pub struct MaintenanceStepRequest {
    /// Flush the visible WAL tail into metadata tables, then run one bounded
    /// reorganization step.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub metadata: Option<MetadataMaintenanceRequest>,
    /// Advance the retention floor to the flushed manifest head. Nothing
    /// surrenders replay history unless this is true.
    #[serde(default)]
    pub advance_retention: bool,
    /// Run one bounded mark-and-sweep garbage-collection pass. Nothing
    /// sweeps unless this is present.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub gc: Option<GcRequest>,
}

/// Overrides for the metadata-upkeep action.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(deny_unknown_fields)]
pub struct MetadataMaintenanceRequest {
    /// Flush the visible WAL tail once it reaches this many segments.
    /// Absent uses the server's default threshold; zero, and any value above
    /// the write-rejection threshold, are rejected as `invalid_request`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub max_wal_tail_segments: Option<u64>,
}

/// What the WAL-flush part of a maintenance step did.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(tag = "outcome", rename_all = "snake_case")]
pub enum WalFlushStepOutcome {
    /// The tail was below the threshold, so there was nothing to flush.
    NotNeeded,
    /// The step flushed the WAL tail and advanced the metadata root.
    Flushed {
        /// Sequence covered by the published manifest.
        manifest_head_seq: ChangeSeq,
    },
    /// The root already covered the attempted sequence — another publisher
    /// got there first.
    Superseded {
        /// Sequence this step attempted to flush through.
        attempted_seq: ChangeSeq,
        /// Manifest the root currently references.
        current_manifest_id: ManifestId,
    },
    /// A concurrent head update won the race.
    RaceLost {
        /// Head sequence observed before the advance attempt.
        observed_head_seq: ChangeSeq,
    },
}

/// What the metadata-reorganization part of a maintenance step did.
///
/// Deliberately coarse: the run counts and byte budgets a reorganization
/// consumes are engine policy, not a wire contract.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(tag = "outcome", rename_all = "snake_case")]
pub enum ReorganizeStepOutcome {
    /// No family group had enough delta runs to merge.
    NotNeeded,
    /// One family group was merged and a manifest published.
    UnitPublished,
    /// A group has outgrown one step, and this step started the background
    /// streaming compaction that rebuilds it. The step published nothing;
    /// the job publishes once, when it finishes.
    CompactionStarted,
    /// A job for this namespace is already running, so this step started
    /// none. One runs at a time per namespace; a later step plans this group
    /// again.
    CompactionRunning,
    /// This step's job holds the namespace's slot and is waiting for a
    /// process compaction permit. It starts when one frees; nothing is
    /// needed to make it.
    CompactionAtCapacity,
    /// A group needs a streaming compaction and this handle schedules no
    /// background work, so nothing will run one until an operator does. The
    /// self-hosting guide names the call.
    CompactionRequired,
    /// Another publisher advanced the root first; a later step retries.
    Superseded,
}

/// Result of one explicit maintenance step.
///
/// One report per action the request selected, and none for an action it
/// did not: an absent field means "not selected", never "ran and found
/// nothing to do". The latter is what the outcomes inside a report say.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
pub struct MaintenanceStepResponse {
    /// Namespace the step ran against.
    pub namespace_id: NamespaceId,
    /// Namespace status observed before the step acted.
    pub status_before: NamespaceStatusResponse,
    /// What the metadata-upkeep action did.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub metadata: Option<MetadataMaintenanceResponse>,
    /// Where the retention floor ended up.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub retention: Option<AdvanceRetentionResponse>,
    /// What the collection pass reclaimed.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub gc: Option<GcResponse>,
}

/// What one metadata-upkeep action did, part by part.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
pub struct MetadataMaintenanceResponse {
    /// What the WAL flush did.
    pub wal_flush: WalFlushStepOutcome,
    /// What the reorganization unit did.
    pub reorganize: ReorganizeStepOutcome,
}

/// Options for one store contract probe. Empty today; a body is still sent
/// so later options do not change the shape of the request. An option this
/// build does not know is rejected rather than ignored, so a caller never
/// believes it selected something.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(deny_unknown_fields)]
pub struct StoreProbeRequest {}

/// What one store contract probe observed, check by check.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
pub struct StoreProbeResponse {
    /// Label the server minted for this run. It scopes the objects the run
    /// wrote, so it identifies the run in provider logs too.
    pub run_id: String,
    /// Every check the run performed, in the order it performed them. A
    /// failed check lives here rather than in an error: the probe answered
    /// the question, and the answer is that the store is wrong.
    pub checks: Vec<StoreProbeCheckResult>,
}

/// One named contract check and what the store did with it.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
pub struct StoreProbeCheckResult {
    /// Stable check name.
    pub name: String,
    /// What the store did.
    pub outcome: StoreProbeCheckOutcome,
    /// What was expected and what happened instead. Present only on
    /// `failed`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub message: Option<String>,
}

/// What one contract check concluded about the store.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "snake_case")]
pub enum StoreProbeCheckOutcome {
    /// The store behaved as the contract requires.
    Passed,
    /// The store declares it cannot do this at all. Only the optional
    /// capabilities answer this way, and it is an answer rather than a
    /// fault.
    Unsupported,
    /// The store did something the contract forbids, or the operation
    /// failed outright.
    Failed,
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::ContentId;

    fn path(value: &str) -> AbsolutePath {
        AbsolutePath::parse(value).expect("valid test path")
    }

    fn attribute_key(value: &str) -> AttributeKey {
        AttributeKey::parse(value).expect("valid test attribute key")
    }

    fn sample_content_ref() -> ContentRef {
        ContentRef::blob_v1(
            ContentId::parse("con_0123456789abcdef0123456789abcdef").expect("valid content id"),
            b"hello",
        )
    }

    #[test]
    fn namespace_create_and_fork_responses_use_the_status_shape() {
        let create = NamespaceStatusResponse {
            namespace_id: NamespaceId::parse("demo").expect("namespace id"),
            head_seq: ChangeSeq(0),
            current_manifest_id: None,
            wal_tail_segments: 0,
            retention_floor_seq: ChangeSeq(0),
        };
        assert_eq!(
            serde_json::to_value(create).expect("serialize create response"),
            serde_json::json!({
                "namespace_id": "demo",
                "head_seq": 0,
                "wal_tail_segments": 0,
                "retention_floor_seq": 0
            })
        );

        let fork = NamespaceStatusResponse {
            namespace_id: NamespaceId::parse("demo-branch").expect("namespace id"),
            head_seq: ChangeSeq(7),
            current_manifest_id: None,
            wal_tail_segments: 0,
            retention_floor_seq: ChangeSeq(7),
        };
        assert_eq!(
            serde_json::to_value(fork).expect("serialize fork response"),
            serde_json::json!({
                "namespace_id": "demo-branch",
                "head_seq": 7,
                "wal_tail_segments": 0,
                "retention_floor_seq": 7
            })
        );
    }

    #[test]
    fn behavior_enums_use_snake_case_wire_values() {
        assert_eq!(
            DestinationBehavior::default(),
            DestinationBehavior::NoReplace
        );
        assert_eq!(
            DeleteDirectoryBehavior::default(),
            DeleteDirectoryBehavior::NonRecursive
        );
        assert_eq!(
            serde_json::to_value(DestinationBehavior::NoReplace)
                .expect("destination behavior json"),
            serde_json::json!("no_replace")
        );
        assert_eq!(
            serde_json::to_value(DestinationBehavior::Replace).expect("destination behavior json"),
            serde_json::json!("replace")
        );
        assert_eq!(
            serde_json::to_value(DeleteDirectoryBehavior::NonRecursive)
                .expect("delete behavior json"),
            serde_json::json!("non_recursive")
        );
        assert_eq!(
            serde_json::to_value(DeleteDirectoryBehavior::Recursive).expect("delete behavior json"),
            serde_json::json!("recursive")
        );
    }

    #[test]
    fn filesystem_delete_and_move_operations_use_behavior_field() {
        let create_directory = FilesystemOperation::CreateDirectory {
            path: path("/docs"),
            parents: false,
        };
        assert_eq!(
            serde_json::to_value(&create_directory).expect("create directory op json"),
            serde_json::json!({
                "kind": "create_directory",
                "path": "/docs"
            })
        );

        let create_directory_with_parents = FilesystemOperation::CreateDirectory {
            path: path("/docs/notes"),
            parents: true,
        };
        assert_eq!(
            serde_json::to_value(&create_directory_with_parents)
                .expect("create directory with parents op json"),
            serde_json::json!({
                "kind": "create_directory",
                "path": "/docs/notes",
                "parents": true
            })
        );

        let delete = FilesystemOperation::DeletePath {
            path: path("/docs"),
            behavior: DeleteDirectoryBehavior::Recursive,
            expected_inode_id: None,
        };
        assert_eq!(
            serde_json::to_value(&delete).expect("delete op json"),
            serde_json::json!({
                "kind": "delete_path",
                "path": "/docs",
                "behavior": "recursive"
            })
        );

        let move_path = FilesystemOperation::MovePath {
            from_path: path("/docs/a.txt"),
            to_path: path("/docs/b.txt"),
            behavior: DestinationBehavior::Replace,
        };
        assert_eq!(
            serde_json::to_value(&move_path).expect("move op json"),
            serde_json::json!({
                "kind": "move_path",
                "from_path": "/docs/a.txt",
                "to_path": "/docs/b.txt",
                "behavior": "replace"
            })
        );

        let copy_path = FilesystemOperation::CopyPath {
            from_path: path("/docs/a.txt"),
            to_path: path("/docs/b.txt"),
            behavior: DestinationBehavior::Replace,
        };
        assert_eq!(
            serde_json::to_value(&copy_path).expect("copy op json"),
            serde_json::json!({
                "kind": "copy_path",
                "from_path": "/docs/a.txt",
                "to_path": "/docs/b.txt",
                "behavior": "replace"
            })
        );

        let update_attributes = FilesystemOperation::UpdateAttributes {
            path: path("/docs/a.txt"),
            set: BTreeMap::from([(
                attribute_key("owner"),
                AttributeValue::parse("ada").expect("valid attribute value"),
            )]),
            remove: vec![attribute_key("draft")],
            expected_inode_id: Some(InodeId(7)),
            expected_attributes_revision_no: Some(AttributeRevisionNo(3)),
        };
        assert_eq!(
            serde_json::to_value(&update_attributes).expect("update attributes op json"),
            serde_json::json!({
                "kind": "update_attributes",
                "path": "/docs/a.txt",
                "set": {"owner": "ada"},
                "remove": ["draft"],
                "expected_inode_id": "ino_7",
                "expected_attributes_revision_no": 3
            })
        );
    }

    #[test]
    fn update_attributes_omits_empty_collections_and_absent_guards() {
        let set_only = FilesystemOperation::UpdateAttributes {
            path: path("/docs/a.txt"),
            set: BTreeMap::from([(
                attribute_key("owner"),
                AttributeValue::parse("ada,grace").expect("valid attribute value"),
            )]),
            remove: Vec::new(),
            expected_inode_id: None,
            expected_attributes_revision_no: None,
        };
        assert_eq!(
            serde_json::to_value(&set_only).expect("set-only op json"),
            serde_json::json!({
                "kind": "update_attributes",
                "path": "/docs/a.txt",
                "set": {"owner": "ada,grace"}
            })
        );

        let decoded: FilesystemOperation = serde_json::from_value(serde_json::json!({
            "kind": "update_attributes",
            "path": "/docs/a.txt",
            "remove": ["draft"]
        }))
        .expect("remove-only op defaults the set map and both guards");
        assert_eq!(
            decoded,
            FilesystemOperation::UpdateAttributes {
                path: path("/docs/a.txt"),
                set: BTreeMap::new(),
                remove: vec![attribute_key("draft")],
                expected_inode_id: None,
                expected_attributes_revision_no: None,
            }
        );
    }

    #[test]
    fn update_attributes_validates_keys_and_values_during_deserialization() {
        // The key grammar and the value shape are enforced on the way in, so
        // a malformed update never reaches planning.
        for encoded in [
            serde_json::json!({
                "kind": "update_attributes",
                "path": "/docs/a.txt",
                "set": {"": "ada"}
            }),
            serde_json::json!({
                "kind": "update_attributes",
                "path": "/docs/a.txt",
                "set": {"owner": {"kind": "string", "value": "ada"}}
            }),
            serde_json::json!({
                "kind": "update_attributes",
                "path": "/docs/a.txt",
                "remove": ["a\u{0}b"]
            }),
        ] {
            assert!(serde_json::from_value::<FilesystemOperation>(encoded).is_err());
        }
    }

    #[test]
    fn filesystem_operations_default_omitted_behavior_fields() {
        let put: FilesystemOperation = serde_json::from_value(serde_json::json!({
            "kind": "put_file",
            "path": "/docs/a.txt",
            "content_ref": {
                "kind": "blob_v1",
                "content_id": "con_0123456789abcdef0123456789abcdef",
                "size_bytes": 1,
                "checksum": {
                    "algorithm": "sha256",
                    "value": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
                }
            }
        }))
        .expect("put op defaults behavior");
        assert!(matches!(
            put,
            FilesystemOperation::PutFile {
                behavior: DestinationBehavior::NoReplace,
                expected_revision_no: None,
                ..
            }
        ));

        let delete: FilesystemOperation = serde_json::from_value(serde_json::json!({
            "kind": "delete_path",
            "path": "/docs"
        }))
        .expect("delete op defaults behavior");
        assert_eq!(
            delete,
            FilesystemOperation::DeletePath {
                path: path("/docs"),
                behavior: DeleteDirectoryBehavior::NonRecursive,
                expected_inode_id: None,
            }
        );

        let move_path: FilesystemOperation = serde_json::from_value(serde_json::json!({
            "kind": "move_path",
            "from_path": "/docs/a.txt",
            "to_path": "/docs/b.txt"
        }))
        .expect("move op defaults behavior");
        assert_eq!(
            move_path,
            FilesystemOperation::MovePath {
                from_path: path("/docs/a.txt"),
                to_path: path("/docs/b.txt"),
                behavior: DestinationBehavior::NoReplace,
            }
        );

        let copy_path: FilesystemOperation = serde_json::from_value(serde_json::json!({
            "kind": "copy_path",
            "from_path": "/docs/a.txt",
            "to_path": "/docs/b.txt"
        }))
        .expect("copy op defaults behavior");
        assert_eq!(
            copy_path,
            FilesystemOperation::CopyPath {
                from_path: path("/docs/a.txt"),
                to_path: path("/docs/b.txt"),
                behavior: DestinationBehavior::NoReplace,
            }
        );
    }

    #[test]
    fn filesystem_operation_paths_keep_the_plain_string_wire_shape() {
        let content_ref = ContentRef::blob_v1(ContentId::generate(), b"hello");
        let cases = [
            (
                FilesystemOperation::PutFile {
                    path: path("/docs/a.txt"),
                    content_ref: content_ref.clone(),
                    behavior: DestinationBehavior::NoReplace,
                    expected_revision_no: None,
                },
                serde_json::json!({
                    "kind": "put_file",
                    "path": "/docs/a.txt",
                    "content_ref": content_ref,
                    "behavior": "no_replace"
                }),
            ),
            (
                FilesystemOperation::Undelete {
                    inode_id: InodeId(7),
                    deletion_seq: ChangeSeq(8),
                    path: Some(path("/docs/restored")),
                },
                serde_json::json!({
                    "kind": "undelete",
                    "inode_id": "ino_7",
                    "deletion_seq": 8,
                    "path": "/docs/restored"
                }),
            ),
            (
                FilesystemOperation::RestoreRevision {
                    path: path("/docs/a.txt"),
                    source_revision_no: RevisionNo(2),
                },
                serde_json::json!({
                    "kind": "restore_revision",
                    "path": "/docs/a.txt",
                    "source_revision_no": 2
                }),
            ),
            (
                FilesystemOperation::UpdateAttributes {
                    path: path("/docs/a.txt"),
                    set: BTreeMap::new(),
                    remove: vec![attribute_key("draft")],
                    expected_inode_id: None,
                    expected_attributes_revision_no: None,
                },
                serde_json::json!({
                    "kind": "update_attributes",
                    "path": "/docs/a.txt",
                    "remove": ["draft"]
                }),
            ),
        ];

        for (operation, string_shaped_json) in cases {
            assert_eq!(
                serde_json::to_value(operation).expect("serialize filesystem operation"),
                string_shaped_json
            );
        }
    }

    #[test]
    fn filesystem_operation_paths_validate_during_deserialization() {
        for encoded in [
            serde_json::json!({"kind": "create_directory", "path": "relative", "parents": false}),
            serde_json::json!({
                "kind": "put_file",
                "path": "relative",
                "content_ref": ContentRef::blob_v1(ContentId::generate(), b"hello")
            }),
            serde_json::json!({"kind": "delete_path", "path": "relative"}),
            serde_json::json!({
                "kind": "move_path",
                "from_path": "relative",
                "to_path": "/target"
            }),
            serde_json::json!({
                "kind": "copy_path",
                "from_path": "/source",
                "to_path": "relative"
            }),
            serde_json::json!({
                "kind": "undelete",
                "inode_id": "ino_7",
                "deletion_seq": 8,
                "path": "relative"
            }),
            serde_json::json!({
                "kind": "restore_revision",
                "path": "relative",
                "source_revision_no": 2
            }),
            serde_json::json!({
                "kind": "update_attributes",
                "path": "relative",
                "remove": ["draft"]
            }),
        ] {
            assert!(serde_json::from_value::<FilesystemOperation>(encoded).is_err());
        }
    }

    #[test]
    fn inode_request_fields_accept_only_the_public_format() {
        let operations = [
            serde_json::json!({
                "kind": "delete_path",
                "path": "/docs/a.txt",
                "expected_inode_id": "ino_27"
            }),
            serde_json::json!({
                "kind": "undelete",
                "inode_id": "ino_27",
                "deletion_seq": 8
            }),
            serde_json::json!({
                "kind": "update_attributes",
                "path": "/docs/a.txt",
                "expected_inode_id": "ino_27"
            }),
        ];

        for operation in operations {
            serde_json::from_value::<FilesystemOperation>(operation.clone())
                .expect("valid public inode ID");

            let inode_key = if operation["kind"] == "undelete" {
                "inode_id"
            } else {
                "expected_inode_id"
            };
            for invalid in [serde_json::json!(27), serde_json::json!("27")] {
                let mut invalid_operation = operation.clone();
                invalid_operation[inode_key] = invalid;
                assert!(
                    serde_json::from_value::<FilesystemOperation>(invalid_operation).is_err(),
                    "{inode_key} accepted an invalid inode ID"
                );
            }
        }
    }

    /// A guard the server never saw must fail the request, not the
    /// precondition. Every guard is optional, so a misspelled one used to
    /// decode to `None` and let the write apply unguarded.
    #[test]
    fn a_misspelled_guard_does_not_decode() {
        let put = |guard: &str| {
            let mut operation = serde_json::json!({
                "kind": "put_file",
                "path": "/docs/a.txt",
                "content_ref": sample_content_ref(),
                "behavior": "replace"
            });
            operation[guard] = serde_json::json!(3);
            serde_json::json!({
                "commit_id": "guarded-put",
                "actor": crate::ActorRef::loonfs_system(),
                "operations": [operation]
            })
        };

        let spelled: CommitRequest = serde_json::from_value(put("expected_revision_no"))
            .expect("the guard spelled correctly decodes");
        assert!(matches!(
            spelled.operations.as_slice(),
            [FilesystemOperation::PutFile {
                expected_revision_no: Some(RevisionNo(3)),
                ..
            }]
        ));

        for misspelling in ["expected_revsion_no", "expectedRevisionNo"] {
            assert!(
                serde_json::from_value::<CommitRequest>(put(misspelling)).is_err(),
                "`{misspelling}` decoded instead of failing the request"
            );
        }
    }

    #[test]
    fn expected_revision_no_must_fit_the_public_integer_range() {
        let body = |expected_revision_no: u64| {
            serde_json::json!({
                "commit_id": "bounded-revision-guard",
                "actor": crate::ActorRef::loonfs_system(),
                "operations": [{
                    "kind": "put_file",
                    "path": "/docs/a.txt",
                    "content_ref": sample_content_ref(),
                    "behavior": "replace",
                    "expected_revision_no": expected_revision_no
                }]
            })
        };

        let request: CommitRequest = serde_json::from_value(body(crate::MAX_PUBLIC_INTEGER))
            .expect("deserialize the maximum revision number");
        assert!(matches!(
            request.operations.as_slice(),
            [FilesystemOperation::PutFile {
                expected_revision_no: Some(RevisionNo(value)),
                ..
            }] if *value == crate::MAX_PUBLIC_INTEGER
        ));

        let error = serde_json::from_value::<CommitRequest>(body(crate::MAX_PUBLIC_INTEGER + 1))
            .expect_err("reject a revision number above the public limit");
        assert!(
            error
                .to_string()
                .contains("must be an integer from 0 through 9007199254740991"),
            "unexpected range error: {error}"
        );
    }

    /// The whole commit request tree is strict, not just its root: a typo one
    /// level down hides the same guards.
    #[test]
    fn a_commit_request_rejects_unknown_fields_at_every_level() {
        let valid = || {
            serde_json::json!({
                "commit_id": "strict-commit",
                "actor": crate::ActorRef::loonfs_system(),
                "content_tokens": [{
                    "content_ref": sample_content_ref(),
                    "token": "opaque-proof"
                }],
                "operations": [{
                    "kind": "update_attributes",
                    "path": "/docs/a.txt",
                    "set": {"owner": "ada"},
                    "expected_inode_id": "ino_7"
                }]
            })
        };
        serde_json::from_value::<CommitRequest>(valid())
            .expect("the same body without a typo decodes");

        let mut at_root = valid();
        at_root["mesage"] = serde_json::json!("a note");

        let mut in_operation = valid();
        in_operation["operations"][0]["expectedAttributesRevisionNo"] = serde_json::json!(3);

        let mut in_content_token = valid();
        in_content_token["content_tokens"][0]["expires_at_ms"] = serde_json::json!(1);

        let mut in_content_ref = valid();
        in_content_ref["content_tokens"][0]["content_ref"]["sizeBytes"] = serde_json::json!(5);

        for (level, body) in [
            ("the request root", at_root),
            ("an operation variant", in_operation),
            ("a nested content token", in_content_token),
            ("a content ref below that", in_content_ref),
        ] {
            assert!(
                serde_json::from_value::<CommitRequest>(body).is_err(),
                "an unknown field in {level} decoded instead of failing the request"
            );
        }
    }

    #[test]
    fn checkpoint_responses_use_one_checkpoint_wire_object() {
        let namespace_id = NamespaceId::parse("demo").expect("namespace id");
        let checkpoint = Checkpoint {
            checkpoint_id: CheckpointId::parse("chk_00000000000000000000000000000001")
                .expect("checkpoint id"),
            owner: CheckpointOwnerSummary::User {
                name: "release".to_owned(),
            },
            created_at_ms: 1_752_623_000_000,
            expires_at_ms: Some(1_752_626_600_000),
            checkpoint_seq: ChangeSeq(12),
            manifest_id: ManifestId(9),
        };
        let checkpoint_json = serde_json::json!({
            "checkpoint_id": "chk_00000000000000000000000000000001",
            "owner": {"kind": "user", "name": "release"},
            "created_at_ms": 1_752_623_000_000_u64,
            "expires_at_ms": 1_752_626_600_000_u64,
            "checkpoint_seq": 12,
            "manifest_id": 9,
        });
        let mut create_json = checkpoint_json.clone();
        create_json["namespace_id"] = serde_json::json!("demo");
        assert_eq!(
            serde_json::to_value(CreateCheckpointResponse {
                namespace_id: namespace_id.clone(),
                checkpoint: checkpoint.clone(),
            })
            .expect("serialize create checkpoint response"),
            create_json,
        );
        assert_eq!(
            serde_json::to_value(ListCheckpointsResponse {
                namespace_id: namespace_id.clone(),
                checkpoints: vec![checkpoint.clone()],
                next_cursor: None,
            })
            .expect("serialize list checkpoints response"),
            serde_json::json!({
                "namespace_id": "demo",
                "checkpoints": [checkpoint_json],
            }),
        );
        assert_eq!(
            serde_json::to_value(ReleaseCheckpointResponse {
                namespace_id,
                checkpoint_id: checkpoint.checkpoint_id,
            })
            .expect("serialize release checkpoint response"),
            serde_json::json!({
                "namespace_id": "demo",
                "checkpoint_id": "chk_00000000000000000000000000000001",
            }),
        );
    }

    #[test]
    fn optional_response_fields_are_omitted_and_default_when_absent() {
        let checkpoint_json = serde_json::to_value(Checkpoint {
            checkpoint_id: CheckpointId::parse("chk_00000000000000000000000000000001")
                .expect("checkpoint id"),
            owner: CheckpointOwnerSummary::User {
                name: "release".to_owned(),
            },
            created_at_ms: 1_752_623_000_000,
            expires_at_ms: None,
            checkpoint_seq: ChangeSeq(3),
            manifest_id: ManifestId(3),
        })
        .expect("serialize checkpoint");
        assert!(checkpoint_json.get("expires_at_ms").is_none());
        let checkpoint: Checkpoint = serde_json::from_value(checkpoint_json)
            .expect("decode checkpoint without optional fields");
        assert_eq!(checkpoint.expires_at_ms, None);

        let gc = GcResponse::empty(NamespaceId::parse("demo").expect("namespace id"));
        let gc_json = serde_json::to_value(gc).expect("serialize gc response");
        assert!(gc_json.get("next_reclamation_at_ms").is_none());
        let gc: GcResponse =
            serde_json::from_value(gc_json).expect("decode gc response without optional fields");
        assert_eq!(gc.next_reclamation_at_ms, None);
    }

    #[test]
    fn maintenance_step_outcomes_use_the_outcome_tag() {
        assert_eq!(
            serde_json::to_value(WalFlushStepOutcome::Flushed {
                manifest_head_seq: ChangeSeq(9),
            })
            .expect("serialize WAL flush outcome"),
            serde_json::json!({"outcome": "flushed", "manifest_head_seq": 9})
        );
        assert_eq!(
            serde_json::to_value(ReorganizeStepOutcome::UnitPublished)
                .expect("serialize reorganize outcome"),
            serde_json::json!({"outcome": "unit_published"})
        );
    }

    /// The maintenance bodies are optional selectors and overrides all the
    /// way down, so a typo would run a different step than the caller asked
    /// for and report the difference as "nothing to do".
    #[test]
    fn maintenance_request_bodies_reject_unknown_fields() {
        serde_json::from_value::<MaintenanceStepRequest>(serde_json::json!({
            "metadata": {"max_wal_tail_segments": 4},
            "advance_retention": true,
            "gc": {"grace_window_ms": 1_800_000, "max_objects": 32}
        }))
        .expect("the same body without a typo decodes");

        for body in [
            serde_json::json!({"advance_retenton": true}),
            serde_json::json!({"metadata": {"maxWalTailSegments": 4}}),
            serde_json::json!({"gc": {"max_object": 32}}),
        ] {
            assert!(
                serde_json::from_value::<MaintenanceStepRequest>(body.clone()).is_err(),
                "an unknown field decoded instead of failing the step: {body}"
            );
        }

        serde_json::from_value::<CreateCheckpointRequest>(
            serde_json::json!({"name": "nightly", "ttl_ms": 60_000}),
        )
        .expect("the same checkpoint body without a typo decodes");
        assert!(serde_json::from_value::<CreateCheckpointRequest>(
            serde_json::json!({"name": "nightly", "ttlMs": 60_000})
        )
        .is_err());

        // The probe body carries no options yet, so an unknown one is the
        // only thing it can be sent.
        serde_json::from_value::<StoreProbeRequest>(serde_json::json!({}))
            .expect("an empty probe body decodes");
        assert!(
            serde_json::from_value::<StoreProbeRequest>(serde_json::json!({"deep": true})).is_err()
        );

        serde_json::from_value::<CreateNamespaceRequest>(serde_json::json!({
            "namespace_id": "demo"
        }))
        .expect("the same create body without a typo decodes");
        assert!(
            serde_json::from_value::<CreateNamespaceRequest>(serde_json::json!({
                "namespace_id": "demo",
                "fork_of": "other"
            }))
            .is_err()
        );
        assert!(
            serde_json::from_value::<ForkNamespaceRequest>(serde_json::json!({
                "new_namespace_id": "demo",
                "source_namespace_id": "other"
            }))
            .is_err()
        );
    }
}