fix(authorization): close lease-transition CAS race and log lease-read failures

H4 (HIGH): The four lease state transitions (`claim`, `consume`,
`revoke`) wrote with `CasExpectation::Any`, relying on a per-owner
process-local `mutation_lock` for atomicity. Two host processes
sharing the same backend root could observe identical pre-transition
state and clobber each other — silently double-consuming a one-shot
fingerprinted lease, or overwriting a `Consumed` marker with `Revoked`.

Introduce `update_lease_cas`: read the lease with its `RecordVersion`,
let the caller mutate, write with `CasExpectation::Version(_)`, retry
up to `CAS_RETRY_ATTEMPTS` (3) on `VersionMismatch`. `revoke`, `claim`,
`consume` are migrated through this helper; the inline state machine
in `consume` is preserved. `ensure_claimable`/`ensure_consumable` re-
run inside the loop so a race that flips status surfaces the proper
typed error rather than a clobber.

Backends without per-row versioning (`LocalFilesystem` and legacy
byte-only mounts) reject `CasExpectation::Version(_)` with
`Unsupported`. For those, `write_lease_raw` falls back to
`CasExpectation::Any` and carries the safety invariant via the
existing `mutation_lock` — same trade-off documented on
`FilesystemCapabilityLeaseStore` and matched by
`ironclaw_processes::put_with_byte_fallback`.

Two new `CapabilityLeaseError` variants:
- `VersionMismatch` (internal CAS-loop signal; never escapes the
  public API)
- `CasExhausted` (transient; caller may retry at a higher level)

H5 (HIGH): `LibSqlCapabilityLeaseStore::leases_for_scope` and
`PostgresCapabilityLeaseStore::leases_for_scope` previously used
`unwrap_or_default()` to collapse DB errors into an empty `Vec`.
The trait signature (`-> Vec<CapabilityLease>`) provides no Result
channel, so propagation requires a wider refactor. As a tactical
mitigation, the error path now `tracing::warn!`s with the underlying
error so a DB outage is visible to operators instead of masquerading
as "no leases for this scope" (which still fails closed for dispatch,
but did so invisibly). Each warn site is annotated `// silent-ok:`
per `.claude/rules/error-handling.md`.

Audit findings F1 (HIGH) and F2 (HIGH) from the ironclaw_authorization
crate audit.

crates/ironclaw_authorization/src/db.rs

