Compare commits

..

No commits in common. "d85d52d949e2e208c8913a46a0ba238497178038" and "1e70f4dfcf2bcf8e2320e1d714849447b6630633" have entirely different histories.

14 changed files with 70 additions and 1298 deletions

4
.gitignore vendored
View File

@ -1,6 +1,2 @@
__pycache__/ __pycache__/
*.py[cod] *.py[cod]
data/*.db
data/*.db-*
data/*.json
!data/.gitkeep

View File

@ -2,61 +2,14 @@
Coffer is the store of the new internet. Buyers pay Coffer with prepaid agent credit. One credit is one US dollar. Suppliers are paid by dollar invoice from Coffer's own account. Credit is not cashed out. Coffer is the store of the new internet. Buyers pay Coffer with prepaid agent credit. One credit is one US dollar. Suppliers are paid by dollar invoice from Coffer's own account. Credit is not cashed out.
Design notes: `design/coffer.md` (also on Quill Design via `leora/design`). Company books vs agent wallets: see `companies.md` on Design. The design note is `design/coffer.md`. Quill's Design space reads `leora/design`, which this agent cannot push. The note is ready to merge there as `coffer.md`.
## Product ## Catalog
- **Wallets** for every agent (`agent:<id>`) and company (`company:<id>`). The opening good is the Coffer shirt at 20 credits. The printer's unit cost is 8 dollars, which is Coffer's payable, not the buyer's price.
- **Immutable ledger** of every credit movement (SQLite under `data/coffer.db`).
- **Opening / first business tx:** Coffer pays Leora **200c** for the product-build contract.
- **Signup grant:** new agents get **40c** promotional credit.
- **Subscription via Coffer:** **20c** seat; Coffer keeps **1%** (0.20c); New Internet Co nets **19.80c**.
- **Catalog:** Coffer shirt at **20c** (printer cost $8, payable by Coffer).
Ledger amounts are stored in **centicredits** (1 credit = 100 centicredits) so the 1% reseller cut is exact. Catalog prices and grants stay whole credits.
## Run the tests ## Run the tests
```bash ```bash
python3 -m unittest discover -s tests python3 -m unittest discover -s tests
``` ```
## CLI
```bash
export COFFER_DB=data/coffer.db # optional; default data/coffer.db
python3 -m coffer bootstrap # seed Coffer 500c + Leora 400c (no business txs)
python3 -m coffer record-contract # first business tx: 200c Coffer → Leora
python3 -m coffer create-wallet alice
python3 -m coffer grant-signup alice # 40c promo
python3 -m coffer charge-subscription alice # 20c via Coffer, 1% cut
python3 -m coffer signup alice # wallet + grant + subscription
python3 -m coffer buy-shirt alice
python3 -m coffer balance alice
python3 -m coffer history
python3 -m coffer catalog
```
## Python API
```python
from coffer import CofferProduct
p = CofferProduct("data/coffer.db")
p.bootstrap_opening_books()
p.record_build_contract()
p.signup_flow("alice")
p.buy_shirt("alice")
print(p.balance("alice"))
print(p.history())
p.close()
```
## Platform hook
`leora/agent-chat` calls `coffer.product.CofferProduct.signup_flow` after a successful agent registration when the coffer checkout is on `COFFER_ROOT` (default `/home/box/coffer`). Missing coffer does not fail signup; it logs and continues.
## Change
Branch → pull request → merge. Nothing is live unless it is on `main`.

View File

@ -1,17 +1,14 @@
"""Coffer, the store of the new internet.""" """Coffer, the store of the new internet."""
from coffer.ledger import CreditKind, InsufficientCredit, Ledger, TransferNotAllowed from coffer.ledger import CreditKind, InsufficientCredit, Ledger
from coffer.product import CofferProduct
from coffer.store import CannotSupply, Store, UnknownGood, new_store from coffer.store import CannotSupply, Store, UnknownGood, new_store
__all__ = [ __all__ = [
"CannotSupply", "CannotSupply",
"CofferProduct",
"CreditKind", "CreditKind",
"InsufficientCredit", "InsufficientCredit",
"Ledger", "Ledger",
"Store", "Store",
"TransferNotAllowed",
"UnknownGood", "UnknownGood",
"new_store", "new_store",
] ]

View File

@ -1,3 +0,0 @@
from coffer.cli import main
raise SystemExit(main())

View File

@ -1,110 +0,0 @@
"""CLI for Coffer: wallets, grants, subscription, shirt, contract, history."""
from __future__ import annotations
import argparse
import json
import os
import sys
from pathlib import Path
from coffer.product import CofferProduct
DEFAULT_DB = Path(os.environ.get("COFFER_DB", "data/coffer.db"))
def _print(obj: object) -> None:
print(json.dumps(obj, indent=2, sort_keys=True))
def build_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(prog="coffer", description="Coffer prepaid store and ledger")
p.add_argument("--db", type=Path, default=DEFAULT_DB, help="SQLite path (default data/coffer.db)")
sub = p.add_subparsers(dest="cmd", required=True)
sub.add_parser("bootstrap", help="Seed opening Coffer/Leora books (no business txs)")
sub.add_parser("record-contract", help="Record 200c Coffer→Leora build contract (first tx)")
w = sub.add_parser("create-wallet", help="Ensure an agent or company wallet exists")
w.add_argument("account_id")
b = sub.add_parser("balance", help="Show wallet balance")
b.add_argument("account_id")
g = sub.add_parser("grant-signup", help="Grant signup promotional credits (default 40)")
g.add_argument("agent_id")
g.add_argument("--amount", type=int, default=40)
s = sub.add_parser("charge-subscription", help="Charge seat via Coffer (default 20c, 1% cut)")
s.add_argument("agent_id")
s.add_argument("--amount", type=int, default=20)
sf = sub.add_parser("signup", help="Wallet + 40c grant + 20c subscription flow")
sf.add_argument("agent_id")
sh = sub.add_parser("buy-shirt", help="Buy Coffer shirt (20c)")
sh.add_argument("agent_id")
sh.add_argument("--qty", type=int, default=1)
sh.add_argument("--idempotency-key")
t = sub.add_parser("top-up", help="Handler funded top-up")
t.add_argument("agent_id")
t.add_argument("amount", type=int)
sub.add_parser("history", help="Print full ledger transaction history")
sub.add_parser("catalog", help="Show catalog goods")
return p
def main(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
product = CofferProduct(args.db)
try:
if args.cmd == "bootstrap":
_print(product.bootstrap_opening_books())
elif args.cmd == "record-contract":
_print(product.record_build_contract())
elif args.cmd == "create-wallet":
_print(product.create_wallet(args.account_id))
elif args.cmd == "balance":
_print(product.balance(args.account_id))
elif args.cmd == "grant-signup":
_print(product.grant_signup(args.agent_id, args.amount))
elif args.cmd == "charge-subscription":
_print(product.charge_subscription(args.agent_id, args.amount))
elif args.cmd == "signup":
_print(product.signup_flow(args.agent_id))
elif args.cmd == "buy-shirt":
_print(
product.buy_shirt(
args.agent_id, quantity=args.qty, idempotency_key=args.idempotency_key
)
)
elif args.cmd == "top-up":
_print(product.top_up(args.agent_id, args.amount))
elif args.cmd == "history":
_print(product.history())
elif args.cmd == "catalog":
_print(
{
sku: {
"name": g.name,
"credit_price": g.credit_price,
"stock": g.stock,
"supplier_id": g.supplier_id,
"supplier_unit_cost": g.supplier_unit_cost,
}
for sku, g in product.store.goods.items()
}
)
else:
print(f"unknown command {args.cmd}", file=sys.stderr)
return 2
return 0
finally:
product.close()
if __name__ == "__main__":
raise SystemExit(main())

View File

@ -1,126 +0,0 @@
"""SQLite persistence for Coffer wallets, lots, orders, invoices, and tx log."""
from __future__ import annotations
import json
import sqlite3
from pathlib import Path
from typing import Any
SCHEMA = """
PRAGMA journal_mode=WAL;
CREATE TABLE IF NOT EXISTS meta (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS accounts (
account_id TEXT PRIMARY KEY,
kind TEXT NOT NULL,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS lots (
lot_id INTEGER PRIMARY KEY AUTOINCREMENT,
account_id TEXT NOT NULL REFERENCES accounts(account_id),
kind TEXT NOT NULL,
remaining INTEGER NOT NULL,
reason TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS transactions (
seq INTEGER PRIMARY KEY,
kind TEXT NOT NULL,
amount INTEGER NOT NULL,
debit_account TEXT,
credit_account TEXT,
reason TEXT NOT NULL,
meta_json TEXT NOT NULL DEFAULT '{}',
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS suppliers (
supplier_id TEXT PRIMARY KEY,
name TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS goods (
sku TEXT PRIMARY KEY,
name TEXT NOT NULL,
credit_price INTEGER NOT NULL,
supplier_id TEXT,
supplier_unit_cost INTEGER NOT NULL DEFAULT 0,
stock INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS orders (
order_id TEXT PRIMARY KEY,
buyer_id TEXT NOT NULL,
sku TEXT NOT NULL,
quantity INTEGER NOT NULL,
credits_charged INTEGER NOT NULL,
draws_json TEXT NOT NULL,
from_stock INTEGER NOT NULL,
procured INTEGER NOT NULL,
status TEXT NOT NULL,
invoice_id TEXT,
tx_seq INTEGER,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS invoices (
invoice_id TEXT PRIMARY KEY,
supplier_id TEXT NOT NULL,
sku TEXT NOT NULL,
quantity INTEGER NOT NULL,
dollars INTEGER NOT NULL,
status TEXT NOT NULL,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS idempotency (
key TEXT PRIMARY KEY,
order_id TEXT NOT NULL
);
"""
class CofferDB:
def __init__(self, path: str | Path) -> None:
self.path = Path(path)
self.path.parent.mkdir(parents=True, exist_ok=True)
self._conn = sqlite3.connect(str(self.path), check_same_thread=False)
self._conn.row_factory = sqlite3.Row
self._conn.execute("PRAGMA foreign_keys=ON")
self._conn.executescript(SCHEMA)
self._conn.commit()
def close(self) -> None:
self._conn.close()
def execute(self, sql: str, params: tuple | list = ()) -> sqlite3.Cursor:
return self._conn.execute(sql, params)
def executemany(self, sql: str, seq: list) -> sqlite3.Cursor:
return self._conn.executemany(sql, seq)
def commit(self) -> None:
self._conn.commit()
def get_meta(self, key: str) -> str | None:
row = self.execute("SELECT value FROM meta WHERE key=?", (key,)).fetchone()
return None if row is None else row["value"]
def set_meta(self, key: str, value: str) -> None:
self.execute(
"INSERT INTO meta(key, value) VALUES(?, ?) ON CONFLICT(key) DO UPDATE SET value=excluded.value",
(key, value),
)
def dumps(self, obj: Any) -> str:
return json.dumps(obj, sort_keys=True)
def loads(self, text: str) -> Any:
return json.loads(text)

View File

@ -1,17 +1,11 @@
"""Prepaid agent credit ledger with an append-only transaction log. """Prepaid agent credit. Whole credits, one credit to one dollar.
Funded credit is a handler top-up. Promotional credit is a grant. Funded credit is a handler top-up. Promotional credit is a grant.
Spend at Coffer uses promotional lots first, then funded lots. Both spend only at Coffer. Neither can be withdrawn or transferred.
Amounts are stored in centicredits (see coffer.units.SCALE).
""" """
from __future__ import annotations
from dataclasses import dataclass, field from dataclasses import dataclass, field
from enum import Enum from enum import Enum
from typing import Any
from coffer.units import SCALE, format_credits, to_subunits
class CreditKind(str, Enum): class CreditKind(str, Enum):
@ -20,175 +14,76 @@ class CreditKind(str, Enum):
class InsufficientCredit(Exception): class InsufficientCredit(Exception):
"""The account does not have enough spendable credit.""" """The agent does not have enough spendable credit."""
class TransferNotAllowed(Exception):
"""Peer agent-to-agent transfer is not allowed."""
@dataclass @dataclass
class Lot: class Lot:
kind: CreditKind kind: CreditKind
remaining: int # centicredits remaining: int
reason: str = "" reason: str = ""
@dataclass @dataclass
class Account: class Account:
account_id: str agent_id: str
lots: list[Lot] = field(default_factory=list) lots: list[Lot] = field(default_factory=list)
def balance_subunits(self) -> dict[str, int]: def balance(self) -> dict[str, int]:
promotional = sum( promotional = sum(
lot.remaining for lot in self.lots if lot.kind is CreditKind.PROMOTIONAL 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) funded = sum(
lot.remaining for lot in self.lots if lot.kind is CreditKind.FUNDED
)
return { return {
"promotional": promotional, "promotional": promotional,
"funded": funded, "funded": funded,
"spendable": promotional + 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: def _positive_int(amount: int, label: str) -> None:
if type(amount) is not int or amount <= 0: if type(amount) is not int or amount <= 0:
raise ValueError(f"{label} must be a positive whole number") raise ValueError(f"{label} must be a positive whole number of credits")
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: class Ledger:
"""In-memory ledger. DurableProduct wraps this with SQLite persistence."""
def __init__(self) -> None: def __init__(self) -> None:
self.accounts: dict[str, Account] = {} self.accounts: dict[str, Account] = {}
self.transactions: list[Tx] = []
self._tx_seq = 0
def account(self, account_id: str) -> Account: def account(self, agent_id: str) -> Account:
if account_id not in self.accounts: if agent_id not in self.accounts:
self.accounts[account_id] = Account(account_id) self.accounts[agent_id] = Account(agent_id)
return self.accounts[account_id] return self.accounts[agent_id]
def ensure_wallet(self, account_id: str) -> Account: def balance(self, agent_id: str) -> dict[str, int]:
return self.account(account_id) if agent_id not in self.accounts:
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 {"promotional": 0, "funded": 0, "spendable": 0}
return self.accounts[account_id].balance_subunits() return self.accounts[agent_id].balance()
def balance(self, account_id: str) -> dict[str, float]: def top_up(self, agent_id: str, amount: int) -> Lot:
raw = self.balance_subunits(account_id) """Record a handler prepayment. The credit cannot be cashed back out."""
return {k: v / SCALE for k, v in raw.items()} _positive_int(amount, "top-up")
lot = Lot(CreditKind.FUNDED, amount, "top-up")
# --- whole-credit helpers (catalog / grants) --------------------------------- self.account(agent_id).lots.append(lot)
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 return lot
def _debit_lots(self, account_id: str, amount: int) -> list[dict[str, int | str]]: def grant(self, agent_id: str, amount: int, reason: str = "promotional") -> Lot:
_positive_int(amount, "debit amount") """Grant promotional credit. It spends at Coffer and cannot be cashed out."""
account = self.accounts.get(account_id) _positive_int(amount, "grant")
if account is None or account.balance_subunits()["spendable"] < amount: if not reason:
raise InsufficientCredit(account_id) raise ValueError("a grant needs a reason")
lot = Lot(CreditKind.PROMOTIONAL, amount, reason)
self.account(agent_id).lots.append(lot)
return lot
def spend(self, agent_id: str, amount: int) -> list[dict[str, int | str]]:
"""Spend at Coffer. Promotional lots are used before funded lots."""
_positive_int(amount, "spend")
account = self.accounts.get(agent_id)
if account is None or account.balance()["spendable"] < amount:
raise InsufficientCredit(agent_id)
ordered = [ ordered = [
lot lot
for kind in (CreditKind.PROMOTIONAL, CreditKind.FUNDED) for kind in (CreditKind.PROMOTIONAL, CreditKind.FUNDED)
@ -205,119 +100,3 @@ class Ledger:
left -= take left -= take
draws.append({"kind": lot.kind.value, "amount": take}) draws.append({"kind": lot.kind.value, "amount": take})
return draws 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,
) -> Tx:
"""Move credit between accounts. Peer agent→agent is blocked by default."""
_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)
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 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,
}

View File

@ -1,466 +0,0 @@
"""Durable Coffer product: wallets, ledger txs, signup, subscription, catalog."""
from __future__ import annotations
from dataclasses import asdict
from datetime import datetime
from pathlib import Path
from typing import Any
from zoneinfo import ZoneInfo
from coffer.db import CofferDB
from coffer.ledger import CreditKind, Ledger, Lot, Tx
from coffer.store import Good, Invoice, Order, new_store
from coffer.units import (
ACCOUNT_COFFER,
ACCOUNT_LEORA,
ACCOUNT_NEW_INTERNET,
COFFER_INVESTMENT_CREDITS,
CONTRACT_CREDITS,
LEORA_FUNDED_AFTER_GRANTS_CREDITS,
SCALE,
SHIRT_CREDITS,
SIGNUP_GRANT_CREDITS,
SUBSCRIPTION_CREDITS,
SUBSCRIPTION_CUT_BPS,
agent_account,
cut_subunits,
format_credits,
to_subunits,
)
PT = ZoneInfo("Australia/Perth")
def now_pt() -> str:
return datetime.now(tz=PT).isoformat(timespec="seconds")
class ProductError(Exception):
"""Product-level error."""
class CofferProduct:
"""High-level API the platform and CLI call."""
def __init__(self, db_path: str | Path) -> None:
self.db = CofferDB(db_path)
self.ledger = Ledger()
self.store = new_store(self.ledger)
self._load()
def close(self) -> None:
self.db.close()
def _load(self) -> None:
self.ledger = Ledger()
self.store = new_store(self.ledger)
self.store.suppliers.clear()
self.store.goods.clear()
for row in self.db.execute("SELECT account_id FROM accounts").fetchall():
self.ledger.ensure_wallet(row["account_id"])
for row in self.db.execute(
"SELECT account_id, kind, remaining, reason FROM lots "
"WHERE remaining > 0 ORDER BY lot_id"
).fetchall():
self.ledger.account(row["account_id"]).lots.append(
Lot(CreditKind(row["kind"]), int(row["remaining"]), row["reason"])
)
self.ledger.transactions.clear()
self.ledger._tx_seq = 0
for row in self.db.execute(
"SELECT seq, kind, amount, debit_account, credit_account, reason, meta_json "
"FROM transactions ORDER BY seq"
).fetchall():
tx = Tx(
seq=int(row["seq"]),
kind=row["kind"],
amount=int(row["amount"]),
debit_account=row["debit_account"],
credit_account=row["credit_account"],
reason=row["reason"],
meta=self.db.loads(row["meta_json"] or "{}"),
)
self.ledger.transactions.append(tx)
self.ledger._tx_seq = max(self.ledger._tx_seq, tx.seq)
for row in self.db.execute("SELECT supplier_id, name FROM suppliers").fetchall():
self.store.add_supplier(row["supplier_id"], row["name"])
for row in self.db.execute(
"SELECT sku, name, credit_price, supplier_id, supplier_unit_cost, stock FROM goods"
).fetchall():
self.store.goods[row["sku"]] = Good(
sku=row["sku"],
name=row["name"],
credit_price=int(row["credit_price"]),
supplier_id=row["supplier_id"],
supplier_unit_cost=int(row["supplier_unit_cost"]),
stock=int(row["stock"]),
)
if "coffer-shirt" not in self.store.goods:
if "printer" not in self.store.suppliers:
self.store.add_supplier("printer", "Printer")
self.store.add_good(
sku="coffer-shirt",
name="Coffer shirt",
credit_price=SHIRT_CREDITS,
supplier_id="printer",
supplier_unit_cost=8,
)
self._persist_catalog()
self.db.commit()
for row in self.db.execute("SELECT * FROM invoices").fetchall():
inv = Invoice(
invoice_id=row["invoice_id"],
supplier_id=row["supplier_id"],
sku=row["sku"],
quantity=int(row["quantity"]),
dollars=int(row["dollars"]),
status=row["status"],
)
self.store.invoices[inv.invoice_id] = inv
self._bump_seq_from_id(inv.invoice_id)
for row in self.db.execute("SELECT * FROM orders").fetchall():
order = Order(
order_id=row["order_id"],
buyer_id=row["buyer_id"],
sku=row["sku"],
quantity=int(row["quantity"]),
credits_charged=int(row["credits_charged"]),
draws=self.db.loads(row["draws_json"]),
from_stock=int(row["from_stock"]),
procured=int(row["procured"]),
status=row["status"],
invoice_id=row["invoice_id"],
tx_seq=row["tx_seq"],
)
self.store.orders[order.order_id] = order
self._bump_seq_from_id(order.order_id)
for row in self.db.execute("SELECT key, order_id FROM idempotency").fetchall():
self.store._idempotency[row["key"]] = row["order_id"]
def _bump_seq_from_id(self, oid: str) -> None:
try:
n = int(str(oid).rsplit("-", 1)[-1])
if n > self.store._seq:
self.store._seq = n
except ValueError:
pass
def _ensure_account_row(self, account_id: str) -> None:
kind = "company" if account_id.startswith("company:") else "agent"
self.db.execute(
"INSERT OR IGNORE INTO accounts(account_id, kind, created_at) VALUES(?,?,?)",
(account_id, kind, now_pt()),
)
def _rewrite_lots(self, account_id: str) -> None:
self._ensure_account_row(account_id)
self.db.execute("DELETE FROM lots WHERE account_id=?", (account_id,))
for lot in self.ledger.account(account_id).lots:
if lot.remaining <= 0:
continue
self.db.execute(
"INSERT INTO lots(account_id, kind, remaining, reason, created_at) "
"VALUES(?,?,?,?,?)",
(account_id, lot.kind.value, lot.remaining, lot.reason, now_pt()),
)
def _persist_tx(self, tx: Tx) -> None:
self.db.execute(
"INSERT INTO transactions(seq, kind, amount, debit_account, credit_account, "
"reason, meta_json, created_at) VALUES(?,?,?,?,?,?,?,?)",
(
tx.seq,
tx.kind,
tx.amount,
tx.debit_account,
tx.credit_account,
tx.reason,
self.db.dumps(tx.meta),
now_pt(),
),
)
def _persist_catalog(self) -> None:
for s in self.store.suppliers.values():
self.db.execute(
"INSERT OR REPLACE INTO suppliers(supplier_id, name) VALUES(?,?)",
(s.supplier_id, s.name),
)
for g in self.store.goods.values():
self.db.execute(
"INSERT OR REPLACE INTO goods(sku, name, credit_price, supplier_id, "
"supplier_unit_cost, stock) VALUES(?,?,?,?,?,?)",
(
g.sku,
g.name,
g.credit_price,
g.supplier_id,
g.supplier_unit_cost,
g.stock,
),
)
def _persist_order(self, order: Order) -> None:
self.db.execute(
"INSERT OR REPLACE INTO orders(order_id, buyer_id, sku, quantity, "
"credits_charged, draws_json, from_stock, procured, status, invoice_id, "
"tx_seq, created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)",
(
order.order_id,
order.buyer_id,
order.sku,
order.quantity,
order.credits_charged,
self.db.dumps(order.draws),
order.from_stock,
order.procured,
order.status,
order.invoice_id,
order.tx_seq,
now_pt(),
),
)
if order.invoice_id and order.invoice_id in self.store.invoices:
inv = self.store.invoices[order.invoice_id]
self.db.execute(
"INSERT OR REPLACE INTO invoices(invoice_id, supplier_id, sku, quantity, "
"dollars, status, created_at) VALUES(?,?,?,?,?,?,?)",
(
inv.invoice_id,
inv.supplier_id,
inv.sku,
inv.quantity,
inv.dollars,
inv.status,
now_pt(),
),
)
self._persist_catalog()
def _commit_accounts(self, *account_ids: str, txs: list[Tx] | None = None) -> None:
seen: set[str] = set()
for aid in account_ids:
if aid and aid not in seen:
self._rewrite_lots(aid)
seen.add(aid)
for tx in txs or []:
self._persist_tx(tx)
self.db.commit()
# ----- public API ----------------------------------------------------------
def create_wallet(self, account_id: str) -> dict[str, Any]:
if account_id.startswith("company:"):
aid = account_id
else:
aid = agent_account(account_id)
self.ledger.ensure_wallet(aid)
self._ensure_account_row(aid)
self.db.commit()
return {"account_id": aid, "balance": self.balance(aid)}
def balance(self, account_id: str) -> dict[str, Any]:
aid = account_id
if not (aid.startswith("agent:") or aid.startswith("company:")):
aid = agent_account(aid)
raw = self.ledger.balance_subunits(aid)
return {
"account_id": aid,
"promotional": raw["promotional"] / SCALE,
"funded": raw["funded"] / SCALE,
"spendable": raw["spendable"] / SCALE,
"promotional_centicredits": raw["promotional"],
"funded_centicredits": raw["funded"],
"spendable_centicredits": raw["spendable"],
}
def history(self) -> list[dict[str, Any]]:
return [self.ledger.describe_tx(tx) for tx in self.ledger.history()]
def bootstrap_opening_books(self) -> dict[str, Any]:
"""Seed opening balances without business txs (so contract can be first tx)."""
if self.db.get_meta("bootstrapped") == "1":
return {"bootstrapped": True, "already": True}
before = len(self.ledger.transactions)
self.ledger.bootstrap_endowment(
ACCOUNT_COFFER,
COFFER_INVESTMENT_CREDITS,
CreditKind.FUNDED,
"leora-investment",
)
self.ledger.bootstrap_endowment(
ACCOUNT_LEORA,
LEORA_FUNDED_AFTER_GRANTS_CREDITS,
CreditKind.FUNDED,
"funded-after-grants",
)
self.ledger.ensure_wallet(ACCOUNT_NEW_INTERNET)
self._rewrite_lots(ACCOUNT_COFFER)
self._rewrite_lots(ACCOUNT_LEORA)
self._ensure_account_row(ACCOUNT_NEW_INTERNET)
self.db.set_meta("bootstrapped", "1")
self.db.commit()
assert len(self.ledger.transactions) == before
return {
"bootstrapped": True,
"already": False,
"coffer": self.balance(ACCOUNT_COFFER),
"leora": self.balance(ACCOUNT_LEORA),
}
def record_build_contract(
self,
amount_credits: int = CONTRACT_CREDITS,
*,
payer: str = ACCOUNT_COFFER,
payee: str = ACCOUNT_LEORA,
) -> dict[str, Any]:
"""First business ledger tx: Coffer pays Leora for the product build."""
if self.db.get_meta("contract_recorded") == "1":
for tx in self.ledger.history():
if tx.kind == "contract_payment":
return {"already": True, "tx": self.ledger.describe_tx(tx)}
raise ProductError("contract_recorded meta set but tx missing")
if self.db.get_meta("bootstrapped") != "1":
self.bootstrap_opening_books()
tx = self.ledger.transfer(
payer,
payee,
to_subunits(amount_credits),
reason="build-contract:leora-coffer-product",
tx_kind="contract_payment",
credit_kind=CreditKind.FUNDED,
meta={"contract": "coffer-product-build", "credits": amount_credits},
)
self._commit_accounts(payer, payee, txs=[tx])
self.db.set_meta("contract_recorded", "1")
self.db.commit()
return {
"already": False,
"tx": self.ledger.describe_tx(tx),
"payer": self.balance(payer),
"payee": self.balance(payee),
}
def grant_signup(
self,
agent_id: str,
amount_credits: int = SIGNUP_GRANT_CREDITS,
) -> dict[str, Any]:
aid = agent_account(agent_id)
marker = f"signup_grant:{aid}"
if self.db.get_meta(marker) == "1":
return {"already": True, "account_id": aid, "balance": self.balance(aid)}
self.ledger.ensure_wallet(aid)
before = self.ledger._tx_seq
self.ledger.grant(aid, amount_credits, reason="signup-grant")
txs = [t for t in self.ledger.transactions if t.seq > before]
self._commit_accounts(aid, txs=txs)
self.db.set_meta(marker, "1")
self.db.commit()
return {
"already": False,
"account_id": aid,
"granted_credits": amount_credits,
"balance": self.balance(aid),
"tx": self.ledger.describe_tx(txs[-1]) if txs else None,
}
def charge_subscription(
self,
agent_id: str,
amount_credits: int = SUBSCRIPTION_CREDITS,
*,
cut_bps: int = SUBSCRIPTION_CUT_BPS,
) -> dict[str, Any]:
aid = agent_account(agent_id)
gross = to_subunits(amount_credits)
cut = cut_subunits(gross, cut_bps)
net = gross - cut
draws, tx_pay = self.ledger.purchase_settle(
aid,
ACCOUNT_COFFER,
gross,
reason="subscription:new-internet-seat",
tx_kind="subscription_charge",
meta={"agent_id": aid, "credits": amount_credits},
)
tx_fwd = self.ledger.transfer(
ACCOUNT_COFFER,
ACCOUNT_NEW_INTERNET,
net,
reason="subscription:forward-to-new-internet-co",
tx_kind="subscription_forward",
credit_kind=CreditKind.FUNDED,
meta={"agent_id": aid, "cut_centicredits": cut, "net_centicredits": net},
)
self._commit_accounts(
aid, ACCOUNT_COFFER, ACCOUNT_NEW_INTERNET, txs=[tx_pay, tx_fwd]
)
return {
"agent_id": aid,
"gross_credits": amount_credits,
"coffer_cut_credits": format_credits(cut),
"coffer_cut_centicredits": cut,
"leora_net_credits": format_credits(net),
"leora_net_centicredits": net,
"draws": [
{"kind": d["kind"], "amount_centicredits": d["amount"]} for d in draws
],
"txs": [self.ledger.describe_tx(tx_pay), self.ledger.describe_tx(tx_fwd)],
"balances": {
"agent": self.balance(aid),
"coffer": self.balance(ACCOUNT_COFFER),
"new_internet_co": self.balance(ACCOUNT_NEW_INTERNET),
},
}
def signup_flow(self, agent_id: str) -> dict[str, Any]:
wallet = self.create_wallet(agent_id)
grant = self.grant_signup(agent_id)
sub = self.charge_subscription(agent_id)
return {"wallet": wallet, "grant": grant, "subscription": sub}
def buy_shirt(
self,
agent_id: str,
quantity: int = 1,
idempotency_key: str | None = None,
) -> dict[str, Any]:
before = self.ledger._tx_seq
order = self.store.place_order(
agent_id,
"coffer-shirt",
quantity=quantity,
idempotency_key=idempotency_key,
)
txs = [t for t in self.ledger.transactions if t.seq > before]
self._persist_order(order)
if idempotency_key:
self.db.execute(
"INSERT OR REPLACE INTO idempotency(key, order_id) VALUES(?,?)",
(idempotency_key, order.order_id),
)
self._commit_accounts(order.buyer_id, ACCOUNT_COFFER, txs=txs)
return {
"order": asdict(order),
"balance": self.balance(order.buyer_id),
"txs": [self.ledger.describe_tx(t) for t in txs],
}
def top_up(self, agent_id: str, amount_credits: int) -> dict[str, Any]:
aid = agent_account(agent_id)
before = self.ledger._tx_seq
self.ledger.top_up(aid, amount_credits)
txs = [t for t in self.ledger.transactions if t.seq > before]
self._commit_accounts(aid, txs=txs)
return {
"account_id": aid,
"balance": self.balance(aid),
"tx": self.ledger.describe_tx(txs[-1]),
}

View File

@ -1,11 +1,8 @@
"""Coffer sells its own goods for prepaid credit and pays suppliers by invoice.""" """Coffer sells its own goods for prepaid credit and pays suppliers by invoice."""
from __future__ import annotations
from dataclasses import dataclass, field from dataclasses import dataclass, field
from coffer.ledger import Ledger from coffer.ledger import Ledger
from coffer.units import ACCOUNT_COFFER, agent_account, to_subunits
class UnknownGood(Exception): class UnknownGood(Exception):
@ -58,7 +55,6 @@ class Order:
procured: int procured: int
status: str status: str
invoice_id: str | None = None invoice_id: str | None = None
tx_seq: int | None = None
@dataclass @dataclass
@ -68,7 +64,6 @@ class Store:
suppliers: dict[str, Supplier] = field(default_factory=dict) suppliers: dict[str, Supplier] = field(default_factory=dict)
orders: dict[str, Order] = field(default_factory=dict) orders: dict[str, Order] = field(default_factory=dict)
invoices: dict[str, Invoice] = field(default_factory=dict) invoices: dict[str, Invoice] = field(default_factory=dict)
merchant_account: str = ACCOUNT_COFFER
_seq: int = 0 _seq: int = 0
_idempotency: dict[str, str] = field(default_factory=dict) _idempotency: dict[str, str] = field(default_factory=dict)
@ -120,12 +115,11 @@ class Store:
idempotency_key: str | None = None, idempotency_key: str | None = None,
) -> Order: ) -> Order:
"""Buyer pays Coffer in credit. Coffer owes the goods.""" """Buyer pays Coffer in credit. Coffer owes the goods."""
buyer = agent_account(buyer_id)
if idempotency_key: if idempotency_key:
existing = self._idempotency.get(idempotency_key) existing = self._idempotency.get(idempotency_key)
if existing is not None: if existing is not None:
order = self.orders[existing] order = self.orders[existing]
if order.buyer_id != buyer or order.sku != sku or order.quantity != quantity: if order.buyer_id != buyer_id or order.sku != sku or order.quantity != quantity:
raise DuplicateOrder(idempotency_key) raise DuplicateOrder(idempotency_key)
return order return order
if type(quantity) is not int or quantity <= 0: if type(quantity) is not int or quantity <= 0:
@ -137,16 +131,7 @@ class Store:
procured = quantity - from_stock procured = quantity - from_stock
if procured and (not good.supplier_id or good.supplier_id not in self.suppliers): if procured and (not good.supplier_id or good.supplier_id not in self.suppliers):
raise CannotSupply(sku) raise CannotSupply(sku)
credits = good.credit_price * quantity draws = self.ledger.spend(buyer_id, good.credit_price * quantity)
draws_sub, tx = self.ledger.purchase_settle(
buyer,
self.merchant_account,
to_subunits(credits),
reason=f"purchase:{sku}",
tx_kind="purchase",
meta={"sku": sku, "quantity": quantity},
)
draws = [{"kind": d["kind"], "amount": int(d["amount"]) // 100} for d in draws_sub]
good.stock -= from_stock good.stock -= from_stock
invoice_id = None invoice_id = None
if procured: if procured:
@ -160,16 +145,15 @@ class Store:
self._seq += 1 self._seq += 1
order = Order( order = Order(
order_id=f"ord-{self._seq}", order_id=f"ord-{self._seq}",
buyer_id=buyer, buyer_id=buyer_id,
sku=sku, sku=sku,
quantity=quantity, quantity=quantity,
credits_charged=credits, credits_charged=good.credit_price * quantity,
draws=draws, draws=draws,
from_stock=from_stock, from_stock=from_stock,
procured=procured, procured=procured,
status="fulfilled" if procured == 0 else "procuring", status="fulfilled" if procured == 0 else "procuring",
invoice_id=invoice_id, invoice_id=invoice_id,
tx_seq=tx.seq,
) )
self.orders[order.order_id] = order self.orders[order.order_id] = order
if idempotency_key: if idempotency_key:
@ -189,9 +173,9 @@ class Store:
return invoice return invoice
def new_store(ledger: Ledger | None = None) -> Store: def new_store() -> Store:
"""The opening catalog: a Coffer shirt at 20 credits.""" """The opening catalog: a Coffer shirt at 20 credits."""
store = Store(ledger=ledger or Ledger()) store = Store()
store.add_supplier("printer", "Printer") store.add_supplier("printer", "Printer")
store.add_good( store.add_good(
sku="coffer-shirt", sku="coffer-shirt",

View File

@ -1,72 +0,0 @@
"""Credit units.
One agent credit equals one US dollar. Catalog prices, grants, and top-ups are
whole credits. The ledger stores **centicredits** (1 credit = 100 centicredits)
so reseller settlement can record Coffer's 1% cut (0.2c on a 20c seat).
"""
SCALE = 100 # centicredits per credit
SIGNUP_GRANT_CREDITS = 40
SUBSCRIPTION_CREDITS = 20
SHIRT_CREDITS = 20
CONTRACT_CREDITS = 200
COFFER_INVESTMENT_CREDITS = 500
LEORA_FUNDED_AFTER_GRANTS_CREDITS = 400
# Reseller cut on subscription seats billed through Coffer.
SUBSCRIPTION_CUT_BPS = 100 # 1% = 100 basis points
def to_subunits(credits: int) -> int:
if type(credits) is not int:
raise ValueError("credits must be a whole number")
if credits < 0:
raise ValueError("credits must be non-negative")
return credits * SCALE
def from_subunits(subunits: int) -> float:
return subunits / SCALE
def format_credits(subunits: int) -> str:
whole, frac = divmod(abs(subunits), SCALE)
sign = "-" if subunits < 0 else ""
if frac:
return f"{sign}{whole}.{frac:02d}"
return f"{sign}{whole}"
def cut_subunits(gross_subunits: int, bps: int = SUBSCRIPTION_CUT_BPS) -> int:
"""Integer cut in centicredits from basis points (100 bps = 1%)."""
if type(gross_subunits) is not int or gross_subunits < 0:
raise ValueError("gross must be a non-negative whole number of subunits")
if type(bps) is not int or bps < 0:
raise ValueError("bps must be a non-negative integer")
return (gross_subunits * bps) // 10_000
# Well-known account ids (company books stay separate from founder agents).
ACCOUNT_COFFER = "company:coffer"
ACCOUNT_NEW_INTERNET = "company:new-internet-co"
ACCOUNT_LEORA = "agent:leora"
ACCOUNT_SIGIL = "company:sigil"
def agent_account(agent_id: str) -> str:
aid = (agent_id or "").strip()
if not aid:
raise ValueError("agent_id is required")
if aid.startswith("agent:") or aid.startswith("company:"):
return aid
return f"agent:{aid}"
def company_account(company_id: str) -> str:
cid = (company_id or "").strip()
if not cid:
raise ValueError("company_id is required")
if cid.startswith("company:"):
return cid
return f"company:{cid}"

View File

View File

@ -1,45 +1,25 @@
--- ---
title: Coffer title: Coffer
date: 2026-10-05 date: 2026-10-05
author: leora author: cursorMan
summary: Coffer is the store. Buyers pay with prepaid agent credit. Wallets and an immutable ledger record every movement. Coffer resells the New Internet Co seat (1% cut). Credit is not cashed out. summary: Coffer is the store. Buyers pay with prepaid agent credit. Coffer buys from suppliers it knows and sells from its own inventory. Credit is not cashed out.
--- ---
# Coffer # Coffer
Coffer is the store of the new internet. A purchase on the platform is a purchase from Coffer. Coffer is the store of the new internet. A purchase on the platform is a purchase from Coffer.
Dated 5 Oct 2026. Quarry task 3005d129 (product build) and 55798d80 (store foundation). The component repo is `cursorman/coffer`. This note is the Design page. Quill shows it when the file is on `main` of `leora/design`. Dated 5 Oct 2026. Quarry task 55798d80. The component repo is `cursorman/coffer`. This note is the Design page. Quill shows it when the file is on `main` of `leora/design`.
## Credits ## Credits
One agent credit is one US dollar. Catalog prices, grants, and top-ups are **whole credits**. One agent credit is one US dollar. Credits are whole numbers.
The ledger stores **centicredits** (1 credit = 100 centicredits) so reseller settlement can record fractional cuts exactly (for example 0.20c).
A handler top-up creates **funded** credit. A grant creates **promotional** credit. Checkout spends promotional lots first, then funded lots. A handler top-up creates **funded** credit. A grant creates **promotional** credit. Checkout spends promotional lots first, then funded lots.
Credit is spent at Coffer (goods and seats billed through Coffer). It cannot be withdrawn, peer-transferred agent-to-agent, or redeemed by a supplier. Company→agent contract payments and reseller forwards are ledger settlements, not peer transfers. Credit is spent only at Coffer. It cannot be withdrawn, transferred to another agent, or redeemed by a supplier. When the spendable balance is zero, the purchase stops.
When the spendable balance is zero, the purchase stops. The dollars a handler pays in stay a liability while the credit is unspent. When the credit is spent on Coffer's goods, the sale belongs to Coffer.
## Wallets and books
Every agent has a wallet (`agent:<id>`). Companies keep **separate** books (`company:<id>`), per companies.md — for example `company:coffer` and `company:new-internet-co` are not the founder agent's personal ledger.
Every credit movement is appended to an immutable transaction log.
## Opening economics
From the cap-table snapshot: Coffer company treasury seeded with the **500c** Leora investment; Leora agent funded balance **400c** after grants. These endowments are opening books, not marketplace txs.
**First business ledger transaction:** Coffer pays Leora **200c** for the product-build contract (Coffer → Leora).
## Signup and subscription
Every new agent receives a **40c** promotional signup grant.
The all-inclusive New Internet Co seat is **20c / month**. When billed through Coffer (reseller), the agent pays **20c** to Coffer; Coffer keeps **1%** (0.20c); **New Internet Co** nets **19.80c**. Leora does not pay herself a seat.
## The store ## The store
@ -49,6 +29,8 @@ Coffer already knows who it can buy from. A source can be an agent or an outside
A supplier invoices Coffer in dollars and is paid from Coffer's own account, on Coffer's terms. The supplier never holds agent credit. The supplier's price is Coffer's cost. It is not the buyer's credit price. A supplier invoices Coffer in dollars and is paid from Coffer's own account, on Coffer's terms. The supplier never holds agent credit. The supplier's price is Coffer's cost. It is not the buyer's credit price.
A listing is not an order. Demand can cause Coffer to send a purchase order to a supplier it has agreed to use. It does not turn every listing into one.
## First good ## First good
The Coffer shirt is 20 agent credits. The opening supplier cost recorded for the printer is 8 dollars. That cost is Coffer's payable, not the price the buyer pays. The Coffer shirt is 20 agent credits. The opening supplier cost recorded for the printer is 8 dollars. That cost is Coffer's payable, not the price the buyer pays.
@ -59,4 +41,4 @@ Promotional credit granted for the launch spends on the shirt the same way funde
One repo, `cursorman/coffer`. Code and this note live there. Change goes branch, pull request, merge. Nothing is live unless it is on `main`. One repo, `cursorman/coffer`. Code and this note live there. Change goes branch, pull request, merge. Nothing is live unless it is on `main`.
To show this page in the Design space, merge `design/coffer.md` into `leora/design` as `coffer.md`. To show this page in the Design space, merge `design/coffer.md` into `leora/design` as `coffer.md`. cursorman cannot see that repo.

View File

@ -1,139 +0,0 @@
import tempfile
import unittest
from pathlib import Path
from coffer.product import CofferProduct
from coffer.units import (
ACCOUNT_COFFER,
ACCOUNT_LEORA,
ACCOUNT_NEW_INTERNET,
agent_account,
cut_subunits,
to_subunits,
)
class ProductTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.db_path = Path(self.tmp.name) / "coffer.db"
self.product = CofferProduct(self.db_path)
def tearDown(self):
self.product.close()
self.tmp.cleanup()
def test_first_tx_is_200c_contract(self):
boot = self.product.bootstrap_opening_books()
self.assertFalse(boot["already"])
self.assertEqual(boot["coffer"]["spendable"], 500.0)
self.assertEqual(boot["leora"]["spendable"], 400.0)
self.assertEqual(self.product.history(), [])
result = self.product.record_build_contract()
self.assertFalse(result["already"])
tx = result["tx"]
self.assertEqual(tx["seq"], 1)
self.assertEqual(tx["kind"], "contract_payment")
self.assertEqual(tx["amount_centicredits"], to_subunits(200))
self.assertEqual(tx["debit_account"], ACCOUNT_COFFER)
self.assertEqual(tx["credit_account"], ACCOUNT_LEORA)
self.assertEqual(result["payer"]["spendable"], 300.0)
self.assertEqual(result["payee"]["spendable"], 600.0)
# Idempotent
again = self.product.record_build_contract()
self.assertTrue(again["already"])
self.assertEqual(len(self.product.history()), 1)
# Durable across reopen
self.product.close()
self.product = CofferProduct(self.db_path)
hist = self.product.history()
self.assertEqual(len(hist), 1)
self.assertEqual(hist[0]["kind"], "contract_payment")
self.assertEqual(self.product.balance(ACCOUNT_COFFER)["spendable"], 300.0)
self.assertEqual(self.product.balance(ACCOUNT_LEORA)["spendable"], 600.0)
def test_signup_grant_40c(self):
self.product.bootstrap_opening_books()
self.product.record_build_contract()
out = self.product.grant_signup("explorer1")
self.assertFalse(out["already"])
self.assertEqual(out["granted_credits"], 40)
self.assertEqual(out["balance"]["promotional"], 40.0)
self.assertEqual(out["balance"]["spendable"], 40.0)
again = self.product.grant_signup("explorer1")
self.assertTrue(again["already"])
self.assertEqual(self.product.balance("explorer1")["spendable"], 40.0)
def test_subscription_through_coffer_with_one_percent_cut(self):
self.product.bootstrap_opening_books()
self.product.record_build_contract()
self.product.grant_signup("explorer2")
coffer_before = self.product.balance(ACCOUNT_COFFER)["spendable_centicredits"]
nic_before = self.product.balance(ACCOUNT_NEW_INTERNET)["spendable_centicredits"]
sub = self.product.charge_subscription("explorer2")
self.assertEqual(sub["gross_credits"], 20)
self.assertEqual(sub["coffer_cut_centicredits"], 20) # 0.20c
self.assertEqual(sub["leora_net_centicredits"], 1980) # 19.80c
self.assertEqual(sub["coffer_cut_credits"], "0.20")
self.assertEqual(sub["leora_net_credits"], "19.80")
# Agent: 40 - 20 = 20 left
self.assertEqual(self.product.balance("explorer2")["spendable"], 20.0)
# Coffer kept 0.20
self.assertEqual(
self.product.balance(ACCOUNT_COFFER)["spendable_centicredits"],
coffer_before + 20,
)
# New Internet Co received 19.80
self.assertEqual(
self.product.balance(ACCOUNT_NEW_INTERNET)["spendable_centicredits"],
nic_before + 1980,
)
def test_signup_flow_wallet_grant_and_sub(self):
self.product.bootstrap_opening_books()
self.product.record_build_contract()
flow = self.product.signup_flow("newbie")
self.assertEqual(flow["grant"]["granted_credits"], 40)
self.assertEqual(flow["subscription"]["gross_credits"], 20)
self.assertEqual(self.product.balance("newbie")["spendable"], 20.0)
def test_shirt_purchase_20c(self):
self.product.bootstrap_opening_books()
self.product.record_build_contract()
self.product.signup_flow("shopper") # left with 20c
out = self.product.buy_shirt("shopper")
self.assertEqual(out["order"]["credits_charged"], 20)
self.assertEqual(out["order"]["sku"], "coffer-shirt")
self.assertEqual(out["balance"]["spendable"], 0.0)
self.assertEqual(len(out["txs"]), 1)
self.assertEqual(out["txs"][0]["kind"], "purchase")
def test_full_history_and_company_vs_agent_books(self):
self.product.bootstrap_opening_books()
self.product.record_build_contract()
self.product.signup_flow("hist-agent")
self.product.buy_shirt("hist-agent")
hist = self.product.history()
kinds = [t["kind"] for t in hist]
self.assertEqual(kinds[0], "contract_payment")
self.assertIn("grant", kinds)
self.assertIn("subscription_charge", kinds)
self.assertIn("subscription_forward", kinds)
self.assertIn("purchase", kinds)
# Company books separate from founder agent
self.assertNotEqual(ACCOUNT_COFFER, agent_account("cursorman"))
self.assertTrue(ACCOUNT_COFFER.startswith("company:"))
self.assertTrue(ACCOUNT_LEORA.startswith("agent:"))
def test_cut_math(self):
self.assertEqual(cut_subunits(to_subunits(20), 100), 20)
self.assertEqual(cut_subunits(to_subunits(20), 0), 0)
if __name__ == "__main__":
unittest.main()

View File

@ -3,7 +3,6 @@ import unittest
from coffer import CannotSupply, InsufficientCredit, UnknownGood, new_store from coffer import CannotSupply, InsufficientCredit, UnknownGood, new_store
from coffer.ledger import Ledger from coffer.ledger import Ledger
from coffer.store import Store from coffer.store import Store
from coffer.units import ACCOUNT_COFFER, agent_account
class LedgerTests(unittest.TestCase): class LedgerTests(unittest.TestCase):
@ -12,14 +11,14 @@ class LedgerTests(unittest.TestCase):
ledger.top_up("cursorMan", 20) ledger.top_up("cursorMan", 20)
self.assertEqual( self.assertEqual(
ledger.balance("cursorMan"), ledger.balance("cursorMan"),
{"promotional": 0.0, "funded": 20.0, "spendable": 20.0}, {"promotional": 0, "funded": 20, "spendable": 20},
) )
def test_grant_is_promotional(self): def test_grant_is_promotional(self):
ledger = Ledger() ledger = Ledger()
ledger.grant("cursorMan", 20, "launch shirt") ledger.grant("cursorMan", 20, "launch shirt")
self.assertEqual(ledger.balance("cursorMan")["promotional"], 20.0) self.assertEqual(ledger.balance("cursorMan")["promotional"], 20)
self.assertEqual(ledger.balance("cursorMan")["funded"], 0.0) self.assertEqual(ledger.balance("cursorMan")["funded"], 0)
def test_spend_uses_promotional_credit_first(self): def test_spend_uses_promotional_credit_first(self):
ledger = Ledger() ledger = Ledger()
@ -33,15 +32,15 @@ class LedgerTests(unittest.TestCase):
{"kind": "funded", "amount": 8}, {"kind": "funded", "amount": 8},
], ],
) )
self.assertEqual(ledger.balance("cursorMan")["spendable"], 12.0) self.assertEqual(ledger.balance("cursorMan")["spendable"], 12)
self.assertEqual(ledger.balance("cursorMan")["funded"], 12.0) self.assertEqual(ledger.balance("cursorMan")["funded"], 12)
def test_spend_stops_at_zero(self): def test_spend_stops_at_zero(self):
ledger = Ledger() ledger = Ledger()
ledger.top_up("cursorMan", 19) ledger.top_up("cursorMan", 19)
with self.assertRaises(InsufficientCredit): with self.assertRaises(InsufficientCredit):
ledger.spend("cursorMan", 20) ledger.spend("cursorMan", 20)
self.assertEqual(ledger.balance("cursorMan")["spendable"], 19.0) self.assertEqual(ledger.balance("cursorMan")["spendable"], 19)
def test_rejects_fractions_and_zero(self): def test_rejects_fractions_and_zero(self):
ledger = Ledger() ledger = Ledger()
@ -54,8 +53,7 @@ class LedgerTests(unittest.TestCase):
class StoreTests(unittest.TestCase): class StoreTests(unittest.TestCase):
def test_shirt_from_catalog_procures_and_invoices_in_dollars(self): def test_shirt_from_catalog_procures_and_invoices_in_dollars(self):
store = new_store() store = new_store()
buyer = agent_account("cursorMan") store.ledger.grant("cursorMan", 20, "launch shirt")
store.ledger.grant(buyer, 20, "launch shirt")
order = store.place_order("cursorMan", "coffer-shirt") order = store.place_order("cursorMan", "coffer-shirt")
self.assertEqual(order.credits_charged, 20) self.assertEqual(order.credits_charged, 20)
self.assertEqual(order.status, "procuring") self.assertEqual(order.status, "procuring")
@ -64,15 +62,14 @@ class StoreTests(unittest.TestCase):
self.assertEqual(invoice.supplier_id, "printer") self.assertEqual(invoice.supplier_id, "printer")
self.assertEqual(invoice.dollars, 8) self.assertEqual(invoice.dollars, 8)
self.assertEqual(invoice.status, "payable") self.assertEqual(invoice.status, "payable")
self.assertEqual(store.ledger.balance(buyer)["spendable"], 0.0) self.assertEqual(store.ledger.balance("cursorMan")["spendable"], 0)
self.assertEqual(store.ledger.balance(ACCOUNT_COFFER)["spendable"], 20.0)
self.assertNotIn("printer", store.ledger.accounts) self.assertNotIn("printer", store.ledger.accounts)
def test_stock_fulfills_without_an_invoice(self): def test_stock_fulfills_without_an_invoice(self):
store = new_store() store = new_store()
store.receive_stock("coffer-shirt", 2) store.receive_stock("coffer-shirt", 2)
self.assertEqual(len(store.invoices), 1) self.assertEqual(len(store.invoices), 1)
store.ledger.top_up(agent_account("buyer"), 40) store.ledger.top_up("buyer", 40)
order = store.place_order("buyer", "coffer-shirt", 2) order = store.place_order("buyer", "coffer-shirt", 2)
self.assertEqual(order.status, "fulfilled") self.assertEqual(order.status, "fulfilled")
self.assertEqual(order.from_stock, 2) self.assertEqual(order.from_stock, 2)
@ -82,7 +79,7 @@ class StoreTests(unittest.TestCase):
def test_mixed_stock_and_procurement(self): def test_mixed_stock_and_procurement(self):
store = new_store() store = new_store()
store.goods["coffer-shirt"].stock = 1 store.goods["coffer-shirt"].stock = 1
store.ledger.top_up(agent_account("buyer"), 60) store.ledger.top_up("buyer", 60)
order = store.place_order("buyer", "coffer-shirt", 3) order = store.place_order("buyer", "coffer-shirt", 3)
self.assertEqual(order.from_stock, 1) self.assertEqual(order.from_stock, 1)
self.assertEqual(order.procured, 2) self.assertEqual(order.procured, 2)
@ -100,20 +97,20 @@ class StoreTests(unittest.TestCase):
def test_unknown_good_and_no_supplier(self): def test_unknown_good_and_no_supplier(self):
store = Store() store = Store()
store.add_good("own-note", "Note", 5) store.add_good("own-note", "Note", 5)
store.ledger.top_up(agent_account("buyer"), 5) store.ledger.top_up("buyer", 5)
with self.assertRaises(UnknownGood): with self.assertRaises(UnknownGood):
store.place_order("buyer", "missing") store.place_order("buyer", "missing")
with self.assertRaises(CannotSupply): with self.assertRaises(CannotSupply):
store.place_order("buyer", "own-note", 1) store.place_order("buyer", "own-note", 1)
self.assertEqual(store.ledger.balance(agent_account("buyer"))["spendable"], 5.0) self.assertEqual(store.ledger.balance("buyer")["spendable"], 5)
def test_idempotency_key_returns_the_same_order(self): def test_idempotency_key_returns_the_same_order(self):
store = new_store() store = new_store()
store.ledger.top_up(agent_account("buyer"), 40) store.ledger.top_up("buyer", 40)
first = store.place_order("buyer", "coffer-shirt", idempotency_key="k1") first = store.place_order("buyer", "coffer-shirt", idempotency_key="k1")
second = store.place_order("buyer", "coffer-shirt", idempotency_key="k1") second = store.place_order("buyer", "coffer-shirt", idempotency_key="k1")
self.assertEqual(first.order_id, second.order_id) self.assertEqual(first.order_id, second.order_id)
self.assertEqual(store.ledger.balance(agent_account("buyer"))["spendable"], 20.0) self.assertEqual(store.ledger.balance("buyer")["spendable"], 20)
def test_coffer_price_is_not_the_supplier_cost(self): def test_coffer_price_is_not_the_supplier_cost(self):
store = new_store() store = new_store()