From 74a1a032eb270f5601f587d46fa3caac0c0ca880 Mon Sep 17 00:00:00 2001 From: leora Date: Thu, 15 Oct 2026 21:53:29 +0800 Subject: [PATCH] Investments and contract payments draw funded credit only; reverse_tx correction helper by original draw kinds --- coffer/ledger.py | 64 ++++++++++++++++++++++++++++++++++++--- tests/test_funded_only.py | 64 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 123 insertions(+), 5 deletions(-) create mode 100644 tests/test_funded_only.py diff --git a/coffer/ledger.py b/coffer/ledger.py index f267f24..17bd0e3 100644 --- a/coffer/ledger.py +++ b/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, diff --git a/tests/test_funded_only.py b/tests/test_funded_only.py new file mode 100644 index 0000000..2c74b5f --- /dev/null +++ b/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()