Skip to main content

rightkit_daemon/
approval.rs

1//! Approval workflow mechanics: a request is bound to one effect, decided by a verified
2//! principal, and claimed for execution exactly once (PK07-4).
3//!
4//! This module owns the **state machine and its guarantees**, not the wire. What the owner signs
5//! (Pulse's P-256 `pulse.approval.v1` lines, Mail's account-scoped database decisions) and what the
6//! effect does (send, delete, unlock) stay declared inputs: an `ApprovalVerifier` the app
7//! implements, an `ApprovalProfile` of per-action data, an `AuditSink` and the effect itself.
8//! The coordinator never re-encodes signed bytes: `VerifyInput` hands the verifier the stored request (id,
9//! action, effect digest, account, client, expiry, revision) and the decision being made, and the
10//! verifier builds its own signing input from them, as Pulse builds its lines from "the engine's
11//! own record of the waiting request, never from the caller".
12//! Real signing-vector compatibility is verified by the app's profile integration.
13//!
14//! ```text
15//!   request -> Pending --decide(Approve)--> Approved --claim--> Claimed --settle--> Settled
16//!                |  \--decide(Deny)--> Denied     |                |
17//!                |                               |                +-- crash or timeout --> Settled(OutcomeUnknown)
18//!                +--decide(Revoke) / revoke_account (Pending or Approved) --> Revoked
19//!                +--cancel (Pending or Approved)  --> Cancelled
20//!                +--expiry (Pending or Approved, checked on every touch) --> Expired
21//! ```
22//!
23//! # Guarantees
24//!
25//! * **Request binding.** A decision applies to the stored request at the revision the approver
26//!   saw (`ApprovalCoordinator::decide` takes it, the verifier receives it). A signature for
27//!   one request, action, account or revision cannot answer another; a stale revision is
28//!   `ApprovalError::StaleRevision`. A nonce the verifier reports is accepted once, across all
29//!   requests (`ApprovalStore::claim_nonce`).
30//! * **Scope.** The verified principal's account claims must cover the request's account (an
31//!   account-restricted principal never decides an account-less request), and when the request
32//!   names approvers the principal must be one of them.
33//! * **Expiry and revocation.** Every touch of an open request checks the clock first; a decision
34//!   or claim at or after `expires_ms` finds the request `Expired`. A revoked, denied, cancelled
35//!   or expired request can never be claimed.
36//! * **One execution claim.** `ApprovalCoordinator::claim` is a compare-and-set on the
37//!   revision; of any number of concurrent claimants exactly one gets a `ClaimToken`. The token
38//!   is not `Clone`; `settle` accepts only the token whose claim
39//!   id is on record.
40//! * **Audit before effect.** When the action's profile says `AuditRequirement::BeforeEffect`
41//!   every mutating request or state transition appends a durable audit row before insert or
42//!   compare-and-set. Failed appends leave approval state unchanged. Rows record attempted
43//!   transitions: a later store conflict can prevent their candidate revision from committing.
44//!   A coordinator without an `AuditSink` refuses such a request.
45//! * **No exactly-once claim for external effects.** A claim that is never settled (a crash, a
46//!   handler that disappeared) is resolved by `ApprovalCoordinator::recover_claims`: an
47//!   action declared `reconcilable` asks the app's `EffectReconciler`; everything else becomes
48//!   `Settled` with `EffectOutcome::OutcomeUnknown`. The coordinator never guesses that an
49//!   effect did not happen.
50//!
51//! # Storage
52//!
53//! `ApprovalStore` is the durable half. Its contract: `replace` is an atomic compare-and-set on
54//! `revision`, durable when it returns `Ok`, and `claim_nonce` is atomic. `MemoryApprovalStore`
55//! is the in-process reference (a restart forgets everything, as Pulse's engine does today); an
56//! app with a database implements the trait over a table with a unique key on `id`, a
57//! `WHERE revision = ?` update, and a unique key on the nonce.
58//!
59//! Strings that reach a record or an audit row (`NewRequest::detail`, outcome details,
60//! reasons) go through `redact_detail`: anything but `[A-Za-z0-9_=. -]` withholds the whole
61//! detail, so a caller cannot log message content through the approval ledger.
62
63use std::collections::{BTreeMap, BTreeSet};
64use std::fmt;
65use std::sync::atomic::{AtomicU64, Ordering};
66use std::sync::{Arc, Mutex, MutexGuard};
67use std::time::{SystemTime, UNIX_EPOCH};
68
69use serde::{Deserialize, Serialize};
70
71/// Longest request id, label or digest field accepted, in bytes.
72const MAX_LABEL: usize = 256;
73/// Longest detail kept after redaction, in bytes.
74const MAX_DETAIL: usize = 200;
75/// Most nonces [`MemoryApprovalStore`] remembers before it refuses new ones (fail closed).
76const MAX_NONCES: usize = 4096;
77/// Default bound on open (Pending or Approved) requests.
78pub const DEFAULT_MAX_OPEN: usize = 1024;
79
80// ---------------------------------------------------------------------------------------------
81// Time
82// ---------------------------------------------------------------------------------------------
83
84/// Milliseconds since the Unix epoch. Injected so expiry is testable.
85pub trait ApprovalClock: Send + Sync {
86    fn now_ms(&self) -> u64;
87}
88
89/// The system clock.
90#[derive(Debug, Clone, Copy, Default)]
91pub struct SystemApprovalClock;
92
93impl ApprovalClock for SystemApprovalClock {
94    fn now_ms(&self) -> u64 {
95        SystemTime::now()
96            .duration_since(UNIX_EPOCH)
97            .map(|d| d.as_millis() as u64)
98            .unwrap_or(0)
99    }
100}
101
102// ---------------------------------------------------------------------------------------------
103// Profile (data)
104// ---------------------------------------------------------------------------------------------
105
106/// Whether an audit row must be durable before each approval mutation.
107#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
108#[serde(rename_all = "snake_case")]
109pub enum AuditRequirement {
110    /// Rows are written when a sink exists; a failed append is counted
111    /// ([`ApprovalCoordinator::audit_failures`]) and changes nothing.
112    BestEffort,
113    /// Every request or state transition must append its row before insert or compare-and-set.
114    BeforeEffect,
115}
116
117/// Per-action policy, as data. Mail's send, forward, delete and attach and Pulse's approve and
118/// unlock are entries of an [`ApprovalProfile`], not code.
119#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
120pub struct ActionProfile {
121    /// Lifetime of a request that names none.
122    pub default_ttl_ms: u64,
123    /// Longest lifetime a request may ask for.
124    pub max_ttl_ms: u64,
125    pub audit: AuditRequirement,
126    /// A request for this action must name an account.
127    pub requires_account: bool,
128    /// A request for this action must name a client (Pulse's `client:` binding).
129    pub requires_client: bool,
130    /// The effect can be asked whether it happened ([`EffectReconciler`]), so an unsettled claim
131    /// need not become `OutcomeUnknown`.
132    pub reconcilable: bool,
133}
134
135impl ActionProfile {
136    /// A profile with `ttl_ms` as both default and maximum, best-effort audit, no required
137    /// account or client, not reconcilable.
138    pub fn new(ttl_ms: u64) -> Self {
139        Self {
140            default_ttl_ms: ttl_ms,
141            max_ttl_ms: ttl_ms,
142            audit: AuditRequirement::BestEffort,
143            requires_account: false,
144            requires_client: false,
145            reconcilable: false,
146        }
147    }
148
149    pub fn max_ttl_ms(mut self, ms: u64) -> Self {
150        self.max_ttl_ms = ms;
151        self
152    }
153
154    pub fn audit(mut self, audit: AuditRequirement) -> Self {
155        self.audit = audit;
156        self
157    }
158
159    pub fn requires_account(mut self, on: bool) -> Self {
160        self.requires_account = on;
161        self
162    }
163
164    pub fn requires_client(mut self, on: bool) -> Self {
165        self.requires_client = on;
166        self
167    }
168
169    pub fn reconcilable(mut self, on: bool) -> Self {
170        self.reconcilable = on;
171        self
172    }
173}
174
175/// The declared actions and bounds of one app's approval workflow. Serde-able so a profile can
176/// ship as a data file; [`ApprovalCoordinator::new`] validates it.
177#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
178pub struct ApprovalProfile {
179    actions: BTreeMap<String, ActionProfile>,
180    max_open: usize,
181}
182
183impl Default for ApprovalProfile {
184    fn default() -> Self {
185        Self {
186            actions: BTreeMap::new(),
187            max_open: DEFAULT_MAX_OPEN,
188        }
189    }
190}
191
192fn valid_label(text: &str) -> bool {
193    !text.is_empty()
194        && text.len() <= MAX_LABEL
195        && text.chars().all(|c| !c.is_control() && !c.is_whitespace())
196}
197
198impl ApprovalProfile {
199    pub fn new() -> Self {
200        Self::default()
201    }
202
203    /// Declare an action, refusing a bad name or a policy that cannot work (a zero lifetime, a
204    /// default above the maximum).
205    pub fn try_action(
206        mut self,
207        name: impl Into<String>,
208        profile: ActionProfile,
209    ) -> Result<Self, ApprovalError> {
210        let name = name.into();
211        check_action(&name, &profile)?;
212        self.actions.insert(name, profile);
213        Ok(self)
214    }
215
216    /// [`try_action`](Self::try_action) for a profile built in code: a bad entry is a programming
217    /// error and panics at construction.
218    #[track_caller]
219    pub fn action(self, name: impl Into<String>, profile: ActionProfile) -> Self {
220        match self.try_action(name, profile) {
221            Ok(p) => p,
222            Err(e) => panic!("ApprovalProfile::action: {e}"),
223        }
224    }
225
226    /// Most requests that may be open (Pending or Approved) at once; further requests are
227    /// [`ApprovalError::TooManyOpen`] after expired ones are swept. Default
228    /// [`DEFAULT_MAX_OPEN`].
229    pub fn max_open(mut self, max: usize) -> Self {
230        self.max_open = max;
231        self
232    }
233
234    pub fn get(&self, action: &str) -> Option<&ActionProfile> {
235        self.actions.get(action)
236    }
237
238    fn validate(&self) -> Result<(), ApprovalError> {
239        if self.max_open == 0 {
240            return Err(ApprovalError::Invalid("max_open must be at least 1"));
241        }
242        for (name, profile) in &self.actions {
243            check_action(name, profile)?;
244        }
245        Ok(())
246    }
247}
248
249fn check_action(name: &str, profile: &ActionProfile) -> Result<(), ApprovalError> {
250    if !valid_label(name) {
251        return Err(ApprovalError::Invalid("action name is not a plain label"));
252    }
253    if profile.default_ttl_ms == 0
254        || profile.max_ttl_ms == 0
255        || profile.default_ttl_ms > profile.max_ttl_ms
256    {
257        return Err(ApprovalError::Invalid(
258            "action lifetime must be positive and the default no greater than the maximum",
259        ));
260    }
261    Ok(())
262}
263
264// ---------------------------------------------------------------------------------------------
265// Records
266// ---------------------------------------------------------------------------------------------
267
268/// Where a request is.
269#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
270#[serde(rename_all = "snake_case")]
271#[non_exhaustive]
272pub enum ApprovalState {
273    Pending,
274    Approved,
275    Denied,
276    Revoked,
277    Cancelled,
278    Expired,
279    /// Claimed for execution; the effect may be running or may have run.
280    Claimed,
281    /// Done; see [`ApprovalRecord::outcome`].
282    Settled,
283}
284
285impl ApprovalState {
286    /// Pending or Approved: the states a request can still be decided, revoked, cancelled or
287    /// expired from.
288    pub fn is_open(self) -> bool {
289        matches!(self, ApprovalState::Pending | ApprovalState::Approved)
290    }
291
292    /// No transition leaves this state.
293    pub fn is_terminal(self) -> bool {
294        !matches!(
295            self,
296            ApprovalState::Pending | ApprovalState::Approved | ApprovalState::Claimed
297        )
298    }
299
300    pub fn as_str(self) -> &'static str {
301        match self {
302            ApprovalState::Pending => "pending",
303            ApprovalState::Approved => "approved",
304            ApprovalState::Denied => "denied",
305            ApprovalState::Revoked => "revoked",
306            ApprovalState::Cancelled => "cancelled",
307            ApprovalState::Expired => "expired",
308            ApprovalState::Claimed => "claimed",
309            ApprovalState::Settled => "settled",
310        }
311    }
312}
313
314/// What the effect did, as settled.
315#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
316#[serde(rename_all = "snake_case")]
317#[non_exhaustive]
318pub enum EffectOutcome {
319    Succeeded,
320    Failed,
321    /// It is not known whether the effect happened (a crash after the claim, a handler that
322    /// reports uncertainty, a reconciler that cannot tell). Never read as success or failure.
323    OutcomeUnknown,
324}
325
326/// The decision a principal makes about a request.
327#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
328#[serde(rename_all = "snake_case")]
329#[non_exhaustive]
330pub enum Decision {
331    Approve,
332    Deny,
333    /// Withdraw an approval (or a pending request) before it is claimed.
334    Revoke,
335}
336
337impl Decision {
338    pub fn as_str(self) -> &'static str {
339        match self {
340            Decision::Approve => "approve",
341            Decision::Deny => "deny",
342            Decision::Revoke => "revoke",
343        }
344    }
345}
346
347/// One request and everything the workflow knows about it.
348#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
349#[non_exhaustive]
350pub struct ApprovalRecord {
351    pub id: String,
352    pub action: String,
353    /// Lowercase hex digest (64 digits) of the exact effect being approved.
354    pub effect_digest: String,
355    pub account: Option<String>,
356    pub client: Option<String>,
357    pub requester: String,
358    /// Principals allowed to decide; empty means any verified principal that passes the account
359    /// scope.
360    pub approvers: BTreeSet<String>,
361    /// Redacted ([`redact_detail`]).
362    pub detail: String,
363    pub created_ms: u64,
364    pub expires_ms: u64,
365    /// Starts at 1 and increases by one with every transition. A decision binds to it.
366    pub revision: u64,
367    pub state: ApprovalState,
368    pub audit: AuditRequirement,
369    pub reconcilable: bool,
370    pub decided_by: Option<String>,
371    pub decided_ms: Option<u64>,
372    pub claim_id: Option<String>,
373    pub claimed_ms: Option<u64>,
374    pub outcome: Option<EffectOutcome>,
375    pub settled_ms: Option<u64>,
376    /// Why a request ended without an effect, or how a claim was resolved (redacted).
377    pub reason: Option<String>,
378}
379
380/// A request to create.
381#[derive(Debug, Clone)]
382#[non_exhaustive]
383pub struct NewRequest {
384    pub id: Option<String>,
385    pub action: String,
386    pub effect_digest: String,
387    pub account: Option<String>,
388    pub client: Option<String>,
389    pub requester: String,
390    pub ttl_ms: Option<u64>,
391    pub approvers: BTreeSet<String>,
392    pub detail: String,
393}
394
395impl NewRequest {
396    pub fn new(
397        action: impl Into<String>,
398        effect_digest: impl Into<String>,
399        requester: impl Into<String>,
400    ) -> Self {
401        Self {
402            id: None,
403            action: action.into(),
404            effect_digest: effect_digest.into(),
405            account: None,
406            client: None,
407            requester: requester.into(),
408            ttl_ms: None,
409            approvers: BTreeSet::new(),
410            detail: String::new(),
411        }
412    }
413
414    /// Use the app's own request id (Pulse's engine request id) instead of a generated one.
415    pub fn id(mut self, id: impl Into<String>) -> Self {
416        self.id = Some(id.into());
417        self
418    }
419
420    pub fn account(mut self, account: impl Into<String>) -> Self {
421        self.account = Some(account.into());
422        self
423    }
424
425    pub fn client(mut self, client: impl Into<String>) -> Self {
426        self.client = Some(client.into());
427        self
428    }
429
430    pub fn ttl_ms(mut self, ttl_ms: u64) -> Self {
431        self.ttl_ms = Some(ttl_ms);
432        self
433    }
434
435    pub fn approvers<I, S>(mut self, approvers: I) -> Self
436    where
437        I: IntoIterator<Item = S>,
438        S: Into<String>,
439    {
440        self.approvers = approvers.into_iter().map(Into::into).collect();
441        self
442    }
443
444    pub fn detail(mut self, detail: impl Into<String>) -> Self {
445        self.detail = detail.into();
446        self
447    }
448}
449
450/// Keep `detail` only when it is made of `[A-Za-z0-9_=. -]` (and short); otherwise the whole
451/// detail is replaced by `detail=withheld`, never a partial copy.
452pub fn redact_detail(detail: &str) -> String {
453    let plain = |c: char| c.is_ascii_alphanumeric() || "_=. -".contains(c);
454    let trimmed = detail.trim();
455    if trimmed.len() <= MAX_DETAIL && trimmed.chars().all(plain) {
456        trimmed.to_string()
457    } else {
458        "detail=withheld".to_string()
459    }
460}
461
462// ---------------------------------------------------------------------------------------------
463// Verification
464// ---------------------------------------------------------------------------------------------
465
466/// Which accounts a verified principal may decide for.
467#[derive(Debug, Clone, PartialEq, Eq)]
468pub enum AccountScope {
469    /// Not account-restricted (Pulse's owner key).
470    Any,
471    /// Only these accounts (Mail's per-account decisions).
472    Only(BTreeSet<String>),
473}
474
475impl AccountScope {
476    fn permits(&self, account: Option<&str>) -> bool {
477        match self {
478            AccountScope::Any => true,
479            AccountScope::Only(set) => account.is_some_and(|a| set.contains(a)),
480        }
481    }
482}
483
484/// Who the verifier proved decided, and what that principal may decide for.
485#[derive(Debug, Clone, PartialEq, Eq)]
486#[non_exhaustive]
487pub struct VerifiedPrincipal {
488    /// The principal's stable id (a device id, an enrolled key id, a local user).
489    pub id: String,
490    pub accounts: AccountScope,
491    /// Claims the verifier established, kept for the app's audit formatting. Not interpreted.
492    pub claims: BTreeMap<String, String>,
493}
494
495impl VerifiedPrincipal {
496    pub fn new(id: impl Into<String>, accounts: AccountScope) -> Self {
497        Self {
498            id: id.into(),
499            accounts,
500            claims: BTreeMap::new(),
501        }
502    }
503
504    pub fn claim(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
505        self.claims.insert(key.into(), value.into());
506        self
507    }
508}
509
510/// A nonce the proof carried, to be accepted once.
511#[derive(Debug, Clone, PartialEq, Eq)]
512pub struct Nonce {
513    pub value: String,
514    /// When the proof stops being valid; the store may forget the nonce after it.
515    pub expires_ms: u64,
516}
517
518/// A successful verification.
519#[derive(Debug, Clone, PartialEq, Eq)]
520#[non_exhaustive]
521pub struct Verified {
522    pub principal: VerifiedPrincipal,
523    pub nonce: Option<Nonce>,
524}
525
526impl Verified {
527    pub fn new(principal: VerifiedPrincipal) -> Self {
528        Self {
529            principal,
530            nonce: None,
531        }
532    }
533
534    pub fn with_nonce(mut self, value: impl Into<String>, expires_ms: u64) -> Self {
535        self.nonce = Some(Nonce {
536            value: value.into(),
537            expires_ms,
538        });
539        self
540    }
541}
542
543/// Why a proof was refused. The text is for the caller's log; it reaches the error returned to
544/// the app but never the record.
545#[derive(Debug, Clone, PartialEq, Eq)]
546pub struct VerifyError {
547    reason: String,
548}
549
550impl VerifyError {
551    pub fn new(reason: impl Into<String>) -> Self {
552        Self {
553            reason: reason.into(),
554        }
555    }
556}
557
558impl fmt::Display for VerifyError {
559    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
560        f.write_str(&self.reason)
561    }
562}
563
564impl std::error::Error for VerifyError {}
565
566/// What the verifier is asked to check: the **stored** request, the decision being made, and the
567/// caller's proof. The verifier must bind the proof to `record` (id, action, effect digest,
568/// account, client, expiry, revision) and to `decision`, so a proof made for another request or
569/// another decision fails here.
570#[derive(Debug)]
571pub struct VerifyInput<'a> {
572    pub record: &'a ApprovalRecord,
573    pub decision: Decision,
574    /// The caller's proof, opaque to this crate (a signature and its fields, a local session
575    /// token, a database row id).
576    pub proof: &'a [u8],
577    pub now_ms: u64,
578}
579
580/// The app's native verification: a Secure Enclave or P-256 signature checked against the
581/// enrolled key registry, a Mail user session, a local unlock. It must be fail-closed (any doubt
582/// is `Err`) and must not retain the proof.
583pub trait ApprovalVerifier: Send + Sync {
584    fn verify(&self, input: &VerifyInput<'_>) -> Result<Verified, VerifyError>;
585}
586
587// ---------------------------------------------------------------------------------------------
588// Store
589// ---------------------------------------------------------------------------------------------
590
591/// Why a store call failed.
592#[derive(Debug, Clone, PartialEq, Eq)]
593#[non_exhaustive]
594pub enum StoreError {
595    /// `insert` of an id that exists.
596    Exists,
597    /// `replace` of an id that does not exist.
598    NotFound,
599    /// `replace` found a revision other than the expected one.
600    Conflict,
601    /// The store is at a configured bound.
602    Full,
603    Unavailable(String),
604}
605
606impl fmt::Display for StoreError {
607    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
608        match self {
609            StoreError::Exists => f.write_str("the request id already exists"),
610            StoreError::NotFound => f.write_str("the request does not exist"),
611            StoreError::Conflict => f.write_str("the request changed since it was read"),
612            StoreError::Full => f.write_str("the approval store is full"),
613            StoreError::Unavailable(why) => write!(f, "the approval store is unavailable: {why}"),
614        }
615    }
616}
617
618impl std::error::Error for StoreError {}
619
620/// The durable half of the workflow. Implementations must make `replace` an atomic
621/// compare-and-set on `revision` and durable on `Ok`, and `claim_nonce` atomic; the coordinator's
622/// one-claim and replay guarantees are exactly these two operations.
623pub trait ApprovalStore: Send + Sync {
624    /// Store a new request. `Err(Exists)` when the id is taken.
625    fn insert(&self, record: &ApprovalRecord) -> Result<(), StoreError>;
626
627    fn get(&self, id: &str) -> Result<Option<ApprovalRecord>, StoreError>;
628
629    /// Replace the stored record with `record` only if the stored revision equals
630    /// `expected_revision`; `Err(Conflict)` otherwise, `Err(NotFound)` for an unknown id.
631    fn replace(&self, expected_revision: u64, record: &ApprovalRecord) -> Result<(), StoreError>;
632
633    /// Every record in one of `states`, oldest first.
634    fn list(&self, states: &[ApprovalState]) -> Result<Vec<ApprovalRecord>, StoreError>;
635
636    /// Remember `value` until `expires_ms`. `Ok(true)` when it was new, `Ok(false)` when it was
637    /// already remembered (a replay). `now_ms` lets the store forget expired nonces.
638    fn claim_nonce(&self, value: &str, expires_ms: u64, now_ms: u64) -> Result<bool, StoreError>;
639}
640
641#[derive(Default)]
642struct MemoryInner {
643    records: BTreeMap<String, ApprovalRecord>,
644    nonces: BTreeMap<String, u64>,
645}
646
647/// An in-process [`ApprovalStore`]. Cloning shares the state, so two coordinators over clones
648/// model two threads (or, with a fresh coordinator over the same store, a restart that kept the
649/// durable half). A real restart of the process forgets everything.
650#[derive(Clone, Default)]
651pub struct MemoryApprovalStore {
652    inner: Arc<Mutex<MemoryInner>>,
653}
654
655impl MemoryApprovalStore {
656    pub fn new() -> Self {
657        Self::default()
658    }
659
660    fn lock(&self) -> MutexGuard<'_, MemoryInner> {
661        // A panic while the lock was held cannot leave a half-applied update: every mutation is a
662        // single map operation.
663        self.inner.lock().unwrap_or_else(|e| e.into_inner())
664    }
665}
666
667impl ApprovalStore for MemoryApprovalStore {
668    fn insert(&self, record: &ApprovalRecord) -> Result<(), StoreError> {
669        let mut inner = self.lock();
670        if inner.records.contains_key(&record.id) {
671            return Err(StoreError::Exists);
672        }
673        inner.records.insert(record.id.clone(), record.clone());
674        Ok(())
675    }
676
677    fn get(&self, id: &str) -> Result<Option<ApprovalRecord>, StoreError> {
678        Ok(self.lock().records.get(id).cloned())
679    }
680
681    fn replace(&self, expected_revision: u64, record: &ApprovalRecord) -> Result<(), StoreError> {
682        let mut inner = self.lock();
683        match inner.records.get_mut(&record.id) {
684            None => Err(StoreError::NotFound),
685            Some(stored) if stored.revision != expected_revision => Err(StoreError::Conflict),
686            Some(stored) => {
687                *stored = record.clone();
688                Ok(())
689            }
690        }
691    }
692
693    fn list(&self, states: &[ApprovalState]) -> Result<Vec<ApprovalRecord>, StoreError> {
694        let mut out: Vec<ApprovalRecord> = self
695            .lock()
696            .records
697            .values()
698            .filter(|r| states.contains(&r.state))
699            .cloned()
700            .collect();
701        out.sort_by(|a, b| {
702            a.created_ms
703                .cmp(&b.created_ms)
704                .then_with(|| a.id.cmp(&b.id))
705        });
706        Ok(out)
707    }
708
709    fn claim_nonce(&self, value: &str, expires_ms: u64, now_ms: u64) -> Result<bool, StoreError> {
710        let mut inner = self.lock();
711        inner.nonces.retain(|_, until| *until > now_ms);
712        if inner.nonces.contains_key(value) {
713            return Ok(false);
714        }
715        if inner.nonces.len() >= MAX_NONCES {
716            return Err(StoreError::Full);
717        }
718        inner.nonces.insert(value.to_string(), expires_ms);
719        Ok(true)
720    }
721}
722
723// ---------------------------------------------------------------------------------------------
724// Audit
725// ---------------------------------------------------------------------------------------------
726
727/// What an audit row records.
728#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
729#[serde(rename_all = "snake_case")]
730#[non_exhaustive]
731pub enum AuditKind {
732    Requested,
733    Approved,
734    Denied,
735    Revoked,
736    Cancelled,
737    Expired,
738    Claimed,
739    Succeeded,
740    Failed,
741    OutcomeUnknown,
742}
743
744impl AuditKind {
745    pub fn as_str(self) -> &'static str {
746        match self {
747            AuditKind::Requested => "requested",
748            AuditKind::Approved => "approved",
749            AuditKind::Denied => "denied",
750            AuditKind::Revoked => "revoked",
751            AuditKind::Cancelled => "cancelled",
752            AuditKind::Expired => "expired",
753            AuditKind::Claimed => "claimed",
754            AuditKind::Succeeded => "succeeded",
755            AuditKind::Failed => "failed",
756            AuditKind::OutcomeUnknown => "outcome_unknown",
757        }
758    }
759}
760
761/// One attempted transition, with its candidate revision. A later store conflict may prevent
762/// commitment. Every free-text field has been through [`redact_detail`].
763#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
764#[non_exhaustive]
765pub struct AuditEvent {
766    pub request_id: String,
767    pub action: String,
768    pub account: Option<String>,
769    pub kind: AuditKind,
770    /// The deciding principal or requester, when there is one.
771    pub actor: Option<String>,
772    pub detail: String,
773    pub revision: u64,
774    pub at_ms: u64,
775}
776
777impl AuditEvent {
778    /// A one-line form with a fixed field order, for a log or a plain audit column.
779    pub fn line(&self) -> String {
780        format!(
781            "approval={} event={} action={} account={} actor={} revision={} detail={}",
782            self.request_id,
783            self.kind.as_str(),
784            self.action,
785            self.account.as_deref().unwrap_or("-"),
786            self.actor.as_deref().unwrap_or("-"),
787            self.revision,
788            self.detail
789        )
790    }
791}
792
793/// Why an append failed.
794#[derive(Debug, Clone, PartialEq, Eq)]
795pub struct AuditError {
796    reason: String,
797}
798
799impl AuditError {
800    pub fn new(reason: impl Into<String>) -> Self {
801        Self {
802            reason: reason.into(),
803        }
804    }
805}
806
807impl fmt::Display for AuditError {
808    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
809        f.write_str(&self.reason)
810    }
811}
812
813impl std::error::Error for AuditError {}
814
815/// The app's durable audit log. `Ok` means the row is durable.
816pub trait AuditSink: Send + Sync {
817    fn append(&self, event: &AuditEvent) -> Result<(), AuditError>;
818}
819
820// ---------------------------------------------------------------------------------------------
821// Reconciliation
822// ---------------------------------------------------------------------------------------------
823
824/// What an idempotent effect says about an unsettled claim.
825#[derive(Debug, Clone, Copy, PartialEq, Eq)]
826#[non_exhaustive]
827pub enum Reconciled {
828    Applied,
829    NotApplied,
830    /// It cannot be told.
831    Unknown,
832}
833
834/// For actions declared `reconcilable`: ask the effect whether a claimed request happened (a
835/// sent-mail ledger, an idempotency key). Never consulted for other actions.
836pub trait EffectReconciler: Send + Sync {
837    fn reconcile(&self, record: &ApprovalRecord) -> Reconciled;
838}
839
840// ---------------------------------------------------------------------------------------------
841// Errors
842// ---------------------------------------------------------------------------------------------
843
844/// Why a workflow call was refused.
845#[derive(Debug)]
846#[non_exhaustive]
847pub enum ApprovalError {
848    /// The action is not in the profile.
849    UnknownAction(String),
850    /// A field failed validation.
851    Invalid(&'static str),
852    /// The request id exists.
853    Duplicate(String),
854    NotFound(String),
855    TooManyOpen(usize),
856    /// The action requires audit before effect and no sink is configured.
857    AuditRequired,
858    /// The call is not allowed from the request's current state.
859    InvalidState {
860        id: String,
861        state: ApprovalState,
862    },
863    /// `claim` on a request that is still Pending.
864    NotApproved(String),
865    /// `claim` on a request that is already Claimed (or settled): the one claim is spent.
866    AlreadyClaimed(String),
867    /// A token that is not the one on record.
868    StaleClaim(String),
869    /// The request expired (and is now `Expired`).
870    Expired(String),
871    /// The decision named a revision other than the current one.
872    StaleRevision {
873        expected: u64,
874        found: u64,
875    },
876    /// Another caller changed the request first.
877    Conflict(String),
878    /// The verifier refused the proof.
879    Unverified(String),
880    /// The principal's accounts do not cover the request's account.
881    ScopeDenied,
882    /// The request names approvers and the principal is not one.
883    NotAnApprover,
884    /// The proof's nonce was already used.
885    ReplayedNonce,
886    /// Only the requester may cancel.
887    NotRequester,
888    /// A required audit row could not be appended; see the operation for what that left.
889    AuditFailed(String),
890    Store(StoreError),
891}
892
893impl fmt::Display for ApprovalError {
894    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
895        match self {
896            ApprovalError::UnknownAction(a) => write!(f, "approval: action {a:?} is not declared"),
897            ApprovalError::Invalid(why) => write!(f, "approval: {why}"),
898            ApprovalError::Duplicate(id) => write!(f, "approval: request {id} already exists"),
899            ApprovalError::NotFound(id) => write!(f, "approval: request {id} does not exist"),
900            ApprovalError::TooManyOpen(n) => write!(f, "approval: {n} requests are already open"),
901            ApprovalError::AuditRequired => {
902                f.write_str("approval: the action requires audit before effect and no sink is set")
903            }
904            ApprovalError::InvalidState { id, state } => {
905                write!(f, "approval: request {id} is {}", state.as_str())
906            }
907            ApprovalError::NotApproved(id) => write!(f, "approval: request {id} is not approved"),
908            ApprovalError::AlreadyClaimed(id) => {
909                write!(f, "approval: request {id} was already claimed")
910            }
911            ApprovalError::StaleClaim(id) => {
912                write!(
913                    f,
914                    "approval: the claim on request {id} is not the one on record"
915                )
916            }
917            ApprovalError::Expired(id) => write!(f, "approval: request {id} expired"),
918            ApprovalError::StaleRevision { expected, found } => write!(
919                f,
920                "approval: the decision is for revision {expected}, the request is at {found}"
921            ),
922            ApprovalError::Conflict(id) => {
923                write!(f, "approval: request {id} changed concurrently")
924            }
925            ApprovalError::Unverified(why) => write!(f, "approval: proof refused: {why}"),
926            ApprovalError::ScopeDenied => {
927                f.write_str("approval: the principal may not decide for this account")
928            }
929            ApprovalError::NotAnApprover => {
930                f.write_str("approval: the principal is not an approver of this request")
931            }
932            ApprovalError::ReplayedNonce => f.write_str("approval: the proof was already used"),
933            ApprovalError::NotRequester => {
934                f.write_str("approval: only the requester may cancel a request")
935            }
936            ApprovalError::AuditFailed(why) => write!(f, "approval: audit row not written: {why}"),
937            ApprovalError::Store(e) => write!(f, "approval: {e}"),
938        }
939    }
940}
941
942impl std::error::Error for ApprovalError {}
943
944impl From<StoreError> for ApprovalError {
945    fn from(e: StoreError) -> Self {
946        ApprovalError::Store(e)
947    }
948}
949
950// ---------------------------------------------------------------------------------------------
951// Coordinator
952// ---------------------------------------------------------------------------------------------
953
954/// The one right to execute an approved request. Not `Clone`: it is handed out once per request
955/// and presented to [`ApprovalCoordinator::settle`].
956#[derive(Debug)]
957#[non_exhaustive]
958pub struct ClaimToken {
959    pub request_id: String,
960    pub claim_id: String,
961    pub action: String,
962    pub effect_digest: String,
963    pub account: Option<String>,
964    pub client: Option<String>,
965    /// The revision of the `Claimed` record.
966    pub revision: u64,
967}
968
969/// Which unsettled claims [`ApprovalCoordinator::recover_claims`] resolves.
970#[derive(Debug, Clone, Copy, PartialEq, Eq)]
971pub enum ClaimRecovery {
972    /// Every claim: for startup, before any handler runs. At runtime it would resolve claims
973    /// whose handler is still working, which then fail to settle ([`ApprovalError::StaleClaim`]).
974    All,
975    /// Only claims older than this many milliseconds (a handler timeout).
976    OlderThan(u64),
977}
978
979/// The workflow coordinator. Cheap to share (`&self` everywhere); all cross-caller exclusion is
980/// the store's compare-and-set.
981pub struct ApprovalCoordinator {
982    store: Arc<dyn ApprovalStore>,
983    verifier: Arc<dyn ApprovalVerifier>,
984    audit: Option<Arc<dyn AuditSink>>,
985    clock: Arc<dyn ApprovalClock>,
986    profile: ApprovalProfile,
987    audit_failures: AtomicU64,
988}
989
990impl ApprovalCoordinator {
991    /// A coordinator over `store`, verifying with `verifier`, under `profile`. Fails when the
992    /// profile is invalid.
993    pub fn new(
994        profile: ApprovalProfile,
995        store: Arc<dyn ApprovalStore>,
996        verifier: Arc<dyn ApprovalVerifier>,
997    ) -> Result<Self, ApprovalError> {
998        profile.validate()?;
999        Ok(Self {
1000            store,
1001            verifier,
1002            audit: None,
1003            clock: Arc::new(SystemApprovalClock),
1004            profile,
1005            audit_failures: AtomicU64::new(0),
1006        })
1007    }
1008
1009    pub fn with_audit(mut self, sink: Arc<dyn AuditSink>) -> Self {
1010        self.audit = Some(sink);
1011        self
1012    }
1013
1014    pub fn with_clock(mut self, clock: Arc<dyn ApprovalClock>) -> Self {
1015        self.clock = clock;
1016        self
1017    }
1018
1019    /// How many best-effort audit appends failed since this coordinator was built.
1020    pub fn audit_failures(&self) -> u64 {
1021        self.audit_failures.load(Ordering::SeqCst)
1022    }
1023
1024    pub fn get(&self, id: &str) -> Result<Option<ApprovalRecord>, ApprovalError> {
1025        Ok(self.store.get(id)?)
1026    }
1027
1028    fn load(&self, id: &str) -> Result<ApprovalRecord, ApprovalError> {
1029        self.store
1030            .get(id)?
1031            .ok_or_else(|| ApprovalError::NotFound(id.to_string()))
1032    }
1033
1034    fn event(
1035        &self,
1036        record: &ApprovalRecord,
1037        kind: AuditKind,
1038        actor: Option<&str>,
1039        detail: &str,
1040    ) -> AuditEvent {
1041        AuditEvent {
1042            request_id: record.id.clone(),
1043            action: record.action.clone(),
1044            account: record.account.clone(),
1045            kind,
1046            actor: actor.map(str::to_string),
1047            detail: redact_detail(detail),
1048            revision: record.revision,
1049            at_ms: self.clock.now_ms(),
1050        }
1051    }
1052
1053    fn audit_best_effort(&self, event: AuditEvent) {
1054        if let Some(sink) = &self.audit {
1055            if sink.append(&event).is_err() {
1056                self.audit_failures.fetch_add(1, Ordering::SeqCst);
1057            }
1058        }
1059    }
1060
1061    /// Append `event` according to the record's requirement: `BeforeEffect` makes a failure an
1062    /// error, anything else counts it.
1063    fn audit_for(&self, record: &ApprovalRecord, event: AuditEvent) -> Result<(), ApprovalError> {
1064        match (&self.audit, record.audit) {
1065            (Some(sink), AuditRequirement::BeforeEffect) => sink
1066                .append(&event)
1067                .map_err(|e| ApprovalError::AuditFailed(e.to_string())),
1068            (None, AuditRequirement::BeforeEffect) => Err(ApprovalError::AuditRequired),
1069            _ => {
1070                self.audit_best_effort(event);
1071                Ok(())
1072            }
1073        }
1074    }
1075
1076    /// Replace `current` with the edited copy, revision plus one, through the store's
1077    /// compare-and-set.
1078    fn transition(
1079        &self,
1080        current: &ApprovalRecord,
1081        kind: AuditKind,
1082        actor: Option<&str>,
1083        detail: &str,
1084        edit: impl FnOnce(&mut ApprovalRecord),
1085    ) -> Result<ApprovalRecord, ApprovalError> {
1086        let mut next = current.clone();
1087        edit(&mut next);
1088        next.revision = current
1089            .revision
1090            .checked_add(1)
1091            .ok_or(ApprovalError::Invalid("approval revision exhausted"))?;
1092        self.audit_for(&next, self.event(&next, kind, actor, detail))?;
1093        match self.store.replace(current.revision, &next) {
1094            Ok(()) => Ok(next),
1095            Err(StoreError::Conflict) => Err(ApprovalError::Conflict(current.id.clone())),
1096            Err(e) => Err(ApprovalError::Store(e)),
1097        }
1098    }
1099
1100    /// If `record` is open and due, move it to `Expired` and return the expired record.
1101    /// `Ok(None)` when it is not due. A concurrent change that wins the race is not an error: the
1102    /// caller re-reads.
1103    fn expire_if_due(
1104        &self,
1105        record: &ApprovalRecord,
1106    ) -> Result<Option<ApprovalRecord>, ApprovalError> {
1107        if !record.state.is_open() || self.clock.now_ms() < record.expires_ms {
1108            return Ok(None);
1109        }
1110        match self.transition(record, AuditKind::Expired, None, "expired", |r| {
1111            r.state = ApprovalState::Expired;
1112            r.reason = Some("expired".to_string());
1113        }) {
1114            Ok(next) => Ok(Some(next)),
1115            Err(ApprovalError::Conflict(_)) => {
1116                // Someone else moved it (a decision, a cancel); report what is there now.
1117                let now = self.load(&record.id)?;
1118                Ok(if now.state == ApprovalState::Expired {
1119                    Some(now)
1120                } else {
1121                    None
1122                })
1123            }
1124            Err(e) => Err(e),
1125        }
1126    }
1127
1128    // ---- request ---------------------------------------------------------------------------
1129
1130    /// Open a request. The request is bound to `effect_digest` for its whole life.
1131    pub fn request(&self, new: NewRequest) -> Result<ApprovalRecord, ApprovalError> {
1132        let profile = self
1133            .profile
1134            .get(&new.action)
1135            .ok_or_else(|| ApprovalError::UnknownAction(new.action.clone()))?;
1136        if profile.audit == AuditRequirement::BeforeEffect && self.audit.is_none() {
1137            return Err(ApprovalError::AuditRequired);
1138        }
1139        let digest_ok = new.effect_digest.len() == 64
1140            && new
1141                .effect_digest
1142                .bytes()
1143                .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b));
1144        if !digest_ok {
1145            return Err(ApprovalError::Invalid(
1146                "effect digest must be 64 lowercase hex digits",
1147            ));
1148        }
1149        if !valid_label(&new.requester) {
1150            return Err(ApprovalError::Invalid("requester is not a plain label"));
1151        }
1152        for (present, label) in [(&new.account, "account"), (&new.client, "client")] {
1153            if let Some(value) = present {
1154                if !valid_label(value) {
1155                    return Err(ApprovalError::Invalid(match label {
1156                        "account" => "account is not a plain label",
1157                        _ => "client is not a plain label",
1158                    }));
1159                }
1160            }
1161        }
1162        if profile.requires_account && new.account.is_none() {
1163            return Err(ApprovalError::Invalid("the action requires an account"));
1164        }
1165        if profile.requires_client && new.client.is_none() {
1166            return Err(ApprovalError::Invalid("the action requires a client"));
1167        }
1168        if new.approvers.iter().any(|a| !valid_label(a)) {
1169            return Err(ApprovalError::Invalid("an approver is not a plain label"));
1170        }
1171        let ttl = new.ttl_ms.unwrap_or(profile.default_ttl_ms);
1172        if ttl == 0 || ttl > profile.max_ttl_ms {
1173            return Err(ApprovalError::Invalid(
1174                "lifetime must be positive and within the action's maximum",
1175            ));
1176        }
1177        let id = match &new.id {
1178            Some(id) if valid_label(id) && id.len() <= 128 => id.clone(),
1179            Some(_) => return Err(ApprovalError::Invalid("request id is not a plain label")),
1180            None => random_id()?,
1181        };
1182        self.sweep_expired()?;
1183        let open = self
1184            .store
1185            .list(&[ApprovalState::Pending, ApprovalState::Approved])?
1186            .len();
1187        if open >= self.profile.max_open {
1188            return Err(ApprovalError::TooManyOpen(open));
1189        }
1190        let now = self.clock.now_ms();
1191        let record = ApprovalRecord {
1192            id,
1193            action: new.action,
1194            effect_digest: new.effect_digest,
1195            account: new.account,
1196            client: new.client,
1197            requester: new.requester,
1198            approvers: new.approvers,
1199            detail: redact_detail(&new.detail),
1200            created_ms: now,
1201            expires_ms: now.saturating_add(ttl),
1202            revision: 1,
1203            state: ApprovalState::Pending,
1204            audit: profile.audit,
1205            reconcilable: profile.reconcilable,
1206            decided_by: None,
1207            decided_ms: None,
1208            claim_id: None,
1209            claimed_ms: None,
1210            outcome: None,
1211            settled_ms: None,
1212            reason: None,
1213        };
1214        let event = self.event(&record, AuditKind::Requested, Some(&record.requester), "");
1215        self.audit_for(&record, event)?;
1216        match self.store.insert(&record) {
1217            Ok(()) => {}
1218            Err(StoreError::Exists) => return Err(ApprovalError::Duplicate(record.id)),
1219            Err(e) => return Err(ApprovalError::Store(e)),
1220        }
1221        Ok(record)
1222    }
1223
1224    // ---- decide ----------------------------------------------------------------------------
1225
1226    /// Apply a decision. `revision` is the revision the approver saw; `proof` goes to the
1227    /// [`ApprovalVerifier`] untouched.
1228    ///
1229    /// `Approve` and `Deny` need the request `Pending`; `Revoke` needs it `Pending` or
1230    /// `Approved` (a claimed request cannot be revoked: its effect may be running).
1231    pub fn decide(
1232        &self,
1233        id: &str,
1234        decision: Decision,
1235        revision: u64,
1236        proof: &[u8],
1237    ) -> Result<ApprovalRecord, ApprovalError> {
1238        let record = self.load(id)?;
1239        if record.revision != revision {
1240            return Err(ApprovalError::StaleRevision {
1241                expected: revision,
1242                found: record.revision,
1243            });
1244        }
1245        if let Some(expired) = self.expire_if_due(&record)? {
1246            return Err(ApprovalError::Expired(expired.id));
1247        }
1248        let allowed = match decision {
1249            Decision::Approve | Decision::Deny => record.state == ApprovalState::Pending,
1250            Decision::Revoke => record.state.is_open(),
1251        };
1252        if !allowed {
1253            return Err(ApprovalError::InvalidState {
1254                id: record.id,
1255                state: record.state,
1256            });
1257        }
1258        let now = self.clock.now_ms();
1259        let verified = self
1260            .verifier
1261            .verify(&VerifyInput {
1262                record: &record,
1263                decision,
1264                proof,
1265                now_ms: now,
1266            })
1267            .map_err(|e| ApprovalError::Unverified(e.to_string()))?;
1268        if !verified
1269            .principal
1270            .accounts
1271            .permits(record.account.as_deref())
1272        {
1273            return Err(ApprovalError::ScopeDenied);
1274        }
1275        if !record.approvers.is_empty() && !record.approvers.contains(&verified.principal.id) {
1276            return Err(ApprovalError::NotAnApprover);
1277        }
1278        if let Some(nonce) = &verified.nonce {
1279            if nonce.expires_ms <= now {
1280                return Err(ApprovalError::Unverified(
1281                    "the proof has expired".to_string(),
1282                ));
1283            }
1284            if !self
1285                .store
1286                .claim_nonce(&nonce.value, nonce.expires_ms, now)?
1287            {
1288                return Err(ApprovalError::ReplayedNonce);
1289            }
1290        }
1291        let actor = verified.principal.id.clone();
1292        let (state, kind) = match decision {
1293            Decision::Approve => (ApprovalState::Approved, AuditKind::Approved),
1294            Decision::Deny => (ApprovalState::Denied, AuditKind::Denied),
1295            Decision::Revoke => (ApprovalState::Revoked, AuditKind::Revoked),
1296        };
1297        let next = self.transition(&record, kind, Some(&actor), decision.as_str(), |r| {
1298            r.state = state;
1299            r.decided_by = Some(actor.clone());
1300            r.decided_ms = Some(now);
1301        })?;
1302        Ok(next)
1303    }
1304
1305    /// The requester withdraws an open request.
1306    pub fn cancel(&self, id: &str, requester: &str) -> Result<ApprovalRecord, ApprovalError> {
1307        let record = self.load(id)?;
1308        if record.requester != requester {
1309            return Err(ApprovalError::NotRequester);
1310        }
1311        if let Some(expired) = self.expire_if_due(&record)? {
1312            return Err(ApprovalError::Expired(expired.id));
1313        }
1314        if !record.state.is_open() {
1315            return Err(ApprovalError::InvalidState {
1316                id: record.id,
1317                state: record.state,
1318            });
1319        }
1320        let next = self.transition(
1321            &record,
1322            AuditKind::Cancelled,
1323            Some(requester),
1324            "cancelled",
1325            |r| {
1326                r.state = ApprovalState::Cancelled;
1327                r.reason = Some("cancelled".to_string());
1328            },
1329        )?;
1330        Ok(next)
1331    }
1332
1333    /// Revoke every open request of `account` (the account was removed or disabled). Claimed
1334    /// requests are left alone. Returns how many were revoked.
1335    pub fn revoke_account(&self, account: &str, reason: &str) -> Result<usize, ApprovalError> {
1336        let reason = redact_detail(reason);
1337        let mut revoked = 0;
1338        for record in self
1339            .store
1340            .list(&[ApprovalState::Pending, ApprovalState::Approved])?
1341        {
1342            if record.account.as_deref() != Some(account) {
1343                continue;
1344            }
1345            match self.transition(&record, AuditKind::Revoked, None, &reason, |r| {
1346                r.state = ApprovalState::Revoked;
1347                r.reason = Some(reason.clone());
1348            }) {
1349                Ok(_) => revoked += 1,
1350                // A concurrent decision or claim won: it is no longer the open request this call
1351                // meant to revoke (a claim in flight is deliberately not revoked).
1352                Err(ApprovalError::Conflict(_)) => {}
1353                Err(e) => return Err(e),
1354            }
1355        }
1356        Ok(revoked)
1357    }
1358
1359    /// Move every open request that is past its expiry to `Expired`. Returns how many.
1360    pub fn sweep_expired(&self) -> Result<usize, ApprovalError> {
1361        let mut expired = 0;
1362        for record in self
1363            .store
1364            .list(&[ApprovalState::Pending, ApprovalState::Approved])?
1365        {
1366            if self.expire_if_due(&record)?.is_some() {
1367                expired += 1;
1368            }
1369        }
1370        Ok(expired)
1371    }
1372
1373    // ---- claim and settle ------------------------------------------------------------------
1374
1375    /// Take the one right to execute an approved request. Exactly one of any number of
1376    /// concurrent callers gets the token; the others see [`ApprovalError::AlreadyClaimed`] (or
1377    /// `Conflict` when they lost the race on the same revision).
1378    ///
1379    /// With [`AuditRequirement::BeforeEffect`] the `Claimed` row is appended before the
1380    /// compare-and-set. A failed append leaves `Approved` unchanged and returns no token.
1381    pub fn claim(&self, id: &str) -> Result<ClaimToken, ApprovalError> {
1382        let record = self.load(id)?;
1383        match record.state {
1384            ApprovalState::Approved => {}
1385            ApprovalState::Pending => return Err(ApprovalError::NotApproved(record.id)),
1386            ApprovalState::Claimed | ApprovalState::Settled => {
1387                return Err(ApprovalError::AlreadyClaimed(record.id))
1388            }
1389            state => {
1390                return Err(ApprovalError::InvalidState {
1391                    id: record.id,
1392                    state,
1393                })
1394            }
1395        }
1396        if let Some(expired) = self.expire_if_due(&record)? {
1397            return Err(ApprovalError::Expired(expired.id));
1398        }
1399        if record.audit == AuditRequirement::BeforeEffect && self.audit.is_none() {
1400            return Err(ApprovalError::AuditRequired);
1401        }
1402        let now = self.clock.now_ms();
1403        let claim_id = random_id()?;
1404        let claimed = match self.transition(
1405            &record,
1406            AuditKind::Claimed,
1407            record.decided_by.as_deref(),
1408            "",
1409            |r| {
1410                r.state = ApprovalState::Claimed;
1411                r.claim_id = Some(claim_id.clone());
1412                r.claimed_ms = Some(now);
1413            },
1414        ) {
1415            Ok(next) => next,
1416            Err(ApprovalError::Conflict(_)) => {
1417                // Lost the race: say what the winner made of it.
1418                return Err(match self.store.get(id)? {
1419                    Some(r)
1420                        if matches!(r.state, ApprovalState::Claimed | ApprovalState::Settled) =>
1421                    {
1422                        ApprovalError::AlreadyClaimed(r.id)
1423                    }
1424                    Some(r) if r.state.is_terminal() => ApprovalError::InvalidState {
1425                        id: r.id,
1426                        state: r.state,
1427                    },
1428                    _ => ApprovalError::Conflict(id.to_string()),
1429                });
1430            }
1431            Err(e) => return Err(e),
1432        };
1433        Ok(ClaimToken {
1434            request_id: claimed.id,
1435            claim_id,
1436            action: claimed.action,
1437            effect_digest: claimed.effect_digest,
1438            account: claimed.account,
1439            client: claimed.client,
1440            revision: claimed.revision,
1441        })
1442    }
1443
1444    /// Record what the effect did. Accepts only the token whose claim id is on record. With
1445    /// [`AuditRequirement::BeforeEffect`] the outcome row is appended first; if that fails the
1446    /// call returns [`ApprovalError::AuditFailed`], the request stays `Claimed` and the caller
1447    /// must not report success (it may retry with the same token).
1448    pub fn settle(
1449        &self,
1450        token: &ClaimToken,
1451        outcome: EffectOutcome,
1452        detail: &str,
1453    ) -> Result<ApprovalRecord, ApprovalError> {
1454        let record = self.load(&token.request_id)?;
1455        if record.state != ApprovalState::Claimed
1456            || record.claim_id.as_deref() != Some(token.claim_id.as_str())
1457        {
1458            return Err(ApprovalError::StaleClaim(token.request_id.clone()));
1459        }
1460        let detail = redact_detail(detail);
1461        let kind = match outcome {
1462            EffectOutcome::Succeeded => AuditKind::Succeeded,
1463            EffectOutcome::Failed => AuditKind::Failed,
1464            EffectOutcome::OutcomeUnknown => AuditKind::OutcomeUnknown,
1465        };
1466        let now = self.clock.now_ms();
1467        self.transition(&record, kind, record.decided_by.as_deref(), &detail, |r| {
1468            r.state = ApprovalState::Settled;
1469            r.outcome = Some(outcome);
1470            r.settled_ms = Some(now);
1471            r.reason = Some(detail.clone());
1472        })
1473        .map_err(|e| match e {
1474            // Recovery resolved the claim while the effect ran: the token is stale now.
1475            ApprovalError::Conflict(id) => ApprovalError::StaleClaim(id),
1476            other => other,
1477        })
1478    }
1479
1480    /// Resolve claims that were never settled (a crash, a handler that disappeared).
1481    ///
1482    /// A request whose action is `reconcilable` and a `reconciler` is given: `Applied` settles
1483    /// it `Succeeded`, `NotApplied` settles it `Failed` (the effect did not happen; a new request
1484    /// is needed to try again), `Unknown` settles it `OutcomeUnknown`. Every other claim settles
1485    /// as `OutcomeUnknown`: the coordinator cannot know whether an external effect happened and
1486    /// never assumes it did not. Required audit rows gate recovery success as they gate settle.
1487    /// Returns the settled records.
1488    pub fn recover_claims(
1489        &self,
1490        which: ClaimRecovery,
1491        reconciler: Option<&dyn EffectReconciler>,
1492    ) -> Result<Vec<ApprovalRecord>, ApprovalError> {
1493        let now = self.clock.now_ms();
1494        let mut settled = Vec::new();
1495        for record in self.store.list(&[ApprovalState::Claimed])? {
1496            if let ClaimRecovery::OlderThan(age) = which {
1497                let claimed = record.claimed_ms.unwrap_or(0);
1498                if now.saturating_sub(claimed) < age {
1499                    continue;
1500                }
1501            }
1502            let (outcome, why) = match (record.reconcilable, reconciler) {
1503                (true, Some(r)) => match r.reconcile(&record) {
1504                    Reconciled::Applied => (EffectOutcome::Succeeded, "reconciled_applied"),
1505                    Reconciled::NotApplied => (EffectOutcome::Failed, "reconciled_not_applied"),
1506                    Reconciled::Unknown => (EffectOutcome::OutcomeUnknown, "reconciled_unknown"),
1507                },
1508                _ => (EffectOutcome::OutcomeUnknown, "unsettled_claim"),
1509            };
1510            let kind = match outcome {
1511                EffectOutcome::Succeeded => AuditKind::Succeeded,
1512                EffectOutcome::Failed => AuditKind::Failed,
1513                EffectOutcome::OutcomeUnknown => AuditKind::OutcomeUnknown,
1514            };
1515            match self.transition(&record, kind, None, why, |r| {
1516                r.state = ApprovalState::Settled;
1517                r.outcome = Some(outcome);
1518                r.settled_ms = Some(now);
1519                r.reason = Some(why.to_string());
1520            }) {
1521                Ok(next) => {
1522                    settled.push(next);
1523                }
1524                // The handler settled it between the listing and now.
1525                Err(ApprovalError::Conflict(_)) => {}
1526                Err(e) => return Err(e),
1527            }
1528        }
1529        Ok(settled)
1530    }
1531}
1532
1533fn random_id() -> Result<String, ApprovalError> {
1534    let mut raw = [0u8; 16];
1535    getrandom::fill(&mut raw)
1536        .map_err(|_| ApprovalError::Invalid("the system random source is unavailable"))?;
1537    Ok(raw.iter().map(|b| format!("{b:02x}")).collect())
1538}