@@ -122,12 +122,32 @@ impl CapabilityLeaseStore for LibSqlCapabilityLeaseStore {
     }
 
     async fn leases_for_scope(&self, scope: &ResourceScope) -> Vec<CapabilityLease> {
-        let Ok(conn) = self.connect().await else {
-            return Vec::new();
+        let conn = match self.connect().await {
+            Ok(conn) => conn,
+            Err(error) => {
+                // silent-ok: trait signature returns Vec, no Result channel.
+                // Auth-critical reads fail closed (empty Vec ⇒ deny). Log so
+                // a DB outage is at least visible to operators.
+                tracing::warn!(
+                    target: "ironclaw_authorization::lease_store",
+                    ?error,
+                    "libsql leases_for_scope: connect failed; returning empty set (fails closed)"
+                );
+                return Vec::new();
+            }
         };
-        libsql_leases_for_scope(&conn, scope)
-            .await
-            .unwrap_or_default()
+        match libsql_leases_for_scope(&conn, scope).await {
+            Ok(leases) => leases,
+            Err(error) => {
+                // silent-ok: see above — fail closed, log for visibility.
+                tracing::warn!(
+                    target: "ironclaw_authorization::lease_store",
+                    ?error,
+                    "libsql leases_for_scope: query failed; returning empty set (fails closed)"
+                );
+                Vec::new()
+            }
+        }
     }
 
     async fn active_leases_for_context(&self, context: &ExecutionContext) -> Vec<CapabilityLease> {
@@ -240,12 +260,32 @@ impl CapabilityLeaseStore for PostgresCapabilityLeaseStore {
     }
 
     async fn leases_for_scope(&self, scope: &ResourceScope) -> Vec<CapabilityLease> {
-        let Ok(client) = self.pool.get().await else {
-            return Vec::new();
+        let client = match self.pool.get().await {
+            Ok(client) => client,
+            Err(error) => {
+                // silent-ok: trait signature returns Vec, no Result channel.
+                // Auth-critical reads fail closed (empty Vec ⇒ deny). Log so
+                // a DB outage is at least visible to operators.
+                tracing::warn!(
+                    target: "ironclaw_authorization::lease_store",
+                    error = ?error,
+                    "postgres leases_for_scope: pool.get failed; returning empty set (fails closed)"
+                );
+                return Vec::new();
+            }
         };
-        postgres_leases_for_scope(&client, scope)
-            .await
-            .unwrap_or_default()
+        match postgres_leases_for_scope(&client, scope).await {
+            Ok(leases) => leases,
+            Err(error) => {
+                // silent-ok: see above — fail closed, log for visibility.
+                tracing::warn!(
+                    target: "ironclaw_authorization::lease_store",
+                    ?error,
+                    "postgres leases_for_scope: query failed; returning empty set (fails closed)"
+                );
+                Vec::new()
+            }
+        }
     }
 
     async fn active_leases_for_context(&self, context: &ExecutionContext) -> Vec<CapabilityLease> {

crates/ironclaw_authorization/src/lib.rs

@@ -21,8 +21,15 @@ use std::{
 use async_trait::async_trait;
 use chrono::Utc;
 use ironclaw_filesystem::{
-    CasExpectation, ContentType, Entry, FileType, FilesystemError, RootFilesystem,
+    CasExpectation, ContentType, Entry, FileType, FilesystemError, RecordVersion, RootFilesystem,
 };
+
+/// Bounded retry budget for compare-and-swap loops on lease writes.
+///
+/// Each iteration re-reads the current row version and rewrites with
+/// `CasExpectation::Version(_)`; a multi-process race that loses the
+/// CAS retries until either it wins or this budget is exhausted.
+const CAS_RETRY_ATTEMPTS: usize = 3;
 use ironclaw_host_api::{
     AgentId, CapabilityDescriptor, CapabilityGrant, CapabilityGrantId, Decision, DenyReason,
     EffectKind, ExecutionContext, HostApiError, InvocationFingerprint, InvocationId, MissionId,
@@ -210,6 +217,18 @@ pub enum CapabilityLeaseError {
     },
     #[error("capability lease persistence error: {reason}")]
     Persistence { reason: String },
+    /// Internal CAS-loop signal: the lease record was updated between our
+    /// read and write. Surfaces only inside the retry loop and is converted
+    /// to [`CasExhausted`] if the budget is exhausted; callers will not see
+    /// this variant escape the public API.
+    #[doc(hidden)]
+    #[error("capability lease version mismatch (internal retry signal)")]
+    VersionMismatch,
+    /// CAS retry budget exhausted: too many concurrent writers contended on
+    /// the same lease row. Callers should treat this as transient and may
+    /// retry at a higher level.
+    #[error("capability lease compare-and-swap retry budget exhausted")]
+    CasExhausted,
 }
 
 /// Store of active/revoked capability leases.
@@ -416,6 +435,22 @@ where
         scope: &ResourceScope,
         lease_id: CapabilityGrantId,
     ) -> Result<Option<CapabilityLease>, CapabilityLeaseError> {
+        Ok(self
+            .read_lease_versioned(scope, lease_id)
+            .await?
+            .map(|(lease, _)| lease))
+    }
+
+    /// Read the lease together with its current backend record version.
+    ///
+    /// Used by the mutation paths (`revoke`, `claim`, `consume`) to drive a
+    /// `CasExpectation::Version` write, so a concurrent writer from another
+    /// process fails the CAS instead of clobbering this transition.
+    async fn read_lease_versioned(
+        &self,
+        scope: &ResourceScope,
+        lease_id: CapabilityGrantId,
+    ) -> Result<Option<(CapabilityLease, RecordVersion)>, CapabilityLeaseError> {
         let path = lease_path(scope, lease_id)?;
         let Some(versioned) = self
             .filesystem
@@ -425,22 +460,97 @@ where
         else {
             return Ok(None);
         };
-        deserialize(&versioned.entry.body).map(Some)
+        let lease: CapabilityLease = deserialize(&versioned.entry.body)?;
+        Ok(Some((lease, versioned.version)))
     }
 
-    async fn write_lease(&self, lease: &CapabilityLease) -> Result<(), CapabilityLeaseError> {
+    /// Write the lease with the given CAS expectation.
+    ///
+    /// `CasExpectation::Version(_)` is the canonical path used by the mutation
+    /// flows below — a `VersionMismatch` from the backend signals that a
+    /// concurrent writer modified the same row, and the caller's retry loop
+    /// re-reads and tries again. `CasExpectation::Any` remains in use only
+    /// from the issue path, which is paired with the per-owner
+    /// [`mutation_lock`] and writes a freshly-generated lease id that no
+    /// other writer can collide with.
+    /// Write the lease through the backend with the given CAS expectation.
+    ///
+    /// Backends that don't track per-row versions (e.g. `LocalFilesystem`)
+    /// reject `CasExpectation::Version(_)` with `Unsupported`. For those,
+    /// fall back to `CasExpectation::Any` and carry the safety invariant
+    /// via the per-owner `mutation_lock` — same trade-off documented on
+    /// `FilesystemCapabilityLeaseStore` and matched by sibling crates'
+    /// fallback shape (`ironclaw_processes::put_with_byte_fallback`).
+    async fn write_lease_raw(
+        &self,
+        lease: &CapabilityLease,
+        expectation: CasExpectation,
+    ) -> Result<(), CapabilityLeaseError> {
         let path = lease_path(&lease.scope, lease.grant.id)?;
         let body = serialize_pretty(lease)?;
         let entry = Entry::bytes(body).with_content_type(ContentType::json());
-        // `Any` matches the existing semantics — the per-owner `mutation_lock`
-        // serializes claim/consume/revoke within a single instance, so we do
-        // not need backend-side CAS here. Production shared roots still need
-        // a transactional backend or explicit CAS per the crate guardrail.
-        self.filesystem
-            .put(&path, entry, CasExpectation::Any)
-            .await
-            .map(|_| ())
-            .map_err(lease_persistence_error)
+        match self.filesystem.put(&path, entry.clone(), expectation).await {
+            Ok(_) => Ok(()),
+            Err(FilesystemError::Unsupported { .. })
+                if !matches!(expectation, CasExpectation::Any) =>
+            {
+                // Backend has no per-row versioning — degrade to the legacy
+                // single-instance contract under `mutation_lock`.
+                match self.filesystem.put(&path, entry, CasExpectation::Any).await {
+                    Ok(_) => Ok(()),
+                    Err(error) => Err(lease_persistence_error(error)),
+                }
+            }
+            Err(error) => Err(lease_persistence_error(error)),
+        }
+    }
+
+    async fn write_lease(&self, lease: &CapabilityLease) -> Result<(), CapabilityLeaseError> {
+        // Issue path only — see `update_lease_cas` for the mutation pattern.
+        // The per-owner mutation lock + fresh-id invariant in `issue` makes
+        // the CAS race unreachable for first-write of a brand-new lease id.
+        self.write_lease_raw(lease, CasExpectation::Any).await
+    }
+
+    /// Read-modify-write a lease under compare-and-swap.
+    ///
+    /// Reads the current row (and its [`RecordVersion`]), hands the
+    /// deserialized lease to `mutate`, writes the result back with
+    /// `CasExpectation::Version(_)`, and retries up to
+    /// [`CAS_RETRY_ATTEMPTS`] times on `FilesystemError::VersionMismatch`.
+    /// A missing lease maps to [`CapabilityLeaseError::UnknownLease`].
+    ///
+    /// This closes the multi-process race documented on
+    /// [`FilesystemCapabilityLeaseStore`]: even with a shared backend root,
+    /// a concurrent writer that updates the lease between our read and
+    /// write fails our CAS, we re-read, and re-apply the mutation against
+    /// the new state. Net effect is last-writer-wins among logically
+    /// concurrent transitions, with no silent clobber.
+    async fn update_lease_cas<M>(
+        &self,
+        scope: &ResourceScope,
+        lease_id: CapabilityGrantId,
+        mut mutate: M,
+    ) -> Result<CapabilityLease, CapabilityLeaseError>
+    where
+        M: FnMut(&mut CapabilityLease) -> Result<(), CapabilityLeaseError>,
+    {
+        for _ in 0..CAS_RETRY_ATTEMPTS {
+            let Some((mut lease, version)) = self.read_lease_versioned(scope, lease_id).await?
+            else {
+                return Err(CapabilityLeaseError::UnknownLease { lease_id });
+            };
+            mutate(&mut lease)?;
+            match self
+                .write_lease_raw(&lease, CasExpectation::Version(version))
+                .await
+            {
+                Ok(()) => return Ok(lease),
+                Err(CapabilityLeaseError::VersionMismatch) => continue,
+                Err(error) => return Err(error),
+            }
+        }
+        Err(CapabilityLeaseError::CasExhausted)
     }
 
     async fn read_lease_index(
@@ -584,13 +694,13 @@ where
     ) -> Result<CapabilityLease, CapabilityLeaseError> {
         let lock = self.mutation_lock(scope);
         let _guard = lock.lock().await;
-        let mut lease = self
-            .read_lease(scope, lease_id)
-            .await?
-            .ok_or(CapabilityLeaseError::UnknownLease { lease_id })?;
-        lease.status = CapabilityLeaseStatus::Revoked;
-        self.write_lease(&lease).await?;
-        Ok(lease)
+        // CAS-Version retry: a concurrent claim/consume from another process
+        // must not be clobbered. Idempotent on already-Revoked leases.
+        self.update_lease_cas(scope, lease_id, |lease| {
+            lease.status = CapabilityLeaseStatus::Revoked;
+            Ok(())
+        })
+        .await
     }
 
     async fn get(
@@ -609,14 +719,16 @@ where
     ) -> Result<CapabilityLease, CapabilityLeaseError> {
         let lock = self.mutation_lock(scope);
         let _guard = lock.lock().await;
-        let mut lease = self
-            .read_lease(scope, lease_id)
-            .await?
-            .ok_or(CapabilityLeaseError::UnknownLease { lease_id })?;
-        ensure_claimable(&lease, invocation_fingerprint)?;
-        lease.status = CapabilityLeaseStatus::Claimed;
-        self.write_lease(&lease).await?;
-        Ok(lease)
+        // CAS-Version retry: a concurrent claim/consume/revoke must not
+        // race past `ensure_claimable`. Re-validate inside the loop so a
+        // race that flips status (e.g. another claimant got there first)
+        // surfaces as the proper typed error rather than a clobber.
+        self.update_lease_cas(scope, lease_id, |lease| {
+            ensure_claimable(lease, invocation_fingerprint)?;
+            lease.status = CapabilityLeaseStatus::Claimed;
+            Ok(())
+        })
+        .await
     }
 
     async fn consume(
@@ -626,29 +738,32 @@ where
     ) -> Result<CapabilityLease, CapabilityLeaseError> {
         let lock = self.mutation_lock(scope);
         let _guard = lock.lock().await;
-        let mut lease = self
-            .read_lease(scope, lease_id)
-            .await?
-            .ok_or(CapabilityLeaseError::UnknownLease { lease_id })?;
-        let was_claimed = lease.status == CapabilityLeaseStatus::Claimed;
-        ensure_consumable(&lease)?;
-        if lease.invocation_fingerprint.is_some() {
-            if let Some(remaining) = lease.grant.constraints.max_invocations.as_mut() {
-                *remaining = 0;
-            }
-            lease.status = CapabilityLeaseStatus::Consumed;
-        } else if let Some(remaining) = lease.grant.constraints.max_invocations.as_mut() {
-            *remaining -= 1;
-            if *remaining == 0 {
+        // CAS-Version retry: one-shot fingerprinted leases MUST NOT be
+        // consumable twice. Without CAS, two processes can both read
+        // Active/Claimed, both consume, and both succeed — granting double
+        // authority. The retry re-evaluates `ensure_consumable` against
+        // the latest version so the loser sees `InactiveLease`.
+        self.update_lease_cas(scope, lease_id, |lease| {
+            let was_claimed = lease.status == CapabilityLeaseStatus::Claimed;
+            ensure_consumable(lease)?;
+            if lease.invocation_fingerprint.is_some() {
+                if let Some(remaining) = lease.grant.constraints.max_invocations.as_mut() {
+                    *remaining = 0;
+                }
                 lease.status = CapabilityLeaseStatus::Consumed;
+            } else if let Some(remaining) = lease.grant.constraints.max_invocations.as_mut() {
+                *remaining -= 1;
+                if *remaining == 0 {
+                    lease.status = CapabilityLeaseStatus::Consumed;
+                } else if was_claimed {
+                    lease.status = CapabilityLeaseStatus::Active;
+                }
             } else if was_claimed {
                 lease.status = CapabilityLeaseStatus::Active;
             }
-        } else if was_claimed {
-            lease.status = CapabilityLeaseStatus::Active;
-        }
-        self.write_lease(&lease).await?;
-        Ok(lease)
+            Ok(())
+        })
+        .await
     }
 
     async fn leases_for_scope(&self, scope: &ResourceScope) -> Vec<CapabilityLease> {
@@ -1358,6 +1473,13 @@ fn lease_host_api_error(error: HostApiError) -> CapabilityLeaseError {
 }
 
 fn lease_persistence_error(error: FilesystemError) -> CapabilityLeaseError {
+    // Preserve the typed `VersionMismatch` signal so the CAS retry loop in
+    // `update_lease_cas` can detect and retry. Every other backend error
+    // collapses into the opaque `Persistence` variant — the redacted
+    // `FilesystemError::Display` is safe to surface across a tenant boundary.
+    if matches!(error, FilesystemError::VersionMismatch { .. }) {
+        return CapabilityLeaseError::VersionMismatch;
+    }
     CapabilityLeaseError::Persistence {
         reason: error.to_string(),
     }

crates/ironclaw_capabilities/src/helpers.rs

@@ -178,6 +178,8 @@ pub(crate) fn capability_lease_error_kind(error: &CapabilityLeaseError) -> &'sta
         CapabilityLeaseError::FingerprintMismatch { .. } => "FingerprintMismatch",
         CapabilityLeaseError::InactiveLease { .. } => "InactiveLease",
         CapabilityLeaseError::Persistence { .. } => "Persistence",
+        CapabilityLeaseError::VersionMismatch => "VersionMismatch",
+        CapabilityLeaseError::CasExhausted => "CasExhausted",
     }
 }