Give the write-ahead recovery record a single home

Crash recovery was smeared across three files with no owner. The two kinds of
write-ahead record (a commit saga and an account-version transition) had their
type definitions, persistence, phase read, and completion dispatch spread over
commit.rs, transition.rs, and lifecycle.rs, so no single place told the story
of what a pending record is or how it completes. recover() was a god-dispatcher
matching on both record kind and saga phase, and read_pending_phase faked a
keyed lookup by scanning every pending saga and re-deserializing each blob,
because SagaStore had no get_saga.

Concentrate the concept in a new `pending` module that owns the record types,
one decode/save, and PendingRecord::complete (the per-kind, per-phase
completion, absorbing the old recover body and complete_transition). recover()
becomes a loop over complete(). Add a keyed SagaStore::get_saga so the failed
commit path reads one record instead of scanning, with a conformance test on
both backends. finalize_envelope and drive_envelope_saga stay on Ledger because
the live commit path shares them; the pending module sequences them for
recovery.

Every recovery crash-window test passes unchanged, which is the evidence the
seam preserves behavior.

crates/kuatia-storage-sql/src/lib.rs

@@ -1230,6 +1230,23 @@ impl SagaStore for SqlStore {
         Ok(result)
     }
 
+    async fn get_saga(&self, id: &i64) -> Result<Option<Vec<u8>>, StoreError> {
+        let row = sqlx::query("SELECT data FROM sagas WHERE id = $1")
+            .bind(*id)
+            .fetch_optional(&self.pool)
+            .await
+            .map_err(|e| StoreError::Internal(e.to_string()))?;
+        match row {
+            Some(row) => {
+                let data_hex: String = row
+                    .try_get("data")
+                    .map_err(|e| StoreError::Internal(e.to_string()))?;
+                Ok(Some(from_hex(&data_hex)?))
+            }
+            None => Ok(None),
+        }
+    }
+
     async fn delete_saga(&self, id: &i64) -> Result<(), StoreError> {
         sqlx::query("DELETE FROM sagas WHERE id = $1")
             .bind(*id)

crates/kuatia-storage/src/mem_store.rs

@@ -448,6 +448,11 @@ impl SagaStore for InMemoryStore {
         Ok(sagas.iter().map(|(k, v)| (*k, v.clone())).collect())
     }
 
+    async fn get_saga(&self, id: &i64) -> Result<Option<Vec<u8>>, StoreError> {
+        let sagas = self.sagas.read().await;
+        Ok(sagas.get(id).cloned())
+    }
+
     async fn delete_saga(&self, id: &i64) -> Result<(), StoreError> {
         let mut sagas = self.sagas.write().await;
         sagas.remove(id);

crates/kuatia-storage/src/store.rs

@@ -237,6 +237,10 @@ pub trait SagaStore: Send + Sync {
     async fn save_saga(&self, id: &i64, data: Vec<u8>) -> Result<(), StoreError>;
     /// Load all pending (incomplete) saga states.
     async fn list_pending_sagas(&self) -> Result<Vec<(i64, Vec<u8>)>, StoreError>;
+    /// Load one saga state by id, or `None` if no record is stored under `id`.
+    /// A keyed read so a caller checking a single in-flight saga does not scan
+    /// every pending record.
+    async fn get_saga(&self, id: &i64) -> Result<Option<Vec<u8>>, StoreError>;
     /// Delete a completed saga state.
     async fn delete_saga(&self, id: &i64) -> Result<(), StoreError>;
 }

crates/kuatia-storage/src/store_tests.rs

@@ -1144,6 +1144,22 @@ pub async fn delete_saga(store: &(impl Store + 'static)) {
     assert!(pending.is_empty());
 }
 
+/// A keyed `get_saga` returns the stored blob, and `None` for an absent or
+/// deleted id.
+pub async fn get_saga_by_id(store: &(impl Store + 'static)) {
+    let id: i64 = 42;
+    let data = vec![7, 8, 9];
+    assert!(store.get_saga(&id).await.unwrap().is_none());
+
+    store.save_saga(&id, data.clone()).await.unwrap();
+    assert_eq!(store.get_saga(&id).await.unwrap(), Some(data));
+    // A different id is still absent while this one exists.
+    assert!(store.get_saga(&99).await.unwrap().is_none());
+
+    store.delete_saga(&id).await.unwrap();
+    assert!(store.get_saga(&id).await.unwrap().is_none());
+}
+
 // ---------------------------------------------------------------------------
 // EventStore tests
 // ---------------------------------------------------------------------------
@@ -1387,6 +1403,7 @@ macro_rules! store_tests {
             query_transfers_store_wide,
             // SagaStore
             save_and_list_sagas,
+            get_saga_by_id,
             delete_saga,
             // EventStore
             append_and_query_events,

crates/kuatia/src/ledger.rs

@@ -29,6 +29,7 @@ mod envelope_saga;
 mod balance;
 mod commit;
 mod lifecycle;
+mod pending;
 mod projection;
 mod query;
 mod transition;

crates/kuatia/src/ledger/commit.rs

@@ -31,53 +31,7 @@ use crate::saga::{
     FinalizeInput, LedgerCtx, ReserveInput, apply_and_verify, consume_reserved, verify_postings,
 };
 
-/// Phase of an in-flight commit, persisted with the write-ahead record so
-/// recovery knows whether validation has completed.
-#[derive(Clone, Copy, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
-enum SagaPhase {
-    /// Saved before reserve. Validation has not necessarily run, so recovery must
-    /// re-reserve and re-validate before it can commit.
-    Reserving,
-    /// Saved at the start of finalize — after validation passed and just before
-    /// the consumed postings begin being removed from the reserved index (the
-    /// point of no return). Recovery rolls forward without re-validating.
-    Finalizing,
-}
-
-/// Write-ahead record for an in-flight commit, persisted via `SagaStore` before
-/// the saga mutates anything and removed once it reaches a terminal state. On
-/// startup [`Ledger::recover`] completes any that survive a crash.
-#[derive(serde::Serialize, serde::Deserialize)]
-struct PendingSaga {
-    envelope: Envelope,
-    reservation: ReservationId,
-    phase: SagaPhase,
-}
-
-/// Write-ahead record for an in-flight account-version transition
-/// (freeze/unfreeze/close). The transition appends a new account version and then
-/// its lifecycle event; a crash between the two leaves a version bump with no
-/// event. Persisting this before either write lets [`Ledger::recover`] roll the
-/// transition forward, re-appending the (idempotent) event.
-#[derive(serde::Serialize, serde::Deserialize)]
-pub(super) struct PendingTransition {
-    /// The next account version to append: version already bumped, flag flipped.
-    pub next: Account,
-    /// The lifecycle event paired with this version bump. It carries the target
-    /// version, so re-appending it on recovery dedups to the original.
-    pub event: LedgerEventKind,
-}
-
-/// The two kinds of write-ahead record the [`SagaStore`](kuatia_storage::store::SagaStore)
-/// holds, tagged so [`Ledger::recover`] can tell an envelope commit saga from an
-/// account transition and complete each through its own path.
-#[derive(serde::Serialize, serde::Deserialize)]
-enum PendingRecord {
-    /// A two-step envelope commit saga (reserve → finalize).
-    Envelope(PendingSaga),
-    /// A single account-version transition (append version + lifecycle event).
-    Transition(PendingTransition),
-}
+use super::pending::{PendingRecord, SagaPhase};
 
 /// State loaded in phase 1, passed to the pure validation in phase 2.
 struct LoadedState {
@@ -258,7 +212,8 @@ impl Ledger {
         // mutation. The finalize step bumps the phase to Finalizing.
         let reservation = ReservationId::default();
         let saga_id = reservation.0;
-        self.save_pending(&envelope, reservation, SagaPhase::Reserving)
+        PendingRecord::envelope(envelope.clone(), reservation, SagaPhase::Reserving)
+            .save(self, saga_id)
             .await?;
 
         // Commit does not touch the balance projection (ADR-0019): cache points
@@ -272,7 +227,7 @@ impl Ledger {
         // the half-applied commit forward.
         let safe_to_delete = match &result {
             Ok(_) => true,
-            Err(_) => self.read_pending_phase(saga_id).await? != Some(SagaPhase::Finalizing),
+            Err(_) => self.saga_phase(saga_id).await? != Some(SagaPhase::Finalizing),
         };
         if safe_to_delete {
             self.store.delete_saga(&saga_id).await?;
@@ -282,7 +237,7 @@ impl Ledger {
 
     /// Build and run the envelope saga (reserve → finalize) to a terminal
     /// outcome, returning the resulting receipt.
-    async fn drive_envelope_saga(
+    pub(super) async fn drive_envelope_saga(
         self: &Arc<Self>,
         envelope: Envelope,
         reservation: ReservationId,
@@ -319,73 +274,23 @@ impl Ledger {
         }
     }
 
-    /// Complete every pending saga left by a crash. Call on startup; returns how
-    /// many were processed.
+    /// Complete every pending write-ahead record left by a crash. Call on
+    /// startup; returns how many were processed.
     ///
-    /// Recovery branches on the persisted phase. A `Reserving` saga had not
-    /// necessarily validated, so it is re-run through the real saga (which
-    /// re-reserves and **re-validates** — aborting cleanly if the postings were
-    /// taken or an account was frozen meanwhile). A `Finalizing` saga had already
-    /// validated and owns its postings, so it is rolled forward through the
-    /// verified `finalize_envelope`. Either way the record is removed only once
-    /// the work is committed or safely abandoned.
+    /// Each record is decoded and driven to a terminal state by
+    /// `PendingRecord::complete` (in the `pending` module), which owns the
+    /// per-kind completion: a transition rolls forward; an envelope commit
+    /// branches on its persisted phase — a `Reserving` saga is re-run and
+    /// re-validated, a `Finalizing` saga is rolled forward through the verified
+    /// `finalize_envelope`.
     #[instrument(skip(self), name = "ledger.recover")]
     pub async fn recover(self: &Arc<Self>) -> Result<usize, LedgerError> {
         let pending = self.store.list_pending_sagas().await?;
         let count = pending.len();
         for (saga_id, blob) in pending {
-            let record: PendingRecord = serde_json::from_slice(&blob)
-                .map_err(|e| LedgerError::Store(StoreError::Internal(e.to_string())))?;
-
-            match record {
-                PendingRecord::Transition(PendingTransition { next, event }) => {
-                    // Roll the account transition forward: append the version if it
-                    // is not yet present, then (re-)append the idempotent event.
-                    // Both steps no-op when already applied, so this is safe to run
-                    // in any crash window.
-                    self.complete_transition(saga_id, next, event).await?;
-                }
-                PendingRecord::Envelope(PendingSaga {
-                    envelope,
-                    reservation,
-                    phase,
-                }) => {
-                    // The transfer record is durable, but a full commit is more
-                    // than the transfer row: it also includes the committed event,
-                    // appended *after* store_transfer. A crash in that window
-                    // leaves the record present yet the event missing, so repair
-                    // the whole end-state (idempotent) before clearing the record.
-                    let tid = envelope_id(&envelope);
-                    if self.store.get_transfer(&tid).await?.is_some() {
-                        self.append_committed_event(tid).await?;
-                        self.store.delete_saga(&saga_id).await?;
-                        continue;
-                    }
-
-                    match phase {
-                        SagaPhase::Finalizing => {
-                            // Validation passed and the postings are ours; roll
-                            // forward. Keep the record if completion fails so a
-                            // later run retries.
-                            if self.finalize_envelope(&envelope, reservation).await.is_ok() {
-                                self.store.delete_saga(&saga_id).await?;
-                            }
-                        }
-                        SagaPhase::Reserving => {
-                            // Re-run the validating saga. On failure, delete only if
-                            // it did not reach finalize (clean abort); otherwise
-                            // keep for next run.
-                            let result = self.drive_envelope_saga(envelope, reservation).await;
-                            let safe_to_delete = result.is_ok()
-                                || self.read_pending_phase(saga_id).await?
-                                    != Some(SagaPhase::Finalizing);
-                            if safe_to_delete {
-                                self.store.delete_saga(&saga_id).await?;
-                            }
-                        }
-                    }
-                }
-            }
+            PendingRecord::decode(&blob)?
+                .complete(self, saga_id)
+                .await?;
         }
         Ok(count)
     }
@@ -439,7 +344,8 @@ impl Ledger {
         }
 
         // Point of no return: record Finalizing before any posting is consumed.
-        self.save_pending(envelope, reservation, SagaPhase::Finalizing)
+        PendingRecord::envelope(envelope.clone(), reservation, SagaPhase::Finalizing)
+            .save(self, reservation.0)
             .await?;
 
         // The authoritative double-spend guard (see `consume_reserved`): consume
@@ -508,7 +414,7 @@ impl Ledger {
     /// retried finalize both call this to repair the committed end-state.
     /// `append_event` dedups on the transfer id, so calling it more than once for
     /// the same transfer is a no-op.
-    async fn append_committed_event(&self, tid: EnvelopeId) -> Result<(), LedgerError> {
+    pub(super) async fn append_committed_event(&self, tid: EnvelopeId) -> Result<(), LedgerError> {
         self.store
             .append_event(&LedgerEvent {
                 seq: 0,
@@ -519,56 +425,15 @@ impl Ledger {
         Ok(())
     }
 
-    /// Persist the write-ahead pending-saga record (upsert on the reservation id).
-    async fn save_pending(
-        &self,
-        envelope: &Envelope,
-        reservation: ReservationId,
-        phase: SagaPhase,
-    ) -> Result<(), LedgerError> {
-        let blob = serde_json::to_vec(&PendingRecord::Envelope(PendingSaga {
-            envelope: envelope.clone(),
-            reservation,
-            phase,
-        }))
-        .map_err(|e| LedgerError::Store(StoreError::Internal(e.to_string())))?;
-        self.store.save_saga(&reservation.0, blob).await?;
-        Ok(())
-    }
-
-    /// Persist the write-ahead record for an account-version transition, keyed by
-    /// a fresh unique id, and return that id so the caller can delete the record
-    /// once the transition is complete. Shares the reservation-id generator so the
-    /// key never collides with an in-flight commit saga's key.
-    pub(super) async fn save_transition(
-        &self,
-        next: &Account,
-        event: &LedgerEventKind,
-    ) -> Result<i64, LedgerError> {
-        let saga_id = ReservationId::default().0;
-        let blob = serde_json::to_vec(&PendingRecord::Transition(PendingTransition {
-            next: next.clone(),
-            event: event.clone(),
-        }))
-        .map_err(|e| LedgerError::Store(StoreError::Internal(e.to_string())))?;
-        self.store.save_saga(&saga_id, blob).await?;
-        Ok(saga_id)
-    }
-
-    /// Read the persisted phase of a pending *envelope* saga, if one exists under
-    /// `saga_id`. A transition record (no phase) reads as `None`.
-    async fn read_pending_phase(&self, saga_id: i64) -> Result<Option<SagaPhase>, LedgerError> {
-        for (id, blob) in self.store.list_pending_sagas().await? {
-            if id == saga_id {
-                let record: PendingRecord = serde_json::from_slice(&blob)
-                    .map_err(|e| LedgerError::Store(StoreError::Internal(e.to_string())))?;
-                return Ok(match record {
-                    PendingRecord::Envelope(s) => Some(s.phase),
-                    PendingRecord::Transition(_) => None,
-                });
-            }
+    /// The persisted commit phase of the record stored under `saga_id`, via a
+    /// keyed read. `None` when no record exists there or it is a transition
+    /// (which has no phase). Used to decide whether a failed commit reached the
+    /// point of no return.
+    pub(super) async fn saga_phase(&self, saga_id: i64) -> Result<Option<SagaPhase>, LedgerError> {
+        match self.store.get_saga(&saga_id).await? {
+            Some(blob) => Ok(PendingRecord::decode(&blob)?.envelope_phase()),
+            None => Ok(None),
         }
-        Ok(None)
     }
 
     // -----------------------------------------------------------------------
@@ -637,6 +502,7 @@ impl Ledger {
 
 #[cfg(test)]
 mod recovery_tests {
+    use super::super::pending::{PendingSaga, PendingTransition};
     use super::*;
     use kuatia_core::{Account, AccountFlags, ReservationId, TransferBuilder};
     use kuatia_storage::mem_store::InMemoryStore;

crates/kuatia/src/ledger/pending.rs

@@ -0,0 +1,217 @@
+//! The write-ahead record awaiting recovery, and how each kind completes.
+//!
+//! A commit (reserve → finalize) and an account-version transition (append
+//! version → append event) are each more than one store write with no shared
+//! transaction, so a crash mid-sequence can leave a half-applied state. Before
+//! either mutates anything it persists a [`PendingRecord`] via `SagaStore`; on
+//! startup [`Ledger::recover`](super::Ledger::recover) loads every surviving
+//! record and drives it to a terminal state through [`PendingRecord::complete`].
+//!
+//! This module owns the whole write-ahead concept behind one seam: what a
+//! pending record *is*, how it is (de)serialized, how it is persisted, and how
+//! each kind completes. The completion primitives it calls (`finalize_envelope`,
+//! `drive_envelope_saga`) stay on [`Ledger`] because the live commit path shares
+//! them; this module sequences them for the recovery path.
+
+use std::sync::Arc;
+
+use kuatia_core::{Account, Envelope, ReservationId, envelope_id};
+use kuatia_storage::error::StoreError;
+use kuatia_storage::events::{LedgerEvent, LedgerEventKind};
+
+use super::{Ledger, now_millis};
+use crate::error::LedgerError;
+
+/// Phase of an in-flight commit, persisted with the write-ahead record so
+/// recovery knows whether validation has completed.
+#[derive(Clone, Copy, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
+pub(super) enum SagaPhase {
+    /// Saved before reserve. Validation has not necessarily run, so recovery must
+    /// re-reserve and re-validate before it can commit.
+    Reserving,
+    /// Saved at the start of finalize — after validation passed and just before
+    /// the consumed postings begin being removed from the reserved index (the
+    /// point of no return). Recovery rolls forward without re-validating.
+    Finalizing,
+}
+
+/// Write-ahead record for an in-flight commit (reserve → finalize). Persisted
+/// before the saga mutates anything and removed once it reaches a terminal
+/// state.
+#[derive(serde::Serialize, serde::Deserialize)]
+pub(super) struct PendingSaga {
+    pub(super) envelope: Envelope,
+    pub(super) reservation: ReservationId,
+    pub(super) phase: SagaPhase,
+}
+
+/// Write-ahead record for an in-flight account-version transition
+/// (freeze/unfreeze/close). The transition appends a new account version and then
+/// its lifecycle event; a crash between the two leaves a version bump with no
+/// event. Persisting this before either write lets recovery roll the transition
+/// forward, re-appending the (idempotent) event.
+#[derive(serde::Serialize, serde::Deserialize)]
+pub(super) struct PendingTransition {
+    /// The next account version to append: version already bumped, flag flipped.
+    pub(super) next: Account,
+    /// The lifecycle event paired with this version bump. It carries the target
+    /// version, so re-appending it on recovery dedups to the original.
+    pub(super) event: LedgerEventKind,
+}
+
+/// The two kinds of write-ahead record the [`SagaStore`] holds, tagged so
+/// recovery can tell an envelope commit saga from an account transition and
+/// complete each through its own path.
+#[derive(serde::Serialize, serde::Deserialize)]
+pub(super) enum PendingRecord {
+    /// A two-step envelope commit saga (reserve → finalize).
+    Envelope(PendingSaga),
+    /// A single account-version transition (append version + lifecycle event).
+    Transition(PendingTransition),
+}
+
+impl PendingRecord {
+    /// A commit write-ahead record at the given phase.
+    pub(super) fn envelope(
+        envelope: Envelope,
+        reservation: ReservationId,
+        phase: SagaPhase,
+    ) -> Self {
+        Self::Envelope(PendingSaga {
+            envelope,
+            reservation,
+            phase,
+        })
+    }
+
+    /// An account-transition write-ahead record.
+    pub(super) fn transition(next: Account, event: LedgerEventKind) -> Self {
+        Self::Transition(PendingTransition { next, event })
+    }
+
+    /// Decode a record from its stored bytes. The single decoder for the
+    /// write-ahead format, shared by `recover` and the keyed phase read.
+    pub(super) fn decode(blob: &[u8]) -> Result<Self, LedgerError> {
+        serde_json::from_slice(blob)
+            .map_err(|e| LedgerError::Store(StoreError::Internal(e.to_string())))
+    }
+
+    /// The commit phase of an envelope record; `None` for a transition record,
+    /// which has no phase.
+    pub(super) fn envelope_phase(&self) -> Option<SagaPhase> {
+        match self {
+            Self::Envelope(s) => Some(s.phase),
+            Self::Transition(_) => None,
+        }
+    }
+
+    /// Persist this record under `saga_id` (upsert on the id).
+    pub(super) async fn save(&self, ledger: &Ledger, saga_id: i64) -> Result<(), LedgerError> {
+        let blob = serde_json::to_vec(self)
+            .map_err(|e| LedgerError::Store(StoreError::Internal(e.to_string())))?;
+        ledger.store.save_saga(&saga_id, blob).await?;
+        Ok(())
+    }
+
+    /// Drive this record to a terminal state and clear it when safe. Called by
+    /// [`Ledger::recover`](super::Ledger::recover) for every surviving record.
+    ///
+    /// A transition rolls forward (any completion error propagates, so recovery
+    /// retries on the next run). An envelope commit branches on its phase, and
+    /// its drive/finalize failures are absorbed here (the record is kept for a
+    /// later run) rather than aborting recovery of the remaining records.
+    pub(super) async fn complete(
+        self,
+        ledger: &Arc<Ledger>,
+        saga_id: i64,
+    ) -> Result<(), LedgerError> {
+        match self {
+            Self::Transition(PendingTransition { next, event }) => {
+                complete_transition(ledger, saga_id, next, event).await
+            }
+            Self::Envelope(PendingSaga {
+                envelope,
+                reservation,
+                phase,
+            }) => complete_envelope(ledger, saga_id, envelope, reservation, phase).await,
+        }
+    }
+}
+
+/// Roll a crash-interrupted transition forward and clear its write-ahead record.
+///
+/// Idempotent in every crash window: the version append runs only into an empty
+/// version slot (`append_account_version` requires `version == current + 1`, so a
+/// blind retry after it applied would fail), and the event carries its target
+/// version so re-appending it dedups to the original. The empty-slot guard also
+/// subsumes the forward path's is_closed check: a close always bumps the version,
+/// so a since-closed account sits at `version >= next.version` and is skipped.
+async fn complete_transition(
+    ledger: &Ledger,
+    saga_id: i64,
+    next: Account,
+    event: LedgerEventKind,
+) -> Result<(), LedgerError> {
+    // The account is guaranteed to exist here (its version was bumped, or is
+    // about to be), so a read failure is transient or a real invariant breach,
+    // not "not found": surface it verbatim so recovery retries.
+    let current = ledger.store.get_account(&next.id).await?;
+    if current.version < next.version {
+        ledger.store.append_account_version(next).await?;
+    }
+    ledger
+        .store
+        .append_event(&LedgerEvent {
+            seq: 0,
+            timestamp: now_millis()?,
+            kind: event,
+        })
+        .await?;
+    ledger.store.delete_saga(&saga_id).await?;
+    Ok(())
+}
+
+/// Complete a crash-interrupted commit and clear its record when safe.
+async fn complete_envelope(
+    ledger: &Arc<Ledger>,
+    saga_id: i64,
+    envelope: Envelope,
+    reservation: ReservationId,
+    phase: SagaPhase,
+) -> Result<(), LedgerError> {
+    // The transfer record is durable, but a full commit is more than the transfer
+    // row: it also includes the committed event, appended *after* store_transfer.
+    // A crash in that window leaves the record present yet the event missing, so
+    // repair the whole end-state (idempotent) before clearing the record.
+    let tid = envelope_id(&envelope);
+    if ledger.store.get_transfer(&tid).await?.is_some() {
+        ledger.append_committed_event(tid).await?;
+        ledger.store.delete_saga(&saga_id).await?;
+        return Ok(());
+    }
+
+    match phase {
+        SagaPhase::Finalizing => {
+            // Validation passed and the postings are ours; roll forward. Keep the
+            // record if completion fails so a later run retries.
+            if ledger
+                .finalize_envelope(&envelope, reservation)
+                .await
+                .is_ok()
+            {
+                ledger.store.delete_saga(&saga_id).await?;
+            }
+        }
+        SagaPhase::Reserving => {
+            // Re-run the validating saga. On failure, delete only if it did not
+            // reach finalize (clean abort); otherwise keep for the next run.
+            let result = ledger.drive_envelope_saga(envelope, reservation).await;
+            let safe_to_delete =
+                result.is_ok() || ledger.saga_phase(saga_id).await? != Some(SagaPhase::Finalizing);
+            if safe_to_delete {
+                ledger.store.delete_saga(&saga_id).await?;
+            }
+        }
+    }
+    Ok(())
+}

crates/kuatia/src/ledger/transition.rs

@@ -15,9 +15,10 @@
 //! already present, and the event carries its target version so a second append
 //! dedups to the original.
 
-use kuatia_core::{Account, AccountFlags, AccountId};
+use kuatia_core::{AccountFlags, AccountId, ReservationId};
 use kuatia_storage::events::{LedgerEvent, LedgerEventKind};
 
+use super::pending::PendingRecord;
 use super::{Ledger, now_millis};
 use crate::error::LedgerError;
 
@@ -51,8 +52,13 @@ impl Ledger {
         let event = make_event(*id, next.version);
 
         // Write-ahead before either write. A crash between the version append and
-        // the event append is then repaired by recover(), not left dangling.
-        let saga_id = self.save_transition(&next, &event).await?;
+        // the event append is then repaired by recover(), not left dangling. The
+        // key shares the reservation-id generator so it never collides with an
+        // in-flight commit saga's key.
+        let saga_id = ReservationId::default().0;
+        PendingRecord::transition(next.clone(), event.clone())
+            .save(self, saga_id)
+            .await?;
 
         let expected = next.version;
         if self.store.append_account_version(next).await? == 0 {
@@ -74,42 +80,4 @@ impl Ledger {
         self.store.delete_saga(&saga_id).await?;
         Ok(())
     }
-
-    /// Roll a crash-interrupted transition forward and clear its write-ahead
-    /// record. Called by [`recover`](Ledger::recover) for a persisted
-    /// [`PendingTransition`](super::commit).
-    ///
-    /// Idempotent in every crash window: the version append runs only when the
-    /// version is not yet present (`append_account_version` requires
-    /// `version == current + 1`, so a blind retry after it applied would fail),
-    /// and the event carries its target version so re-appending it dedups to the
-    /// original.
-    pub(super) async fn complete_transition(
-        &self,
-        saga_id: i64,
-        next: Account,
-        event: LedgerEventKind,
-    ) -> Result<(), LedgerError> {
-        // The account is guaranteed to exist here (its version was already
-        // bumped, or is about to be), so a read failure is transient or a real
-        // invariant breach, not "not found": surface it verbatim so recovery
-        // retries rather than reporting a misleading domain error.
-        let current = self.store.get_account(&next.id).await?;
-        // Append only into an empty version slot. This also subsumes the
-        // is_closed guard the forward path runs: a close always bumps the
-        // version, so a since-closed account sits at version >= next.version and
-        // this branch is skipped, never appending onto a closed account.
-        if current.version < next.version {
-            self.store.append_account_version(next).await?;
-        }
-        self.store
-            .append_event(&LedgerEvent {
-                seq: 0,
-                timestamp: now_millis()?,
-                kind: event,
-            })
-            .await?;
-        self.store.delete_saga(&saga_id).await?;
-        Ok(())
-    }
 }

doc/architecture.md

@@ -83,6 +83,7 @@ classDiagram
     class SagaStore {
         +save_saga(id, data)
         +list_pending_sagas()
+        +get_saga(id)
         +delete_saga(id)
     }
     class EventStore {

doc/glossary.md

@@ -93,14 +93,21 @@ atomically flips `Active → PendingInactive` stamped with a `ReservationId`,
 so two sagas cannot both claim the same posting. This (not a global
 transaction) is what prevents double-spend.
 
-### PendingSaga / recovery
-
-A write-ahead record `{envelope, reservation, phase}` persisted via
-`SagaStore` before a commit mutates anything. The `phase`
-(`Reserving` → `Finalizing`) tells `Ledger::recover()` (startup) how to
-complete a crashed saga: a `Reserving` saga is re-run and **re-validated**;
-a `Finalizing` saga (already validated, owns its postings) is rolled forward
-through the verified `finalize_envelope`. Roll-forward, not rollback.
+### Write-ahead record (PendingRecord) / recovery
+
+A record persisted via `SagaStore` before a multi-write operation mutates
+anything, so a crash mid-sequence can be completed on the next startup. There
+are two kinds (`PendingRecord`): a **commit saga** `{envelope, reservation,
+phase}`, and an **account transition** `{next, event}` (freeze/unfreeze/close).
+The `pending` module owns the concept: what a record is, how it is
+(de)serialized, and how each kind completes.
+
+`Ledger::recover()` (startup) loads each surviving record and drives it to a
+terminal state through `PendingRecord::complete`. A commit saga's `phase`
+(`Reserving` → `Finalizing`) decides how: a `Reserving` saga is re-run and
+**re-validated**; a `Finalizing` saga (already validated, owns its postings) is
+rolled forward through the verified `finalize_envelope`. A transition rolls its
+version + event forward idempotently. Roll-forward, not rollback.
 
 ### Book