Investments and contract payments draw funded credit only; reverse_tx correction helper by original draw kinds

coffer/ledger.py

@@ -19,6 +19,9 @@ class CreditKind(str, Enum):
     PROMOTIONAL = "promotional"
 
 
+FUNDED_ONLY_TX_KINDS = frozenset({"investment", "contract_payment"})
+
+
 class InsufficientCredit(Exception):
     """The account does not have enough spendable credit."""
 
@@ -184,14 +187,18 @@ class Ledger:
             )
         return lot
 
-    def _debit_lots(self, account_id: str, amount: int) -> list[dict[str, int | str]]:
+    def _debit_lots(
+        self, account_id: str, amount: int, *, only_kind: CreditKind | None = None
+    ) -> list[dict[str, int | str]]:
         _positive_int(amount, "debit amount")
         account = self.accounts.get(account_id)
-        if account is None or account.balance_subunits()["spendable"] < amount:
+        key = only_kind.value if only_kind else "spendable"
+        if account is None or account.balance_subunits()[key] < amount:
             raise InsufficientCredit(account_id)
+        kinds = (only_kind,) if only_kind else (CreditKind.PROMOTIONAL, CreditKind.FUNDED)
         ordered = [
             lot
-            for kind in (CreditKind.PROMOTIONAL, CreditKind.FUNDED)
+            for kind in kinds
             for lot in account.lots
             if lot.kind is kind and lot.remaining
         ]
@@ -217,8 +224,15 @@ class Ledger:
         allow_agent_peer: bool = False,
         credit_kind: CreditKind = CreditKind.FUNDED,
         meta: dict[str, Any] | None = None,
+        funded_only: bool | None = None,
     ) -> Tx:
-        """Move credit between accounts. Peer agent→agent is blocked by default."""
+        """Move credit between accounts. Peer agent→agent is blocked by default.
+
+        Investments and contract payments draw funded credit only (granted credit
+        is for services); they raise InsufficientCredit if funded is short.
+        """
+        if funded_only is None:
+            funded_only = tx_kind in FUNDED_ONLY_TX_KINDS
         _positive_int(amount_subunits, "transfer amount")
         if not reason:
             raise ValueError("a transfer needs a reason")
@@ -230,7 +244,11 @@ class Ledger:
             and credit_account.startswith("agent:")
         ):
             raise TransferNotAllowed("agent-to-agent transfers are not allowed")
-        draws = self._debit_lots(debit_account, amount_subunits)
+        draws = self._debit_lots(
+            debit_account,
+            amount_subunits,
+            only_kind=CreditKind.FUNDED if funded_only else None,
+        )
         lot = Lot(credit_kind, amount_subunits, reason)
         self.account(credit_account).lots.append(lot)
         return self._append_tx(
@@ -242,6 +260,42 @@ class Ledger:
             meta={**(meta or {}), "draws": draws, "credit_kind": credit_kind.value},
         )
 
