Collapse the shallow envelope saga into the commit path

For a single commit the two legend steps were pass-throughs: reserve was
reserve_postings plus a count check, and finalize just called back into
Ledger::finalize_envelope. The saga interface (two steps, retry, LIFO
compensation) was about as complex as the work it wrapped, and the two
steps even sat at different layers, one reaching straight into the store,
the other going through Ledger.

Replace the EnvelopeSaga with a linear Ledger::reserve_and_finalize:
reserve the consumed postings, finalize, and on any failure release the
reservation (the one meaningful compensation, collapsed from LIFO). The
reserve/compensation policy now lives next to finalize_envelope, and a
reader can follow the commit top to bottom. Releasing on a reserve
failure also closes a partial-reservation leak the old step left behind.

This refines ADR-0002 rather than reversing it: legend still drives the
composed multi-transfer sagas (PayMovementStep, DepositMovementStep),
which is where it earns its keep. LedgerCtx loses the envelope-only
fields it no longer carries.

(cherry picked from commit fd1fc9e43f12e3c7a1af2c7332287a225ec12d58)

crates/kuatia/src/ledger/commit.rs

@@ -1,16 +1,15 @@
-//! The write-ahead saga/commit engine: resolve, reserve, finalize, recover.
+//! The write-ahead commit engine: resolve, reserve, finalize, recover.
 //!
-//! This is the deep core of the ledger. Every commit is the two-step envelope
-//! saga (`reserve → finalize`, validation inside finalize) with automatic retry
-//! and LIFO compensation. A phase-tracked write-ahead record ([`PendingSaga`])
-//! lets [`Ledger::recover`] complete or safely abandon a commit interrupted by a
-//! crash.
+//! This is the deep core of the ledger. Every commit is the linear
+//! `reserve → finalize` path (validation inside finalize), which releases its
+//! reservation to compensate a failure before the point of no return. A
+//! phase-tracked write-ahead record ([`PendingSaga`]) lets [`Ledger::recover`]
+//! complete or safely abandon a commit interrupted by a crash.
 
 use std::collections::HashMap;
 use std::collections::hash_map::Entry;
 use std::sync::Arc;
 
