Unify the storage seam on the affected-row-count contract

ADR-0003 made the saga commit primitives dumb: each write applies one
update and returns the number of affected rows, leaving the saga to
interpret counts, idempotency, and compensation. Account and book
creation and account versioning stood outside that contract. They
returned domain verdicts (AlreadyExists, VersionConflict) computed inside
the store, so the same layer spoke two philosophies and a reader could
not tell where a domain decision actually lived.

Bring the three entity-lifecycle writes into the count contract.
create_account, create_book, and append_account_version now return the
number of rows written (1 = applied, 0 = not), and the ledger derives the
domain outcome from that count: AccountAlreadyExists, BookAlreadyExists,
and AccountVersionConflict now live in the ledger layer, and the inflight
path keys InflightAlreadyOpen off the ledger error instead of a store
verdict.

The version-monotonicity guard is preserved as a guarded write, not
relocated to the caller: append_account_version still lands a version
only when it is exactly one past the head, so a stale or gap append
matches nothing and reports 0 without punching a hole in the chain. This
keeps the double-append single-winner property while removing the verdict
from the store.

StoreError drops its AlreadyExists and VersionConflict variants; a write
now reports a count or an I/O fault, nothing else, which makes the "no
verdict variants" promise in the error docs actually hold.

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

@@ -361,7 +361,7 @@ impl AccountStore for SqlStore {
         Ok(result)
     }
 
