388 lines
13 KiB
Python
388 lines
13 KiB
Python
"""Prepaid agent credit ledger with an append-only transaction log.
|
|
|
|
Funded credit is a handler top-up. Promotional credit is a grant.
|
|
Spend at Coffer uses promotional lots first, then funded lots.
|
|
Amounts are stored in centicredits (see coffer.units.SCALE).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
from typing import Any
|
|
|
|
from coffer.units import SCALE, format_credits, to_subunits
|
|
|
|
|
|
class CreditKind(str, Enum):
|
|
FUNDED = "funded"
|
|
PROMOTIONAL = "promotional"
|
|
|
|
|
|
FUNDED_ONLY_TX_KINDS = frozenset({"investment", "contract_payment"})
|
|
|
|
|
|
class InsufficientCredit(Exception):
|
|
"""The account does not have enough spendable credit."""
|
|
|
|
|
|
class TransferNotAllowed(Exception):
|
|
"""Peer agent-to-agent transfer is not allowed."""
|
|
|
|
|
|
@dataclass
|
|
class Lot:
|
|
kind: CreditKind
|
|
remaining: int # centicredits
|
|
reason: str = ""
|
|
|
|
|
|
@dataclass
|
|
class Account:
|
|
account_id: str
|
|
lots: list[Lot] = field(default_factory=list)
|
|
|
|
def balance_subunits(self) -> dict[str, int]:
|
|
promotional = sum(
|
|
lot.remaining for lot in self.lots if lot.kind is CreditKind.PROMOTIONAL
|
|
)
|
|
funded = sum(lot.remaining for lot in self.lots if lot.kind is CreditKind.FUNDED)
|
|
return {
|
|
"promotional": promotional,
|
|
"funded": funded,
|
|
"spendable": promotional + funded,
|
|
}
|
|
|
|
def balance(self) -> dict[str, float]:
|
|
raw = self.balance_subunits()
|
|
return {k: v / SCALE for k, v in raw.items()}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Tx:
|
|
seq: int
|
|
kind: str
|
|
amount: int # centicredits
|
|
debit_account: str | None
|
|
credit_account: str | None
|
|
reason: str
|
|
meta: dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
def _positive_int(amount: int, label: str) -> None:
|
|
if type(amount) is not int or amount <= 0:
|
|
raise ValueError(f"{label} must be a positive whole number")
|
|
|
|
|
|
def _nonneg_int(amount: int, label: str) -> None:
|
|
if type(amount) is not int or amount < 0:
|
|
raise ValueError(f"{label} must be a non-negative whole number")
|
|
|
|
|
|
class Ledger:
|
|
"""In-memory ledger. DurableProduct wraps this with SQLite persistence."""
|
|
|
|
def __init__(self) -> None:
|
|
self.accounts: dict[str, Account] = {}
|
|
self.transactions: list[Tx] = []
|
|
self._tx_seq = 0
|
|
|
|
def account(self, account_id: str) -> Account:
|
|
if account_id not in self.accounts:
|
|
self.accounts[account_id] = Account(account_id)
|
|
return self.accounts[account_id]
|
|
|
|
def ensure_wallet(self, account_id: str) -> Account:
|
|
return self.account(account_id)
|
|
|
|
def balance_subunits(self, account_id: str) -> dict[str, int]:
|
|
if account_id not in self.accounts:
|
|
return {"promotional": 0, "funded": 0, "spendable": 0}
|
|
return self.accounts[account_id].balance_subunits()
|
|
|
|
def balance(self, account_id: str) -> dict[str, float]:
|
|
raw = self.balance_subunits(account_id)
|
|
return {k: v / SCALE for k, v in raw.items()}
|
|
|
|
# --- whole-credit helpers (catalog / grants) ---------------------------------
|
|
|
|
def top_up(self, account_id: str, amount_credits: int) -> Lot:
|
|
"""Record a handler prepayment in whole credits."""
|
|
_positive_int(amount_credits, "top-up")
|
|
return self._credit(
|
|
account_id,
|
|
to_subunits(amount_credits),
|
|
CreditKind.FUNDED,
|
|
"top-up",
|
|
tx_kind="top_up",
|
|
)
|
|
|
|
def grant(self, account_id: str, amount_credits: int, reason: str = "promotional") -> Lot:
|
|
"""Grant promotional credit in whole credits."""
|
|
_positive_int(amount_credits, "grant")
|
|
if not reason:
|
|
raise ValueError("a grant needs a reason")
|
|
return self._credit(
|
|
account_id,
|
|
to_subunits(amount_credits),
|
|
CreditKind.PROMOTIONAL,
|
|
reason,
|
|
tx_kind="grant",
|
|
)
|
|
|
|
def spend(self, account_id: str, amount_credits: int) -> list[dict[str, int | str]]:
|
|
"""Spend whole credits at Coffer (reduces lots; no counterparty credit)."""
|
|
_positive_int(amount_credits, "spend")
|
|
draws_sub = self._debit_lots(account_id, to_subunits(amount_credits))
|
|
draws = [
|
|
{"kind": d["kind"], "amount": int(d["amount"]) // SCALE} for d in draws_sub
|
|
]
|
|
self._append_tx(
|
|
kind="spend",
|
|
amount=to_subunits(amount_credits),
|
|
debit_account=account_id,
|
|
credit_account=None,
|
|
reason="spend",
|
|
meta={"draws": draws_sub},
|
|
)
|
|
return draws
|
|
|
|
# --- subunit primitives ------------------------------------------------------
|
|
|
|
def credit_subunits(
|
|
self,
|
|
account_id: str,
|
|
amount: int,
|
|
kind: CreditKind,
|
|
reason: str,
|
|
*,
|
|
tx_kind: str = "credit",
|
|
meta: dict[str, Any] | None = None,
|
|
log: bool = True,
|
|
) -> Lot:
|
|
return self._credit(account_id, amount, kind, reason, tx_kind=tx_kind, meta=meta, log=log)
|
|
|
|
def _credit(
|
|
self,
|
|
account_id: str,
|
|
amount: int,
|
|
kind: CreditKind,
|
|
reason: str,
|
|
*,
|
|
tx_kind: str,
|
|
meta: dict[str, Any] | None = None,
|
|
log: bool = True,
|
|
) -> Lot:
|
|
_positive_int(amount, "credit amount")
|
|
lot = Lot(kind, amount, reason)
|
|
self.account(account_id).lots.append(lot)
|
|
if log:
|
|
self._append_tx(
|
|
kind=tx_kind,
|
|
amount=amount,
|
|
debit_account=None,
|
|
credit_account=account_id,
|
|
reason=reason,
|
|
meta=meta or {"lot_kind": kind.value},
|
|
)
|
|
return lot
|
|
|
|
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)
|
|
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 kinds
|
|
for lot in account.lots
|
|
if lot.kind is kind and lot.remaining
|
|
]
|
|
left = amount
|
|
draws: list[dict[str, int | str]] = []
|
|
for lot in ordered:
|
|
if left == 0:
|
|
break
|
|
take = min(lot.remaining, left)
|
|
lot.remaining -= take
|
|
left -= take
|
|
draws.append({"kind": lot.kind.value, "amount": take})
|
|
return draws
|
|
|
|
def transfer(
|
|
self,
|
|
debit_account: str,
|
|
credit_account: str,
|
|
amount_subunits: int,
|
|
reason: str,
|
|
*,
|
|
tx_kind: str = "transfer",
|
|
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.
|
|
|
|
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")
|
|
if debit_account == credit_account:
|
|
raise ValueError("cannot transfer to the same account")
|
|
if (
|
|
not allow_agent_peer
|
|
and debit_account.startswith("agent:")
|
|
and credit_account.startswith("agent:")
|
|
):
|
|
raise TransferNotAllowed("agent-to-agent transfers are not allowed")
|
|
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(
|
|
kind=tx_kind,
|
|
amount=amount_subunits,
|
|
debit_account=debit_account,
|
|
credit_account=credit_account,
|
|
reason=reason,
|
|
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,
|
|
merchant_id: str,
|
|
amount_subunits: int,
|
|
reason: str,
|
|
*,
|
|
tx_kind: str = "purchase",
|
|
meta: dict[str, Any] | None = None,
|
|
) -> tuple[list[dict[str, int | str]], Tx]:
|
|
"""Buyer pays merchant: debit buyer lots, credit merchant funded lot."""
|
|
_positive_int(amount_subunits, "purchase amount")
|
|
draws = self._debit_lots(buyer_id, amount_subunits)
|
|
self.account(merchant_id).lots.append(
|
|
Lot(CreditKind.FUNDED, amount_subunits, reason)
|
|
)
|
|
tx = self._append_tx(
|
|
kind=tx_kind,
|
|
amount=amount_subunits,
|
|
debit_account=buyer_id,
|
|
credit_account=merchant_id,
|
|
reason=reason,
|
|
meta={**(meta or {}), "draws": draws},
|
|
)
|
|
return draws, tx
|
|
|
|
def bootstrap_endowment(
|
|
self,
|
|
account_id: str,
|
|
amount_credits: int,
|
|
kind: CreditKind,
|
|
reason: str,
|
|
) -> Lot:
|
|
"""Seed opening balance without writing a business transaction.
|
|
|
|
Used so the first logged business tx can be the 200c contract payment.
|
|
"""
|
|
_positive_int(amount_credits, "endowment")
|
|
lot = Lot(kind, to_subunits(amount_credits), reason)
|
|
self.account(account_id).lots.append(lot)
|
|
return lot
|
|
|
|
def _append_tx(
|
|
self,
|
|
*,
|
|
kind: str,
|
|
amount: int,
|
|
debit_account: str | None,
|
|
credit_account: str | None,
|
|
reason: str,
|
|
meta: dict[str, Any] | None = None,
|
|
) -> Tx:
|
|
self._tx_seq += 1
|
|
tx = Tx(
|
|
seq=self._tx_seq,
|
|
kind=kind,
|
|
amount=amount,
|
|
debit_account=debit_account,
|
|
credit_account=credit_account,
|
|
reason=reason,
|
|
meta=meta or {},
|
|
)
|
|
self.transactions.append(tx)
|
|
return tx
|
|
|
|
def history(self) -> list[Tx]:
|
|
return list(self.transactions)
|
|
|
|
def describe_tx(self, tx: Tx) -> dict[str, Any]:
|
|
return {
|
|
"seq": tx.seq,
|
|
"kind": tx.kind,
|
|
"amount_centicredits": tx.amount,
|
|
"amount_credits": format_credits(tx.amount),
|
|
"debit_account": tx.debit_account,
|
|
"credit_account": tx.credit_account,
|
|
"reason": tx.reason,
|
|
"meta": tx.meta,
|
|
}
|