Make intent resolution pure and preserve typed errors across the saga

Two changes to leverage the pure-core/async split the codebase already
uses for validation.

Extract the intent resolve algorithm into kuatia-core as a sibling of
validate_and_plan. resolve held the real intent-layer logic (net-debit
aggregation, greedy selection with change, and the overdraft-shortfall
offset-posting branch) welded to store reads, so its most interesting
paths were only reachable by standing up a full async Ledger and
committing real deposits. Split it into two sans-IO passes:
draft_movements aggregates movements into output postings and
per-account net debits, telling the async layer exactly what to load;
resolve_envelope selects postings, computes change, and covers an
overdraft shortfall. Ledger::resolve now just loads per-debit state
between the two. The change-making and shortfall branches gain direct
unit tests.

Carry the typed LedgerError across the legend saga seam instead of a
stringified round-trip. The seam converted LedgerError to a String-only
SagaError and back to Store(Internal), so an OverdraftExceeded,
AccountFrozen, or InsufficientFunds detected during commit reached the
caller as an internal storage fault, and the typed variants callers
branch on were unreachable through the commit path. The macro path used
here only requires the step error be Send + Sync + Clone (the ledger
never serializes the legend Execution; it has its own PendingSaga WAL),
so LedgerError becomes the step error type directly and StoreError and
LedgerError gain Clone. Genuine plumbing faults map to Store(Internal)
via a small helper.

crates/kuatia-core/src/lib.rs

@@ -5,6 +5,7 @@
 //! deterministically, and embedded anywhere.
 
 pub mod hash;
+pub mod posting_resolution;
 pub mod posting_selection;
 pub mod validate;
 
@@ -14,5 +15,8 @@ pub use hash::{
     account_canonical_bytes, account_hash, account_snapshot_id, canonical_bytes, content_hash,
     double_sha256, envelope_id,
 };
+pub use posting_resolution::{
+    Debit, MovementDraft, ResolveError, ResolveInput, draft_movements, resolve_envelope,
+};
 pub use posting_selection::{SelectionError, select_postings};
 pub use validate::{Plan, PlanInput, ValidationError, validate_and_plan};

crates/kuatia-core/src/posting_resolution.rs