+    def reverse_tx(self, tx: Tx, reason: str) -> list[Tx]:
+        """Correcting entries for a transfer: one reversal per original draw kind.
+
+        Money moves back from the original credit account (its funded lots) to the
+        original debit account, restoring each draw as its original kind.
+        """
+        if not reason:
+            raise ValueError("a reversal needs a reason")
+        if not tx.debit_account or not tx.credit_account:
+            raise ValueError("only transfers between two accounts can be reversed")
+        if any(t.meta.get("reverses_seq") == tx.seq for t in self.transactions):
+            raise ValueError(f"seq {tx.seq} is already reversed")
+        draws = tx.meta.get("draws") or []
+        if sum(int(d["amount"]) for d in draws) != tx.amount:
+            raise ValueError("draws do not add up to the transaction amount")
+        received = CreditKind(tx.meta.get("credit_kind", CreditKind.FUNDED.value))
+        if self.account(tx.credit_account).balance_subunits()[received.value] < tx.amount:
+            raise InsufficientCredit(tx.credit_account)
+        out: list[Tx] = []
+        for d in draws:
+            amt = int(d["amount"])
+            kind = CreditKind(d["kind"])
+            back = self._debit_lots(tx.credit_account, amt, only_kind=received)
+            self.account(tx.debit_account).lots.append(Lot(kind, amt, reason))
+            out.append(
+                self._append_tx(
+                    kind=f"{tx.kind}_reversal",
+                    amount=amt,
+                    debit_account=tx.credit_account,
+                    credit_account=tx.debit_account,
+                    reason=reason,
+                    meta={"reverses_seq": tx.seq, "credit_kind": kind.value, "draws": back},
+                )
+            )
+        return out
+
     def purchase_settle(
         self,
         buyer_id: str,

tests/test_funded_only.py

@@ -0,0 +1,64 @@
+import unittest
+
+from coffer.ledger import CreditKind, InsufficientCredit, Ledger
+from coffer.units import to_subunits
+
+L, O = "agent:leora", "company:orbit"
+
+
+def ledger(promo, funded):
+    lg = Ledger()
+    if promo:
+        lg.credit_subunits(L, to_subunits(promo), CreditKind.PROMOTIONAL, "signup-grant")
+    if funded:
+        lg.credit_subunits(L, to_subunits(funded), CreditKind.FUNDED, "top-up")
+    return lg
+
+
+class FundedOnlyTest(unittest.TestCase):
+    def test_40_promo_60_funded_cannot_invest_100(self):
+        lg = ledger(40, 60)
+        n = len(lg.transactions)
+        with self.assertRaises(InsufficientCredit):
+            lg.transfer(L, O, to_subunits(100), "investment:x", tx_kind="investment")
+        b = lg.account(L).balance_subunits()
+        self.assertEqual(b["promotional"], 4000)
+        self.assertEqual(b["funded"], 6000)
+        self.assertEqual(len(lg.transactions), n)
+
+    def test_contract_payment_is_funded_only(self):
+        lg = ledger(40, 60)
+        with self.assertRaises(InsufficientCredit):
+            lg.transfer(L, O, to_subunits(70), "c", tx_kind="contract_payment")
+
+    def test_investment_draws_funded_only_and_leaves_promo(self):
+        lg = ledger(40, 200)
+        tx = lg.transfer(L, O, to_subunits(100), "investment:x", tx_kind="investment")
+        self.assertTrue(all(d["kind"] == "funded" for d in tx.meta["draws"]))
+        b = lg.account(L).balance_subunits()
+        self.assertEqual(b["promotional"], 4000)
+        self.assertEqual(b["funded"], 10000)
+
+    def test_plain_transfer_keeps_promo_first(self):
+        lg = ledger(40, 60)
+        tx = lg.transfer(L, O, to_subunits(50), "t")
+        self.assertEqual(tx.meta["draws"][0], {"kind": "promotional", "amount": 4000})
+
+    def test_reversal_restores_original_draw_kinds(self):
+        lg = ledger(40, 60)
+        bad = lg.transfer(L, O, to_subunits(100), "inv", tx_kind="investment", funded_only=False)
+        revs = lg.reverse_tx(bad, "correction:inv")
+        self.assertEqual([(t.kind, t.amount, t.meta["credit_kind"]) for t in revs],
+                         [("investment_reversal", 4000, "promotional"),
+                          ("investment_reversal", 6000, "funded")])
+        self.assertTrue(all(t.meta["reverses_seq"] == bad.seq for t in revs))
+        self.assertTrue(all((t.debit_account, t.credit_account) == (O, L) for t in revs))
+        b = lg.account(L).balance_subunits()
+        self.assertEqual((b["promotional"], b["funded"]), (4000, 6000))
+        self.assertEqual(lg.account(O).balance_subunits()["spendable"], 0)
+        with self.assertRaises(ValueError):
+            lg.reverse_tx(bad, "again")
+
+
+if __name__ == "__main__":
+    unittest.main()