-use legend::ExecutionResult;
 use tracing::instrument;
 
 use kuatia_core::{
@@ -26,10 +25,7 @@ use kuatia_storage::store::EnvelopeRecord;
 
 use super::{Ledger, now_millis};
 use crate::error::LedgerError;
-use crate::saga::{
-    EnvelopeSaga, EnvelopeSagaInputs, FinalizeInput, LedgerCtx, ReserveInput, apply_and_verify,
-    consume_reserved, verify_postings,
-};
+use crate::saga::{apply_and_verify, consume_reserved, verify_postings};
 
 use super::pending::{PendingRecord, PendingSaga, SagaPhase};
 
@@ -186,9 +182,9 @@ impl Ledger {
     /// validate -> finalize). This is the single commit path; `commit()` and
     /// `reverse()` both funnel through it.
     ///
-    /// Before running, the saga (envelope + reservation) is persisted as a
+    /// Before running, the commit (envelope + reservation) is persisted as a
     /// pending record so a crash mid-commit is completed by [`recover`](Self::recover). The
-    /// record is deleted once the saga reaches a terminal state. The commit is
+    /// record is deleted once the commit reaches a terminal state. The commit is
     /// idempotent on the content-addressed transfer id.
     #[instrument(skip(self, envelope), name = "ledger.commit_envelope")]
     pub async fn commit_envelope(
@@ -217,45 +213,73 @@ impl Ledger {
             .await
     }
 
-    /// Build and run the envelope saga (reserve → finalize) to a terminal
-    /// outcome, returning the resulting receipt.
-    pub(super) async fn drive_envelope_saga(
-        self: &Arc<Self>,
-        envelope: Envelope,
+    /// The single-commit core: reserve the consumed postings, then finalize.
+    ///
+    /// This used to be a two-step `legend` saga; for one commit the steps were
+    /// pass-throughs, so the reserve/compensation policy now lives here as a
+    /// linear path next to [`finalize_envelope`](Self::finalize_envelope).
+    /// (`legend` still drives the composed multi-transfer sagas in `saga`.)
+    ///
+    /// On any failure the reservation is released (the LIFO compensation collapsed
+    /// to its one meaningful action). Before the point of no return that returns
+    /// the postings to Active so the caller's `clear_if_safe` can drop the
+    /// write-ahead record; past `Finalizing` the postings are already spent, so
+    /// the release is a no-op and the record is instead kept for roll-forward. The
+    /// typed error (validation / overdraft / frozen) reaches the caller unchanged.
+    pub(super) async fn reserve_and_finalize(
+        &self,
+        envelope: &Envelope,
         reservation: ReservationId,
     ) -> Result<Receipt, LedgerError> {
-        let saga = EnvelopeSaga::new(EnvelopeSagaInputs {
-            reserve: ReserveInput,
-            finalize: FinalizeInput,
-        });
-        let ctx = LedgerCtx::for_envelope(Arc::clone(self), envelope, reservation);
-        let execution = saga.build(ctx);
-
-        match execution.start().await {
-            ExecutionResult::Completed(e) => {
-                let ctx = e.into_context();
-                ctx.receipts.last().cloned().ok_or_else(|| {
-                    LedgerError::Store(StoreError::Internal("saga completed but no receipt".into()))
-                })
-            }
-            // The saga's error type is `LedgerError`, so a validation / overdraft
-            // / frozen failure detected during commit reaches the caller as the
-            // real typed variant instead of a stringified internal fault.
-            ExecutionResult::Failed(_, err) => Err(err),
-            ExecutionResult::CompensationFailed {
-                original_error,
-                compensation_error,
-                ..
-            } => Err(LedgerError::CompensationFailed {
-                original: Box::new(original_error),
-                compensation: Box::new(compensation_error),
-            }),
-            ExecutionResult::Paused(_) => Err(LedgerError::Store(StoreError::Internal(
-                "saga paused unexpectedly".into(),
-            ))),
+        let consumes = envelope.consumes();
+
+        let result = match self.reserve_consumed(consumes, reservation).await {
+            Ok(()) => self.finalize_envelope(envelope, reservation).await,
+            Err(err) => Err(err),
+        };
+
+        match result {
+            Ok(receipt) => Ok(receipt),
+            // Compensate by releasing our reservation. If the release itself fails
+            // we surface both errors, matching the old saga's `CompensationFailed`.
+            Err(err) => match self.store.release_postings(consumes, reservation).await {
+                Ok(_) => Err(err),
+                Err(comp) => Err(LedgerError::CompensationFailed {
+                    original: Box::new(err),
+                    compensation: Box::new(LedgerError::Store(comp)),
+                }),
+            },
         }
     }
 
+    /// Reserve every consumed posting into the reserved index under `reservation`
+    /// (a CAS out of the active index), then check the affected-row count against
+    /// the ADR-0003 contract. A short count is accepted only when the shortfall is
+    /// already reserved by us (an idempotent replay, e.g. recovery re-running a
+    /// `Reserving` saga). A deposit consumes nothing, so this is a no-op.
+    async fn reserve_consumed(
+        &self,
+        consumes: &[PostingId],
+        reservation: ReservationId,
+    ) -> Result<(), LedgerError> {
+        if consumes.is_empty() {
+            return Ok(());
+        }
+        let reserved = self
+            .store
+            .reserve_postings(consumes, reservation)
+            .await
+            .map_err(LedgerError::Store)?;
+        verify_postings(
+            self.store.as_ref(),
+            consumes,
+            reserved,
+            |s| matches!(s, PostingState::Reserved(r) if *r == reservation),
+            "reserve",
+        )
+        .await
+    }
+
     /// Complete every pending write-ahead record left by a crash. Call on
     /// startup; returns how many were processed.
     ///

crates/kuatia/src/ledger/pending.rs

@@ -10,7 +10,7 @@
 //! 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
+//! `reserve_and_finalize`) stay on [`Ledger`] because the live commit path shares
 //! them; this module sequences them for the recovery path.
 
 use std::sync::Arc;
@@ -158,8 +158,8 @@ impl PendingSaga {
             .await
     }
 
-    /// Run a fresh commit end to end: write-ahead at Reserving, drive the saga,
-    /// then clear the record when it is safe. The single home of the commit
+    /// Run a fresh commit end to end: write-ahead at Reserving, reserve then
+    /// finalize, then clear the record when it is safe. The single home of the commit
     /// write-ahead lifecycle; [`commit_envelope`](Ledger::commit_envelope) calls
     /// this and recovery mirrors it.
     pub(super) async fn run(self, ledger: &Arc<Ledger>) -> Result<Receipt, LedgerError> {
@@ -167,7 +167,7 @@ impl PendingSaga {
         // Commit does not touch the balance projection (ADR-0019): cache points
         // are appended lazily on read, once enough credits/debits have accrued.
         let result = ledger
-            .drive_envelope_saga(self.envelope.clone(), self.reservation)
+            .reserve_and_finalize(&self.envelope, self.reservation)
             .await;
         self.clear_if_safe(ledger, result.is_ok()).await?;
         result
@@ -223,7 +223,7 @@ impl PendingSaga {
             // absorbed (record kept for the next run); only infra errors propagate.
             SagaPhase::Reserving => {
                 let result = ledger
-                    .drive_envelope_saga(self.envelope.clone(), self.reservation)
+                    .reserve_and_finalize(&self.envelope, self.reservation)
                     .await;
                 self.clear_if_safe(ledger, result.is_ok()).await
             }

crates/kuatia/src/lib.rs

@@ -1,8 +1,8 @@
 //! Kuatia — async ledger resource built on top of [`kuatia_core`].
 //!
 //! This crate adds IO to the pure decision logic: the [`Store`](kuatia_storage::store::Store) trait
-//! abstracts storage, and the [`Ledger`](crate::ledger::Ledger) struct drives the two-step
-//! commit saga (reserve then finalize, validation inside finalize) behind an async API.
+//! abstracts storage, and the [`Ledger`](crate::ledger::Ledger) struct drives the linear
+//! reserve → finalize commit path (validation inside finalize) behind an async API.
 
 pub mod error;
 pub mod inflight;

crates/kuatia/src/saga.rs

@@ -1,46 +1,31 @@
-//! Legend saga step adapters for the ledger.
+//! Ledger commit helpers and high-level saga steps.
 //!
-//! Provides [`Step`] implementations so the ledger can participate
-//! in multi-resource saga workflows, with automatic LIFO compensation across
-//! resource boundaries.
+//! # Count contract
 //!
-//! # Envelope pipeline saga
-//!
-//! A commit is two saga steps over a pre-resolved [`Envelope`] (resolution runs
-//! before the saga, in `Ledger::commit`):
-//!
-//! 1. **ReservePostingsStep** -- `reserve_postings`: move each consumed posting from the active index into the reserved index under the saga's `ReservationId`; interprets the count via `verify_postings`.
-//! 2. **FinalizeTransferStep** -- delegates to `Ledger::finalize_envelope`, which re-validates against current state (the last-step floor / freeze-close guard), marks the saga `Finalizing`, then runs the dumb primitives (`deactivate_postings` → `insert_postings` → `store_transfer` → `append_event`) verifying every end-state.
-//!
-//! The `EnvelopeSaga` is defined via `legend!` below and driven by
-//! `commit_envelope()`. Crash recovery (`Ledger::recover`) re-completes a
-//! persisted saga using its persisted phase: a `Reserving` saga is re-run
-//! (re-validating); a `Finalizing` saga is rolled forward through the same
-//! verified `finalize_envelope`.
+//! `apply_and_verify`, `verify_postings`, and `consume_reserved` encode the
+//! ADR-0003 affected-row-count rule applied after every dumb write primitive in
+//! the commit path. The commit path itself (reserve → finalize) is a linear
+//! method on [`Ledger`] (`ledger::commit`), not a `legend` saga: for a single
+//! commit the two steps were pass-throughs, so collapsing them keeps the
+//! reserve/compensation policy next to the logic it governs (refines ADR-0002).
 //!
 //! # High-level composition
 //!
-//! High-level steps (`PayMovementStep` and `DepositMovementStep`) compose over
-//! the intent-layer API and can be combined into multi-transfer sagas via `legend!`.
-
-// The `legend!` expansion for `EnvelopeSaga` below emits public fields it does
-// not document. An outer `#[allow]` on the macro call is ignored by the
-// compiler, so the allow is scoped to the whole module.
-#![allow(missing_docs)]
+//! [`PayMovementStep`] and [`DepositMovementStep`] wrap the intent-layer
+//! `Ledger::commit` as `legend` [`Step`]s, so several transfers compose into one
+//! multi-transfer saga (an FX trade, a multi-leg settlement) with LIFO
+//! compensation across the whole workflow. This is where `legend` earns its keep.
 
 use std::fmt;
 use std::future::Future;
 use std::sync::Arc;
 
 use async_trait::async_trait;
-use legend::legend;
-use legend::step::{CompensationOutcome, RetryPolicy, Step, StepOutcome};
+use legend::step::{CompensationOutcome, Step, StepOutcome};
 use serde::{Deserialize, Serialize};
-use tracing::Instrument;
 
 use kuatia_core::{
-    AccountId, AssetId, Cent, Envelope, PostingId, PostingState, Receipt, ReservationId,
-    TransferBuilder,
+    AccountId, AssetId, Cent, PostingId, PostingState, Receipt, ReservationId, TransferBuilder,
 };
 
 use crate::error::LedgerError;
@@ -141,21 +126,15 @@ pub(crate) async fn consume_reserved(
 // Saga context -- carries the ledger handle + state between steps
 // ---------------------------------------------------------------------------
 
-/// Saga context that wraps a ledger and tracks state across steps.
+/// Saga context that wraps a ledger and collects the receipts of the transfers a
+/// multi-transfer saga commits, for LIFO compensation.
 ///
 /// The ledger handle is `#[serde(skip)]`: it is supplied when the context is
 /// constructed and is not part of the serialized form.
 #[derive(Clone, Serialize, Deserialize)]
 pub struct LedgerCtx {
-    /// Receipts collected from completed steps.
+    /// Receipts collected from completed steps, popped in reverse to compensate.
     pub receipts: Vec<Receipt>,
-    /// Posting ids reserved so far (for compensation).
-    pub reserved_postings: Vec<PostingId>,
-    /// Resolved envelope produced by the resolve step.
-    pub envelope: Option<Envelope>,
-    /// Reservation owner token for this saga's reserved postings. Serialized so
-    /// it survives pause/recovery, keeping ownership stable across restarts.
-    pub reservation: ReservationId,
     #[serde(skip)]
     ledger: Option<Arc<Ledger>>,
 }
@@ -164,8 +143,6 @@ impl fmt::Debug for LedgerCtx {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("LedgerCtx")
             .field("receipts", &self.receipts)
-            .field("reserved_postings", &self.reserved_postings.len())
-            .field("has_envelope", &self.envelope.is_some())
             .field("ledger_present", &self.ledger.is_some())
             .finish()
     }
@@ -176,25 +153,6 @@ impl LedgerCtx {
     pub fn new(ledger: Arc<Ledger>) -> Self {
         Self {
             receipts: Vec::new(),
-            reserved_postings: Vec::new(),
-            envelope: None,
-            reservation: ReservationId::default(),
-            ledger: Some(ledger),
-        }
-    }
-
-    /// Create a context for the envelope pipeline (reserve → finalize; finalize re-validates)
-    /// with a pre-resolved envelope and an explicit reservation.
-    pub fn for_envelope(
-        ledger: Arc<Ledger>,
-        envelope: Envelope,
-        reservation: ReservationId,
-    ) -> Self {
-        Self {
-            receipts: Vec::new(),
-            reserved_postings: Vec::new(),
-            envelope: Some(envelope),
-            reservation,
             ledger: Some(ledger),
         }
     }
@@ -216,149 +174,6 @@ impl LedgerCtx {
 }
 
 // ===========================================================================
-// Envelope pipeline steps (reserve -> finalize; resolve runs before the saga, validate inside finalize)
-// ===========================================================================
-
-// ---------------------------------------------------------------------------
-// Step 1: ReservePostingsStep
-// ---------------------------------------------------------------------------
-
-/// Input for the reserve step (posting ids come from ctx.envelope).
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct ReserveInput;
-
-/// Reserves consumed postings by CAS: move each from the active index to the
-/// reserved index (the delete-returns-one picks a single winner).
-///
-/// Gets the posting ids from the resolved envelope in the context.
-/// Compensation releases all reserved postings back to Active.
-pub struct ReservePostingsStep;
-
-#[async_trait]
-impl Step<LedgerCtx, LedgerError> for ReservePostingsStep {
-    type Input = ReserveInput;
-
-    async fn execute(
-        ctx: &mut LedgerCtx,
-        _input: &ReserveInput,
-    ) -> Result<StepOutcome, LedgerError> {
-        async {
-            let posting_ids: Vec<PostingId> = ctx
-                .envelope
-                .as_ref()
-                .ok_or_else(|| internal("no envelope in context -- resolve step must run first"))?
-                .consumes()
-                .to_vec();
-            let rid = ctx.reservation;
-            let ledger = ctx.ledger_arc()?;
-            let store = ledger.store();
-
-            let reserved = store
-                .reserve_postings(&posting_ids, rid)
-                .await
-                .map_err(LedgerError::Store)?;
-            // Storage reports the count; the saga decides. A short count is fine
-            // only if the shortfall is already reserved by us (idempotent replay).
-            verify_postings(
-                store,
-                &posting_ids,
-                reserved,
-                |s| matches!(s, PostingState::Reserved(r) if *r == rid),
-                "reserve",
-            )
-            .await?;
-            ctx.reserved_postings.extend_from_slice(&posting_ids);
-            Ok(StepOutcome::Continue)
-        }
-        .instrument(tracing::info_span!("saga_step", step = "reserve"))
-        .await
-    }
-
-    async fn compensate(
-        ctx: &mut LedgerCtx,
-        _input: &ReserveInput,
-    ) -> Result<CompensationOutcome, LedgerError> {
-        ctx.ledger()?
-            .store()
-            .release_postings(&ctx.reserved_postings, ctx.reservation)
-            .await
-            .map_err(LedgerError::Store)?;
-        ctx.reserved_postings.clear();
-        Ok(CompensationOutcome::Completed)
-    }
-
-    fn retry_policy() -> RetryPolicy {
-        RetryPolicy::retries(3)
-    }
-}
-
-// ---------------------------------------------------------------------------
-// Step 2: FinalizeTransferStep
-// ---------------------------------------------------------------------------
-
-/// Input for the finalize step (envelope comes from ctx).
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct FinalizeInput;
-
-/// Re-validates against current state (the last-step floor / freeze-close guard),
-/// then drives the verified, idempotent commit via `Ledger::finalize_envelope`.
-///
-/// Compensation reverses the finalized envelope (only relevant once committed).
-pub struct FinalizeTransferStep;
-
-#[async_trait]
-impl Step<LedgerCtx, LedgerError> for FinalizeTransferStep {
-    type Input = FinalizeInput;
-
-    async fn execute(
-        ctx: &mut LedgerCtx,
-        _input: &FinalizeInput,
-    ) -> Result<StepOutcome, LedgerError> {
-        async {
-            let envelope = ctx
-                .envelope
-                .clone()
-                .ok_or_else(|| internal("no envelope in context -- resolve step must run first"))?;
-            let rid = ctx.reservation;
-            let ledger = ctx.ledger_arc()?;
-
-            // All commit work (re-validate, mark Finalizing, deactivate/insert/
-            // store/event with end-state verification) lives in `finalize_envelope`
-            // so recovery uses exactly the same path. Its typed error (validation,
-            // overdraft, frozen) reaches the caller unchanged.
-            let receipt = ledger.finalize_envelope(&envelope, rid).await?;
-
-            ctx.receipts.push(receipt);
-            ctx.reserved_postings.clear();
-            Ok(StepOutcome::Continue)
-        }
-        .instrument(tracing::info_span!("saga_step", step = "finalize"))
-        .await
-    }
-
-    async fn compensate(
-        ctx: &mut LedgerCtx,
-        _input: &FinalizeInput,
-    ) -> Result<CompensationOutcome, LedgerError> {
-        if let Some(receipt) = ctx.receipts.pop() {
-            ctx.ledger_arc()?.reverse(&receipt.transfer_id).await?;
-        }
-        Ok(CompensationOutcome::Completed)
-    }
-
-    fn retry_policy() -> RetryPolicy {
-        RetryPolicy::retries(3)
-    }
-}
-
-legend! {
-    EnvelopeSaga<LedgerCtx, LedgerError> {
-        reserve: ReservePostingsStep,
-        finalize: FinalizeTransferStep,
-    }
-}
-
-// ===========================================================================
 // High-level steps (pay / deposit movement steps)
 // ===========================================================================