@@ -0,0 +1,447 @@
+//! Pure intent resolution: turn a [`Transfer`] (movements) into a concrete
+//! [`Envelope`] (postings to consume and create).
+//!
+//! Resolution is two passes, both sans-IO and deterministic:
+//!
+//! 1. [`draft_movements`] aggregates movements into output postings and
+//!    per-(account, asset) net debits. It tells the async layer exactly which
+//!    postings and account policies to load.
+//! 2. [`resolve_envelope`] selects postings for each debit, computes change, and
+//!    covers an overdraft shortfall with a negative offset posting.
+//!
+//! The async ledger loads state; this module decides. The change-making and
+//! shortfall branches are the parts most worth property-testing, and living here
+//! they are reachable without standing up a store.
+
+use std::collections::HashMap;
+
+use crate::posting_selection::{SelectionError, select_postings};
+use kuatia_types::{
+    AccountId, AccountPolicy, AssetId, Cent, Envelope, EnvelopeBuilder, NewPosting, OverflowError,
+    Posting, PostingId, Transfer,
+};
+
+// ---------------------------------------------------------------------------
+// Errors
+// ---------------------------------------------------------------------------
+
+/// Failure from resolution pass 2 ([`resolve_envelope`]).
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ResolveError {
+    /// Posting selection failed: insufficient funds for a non-overdraft account,
+    /// or an overflow while summing candidate postings.
+    Selection(SelectionError),
+    /// Monetary arithmetic overflowed while computing change or a shortfall.
+    Overflow,
+}
+
+impl std::fmt::Display for ResolveError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Selection(e) => write!(f, "selection: {e}"),
+            Self::Overflow => write!(f, "monetary amount overflow"),
+        }
+    }
+}
+
+impl std::error::Error for ResolveError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        match self {
+            Self::Selection(e) => Some(e),
+            Self::Overflow => None,
+        }
+    }
+}
+
+impl From<SelectionError> for ResolveError {
+    fn from(e: SelectionError) -> Self {
+        Self::Selection(e)
+    }
+}
+
+impl From<OverflowError> for ResolveError {
+    fn from(_: OverflowError) -> Self {
+        Self::Overflow
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Pass 1: draft
+// ---------------------------------------------------------------------------
+
+/// A positive net debit on one (account, asset) that pass 2 must cover by
+/// selecting postings.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Debit {
+    /// The account (subaccount) being debited.
+    pub account: AccountId,
+    /// The asset owed.
+    pub asset: AssetId,
+    /// The positive amount owed.
+    pub amount: Cent,
+}
+
+/// Output of resolution pass 1: postings credited straight from movements, plus
+/// the debits that still require posting selection in pass 2.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct MovementDraft {
+    /// Postings credited to movement destinations.
+    pub creates: Vec<NewPosting>,
+    /// Positive net debits, one per (account, asset), each needing selection.
+    pub debits: Vec<Debit>,
+}
+
+/// Pass 1: for each movement create an output posting on its destination and
+/// accumulate the net debit on its source. Debits are returned in a
+/// deterministic (account, asset) order so golden vectors are stable regardless
+/// of `HashMap` iteration order.
+pub fn draft_movements(transfer: &Transfer) -> Result<MovementDraft, OverflowError> {
+    let mut creates: Vec<NewPosting> = Vec::new();
+    let mut net_debits: HashMap<(AccountId, AssetId), Cent> = HashMap::new();
+
+    for m in &transfer.movements {
+        let payer = if m.from != m.to { Some(m.from) } else { None };
+        creates.push(NewPosting {
+            owner: m.to,
+            asset: m.asset,
+            value: m.amount,
+            payer,
+        });
+        let entry = net_debits.entry((m.from, m.asset)).or_insert(Cent::ZERO);
+        *entry = entry.checked_add(m.amount)?;
+    }
+
+    let mut debits: Vec<Debit> = net_debits
+        .into_iter()
+        .filter(|(_, amount)| amount.is_positive())
+        .map(|((account, asset), amount)| Debit {
+            account,
+            asset,
+            amount,
+        })
+        .collect();
+    debits.sort_by_key(|d| (d.account, d.asset));
+
+    Ok(MovementDraft { creates, debits })
+}
+
+// ---------------------------------------------------------------------------
+// Pass 2: resolve
+// ---------------------------------------------------------------------------
+
+/// Pre-loaded state for resolution pass 2. The async layer gathers `available`
+/// and `policies` for the debits produced by [`draft_movements`]; this pass is
+/// pure.
+pub struct ResolveInput<'a> {
+    /// The transfer being resolved (for its book and metadata).
+    pub transfer: &'a Transfer,
+    /// Pass 1 output. Consumed here: its `creates` seed the envelope and its
+    /// `debits` drive selection.
+    pub draft: MovementDraft,
+    /// Active postings available for each debit's (account, asset). A missing or
+    /// empty entry means no positive postings to draw on.
+    pub available: &'a HashMap<(AccountId, AssetId), Vec<Posting>>,
+    /// Policy for each debit's account. A missing entry is treated as "no
+    /// overdraft" — the debit fails with [`SelectionError::InsufficientFunds`]
+    /// rather than granting an offset position on unknown terms.
+    pub policies: &'a HashMap<AccountId, AccountPolicy>,
+}
+
+/// Pass 2: for each debit, either select postings and compute change, or (for an
+/// overdraft account short of funds) consume every positive posting and create a
+/// negative offset posting for the shortfall. The floor is enforced later, in
+/// validation. Returns the concrete envelope with no account snapshots pinned;
+/// the caller pins them.
+pub fn resolve_envelope(input: ResolveInput<'_>) -> Result<Envelope, ResolveError> {
+    let ResolveInput {
+        transfer,
+        draft,
+        available,
+        policies,
+    } = input;
+    let MovementDraft {
+        mut creates,
+        debits,
+    } = draft;
+    let mut consumes: Vec<PostingId> = Vec::new();
+
+    for debit in &debits {
+        let avail: &[Posting] = available
+            .get(&(debit.account, debit.asset))
+            .map(Vec::as_slice)
+            .unwrap_or(&[]);
+        let total_positive = Cent::checked_sum(
+            avail
+                .iter()
+                .filter(|p| p.value.is_positive())
+                .map(|p| p.value),
+        )?;
+
+        if total_positive >= debit.amount {
+            // Enough positive postings: select a subset and compute change.
+            let selected = select_postings(avail, debit.asset, debit.amount)?;
+            let consumed_sum = Cent::checked_sum(
+                avail
+                    .iter()
+                    .filter(|p| selected.contains(&p.id))
+                    .map(|p| p.value),
+            )?;
+            let change = consumed_sum.checked_sub(debit.amount)?;
+
+            consumes.extend_from_slice(&selected);
+            if change.is_positive() {
+                creates.push(NewPosting {
+                    owner: debit.account,
+                    asset: debit.asset,
+                    value: change,
+                    payer: None,
+                });
+            }
+        } else {
+            // Not enough positive postings. Overdraft accounts cover the
+            // shortfall with a negative posting (an offset position); any other
+            // policy — or an unknown one — fails.
+            match policies.get(&debit.account) {
+                Some(AccountPolicy::CappedOverdraft { .. } | AccountPolicy::UncappedOverdraft) => {
+                    let positives: Vec<PostingId> = avail
+                        .iter()
+                        .filter(|p| p.value.is_positive())
+                        .map(|p| p.id)
+                        .collect();
+                    consumes.extend_from_slice(&positives);
+                    let shortfall = debit.amount.checked_sub(total_positive)?;
+                    creates.push(NewPosting {
+                        owner: debit.account,
+                        asset: debit.asset,
+                        value: shortfall.checked_neg()?,
+                        payer: None,
+                    });
+                }
+                _ => {
+                    return Err(ResolveError::Selection(SelectionError::InsufficientFunds {
+                        available: total_positive,
+                        requested: debit.amount,
+                    }));
+                }
+            }
+        }
+    }
+
+    Ok(EnvelopeBuilder::new()
+        .consumes(consumes)
+        .creates(creates)
+        .book(transfer.book)
+        .metadata(transfer.metadata.clone())
+        .build())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use kuatia_types::*;
+
+    fn acct(id: i64) -> AccountId {
+        AccountId::new(id)
+    }
+
+    fn posting(owner: AccountId, index: u16, value: i64) -> Posting {
+        Posting::new(
+            PostingId {
+                transfer: EnvelopeId([index as u8; 32]),
+                index,
+            },
+            owner,
+            AssetId::new(1),
+            Cent::from(value),
+        )
+    }
+
+    fn pay(from: AccountId, to: AccountId, amount: i64) -> Transfer {
+        TransferBuilder::new()
+            .pay(from, to, AssetId::new(1), Cent::from(amount))
+            .build()
+    }
+
+    #[test]
+    fn draft_aggregates_net_debit_and_output() {
+        let draft = draft_movements(&pay(acct(1), acct(2), 100)).unwrap();
+        assert_eq!(draft.creates.len(), 1);
+        assert_eq!(draft.creates[0].owner, acct(2));
+        assert_eq!(draft.creates[0].value, Cent::from(100));
+        assert_eq!(draft.creates[0].payer, Some(acct(1)));
+        assert_eq!(
+            draft.debits,
+            vec![Debit {
+                account: acct(1),
+                asset: AssetId::new(1),
+                amount: Cent::from(100),
+            }]
+        );
+    }
+
+    #[test]
+    fn self_movement_has_no_payer_but_still_debits() {
+        // from == to: the created posting carries no payer, but the source still
+        // owes its own net debit (the credit is a separate created posting, not
+        // an offset), so a debit is produced.
+        let draft = draft_movements(&pay(acct(1), acct(1), 100)).unwrap();
+        assert_eq!(draft.creates[0].payer, None);
+        assert_eq!(
+            draft.debits,
+            vec![Debit {
+                account: acct(1),
+                asset: AssetId::new(1),
+                amount: Cent::from(100),
+            }]
+        );
+    }
+
+    #[test]
+    fn exact_funds_no_change() {
+        let transfer = pay(acct(1), acct(2), 100);
+        let draft = draft_movements(&transfer).unwrap();
+        let available = HashMap::from([(
+            (acct(1), AssetId::new(1)),
+            vec![posting(acct(1), 0, 60), posting(acct(1), 1, 40)],
+        )]);
+        let policies = HashMap::new();
+        let env = resolve_envelope(ResolveInput {
+            transfer: &transfer,
+            draft,
+            available: &available,
+            policies: &policies,
+        })
+        .unwrap();
+        assert_eq!(env.consumes().len(), 2);
+        // Only the destination posting is created — no change.
+        assert_eq!(env.creates().len(), 1);
+        assert_eq!(env.creates()[0].owner, acct(2));
+    }
+
+    #[test]
+    fn overpay_creates_change_posting() {
+        let transfer = pay(acct(1), acct(2), 30);
+        let draft = draft_movements(&transfer).unwrap();
+        let available =
+            HashMap::from([((acct(1), AssetId::new(1)), vec![posting(acct(1), 0, 100)])]);
+        let policies = HashMap::new();
+        let env = resolve_envelope(ResolveInput {
+            transfer: &transfer,
+            draft,
+            available: &available,
+            policies: &policies,
+        })
+        .unwrap();
+        assert_eq!(env.consumes().len(), 1);
+        // Destination posting + a change posting back to the payer.
+        let change: Vec<_> = env
+            .creates()
+            .iter()
+            .filter(|p| p.owner == acct(1))
+            .collect();
+        assert_eq!(change.len(), 1);
+        assert_eq!(change[0].value, Cent::from(70));
+    }
+
+    #[test]
+    fn insufficient_funds_without_overdraft_fails() {
+        let transfer = pay(acct(1), acct(2), 100);
+        let draft = draft_movements(&transfer).unwrap();
+        let available =
+            HashMap::from([((acct(1), AssetId::new(1)), vec![posting(acct(1), 0, 40)])]);
+        let policies = HashMap::from([(acct(1), AccountPolicy::NoOverdraft)]);
+        let err = resolve_envelope(ResolveInput {
+            transfer: &transfer,
+            draft,
+            available: &available,
+            policies: &policies,
+        })
+        .unwrap_err();
+        assert_eq!(
+            err,
+            ResolveError::Selection(SelectionError::InsufficientFunds {
+                available: Cent::from(40),
+                requested: Cent::from(100),
+            })
+        );
+    }
+
+    #[test]
+    fn missing_policy_is_treated_as_no_overdraft() {
+        let transfer = pay(acct(1), acct(2), 100);
+        let draft = draft_movements(&transfer).unwrap();
+        let available = HashMap::new();
+        let policies = HashMap::new();
+        let err = resolve_envelope(ResolveInput {
+            transfer: &transfer,
+            draft,
+            available: &available,
+            policies: &policies,
+        })
+        .unwrap_err();
+        assert_eq!(
+            err,
+            ResolveError::Selection(SelectionError::InsufficientFunds {
+                available: Cent::ZERO,
+                requested: Cent::from(100),
+            })
+        );
+    }
+
+    #[test]
+    fn overdraft_covers_shortfall_with_offset_posting() {
+        let transfer = pay(acct(1), acct(2), 100);
+        let draft = draft_movements(&transfer).unwrap();
+        let available =
+            HashMap::from([((acct(1), AssetId::new(1)), vec![posting(acct(1), 0, 30)])]);
+        let policies = HashMap::from([(acct(1), AccountPolicy::UncappedOverdraft)]);
+        let env = resolve_envelope(ResolveInput {
+            transfer: &transfer,
+            draft,
+            available: &available,
+            policies: &policies,
+        })
+        .unwrap();
+        // The single positive posting is consumed.
+        assert_eq!(env.consumes().len(), 1);
+        // A negative offset posting covers the 70 shortfall on the payer.
+        let offset: Vec<_> = env
+            .creates()
+            .iter()
+            .filter(|p| p.owner == acct(1))
+            .collect();
+        assert_eq!(offset.len(), 1);
+        assert_eq!(offset[0].value, Cent::from(-70));
+    }
+
+    #[test]
+    fn capped_overdraft_covers_shortfall() {
+        // The floor is enforced in validation, not here — resolve still creates
+        // the offset posting for a capped-overdraft account.
+        let transfer = pay(acct(1), acct(2), 100);
+        let draft = draft_movements(&transfer).unwrap();
+        let available = HashMap::new();
+        let policies = HashMap::from([(
+            acct(1),
+            AccountPolicy::CappedOverdraft {
+                floor: Cent::from(-1000),
+            },
+        )]);
+        let env = resolve_envelope(ResolveInput {
+            transfer: &transfer,
+            draft,
+            available: &available,
+            policies: &policies,
+        })
+        .unwrap();
+        assert!(env.consumes().is_empty());
+        assert_eq!(
+            env.creates()
+                .iter()
+                .find(|p| p.owner == acct(1))
+                .unwrap()
+                .value,
+            Cent::from(-100)
+        );
+    }
+}

