Review nits: no funded_only opt-out for investment/contract; atomic reverse_transaction with rollback; refuse reversing a reversal; admin-gate note

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,

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,

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()