-    async fn create_account(&self, account: Account) -> Result<(), StoreError> {
+    async fn create_account(&self, account: Account) -> Result<u64, StoreError> {
         // Pessimistic locking: inside one transaction, lock the account's head
         // row with `SELECT ... FOR UPDATE` so a concurrent creator waits. The
         // head is the single row per account; its `ON CONFLICT (id, subaccount)
@@ -384,10 +384,7 @@ impl AccountStore for SqlStore {
         .await
         .map_err(|e| StoreError::Internal(e.to_string()))?;
         if existing.is_some() {
-            return Err(StoreError::AlreadyExists(format!(
-                "account {:?}",
-                account.id
-            )));
+            return Ok(0);
         }
 
         // Append the immutable first version, then point the head at it.
@@ -414,19 +411,16 @@ impl AccountStore for SqlStore {
         .await
         .map_err(|e| StoreError::Internal(e.to_string()))?;
         if res.rows_affected() == 0 {
-            return Err(StoreError::AlreadyExists(format!(
-                "account {:?}",
-                account.id
-            )));
+            return Ok(0);
         }
 
         tx.commit()
             .await
             .map_err(|e| StoreError::Internal(e.to_string()))?;
-        Ok(())
+        Ok(1)
     }
 
-    async fn append_account_version(&self, account: Account) -> Result<(), StoreError> {
+    async fn append_account_version(&self, account: Account) -> Result<u64, StoreError> {
         // Pessimistic locking: inside one transaction, lock the account's head
         // row with `SELECT ... FOR UPDATE` so a concurrent appender waits here
         // until we commit, then check the version, append the new immutable row,
@@ -441,6 +435,10 @@ impl AccountStore for SqlStore {
             .await
             .map_err(|e| StoreError::Internal(e.to_string()))?;
 
+        // A guarded write: no such account, or a version that is not exactly one
+        // past the head, matches nothing and reports 0. This is what keeps the
+        // chain gap-free (a stale or skipped version never lands) and makes a
+        // replay of an already-applied version a no-op.
         let current = sqlx::query(&format!(
             "SELECT version FROM account_head WHERE id = $1 AND subaccount = $2{lock}"
         ))
@@ -448,8 +446,10 @@ impl AccountStore for SqlStore {
         .bind(account.id.sub)
         .fetch_optional(&mut *tx)
         .await
-        .map_err(|e| StoreError::Internal(e.to_string()))?
-        .ok_or_else(|| StoreError::NotFound(format!("account {:?}", account.id)))?;
+        .map_err(|e| StoreError::Internal(e.to_string()))?;
+        let Some(current) = current else {
+            return Ok(0);
+        };
 
         let current_version: i64 = current
             .try_get("version")
@@ -459,11 +459,7 @@ impl AccountStore for SqlStore {
             .ok_or_else(|| StoreError::Internal("account version overflow".to_string()))?;
 
         if account.version as i64 != expected {
-            return Err(StoreError::VersionConflict {
-                account: account.id,
-                expected: expected as u64,
-                actual: account.version,
-            });
+            return Ok(0);
         }
 
         let res = sqlx::query(
@@ -479,11 +475,7 @@ impl AccountStore for SqlStore {
             .await
             .map_err(|e| StoreError::Internal(e.to_string()))?;
         if res.rows_affected() == 0 {
-            return Err(StoreError::VersionConflict {
-                account: account.id,
-                expected: expected as u64,
-                actual: account.version,
-            });
+            return Ok(0);
         }
 
         // Move the head to the new version (delete + insert, never update).
@@ -504,7 +496,7 @@ impl AccountStore for SqlStore {
         tx.commit()
             .await
             .map_err(|e| StoreError::Internal(e.to_string()))?;
-        Ok(())
+        Ok(1)
     }
 
     async fn get_account_history(&self, id: &AccountId) -> Result<Vec<Account>, StoreError> {
@@ -1335,7 +1327,7 @@ impl EventStore for SqlStore {
 
 #[async_trait]
 impl BookStore for SqlStore {
-    async fn create_book(&self, book: Book) -> Result<(), StoreError> {
+    async fn create_book(&self, book: Book) -> Result<u64, StoreError> {
         // Pessimistic locking, same shape as create_account: lock any existing
         // book row with `SELECT ... FOR UPDATE` inside the transaction, then
         // insert with `ON CONFLICT DO NOTHING` as the portable backstop.
@@ -1353,7 +1345,7 @@ impl BookStore for SqlStore {
             .await
             .map_err(|e| StoreError::Internal(e.to_string()))?;
         if existing.is_some() {
-            return Err(StoreError::AlreadyExists(format!("book {:?}", book.id)));
+            return Ok(0);
         }
 
         let res = sqlx::query(
@@ -1366,13 +1358,13 @@ impl BookStore for SqlStore {
         .await
         .map_err(|e| StoreError::Internal(e.to_string()))?;
         if res.rows_affected() == 0 {
-            return Err(StoreError::AlreadyExists(format!("book {:?}", book.id)));
+            return Ok(0);
         }
 
         tx.commit()
             .await
             .map_err(|e| StoreError::Internal(e.to_string()))?;
-        Ok(())
+        Ok(1)
     }
 
     async fn get_book(&self, id: &BookId) -> Result<Book, StoreError> {

crates/kuatia-storage/src/error.rs

@@ -1,27 +1,16 @@
 //! Error types for storage implementations.
 
-use kuatia_types::AccountId;
-
 /// Errors produced by [`Store`](crate::store::Store) implementations.
 ///
 /// 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.
+/// mismatch"/"cas conflict"/"already exists"/"version conflict" variants — every
+/// caller derives those from counts. The only outcomes a write can report are a
+/// count and an I/O fault.
 #[derive(Debug, Clone)]
 pub enum StoreError {
     /// The requested entity was not found.
     NotFound(String),
-    /// The entity already exists (e.g. duplicate account creation).
-    AlreadyExists(String),
-    /// Optimistic version check failed on an account update.
-    VersionConflict {
-        /// Account that had a version mismatch.
-        account: AccountId,
-        /// Version the caller expected.
-        expected: u64,
-        /// Version the store actually had.
-        actual: u64,
-    },
     /// Catch-all for unexpected internal errors.
     Internal(String),
 }
@@ -30,17 +19,6 @@ impl std::fmt::Display for StoreError {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
             Self::NotFound(msg) => write!(f, "not found: {msg}"),
-            Self::AlreadyExists(msg) => write!(f, "already exists: {msg}"),
-            Self::VersionConflict {
-                account,
-                expected,
-                actual,
-            } => {
-                write!(
-                    f,
-                    "version conflict for {account:?}: expected {expected}, got {actual}"
-                )
-            }
             Self::Internal(msg) => write!(f, "internal error: {msg}"),
         }
     }

crates/kuatia-storage/src/mem_store.rs

@@ -148,35 +148,35 @@ impl AccountStore for InMemoryStore {
         Ok(result)
     }
 
-    async fn create_account(&self, account: Account) -> Result<(), StoreError> {
+    async fn create_account(&self, account: Account) -> Result<u64, StoreError> {
         let id = account.id;
         let mut accounts = self.accounts.write().await;
         if accounts.contains_key(&id) {
-            return Err(StoreError::AlreadyExists(format!("account {id:?}")));
+            return Ok(0);
         }
         accounts.insert(id, vec![account]);
-        Ok(())
+        Ok(1)
     }
 
-    async fn append_account_version(&self, account: Account) -> Result<(), StoreError> {
+    async fn append_account_version(&self, account: Account) -> Result<u64, StoreError> {
         let id = account.id;
         let mut accounts = self.accounts.write().await;
-        let versions = accounts
-            .get_mut(&id)
-            .ok_or_else(|| StoreError::NotFound(format!("account {id:?}")))?;
+        // A guarded write: no such account, or a version that is not exactly one
+        // past the head, matches nothing and reports 0. This is what keeps the
+        // chain gap-free (a stale or skipped version never lands) and makes a
+        // replay of an already-applied version a no-op.
+        let Some(versions) = accounts.get_mut(&id) else {
+            return Ok(0);
+        };
         let current_version = versions.last().map(|a| a.version).unwrap_or(0);
         let expected = current_version
             .checked_add(1)
             .ok_or_else(|| StoreError::Internal("account version overflow".to_string()))?;
         if account.version != expected {
-            return Err(StoreError::VersionConflict {
-                account: account.id,
-                expected,
-                actual: account.version,
-            });
+            return Ok(0);
         }
         versions.push(account);
-        Ok(())
+        Ok(1)
     }
 
     async fn get_account_history(&self, id: &AccountId) -> Result<Vec<Account>, StoreError> {
@@ -498,13 +498,13 @@ impl EventStore for InMemoryStore {
 
 #[async_trait]
 impl BookStore for InMemoryStore {
-    async fn create_book(&self, book: Book) -> Result<(), StoreError> {
+    async fn create_book(&self, book: Book) -> Result<u64, StoreError> {
         let mut books = self.books.write().await;
         if books.contains_key(&book.id) {
-            return Err(StoreError::AlreadyExists(format!("book {:?}", book.id)));
+            return Ok(0);
         }
         books.insert(book.id, book);
-        Ok(())
+        Ok(1)
     }
 
     async fn get_book(&self, id: &BookId) -> Result<Book, StoreError> {

crates/kuatia-storage/src/store.rs

@@ -88,10 +88,19 @@ pub trait AccountStore: Send + Sync {
     async fn get_account(&self, id: &AccountId) -> Result<Account, StoreError>;
     /// Fetch multiple accounts by id.
     async fn get_accounts(&self, ids: &[AccountId]) -> Result<Vec<Account>, StoreError>;
-    /// Persist a new account (version 1).
-    async fn create_account(&self, account: Account) -> Result<(), StoreError>;
-    /// Append a new version to an existing account.
-    async fn append_account_version(&self, account: Account) -> Result<(), StoreError>;
+    /// Persist a new account (version 1). A dumb instruction: insert the account
+    /// if absent and return the **number of accounts created** — **1** if this id
+    /// was newly created, **0** if an account with this id already existed. The
+    /// caller decides what `0` means.
+    async fn create_account(&self, account: Account) -> Result<u64, StoreError>;
+    /// Append a new version to an existing account. A dumb instruction with a
+    /// guarded write: the version lands only when it is exactly one past the
+    /// current head (`version == current + 1`) and not already present, which
+    /// keeps the version chain gap-free and single-winner under contention.
+    /// Returns the **number of versions appended** — **1** if it landed, **0**
+    /// otherwise (no such account, a gap/stale append, or an already-applied
+    /// replay). The caller reads state to disambiguate a `0`.
+    async fn append_account_version(&self, account: Account) -> Result<u64, StoreError>;
     /// Return the full version history for an account.
     async fn get_account_history(&self, id: &AccountId) -> Result<Vec<Account>, StoreError>;
     /// List all accounts (latest version of each).
@@ -235,8 +244,10 @@ pub trait SagaStore: Send + Sync {
 /// Book persistence.
 #[async_trait]
 pub trait BookStore: Send + Sync {
-    /// Create a new book.
-    async fn create_book(&self, book: Book) -> Result<(), StoreError>;
+    /// Create a new book. A dumb instruction: insert the book if absent and
+    /// return the **number of books created** — **1** if newly created, **0** if
+    /// a book with this id already existed. The caller decides what `0` means.
+    async fn create_book(&self, book: Book) -> Result<u64, StoreError>;
     /// Fetch a book by id.
     async fn get_book(&self, id: &BookId) -> Result<Book, StoreError>;
     /// List all books.

crates/kuatia-storage/src/store_tests.rs

@@ -173,7 +173,7 @@ async fn commit_envelope(
 /// Create an account and retrieve it.
 pub async fn create_and_get_account(store: &(impl Store + 'static)) {
     let acc = make_account(1, AccountFlags::DEBIT_MUST_NOT_EXCEED_CREDIT);
-    store.create_account(acc.clone()).await.unwrap();
+    assert_eq!(store.create_account(acc.clone()).await.unwrap(), 1);
     let got = store.get_account(&AccountId::new(1)).await.unwrap();
     assert_eq!(got.id, acc.id);
     assert_eq!(got.version, 1);
@@ -181,12 +181,11 @@ pub async fn create_and_get_account(store: &(impl Store + 'static)) {
     assert_eq!(got.flags, acc.flags);
 }
 
-/// Duplicate account creation fails.
-pub async fn create_duplicate_account_fails(store: &(impl Store + 'static)) {
+/// Duplicate account creation lands nothing and reports 0.
+pub async fn create_duplicate_account_reports_zero(store: &(impl Store + 'static)) {
     let acc = make_account(1, AccountFlags::DEBIT_MUST_NOT_EXCEED_CREDIT);
-    store.create_account(acc.clone()).await.unwrap();
-    let err = store.create_account(acc).await.unwrap_err();
-    assert!(matches!(err, StoreError::AlreadyExists(_)));
+    assert_eq!(store.create_account(acc.clone()).await.unwrap(), 1);
+    assert_eq!(store.create_account(acc).await.unwrap(), 0);
 }
 
 /// Get non-existent account returns NotFound.
@@ -227,15 +226,20 @@ pub async fn append_account_version(store: &(impl Store + 'static)) {
     assert!(got.is_frozen());
 }
 
-/// Appending with wrong version number fails.
-pub async fn append_version_conflict(store: &(impl Store + 'static)) {
+/// Appending a version that would leave a gap (not `current + 1`) lands nothing
+/// and reports 0, keeping the chain contiguous.
+pub async fn append_version_gap_reports_zero(store: &(impl Store + 'static)) {
     let acc = make_account(1, AccountFlags::DEBIT_MUST_NOT_EXCEED_CREDIT);
     store.create_account(acc.clone()).await.unwrap();
 
     let mut bad = acc.clone();
     bad.version = 5;
-    let err = store.append_account_version(bad).await.unwrap_err();
-    assert!(matches!(err, StoreError::VersionConflict { .. }));
+    assert_eq!(store.append_account_version(bad).await.unwrap(), 0);
+
+    // The head is untouched: still the sole version 1.
+    let history = store.get_account_history(&AccountId::new(1)).await.unwrap();
+    assert_eq!(history.len(), 1);
+    assert_eq!(history[0].version, 1);
 }
 
 /// Re-appending an already-taken version is rejected and leaves the history
@@ -250,13 +254,12 @@ pub async fn append_duplicate_version_rejected(store: &(impl Store + 'static)) {
     v2.flags = AccountFlags::FROZEN;
     store.append_account_version(v2).await.unwrap();
 
-    // A second append that also targets version 2 (now the current max) must be
-    // rejected rather than duplicating or overwriting it.
+    // A second append that also targets version 2 (now the current max) lands
+    // nothing (reports 0) rather than duplicating or overwriting it.
     let mut v2_again = acc.clone();
     v2_again.version = 2;
     v2_again.flags = AccountFlags::CLOSED;
-    let err = store.append_account_version(v2_again).await.unwrap_err();
-    assert!(matches!(err, StoreError::VersionConflict { .. }));
+    assert_eq!(store.append_account_version(v2_again).await.unwrap(), 0);
 
     // Exactly one row at version 2, and it is the first (frozen) write.
     let history = store.get_account_history(&AccountId::new(1)).await.unwrap();
@@ -1213,12 +1216,11 @@ pub async fn create_and_get_book(store: &(impl Store + 'static)) {
     assert_eq!(got, book);
 }
 
-/// Duplicate book creation fails.
-pub async fn create_duplicate_book_fails(store: &(impl Store + 'static)) {
+/// Duplicate book creation lands nothing and reports 0.
+pub async fn create_duplicate_book_reports_zero(store: &(impl Store + 'static)) {
     let book = make_book(1, "sales");
-    store.create_book(book.clone()).await.unwrap();
-    let err = store.create_book(book).await.unwrap_err();
-    assert!(matches!(err, StoreError::AlreadyExists(_)));
+    assert_eq!(store.create_book(book.clone()).await.unwrap(), 1);
+    assert_eq!(store.create_book(book).await.unwrap(), 0);
 }
 
 /// Get a non-existent book returns NotFound.
@@ -1339,11 +1341,11 @@ macro_rules! store_tests {
         $crate::store_tests!(@tests $factory,
             // AccountStore
             create_and_get_account,
-            create_duplicate_account_fails,
+            create_duplicate_account_reports_zero,
             get_missing_account_fails,
             get_accounts_batch,
             append_account_version,
-            append_version_conflict,
+            append_version_gap_reports_zero,
             append_duplicate_version_rejected,
             get_account_history,
             list_accounts,
@@ -1391,7 +1393,7 @@ macro_rules! store_tests {
             events_sequence_ordering,
             // BookStore
             create_and_get_book,
-            create_duplicate_book_fails,
+            create_duplicate_book_reports_zero,
             get_missing_book_fails,
             list_books,
             // BalanceProjectionStore

crates/kuatia/src/error.rs

@@ -35,8 +35,21 @@ pub enum LedgerError {
     AccountNotEmpty(AccountId),
     /// The account is already closed.
     AccountAlreadyClosed(AccountId),
+    /// An account with this id already exists.
+    AccountAlreadyExists(AccountId),
+    /// An account-version append did not land: the head was not at
+    /// `expected - 1` when the append was attempted (a concurrent transition
+    /// moved it). Carries the version this attempt targeted.
+    AccountVersionConflict {
+        /// Account whose version append was rejected.
+        account: AccountId,
+        /// Version this attempt targeted (`current + 1` at read time).
+        expected: u64,
+    },
     /// A transfer named a book that does not exist.
     BookNotFound(BookId),
+    /// A book with this id already exists.
+    BookAlreadyExists(BookId),
     /// The referenced inflight transaction does not exist (no authorize record).
     InflightNotFound(EnvelopeId),
     /// The referenced transfer is not an inflight authorize, or its metadata is
@@ -76,7 +89,13 @@ impl std::fmt::Display for LedgerError {
             Self::AccountNotFound(id) => write!(f, "account not found: {id:?}"),
             Self::AccountNotEmpty(id) => write!(f, "account not empty: {id:?}"),
             Self::AccountAlreadyClosed(id) => write!(f, "account already closed: {id:?}"),
+            Self::AccountAlreadyExists(id) => write!(f, "account already exists: {id:?}"),
+            Self::AccountVersionConflict { account, expected } => write!(
+                f,
+                "account version conflict for {account:?}: could not append version {expected}"
+            ),
             Self::BookNotFound(id) => write!(f, "book not found: {id:?}"),
+            Self::BookAlreadyExists(id) => write!(f, "book already exists: {id:?}"),
             Self::InflightNotFound(id) => write!(f, "inflight transaction not found: {id:?}"),
             Self::NotInflightTransaction(id) => {
                 write!(f, "not an inflight authorize transaction: {id:?}")

crates/kuatia/src/inflight.rs

@@ -214,7 +214,7 @@ impl Ledger {
             acct.metadata = meta_map(&InflightMeta::Hold { destination: *dest })?;
             match self.create_account(acct).await {
                 Ok(()) => {}
-                Err(LedgerError::Store(StoreError::AlreadyExists(_))) => {
+                Err(LedgerError::AccountAlreadyExists(_)) => {
                     return Err(LedgerError::InflightAlreadyOpen(*hold));
                 }
                 Err(e) => return Err(e),

crates/kuatia/src/ledger/lifecycle.rs

@@ -21,7 +21,9 @@ impl Ledger {
     /// Create a new account and emit an AccountCreated event.
     pub async fn create_account(&self, account: kuatia_core::Account) -> Result<(), LedgerError> {
         let id = account.id;
-        self.store.create_account(account).await?;
+        if self.store.create_account(account).await? == 0 {
+            return Err(LedgerError::AccountAlreadyExists(id));
+        }
         self.store
             .append_event(&LedgerEvent {
                 seq: 0,

crates/kuatia/src/ledger/query.rs

@@ -85,7 +85,11 @@ impl Ledger {
 
     /// Create a new book.
     pub async fn create_book(&self, book: kuatia_core::Book) -> Result<(), LedgerError> {
-        Ok(self.store.create_book(book).await?)
+        let id = book.id;
+        if self.store.create_book(book).await? == 0 {
+            return Err(LedgerError::BookAlreadyExists(id));
+        }
+        Ok(())
     }
 
     /// Fetch a book by id.

crates/kuatia/src/ledger/transition.rs

@@ -54,7 +54,16 @@ impl Ledger {
         // the event append is then repaired by recover(), not left dangling.
         let saga_id = self.save_transition(&next, &event).await?;
 
-        self.store.append_account_version(next).await?;
+        let expected = next.version;
+        if self.store.append_account_version(next).await? == 0 {
+            // A concurrent transition moved the head; the guarded append matched
+            // nothing. Surface the conflict; the write-ahead record is repaired
+            // (or cleared) by recover() on the next startup.
+            return Err(LedgerError::AccountVersionConflict {
+                account: *id,
+                expected,
+            });
+        }
         self.store
             .append_event(&LedgerEvent {
                 seq: 0,