crates/kuatia-storage/src/error.rs

@@ -7,7 +7,7 @@ use kuatia_types::AccountId;
 /// The store is a dumb instruction follower: writes report affected-row counts,
 /// not semantic verdicts, so there are no "posting not active"/"reservation
 /// mismatch"/"cas conflict" variants — the saga derives those from counts.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub enum StoreError {
     /// The requested entity was not found.
     NotFound(String),

crates/kuatia/src/error.rs

@@ -4,13 +4,20 @@
 //! and from storage, so callers get a single error type from every API.
 
 use kuatia_core::{
-    AccountId, AssetId, BookId, EnvelopeId, OverflowError, PostingId, SelectionError,
+    AccountId, AssetId, BookId, EnvelopeId, OverflowError, PostingId, ResolveError, SelectionError,
     ValidationError,
 };
 use kuatia_storage::error::StoreError;
 
 /// Unified error type for the async ledger API.
-#[derive(Debug)]
+///
+/// `Clone` so the saga engine can carry a typed error across its step seam and
+/// return the real variant to the caller (an [`OverdraftExceeded`] detected
+/// during commit stays an [`OverdraftExceeded`], not a stringified internal
+/// fault).
+///
+/// [`OverdraftExceeded`]: ValidationError::OverdraftExceeded
+#[derive(Debug, Clone)]
 pub enum LedgerError {
     /// A transfer invariant was violated.
     Validation(ValidationError),
@@ -127,6 +134,15 @@ impl From<SelectionError> for LedgerError {
     }
 }
 
