diff --git a/coffer/ledger.py b/coffer/ledger.py index f267f24..a135eb4 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,19 @@ 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 tx_kind in FUNDED_ONLY_TX_KINDS: + if funded_only is False: + raise TransferNotAllowed(f"{tx_kind} must draw funded credit only") + funded_only = True + elif funded_only is None: + funded_only = False _positive_int(amount_subunits, "transfer amount") if not reason: raise ValueError("a transfer needs a reason") @@ -230,7 +248,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 +264,48 @@ 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. + + Operator correction tool only. It has no authorization of its own: an + admin/operator gate is REQUIRED before exposing it through any API, + CLI command or agent-facing route. + """ + 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 tx.kind.endswith("_reversal") or "reverses_seq" in tx.meta: + raise ValueError("a reversal cannot itself 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/coffer/product.py b/coffer/product.py index de96830..3f9270d 100644 --- a/coffer/product.py +++ b/coffer/product.py @@ -350,6 +350,35 @@ class CofferProduct: "payee": self.balance(payee), } + def reverse_transaction(self, seq: int, reason: str) -> dict[str, Any]: + """Reverse ledger tx `seq` by its original draw kinds, saved atomically. + + The in-memory reversal and the SQLite writes (lot rows rewritten, + reversal txs inserted) commit together or not at all; on any failure + the DB is rolled back and the ledger reloaded from it. + Operator-only: requires an admin gate before any API/CLI exposure. + """ + tx = next((t for t in self.ledger.history() if t.seq == seq), None) + if tx is None: + raise ProductError(f"no transaction with seq {seq}") + try: + revs = self.ledger.reverse_tx(tx, reason) + for aid in {tx.debit_account, tx.credit_account}: + self._rewrite_lots(aid) + for r in revs: + self._persist_tx(r) + self.db.commit() + except Exception: + self.db._conn.rollback() + self._load() + raise + self._after_mutation() + return { + "reversals": [self.ledger.describe_tx(r) for r in revs], + tx.debit_account: self.balance(tx.debit_account), + tx.credit_account: self.balance(tx.credit_account), + } + def grant_signup( self, agent_id: str, diff --git a/tests/test_funded_only.py b/tests/test_funded_only.py new file mode 100644 index 0000000..faf56bd --- /dev/null +++ b/tests/test_funded_only.py @@ -0,0 +1,125 @@ +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) + from unittest import mock + # Simulate the pre-fix code path that produced ledger seq 39 (promo drawn). + with mock.patch("coffer.ledger.FUNDED_ONLY_TX_KINDS", frozenset()): + bad = lg.transfer(L, O, to_subunits(100), "inv", tx_kind="investment") + self.assertEqual(bad.meta["draws"][0]["kind"], "promotional") + 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() + + +class FundedOnlyGuardsTest(unittest.TestCase): + def test_cannot_opt_out_of_funded_only(self): + from coffer.ledger import TransferNotAllowed + for kind in ("investment", "contract_payment"): + lg = ledger(40, 200) + with self.assertRaises(TransferNotAllowed): + lg.transfer(L, O, to_subunits(100), "x", tx_kind=kind, funded_only=False) + self.assertEqual(lg.account(L).balance_subunits()["promotional"], 4000) + + def test_reversal_of_reversal_refused(self): + lg = ledger(0, 100) + tx = lg.transfer(L, O, to_subunits(100), "inv", tx_kind="investment") + rev = lg.reverse_tx(tx, "fix")[0] + with self.assertRaises(ValueError): + lg.reverse_tx(rev, "fix again") + + +class ProductReverseAtomicTest(unittest.TestCase): + def setUp(self): + import tempfile, os + from coffer.product import CofferProduct + self.d = tempfile.TemporaryDirectory() + os.environ["COFFER_AUTO_PUBLISH_NEWS"] = "0" + self.path = f"{self.d.name}/c.db" + self.p = CofferProduct(self.path) + self.p.ledger.credit_subunits(L, 10000, CreditKind.FUNDED, "top-up", log=False) + tx = self.p.ledger.transfer(L, O, 10000, "inv", tx_kind="investment") + self.p._commit_accounts(L, O, txs=[tx]) + self.seq = tx.seq + + def tearDown(self): + self.p.close(); self.d.cleanup() + + def test_reverse_persists(self): + from coffer.product import CofferProduct + self.p.reverse_transaction(self.seq, "fix") + q = CofferProduct(self.path) + self.assertEqual(q.balance(L)["funded_centicredits"], 10000) + self.assertEqual(q.balance(O)["spendable_centicredits"], 0) + self.assertEqual(q.ledger.history()[-1].meta["reverses_seq"], self.seq) + q.close() + + def test_failed_save_rolls_back(self): + from unittest import mock + from coffer.product import CofferProduct + n = len(self.p.ledger.history()) + with mock.patch.object(self.p, "_persist_tx", side_effect=RuntimeError("disk")): + with self.assertRaises(RuntimeError): + self.p.reverse_transaction(self.seq, "fix") + self.assertEqual(len(self.p.ledger.history()), n) + self.assertEqual(self.p.balance(O)["spendable_centicredits"], 10000) + q = CofferProduct(self.path) + self.assertEqual(len(q.ledger.history()), n) + self.assertEqual(q.balance(O)["spendable_centicredits"], 10000) + q.close()