Make both stores honor the store_transfer involved set verbatim

The store_transfer(record, involved) contract diverged between backends.
SqlStore trusted the caller's involved set and wrote one transfer_accounts
row per account; InMemoryStore ignored the parameter and re-derived
participation from stored postings at read time. They agreed only when the
passed set matched what the postings implied, and the conformance helper
passed only created owners, so the consumed-owner branch was never checked
differentially. A saga that passed a wrong involved set would look correct
in the fast in-memory suite yet be wrong on SQL, silently.

Dropping the parameter is not viable: consumed-posting owners live nowhere
in EnvelopeRecord (the envelope carries consumed PostingIds only, and the
account snapshots cover created owners), and SQL keeps those ids inside an
opaque JSON blob it cannot join on. So make both backends follow the same
instruction instead. InMemoryStore now records the involved set in an
explicit transfer_accounts index and resolves get_transfers_for_account
from it, mirroring the SQL table; it derives nothing from postings. The
trait doc states the contract, and a new conformance test indexes a
consumed-posting owner distinct from every created owner (its posting
never seeded) so the transfer is retrievable only if the backend trusts
involved, forcing both adapters to agree.

crates/kuatia-storage/src/mem_store.rs

@@ -37,6 +37,10 @@ pub struct InMemoryStore {
     postings: RwLock<PostingTables>,
     accounts: RwLock<HashMap<AccountId, Vec<Account>>>,
     transfers: RwLock<HashMap<EnvelopeId, EnvelopeRecord>>,
+    /// Transfer-participation index: the `involved` set the caller passed to
+    /// `store_transfer`, mirroring the SQL `transfer_accounts` table so both
+    /// backends resolve `get_transfers_for_account` from the same instruction.
+    transfer_accounts: RwLock<HashMap<EnvelopeId, Vec<AccountId>>>,
     sagas: RwLock<HashMap<i64, Vec<u8>>>,
     events: RwLock<Vec<LedgerEvent>>,
     books: RwLock<HashMap<BookId, Book>>,
@@ -56,6 +60,7 @@ impl InMemoryStore {
             postings: RwLock::new(PostingTables::default()),
             accounts: RwLock::new(HashMap::new()),
             transfers: RwLock::new(HashMap::new()),
+            transfer_accounts: RwLock::new(HashMap::new()),
             sagas: RwLock::new(HashMap::new()),
             events: RwLock::new(Vec::new()),
             books: RwLock::new(HashMap::new()),
@@ -312,17 +317,22 @@ impl TransferStore for InMemoryStore {
     async fn store_transfer(
         &self,
         record: EnvelopeRecord,
-        _involved: &[AccountId],
+        involved: &[AccountId],
     ) -> Result<u64, StoreError> {
-        // `_involved` is ignored here: `get_transfers_for_account` derives the
-        // involved accounts from the stored envelope (creates owners + consumed
-        // posting owners), which matches the set the caller passes. The SQL
-        // backend instead persists `involved` into its `transfer_accounts` index.
+        // Index the transfer under exactly the accounts the caller supplied,
+        // deriving nothing — the same instruction the SQL backend follows into
+        // its `transfer_accounts` table. Idempotent: a replay writes the same
+        // set. The returned count reflects only whether the transfer row was
+        // newly inserted; the caller decides what `0` means.
+        let tid = record.receipt.transfer_id;
+        // Lock order transfers → transfer_accounts, matching every other reader.
         let mut transfers = self.transfers.write().await;
-        if transfers.contains_key(&record.receipt.transfer_id) {
+        let mut transfer_accounts = self.transfer_accounts.write().await;
+        transfer_accounts.insert(tid, involved.to_vec());
+        if transfers.contains_key(&tid) {
             return Ok(0);
         }
-        transfers.insert(record.receipt.transfer_id, record);
+        transfers.insert(tid, record);
         Ok(1)
     }
 
@@ -331,27 +341,16 @@ impl TransferStore for InMemoryStore {
         id: i64,
         sub: Option<i64>,
     ) -> Result<Vec<EnvelopeRecord>, StoreError> {
-        // Acquire postings → transfers in a consistent order to avoid an AB–BA
-        // deadlock with any reader that takes both.
-        let postings = self.postings.read().await;
+        // Resolve participation from the `involved` index, not from postings, so
+        // this backend answers exactly what `store_transfer` was told — matching
+        // the SQL `transfer_accounts` join. Lock order transfers → index.
         let transfers = self.transfers.read().await;
+        let transfer_accounts = self.transfer_accounts.read().await;
         let matches = |owner: &AccountId| owner.id == id && sub.is_none_or(|s| owner.sub == s);
-        let mut result: Vec<EnvelopeRecord> = transfers
-            .values()
-            .filter(|record| {
-                record
-                    .envelope
-                    .creates()
-                    .iter()
-                    .any(|np| matches(&np.owner))
-                    || record.envelope.consumes().iter().any(|pid| {
-                        postings
-                            .immutable
-                            .get(pid)
-                            .is_some_and(|p| matches(&p.owner))
-                    })
-            })
-            .cloned()
+        let mut result: Vec<EnvelopeRecord> = transfer_accounts
+            .iter()
+            .filter(|(_, accounts)| accounts.iter().any(matches))
+            .filter_map(|(tid, _)| transfers.get(tid).cloned())
             .collect();
         result.sort_by_key(|r| r.created_at);
         Ok(result)

crates/kuatia-storage/src/store.rs

@@ -189,9 +189,12 @@ pub trait TransferStore: Send + Sync {
     async fn get_transfer(&self, id: &EnvelopeId) -> Result<Option<EnvelopeRecord>, StoreError>;
     /// Persist a transfer record if absent (idempotent) and index it under every
     /// account in `involved` (both created and consumed owners — the caller
-    /// supplies the set so storage computes nothing). A dumb instruction:
-    /// returns **1** if the transfer row was newly inserted, **0** if it already
-    /// existed. The caller decides what `0` means.
+    /// supplies the set so storage computes nothing). Every backend indexes the
+    /// transfer under exactly these accounts and derives participation from
+    /// nowhere else, so `get_transfers_for_account` returns the same set of
+    /// transfers regardless of backend. A dumb instruction: returns **1** if the
+    /// transfer row was newly inserted, **0** if it already existed. The caller
+    /// decides what `0` means.
     async fn store_transfer(
         &self,
         record: EnvelopeRecord,

crates/kuatia-storage/src/store_tests.rs

@@ -818,6 +818,50 @@ pub async fn store_transfer_counts(store: &(impl Store + 'static)) {
     );
 }
 
+/// `store_transfer` indexes the transfer under exactly the `involved` set the
+/// caller supplies and derives participation from nowhere else. A consumed
+/// posting's owner (account 7) is distinct from every created owner (account 8),
+/// and that consumed posting is deliberately never seeded into the store, so the
+/// transfer can be found for account 7 only if the backend trusts `involved`.
+/// This is the case where a derive-from-postings backend and a trust-`involved`
+/// backend would disagree; both must return the transfer for both accounts.
+pub async fn store_transfer_indexes_involved(store: &(impl Store + 'static)) {
+    let consumed = PostingId {
+        transfer: EnvelopeId([9; 32]),
+        index: 0,
+    };
+    let tid = EnvelopeId([7; 32]);
+    let envelope = EnvelopeBuilder::new()
+        .consumes(vec![consumed])
+        .creates(vec![NewPosting {
+            owner: AccountId::new(8),
+            asset: AssetId::new(1),
+            value: Cent::from(100),
+            payer: None,
+        }])
+        .build();
+    let record = EnvelopeRecord {
+        envelope,
+        receipt: Receipt { transfer_id: tid },
+        created_at: 2000,
+    };
+    // Full participation set: created owner 8 plus consumed owner 7. Account 7's
+    // posting is never inserted, so only `involved` can surface the transfer.
+    let involved = [AccountId::new(8), AccountId::new(7)];
+
+    assert_eq!(store.store_transfer(record, &involved).await.unwrap(), 1);
+
+    // Consumed-owner branch: found purely because `involved` said so.
+    let for_consumed = store.get_transfers_for_account(7, None).await.unwrap();
+    assert_eq!(for_consumed.len(), 1);
+    assert_eq!(for_consumed[0].receipt.transfer_id, tid);
+
+    // The created owner is indexed identically.
+    let for_created = store.get_transfers_for_account(8, None).await.unwrap();
+    assert_eq!(for_created.len(), 1);
+    assert_eq!(for_created[0].receipt.transfer_id, tid);
+}
+
 // ---------------------------------------------------------------------------
 // Reservation / double-spend regressions (sequential — the conformance harness
 // holds a single `&store`; the second attempt is what must report zero).
@@ -1177,6 +1221,7 @@ macro_rules! store_tests {
             spent_posting_remains_in_immutable_table,
             get_postings_by_account_live_filter,
             store_transfer_counts,
+            store_transfer_indexes_involved,
             // Reservation / double-spend regressions
             reserve_twice_second_zero,
             deactivate_twice_second_zero,