+impl From<ResolveError> for LedgerError {
+    fn from(e: ResolveError) -> Self {
+        match e {
+            ResolveError::Selection(s) => LedgerError::Selection(s),
+            ResolveError::Overflow => LedgerError::Overflow,
+        }
+    }
+}
+
 impl From<OverflowError> for LedgerError {
     fn from(_: OverflowError) -> Self {
         LedgerError::Overflow

crates/kuatia/src/ledger.rs

@@ -9,8 +9,8 @@ use tracing::instrument;
 use kuatia_core::{
     AccountId, AccountPolicy, AccountSnapshotId, AssetId, Book, Cent, DEFAULT_BOOK, Envelope,
     EnvelopeBuilder, EnvelopeId, NewPosting, PlanInput, Posting, PostingFilter, PostingId,
-    PostingState, Receipt, SelectionError, Transfer, account_snapshot_id, envelope_id,
-    select_postings, validate_and_plan,
+    PostingState, Receipt, ResolveInput, Transfer, account_snapshot_id, draft_movements,
+    envelope_id, resolve_envelope, validate_and_plan,
 };
 
 use crate::error::LedgerError;
@@ -23,7 +23,7 @@ pub(crate) fn now_millis() -> Result<i64, LedgerError> {
         .as_millis() as i64)
 }
 use crate::saga::{
-    FinalizeInput, FinalizeTransferStep, LedgerCtx, ReserveInput, ReservePostingsStep, SagaError,
+    FinalizeInput, FinalizeTransferStep, LedgerCtx, ReserveInput, ReservePostingsStep,
 };
 use kuatia_storage::error::StoreError;
 use kuatia_storage::events::{LedgerEvent, LedgerEventKind};
@@ -164,105 +164,42 @@ impl Ledger {
     /// Convert a [`Transfer`] intent into a concrete [`Envelope`] by selecting
     /// postings for each movement and computing change.
     ///
-    /// Pass 1: create output postings and aggregate net debits per (account, asset).
-    /// Pass 2: for each pair with a positive net debit, select postings and compute change.
+    /// The decision is pure ([`kuatia_core::draft_movements`] +
+    /// [`kuatia_core::resolve_envelope`]); this method only loads the state those
+    /// functions need. Pass 1 aggregates net debits and tells us which postings
+    /// and account policies to load; pass 2 selects postings, computes change,
+    /// and covers any overdraft shortfall.
     #[instrument(skip(self, transfer), name = "ledger.resolve")]
     pub async fn resolve(&self, transfer: &Transfer) -> Result<Envelope, LedgerError> {
-        let mut consumes: Vec<PostingId> = Vec::new();
-        let mut creates: Vec<NewPosting> = Vec::new();
-        let mut net_debits: HashMap<(AccountId, AssetId), Cent> = HashMap::new();
-
-        // Pass 1: output postings + debit aggregation
-        for m in &transfer.movements {
-            let payer = if m.from != m.to { Some(m.from) } else { None };
-            creates.push(NewPosting {
-                owner: m.to,
-                asset: m.asset,
-                value: m.amount,
-                payer,
-            });
-            let entry = net_debits.entry((m.from, m.asset)).or_insert(Cent::ZERO);
-            *entry = entry.checked_add(m.amount)?;
-        }
-
-        // Pass 2: posting selection for accounts with positive net debit
-        for ((account, asset), net_debit) in &net_debits {
-            if !net_debit.is_positive() {
-                continue;
-            }
-            let available = self
+        let draft = draft_movements(transfer)?;
+
+        // Load the active postings and account policy for each debit. A deposit
+        // nets to zero on the system account, so it produces no debit and loads
+        // nothing here.
+        let mut available: HashMap<(AccountId, AssetId), Vec<Posting>> = HashMap::new();
+        let mut policies: HashMap<AccountId, AccountPolicy> = HashMap::new();
+        for debit in &draft.debits {
+            let postings = self
                 .store
                 .get_postings_by_account(
-                    account.id,
-                    Some(account.sub),
-                    Some(asset),
+                    debit.account.id,
+                    Some(debit.account.sub),
+                    Some(&debit.asset),
                     PostingFilter::Active,
                 )
                 .await?;
-            let total_positive = Cent::checked_sum(
-                available
-                    .iter()
-                    .filter(|p| p.value.is_positive())
-                    .map(|p| p.value),
-            )?;
-
-            if total_positive >= *net_debit {
-                // Enough positive postings: select a subset and compute change.
-                let selected = select_postings(&available, *asset, *net_debit)?;
-                let consumed_sum = Cent::checked_sum(
-                    available
-                        .iter()
-                        .filter(|p| selected.contains(&p.id))
-                        .map(|p| p.value),
-                )?;
-                let change = consumed_sum.checked_sub(*net_debit)?;
-
-                consumes.extend_from_slice(&selected);
-                if change.is_positive() {
-                    creates.push(NewPosting {
-                        owner: *account,
-                        asset: *asset,
-                        value: change,
-                        payer: None,
-                    });
-                }
-            } else {
-                // Not enough positive postings. Overdraft accounts cover the
-                // shortfall with a negative posting (an offset position); the
-                // floor is enforced later in validation. Any other policy fails.
-                let policy = self.store.get_account(account).await?.policy;
-                match policy {
-                    AccountPolicy::CappedOverdraft { .. } | AccountPolicy::UncappedOverdraft => {
-                        let positives: Vec<PostingId> = available
-                            .iter()
-                            .filter(|p| p.value.is_positive())
-                            .map(|p| p.id)
-                            .collect();
-                        consumes.extend_from_slice(&positives);
-                        let shortfall = net_debit.checked_sub(total_positive)?;
-                        creates.push(NewPosting {
-                            owner: *account,
-                            asset: *asset,
-                            value: shortfall.checked_neg()?,
-                            payer: None,
-                        });
-                    }
-                    _ => {
-                        return Err(LedgerError::Selection(SelectionError::InsufficientFunds {
-                            available: total_positive,
-                            requested: *net_debit,
-                        }));
-                    }
-                }
+            available.insert((debit.account, debit.asset), postings);
+            if let std::collections::hash_map::Entry::Vacant(slot) = policies.entry(debit.account) {
+                slot.insert(self.store.get_account(&debit.account).await?.policy);
             }
         }
 
-        let mut envelope = EnvelopeBuilder::new()
-            .consumes(consumes)
-            .creates(creates)
-            .book(transfer.book)
-            .metadata(transfer.metadata.clone())
-            .build();
+        let mut envelope = resolve_envelope(ResolveInput {
+            transfer,
+            draft,
+            available: &available,
+            policies: &policies,
+        })?;
 
         // Resolve account snapshots for optimistic concurrency
         let ids = envelope.referenced_accounts();
@@ -355,20 +292,17 @@ impl Ledger {
                     LedgerError::Store(StoreError::Internal("saga completed but no receipt".into()))
                 })
             }
-            ExecutionResult::Failed(_, err) => {
-                Err(LedgerError::Store(StoreError::Internal(err.message)))
-            }
+            // 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(LedgerError::Store(StoreError::Internal(
-                    original_error.message,
-                ))),
-                compensation: Box::new(LedgerError::Store(StoreError::Internal(
-                    compensation_error.message,
-                ))),
+                original: Box::new(original_error),
+                compensation: Box::new(compensation_error),
             }),
             ExecutionResult::Paused(_) => Err(LedgerError::Store(StoreError::Internal(
                 "saga paused unexpectedly".into(),

crates/kuatia/src/ledger/envelope_saga.rs

@@ -1,7 +1,7 @@
 use super::*;
 
 legend! {
-    EnvelopeSaga<LedgerCtx, SagaError> {
+    EnvelopeSaga<LedgerCtx, LedgerError> {
         reserve: ReservePostingsStep,
         finalize: FinalizeTransferStep,
     }

crates/kuatia/src/saga.rs

@@ -37,8 +37,17 @@ use kuatia_core::{
 
 use crate::error::LedgerError;
 use crate::ledger::Ledger;
+use kuatia_storage::error::StoreError;
 use kuatia_storage::store::Store;
 
+/// A saga-internal plumbing fault (missing context, a short row-count that the
+/// end-state does not explain). These are genuine internal invariants, distinct
+/// from the typed domain errors ([`LedgerError::Validation`], overdraft, frozen)
+/// that flow through unchanged, so they map to [`StoreError::Internal`].
+fn internal(message: impl Into<String>) -> LedgerError {
+    LedgerError::Store(StoreError::Internal(message.into()))
+}
+
 /// Interpret a dumb primitive's affected-row `count` against the `ids` it
 /// targeted. `count == ids.len()` is success. A short count is acceptable only if
 /// the shortfall is already in the desired end-state — a prior attempt (or this
@@ -51,50 +60,21 @@ async fn verify_postings(
     count: u64,
     ok: impl Fn(&PostingState) -> bool,
     what: &str,
-) -> Result<(), SagaError> {
+) -> Result<(), LedgerError> {
     if count == ids.len() as u64 {
         return Ok(());
     }
     let states = store
         .get_posting_states(ids)
         .await
-        .map_err(|e| SagaError::from(LedgerError::Store(e)))?;
+        .map_err(LedgerError::Store)?;
     if states.len() == ids.len() && states.iter().all(&ok) {
         return Ok(());
     }
-    Err(SagaError {
-        message: format!(
-            "{what}: storage applied {count}/{} rows and the end-state is not satisfied",
-            ids.len()
-        ),
-    })
-}
-
-// ---------------------------------------------------------------------------
-// Saga error -- serializable + cloneable wrapper
-// ---------------------------------------------------------------------------
-
-/// Serializable error wrapper used across saga steps.
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct SagaError {
-    /// Human-readable error description.
-    pub message: String,
-}
-
-impl std::fmt::Display for SagaError {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{}", self.message)
-    }
-}
-
-impl std::error::Error for SagaError {}
-
-impl From<LedgerError> for SagaError {
-    fn from(e: LedgerError) -> Self {
-        Self {
-            message: e.to_string(),
-        }
-    }
+    Err(internal(format!(
+        "{what}: storage applied {count}/{} rows and the end-state is not satisfied",
+        ids.len()
+    )))
 }
 
 // ---------------------------------------------------------------------------
@@ -166,16 +146,16 @@ impl LedgerCtx {
     }
 
     /// Borrow the ledger, returning an error if not injected.
-    pub fn ledger(&self) -> Result<&Ledger, SagaError> {
-        self.ledger.as_ref().map(|l| l.as_ref()).ok_or(SagaError {
-            message: "ledger not injected -- call inject_ledger() after deserializing".into(),
+    pub fn ledger(&self) -> Result<&Ledger, LedgerError> {
+        self.ledger.as_ref().map(|l| l.as_ref()).ok_or_else(|| {
+            internal("ledger not injected -- call inject_ledger() after deserializing")
         })
     }
 
     /// Clone the ledger `Arc`, returning an error if not injected.
-    pub fn ledger_arc(&self) -> Result<Arc<Ledger>, SagaError> {
-        self.ledger.clone().ok_or(SagaError {
-            message: "ledger not injected -- call inject_ledger() after deserializing".into(),
+    pub fn ledger_arc(&self) -> Result<Arc<Ledger>, LedgerError> {
+        self.ledger.clone().ok_or_else(|| {
+            internal("ledger not injected -- call inject_ledger() after deserializing")
         })
     }
 }
@@ -200,17 +180,18 @@ pub struct ReserveInput;
 pub struct ReservePostingsStep;
 
 #[async_trait]
-impl Step<LedgerCtx, SagaError> for ReservePostingsStep {
+impl Step<LedgerCtx, LedgerError> for ReservePostingsStep {
     type Input = ReserveInput;
 
-    async fn execute(ctx: &mut LedgerCtx, _input: &ReserveInput) -> Result<StepOutcome, SagaError> {
+    async fn execute(
+        ctx: &mut LedgerCtx,
+        _input: &ReserveInput,
+    ) -> Result<StepOutcome, LedgerError> {
         async {
             let posting_ids: Vec<PostingId> = ctx
                 .envelope
                 .as_ref()
-                .ok_or(SagaError {
-                    message: "no envelope in context -- resolve step must run first".into(),
-                })?
+                .ok_or_else(|| internal("no envelope in context -- resolve step must run first"))?
                 .consumes()
                 .to_vec();
             let rid = ctx.reservation;
@@ -220,7 +201,7 @@ impl Step<LedgerCtx, SagaError> for ReservePostingsStep {
             let reserved = store
                 .reserve_postings(&posting_ids, rid)
                 .await
-                .map_err(|e| SagaError::from(LedgerError::Store(e)))?;
+                .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(
@@ -241,12 +222,12 @@ impl Step<LedgerCtx, SagaError> for ReservePostingsStep {
     async fn compensate(
         ctx: &mut LedgerCtx,
         _input: &ReserveInput,
-    ) -> Result<CompensationOutcome, SagaError> {
+    ) -> Result<CompensationOutcome, LedgerError> {
         ctx.ledger()?
             .store()
             .release_postings(&ctx.reserved_postings, ctx.reservation)
             .await
-            .map_err(|e| SagaError::from(LedgerError::Store(e)))?;
+            .map_err(LedgerError::Store)?;
         ctx.reserved_postings.clear();
         Ok(CompensationOutcome::Completed)
     }
@@ -271,27 +252,26 @@ pub struct FinalizeInput;
 pub struct FinalizeTransferStep;
 
 #[async_trait]
-impl Step<LedgerCtx, SagaError> for FinalizeTransferStep {
+impl Step<LedgerCtx, LedgerError> for FinalizeTransferStep {
     type Input = FinalizeInput;
 
     async fn execute(
         ctx: &mut LedgerCtx,
         _input: &FinalizeInput,
-    ) -> Result<StepOutcome, SagaError> {
+    ) -> Result<StepOutcome, LedgerError> {
         async {
-            let envelope = ctx.envelope.clone().ok_or(SagaError {
-                message: "no envelope in context -- resolve step must run first".into(),
-            })?;
+            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.
-            let receipt = ledger
-                .finalize_envelope(&envelope, rid)
-                .await
-                .map_err(SagaError::from)?;
+            // 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();
@@ -304,7 +284,7 @@ impl Step<LedgerCtx, SagaError> for FinalizeTransferStep {
     async fn compensate(
         ctx: &mut LedgerCtx,
         _input: &FinalizeInput,
-    ) -> Result<CompensationOutcome, SagaError> {
+    ) -> Result<CompensationOutcome, LedgerError> {
         if let Some(receipt) = ctx.receipts.pop() {
             ctx.ledger_arc()?.reverse(&receipt.transfer_id).await?;
         }
@@ -350,10 +330,11 @@ pub struct DepositInput {
 // Helpers
 // ---------------------------------------------------------------------------
 
-async fn compensate_last_receipt(ctx: &mut LedgerCtx) -> Result<CompensationOutcome, SagaError> {
-    let receipt = ctx.receipts.pop().ok_or(SagaError {
-        message: "no receipt to compensate".into(),
-    })?;
+async fn compensate_last_receipt(ctx: &mut LedgerCtx) -> Result<CompensationOutcome, LedgerError> {
+    let receipt = ctx
+        .receipts
+        .pop()
+        .ok_or_else(|| internal("no receipt to compensate"))?;
     ctx.ledger_arc()?.reverse(&receipt.transfer_id).await?;
     Ok(CompensationOutcome::Completed)
 }
@@ -366,10 +347,10 @@ async fn compensate_last_receipt(ctx: &mut LedgerCtx) -> Result<CompensationOutc
 pub struct PayMovementStep;
 
 #[async_trait]
-impl Step<LedgerCtx, SagaError> for PayMovementStep {
+impl Step<LedgerCtx, LedgerError> for PayMovementStep {
     type Input = PayInput;
 
-    async fn execute(ctx: &mut LedgerCtx, input: &PayInput) -> Result<StepOutcome, SagaError> {
+    async fn execute(ctx: &mut LedgerCtx, input: &PayInput) -> Result<StepOutcome, LedgerError> {
         let ledger = ctx.ledger_arc()?;
         let transfer = TransferBuilder::new()
             .pay(input.from, input.to, input.asset, input.amount)
@@ -382,7 +363,7 @@ impl Step<LedgerCtx, SagaError> for PayMovementStep {
     async fn compensate(
         ctx: &mut LedgerCtx,
         _input: &PayInput,
-    ) -> Result<CompensationOutcome, SagaError> {
+    ) -> Result<CompensationOutcome, LedgerError> {
         compensate_last_receipt(ctx).await
     }
 }
@@ -391,14 +372,17 @@ impl Step<LedgerCtx, SagaError> for PayMovementStep {
 pub struct DepositMovementStep;
 
 #[async_trait]
-impl Step<LedgerCtx, SagaError> for DepositMovementStep {
+impl Step<LedgerCtx, LedgerError> for DepositMovementStep {
     type Input = DepositInput;
 
-    async fn execute(ctx: &mut LedgerCtx, input: &DepositInput) -> Result<StepOutcome, SagaError> {
+    async fn execute(
+        ctx: &mut LedgerCtx,
+        input: &DepositInput,
+    ) -> Result<StepOutcome, LedgerError> {
         let ledger = ctx.ledger_arc()?;
         let transfer = TransferBuilder::new()
             .deposit(input.to, input.asset, input.amount, input.external)
-            .map_err(|e| SagaError::from(LedgerError::from(e)))?
+            .map_err(LedgerError::from)?
             .build();
         let receipt = ledger.commit(transfer).await?;
         ctx.receipts.push(receipt);
@@ -408,7 +392,7 @@ impl Step<LedgerCtx, SagaError> for DepositMovementStep {
     async fn compensate(
         ctx: &mut LedgerCtx,
         _input: &DepositInput,
-    ) -> Result<CompensationOutcome, SagaError> {
+    ) -> Result<CompensationOutcome, LedgerError> {
         compensate_last_receipt(ctx).await
     }
 }

crates/kuatia/tests/saga.rs

@@ -2,6 +2,7 @@
 
 use std::sync::Arc;
 
+use kuatia::error::LedgerError;
 use kuatia::ledger::Ledger;
 use kuatia::mem_store::InMemoryStore;
 use kuatia::saga::*;
@@ -54,7 +55,7 @@ async fn setup_ledger() -> Arc<Ledger> {
 
 // Define a two-step saga: deposit then pay
 legend! {
-    FundAndPay<LedgerCtx, SagaError> {
+    FundAndPay<LedgerCtx, LedgerError> {
         deposit: DepositMovementStep,
         pay: PayMovementStep,
     }
@@ -105,7 +106,7 @@ async fn saga_happy_path() {
 
 // Define a saga that will fail on the second step and trigger compensation
 legend! {
-    DepositAndOverspend<LedgerCtx, SagaError> {
+    DepositAndOverspend<LedgerCtx, LedgerError> {
         deposit: DepositMovementStep,
         pay: PayMovementStep,
     }
@@ -135,7 +136,17 @@ async fn saga_compensation_on_failure() {
     let execution = saga.build(ctx);
 
     match execution.start().await {
-        ExecutionResult::Failed(_, _err) => {
+        ExecutionResult::Failed(_, err) => {
+            // The saga carries the typed `LedgerError` across its step seam, so
+            // the overspend surfaces as `Selection(InsufficientFunds)` rather
+            // than a stringified `Store(Internal)`.
+            assert!(
+                matches!(
+                    err,
+                    LedgerError::Selection(SelectionError::InsufficientFunds { .. })
+                ),
+                "expected typed InsufficientFunds, got {err:?}"
+            );
             // The deposit should have been compensated (reversed)
             // Note: balances won't be exactly 0 because the deposit reversal
             // creates new postings, but the net effect should be zero
@@ -154,7 +165,7 @@ async fn saga_compensation_on_failure() {
 
 // Three-step saga
 legend! {
-    ThreeStepFlow<LedgerCtx, SagaError> {
+    ThreeStepFlow<LedgerCtx, LedgerError> {
         deposit: DepositMovementStep,
         pay_ab: PayMovementStep,
         pay_bc: PayMovementStep,