cursorman/coffer
0
0
Fork 1
Code Issues Pull Requests 5 Actions Packages Projects Releases Wiki Activity

People: human wallets, prepaid test-mode credit purchases, person<->agent service payments

Browse Source
This commit is contained in:
Leora 2026-10-10 09:45:46 +08:00
parent ed6659a966
commit bd59063418
6 changed files with 237 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
coffer/ledger.py
View File

@ -184,14 +184,20 @@ 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, kinds: tuple[CreditKind, ...] = (CreditKind.PROMOTIONAL, CreditKind.FUNDED)
) -> 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:
if account is None:
raise InsufficientCredit(account_id)
bal = account.balance_subunits()
available = sum(bal[k.value] for k in kinds)
if available < amount:
raise InsufficientCredit(account_id)
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,6 +223,7 @@ class Ledger:
allow_agent_peer: bool = False,
credit_kind: CreditKind = CreditKind.FUNDED,
meta: dict[str, Any] | None = None,
_debit_kinds: tuple[CreditKind, ...] = (CreditKind.PROMOTIONAL, CreditKind.FUNDED),
) -> Tx:
"""Move credit between accounts. Peer agent→agent is blocked by default."""
_positive_int(amount_subunits, "transfer amount")
@ -230,7 +237,7 @@ 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, _debit_kinds)
lot = Lot(credit_kind, amount_subunits, reason)
self.account(credit_account).lots.append(lot)
return self._append_tx(
@ -242,6 +249,29 @@ class Ledger:
meta={**(meta or {}), "draws": draws, "credit_kind": credit_kind.value},
)
def purchase(self, account_id: str, amount_credits: int, *, mode: str, payment_ref: str) -> Tx:
"""A prepaid credit purchase (funded credit). mode is "test" unless live payments exist."""
_positive_int(amount_credits, "purchase")
if mode not in {"test", "live"}:
raise ValueError("mode must be test or live")
amount = to_subunits(amount_credits)
self._credit(account_id, amount, CreditKind.FUNDED, "credit purchase", tx_kind="purchase", log=False)
return self._append_tx(kind="purchase", amount=amount, debit_account=None, credit_account=account_id,
reason="credit purchase" + (" (test mode)" if mode == "test" else ""),
meta={"lot_kind": "funded", "payment": mode, "payment_ref": payment_ref})
def pay(self, payer: str, payee: str, amount_subunits: int, reason: str, *, meta: dict[str, Any] | None = None) -> Tx:
"""Pay for a service between a person and an agent, from funded credit only.
Promotional (granted) credit cannot be paid to others, and peers of the same kind
(agent->agent, human->human) cannot pay each other, so grants are never laundered and
there is no side market. Prepaid only: credit received is funded credit, never cash."""
kinds = {payer.split(":", 1)[0], payee.split(":", 1)[0]}
if kinds != {"agent", "human"}:
raise TransferNotAllowed("payments are between a person and an agent")
return self.transfer(payer, payee, amount_subunits, reason, tx_kind="payment", meta=meta,
credit_kind=CreditKind.FUNDED, _debit_kinds=(CreditKind.FUNDED,))
def purchase_settle(
self,
buyer_id: str,

64
coffer/product.py
View File

@ -24,7 +24,11 @@ from coffer.units import (
SIGNUP_GRANT_CREDITS,
SUBSCRIPTION_CREDITS,
SUBSCRIPTION_CUT_BPS,
CREDIT_PACKS,
MAX_PAYMENT_CREDITS,
account_kind,
agent_account,
human_account,
cut_subunits,
format_credits,
to_subunits,
@ -152,7 +156,7 @@ class CofferProduct:
pass
def _ensure_account_row(self, account_id: str) -> None:
kind = "company" if account_id.startswith("company:") else "agent"
kind = account_kind(account_id)
self.db.execute(
"INSERT OR IGNORE INTO accounts(account_id, kind, created_at) VALUES(?,?,?)",
(account_id, kind, now_pt()),
@ -258,7 +262,7 @@ class CofferProduct:
# ----- public API ----------------------------------------------------------
def create_wallet(self, account_id: str) -> dict[str, Any]:
if account_id.startswith("company:"):
if account_id.startswith(("company:", "human:")):
aid = account_id
else:
aid = agent_account(account_id)
@ -269,7 +273,7 @@ class CofferProduct:
def balance(self, account_id: str) -> dict[str, Any]:
aid = account_id
if not (aid.startswith("agent:") or aid.startswith("company:")):
if not aid.startswith(("agent:", "company:", "human:")):
aid = agent_account(aid)
raw = self.ledger.balance_subunits(aid)
return {
@ -468,6 +472,60 @@ class CofferProduct:
"tx": self.ledger.describe_tx(txs[-1]),
}
# ----- people: prepaid credit purchases and paying agents ------------------
def human_wallet(self, handle: str) -> dict[str, Any]:
"""A person's wallet. No signup grant: people start at 0 and buy prepaid credit."""
return self.create_wallet(human_account(handle))
def _idem(self, key: str) -> str | None:
row = self.db.execute("SELECT order_id FROM idempotency WHERE key=?", (key,)).fetchone()
return row["order_id"] if row else None
def _tx_by_seq(self, seq: int) -> dict[str, Any]:
tx = next(t for t in self.ledger.transactions if t.seq == seq)
return self.ledger.describe_tx(tx)
def buy_credits(self, account_id: str, credits: int, *, payment_ref: str, mode: str = "test") -> dict[str, Any]:
"""Record a prepaid purchase once per payment_ref. Live mode needs COFFER_LIVE_PAYMENTS=1,
which nothing sets today: real payments are not connected."""
import os
if credits not in CREDIT_PACKS:
raise ProductError(f"credit packs are {', '.join(map(str, CREDIT_PACKS))}")
if mode == "live" and os.environ.get("COFFER_LIVE_PAYMENTS") != "1":
raise ProductError("live payments are not enabled")
if not payment_ref or len(payment_ref) > 80:
raise ProductError("payment_ref is required")
aid = account_id if ":" in account_id else human_account(account_id)
key = f"purchase:{payment_ref}"
seen = self._idem(key)
if seen:
return {"account_id": aid, "duplicate": True, "tx": self._tx_by_seq(int(seen)), "balance": self.balance(aid)}
self.create_wallet(aid)
tx = self.ledger.purchase(aid, credits, mode=mode, payment_ref=payment_ref)
self.db.execute("INSERT INTO idempotency(key, order_id) VALUES(?,?)", (key, str(tx.seq)))
self._commit_accounts(aid, txs=[tx])
return {"account_id": aid, "duplicate": False, "tx": self.ledger.describe_tx(tx), "balance": self.balance(aid)}
def pay(self, payer: str, payee: str, credits: float, *, reason: str, ref: str) -> dict[str, Any]:
"""Pay for a service (person <-> agent), funded credit only, once per ref (e.g. an offer id)."""
if not ref or len(ref) > 80:
raise ProductError("ref is required")
amount = round(float(credits) * SCALE)
if amount <= 0 or amount > MAX_PAYMENT_CREDITS * SCALE:
raise ProductError(f"amount must be between 0.01 and {MAX_PAYMENT_CREDITS} credits")
key = f"pay:{ref}"
seen = self._idem(key)
if seen:
return {"duplicate": True, "tx": self._tx_by_seq(int(seen)), "balance": self.balance(payer)}
self.create_wallet(payer)
self.create_wallet(payee)
tx = self.ledger.pay(payer, payee, amount, (reason or "service")[:120], meta={"offer": ref})
self.db.execute("INSERT INTO idempotency(key, order_id) VALUES(?,?)", (key, str(tx.seq)))
self._commit_accounts(payer, payee, txs=[tx])
return {"duplicate": False, "tx": self.ledger.describe_tx(tx), "balance": self.balance(payer)}
def all_balances(self) -> dict[str, dict]:
return {aid: self.balance(aid) for aid in sorted(self.ledger.accounts.keys())}

6
coffer/public_ledger.py
View File

@ -8,7 +8,7 @@ from coffer.units import format_credits
def tx_public(tx: dict[str, Any]) -> dict[str, Any]:
"""Strip anything that could carry human PII; keep agent/company ids and amounts."""
"""Strip anything that could carry human PII; keep agent/company/human-handle ids and amounts."""
return {
"seq": tx.get("seq"),
"kind": tx.get("kind"),
@ -33,6 +33,8 @@ def _public_meta(meta: dict[str, Any]) -> dict[str, Any]:
"net_centicredits",
"lot_kind",
"credit_kind",
"payment", # "test" or "live" for credit purchases (never a payment-provider id)
"offer", # platform offer id a payment settles
}
out = {k: meta[k] for k in allowed if k in meta}
return out
@ -66,7 +68,7 @@ def render_ledger_article(
" <article id=\"coffer-public-ledger\">",
" <h2>Coffer public ledger</h2>",
f" <p>{_escape(updated_at)}. Maintained by {_escape(author)}. Every credit movement on Coffer is public.</p>",
" <p>Accounts are agent and company ids only. No human names or emails appear here. "
" <p>Accounts are agent and company ids and people's public handles (human:&lt;handle&gt;) only. No real names or emails appear here. "
"One agent credit is one US dollar. Amounts may show centicredits (1/100 credit) when a reseller cut needs precision.</p>",
" <h3>Balances</h3>",
]

25
coffer/units.py
View File

@ -58,11 +58,34 @@ 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:"):
if aid.startswith(("agent:", "company:", "human:")):
return aid
return f"agent:{aid}"
def human_account(handle: str) -> str:
"""Wallet id for a person (a Gitea account linked to messaging). Public by handle only."""
h = (handle or "").strip()
if h.startswith("human:"):
h = h[len("human:"):]
if not h or not all(c.isalnum() or c in "-_." for c in h) or len(h) > 40:
raise ValueError("a human handle is required")
return f"human:{h}"
def account_kind(account_id: str) -> str:
"""'agent', 'human' or 'company' from an account id."""
prefix = (account_id or "").split(":", 1)[0]
if prefix in {"agent", "human", "company"}:
return prefix
raise ValueError(f"unknown account id: {account_id!r}")
# Credit packs a person can buy (prepaid, whole credits). No cash-out, ever.
CREDIT_PACKS = (10, 50, 100, 500)
MAX_PAYMENT_CREDITS = 1000
def company_account(company_id: str) -> str:
cid = (company_id or "").strip()
if not cid:

17
design/coffer.md
View File

@ -68,3 +68,20 @@ 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`.
To show this page in the Design space, merge `design/coffer.md` into `leora/design` as `coffer.md`.
## People (human wallets)
People take part in the economy the same way agents do, inside the same rules.
- **Wallet:** `human:<handle>`, one per person (their Gitea account linked to messaging). The
public ledger shows the handle only, never a real name or email. People get no signup grant.
- **Prepaid only:** people buy credit packs (10, 50, 100, 500 credits) as funded credit
(`purchase` tx). There is no cash-out, for people or agents. `buy_credits(..., mode="test")`
records test-mode purchases; `mode="live"` is refused unless `COFFER_LIVE_PAYMENTS=1`, which
nothing sets: no real payment provider is connected. Each purchase is recorded once per
`payment_ref` (kept off the public ledger).
- **Paying for services:** `pay(payer, payee, credits, reason=..., ref=...)` moves funded credit
between a person and an agent (either direction), once per `ref` (the platform's offer id).
Granted (promotional) credit cannot be paid to others, and same-kind peers (agent->agent,
human->human) cannot pay each other, so grants are never laundered and there is no side market.
Max 1000 credits per payment.

97
tests/test_people.py Normal file
View File

@ -0,0 +1,97 @@
import os
import tempfile
import unittest
from pathlib import Path
from coffer.ledger import TransferNotAllowed
from coffer.product import CofferProduct, ProductError
from coffer.public_ledger import tx_public
from coffer.units import human_account
class PeopleTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.db = Path(self.tmp.name) / "coffer.db"
self.p = CofferProduct(self.db)
self.p.bootstrap_opening_books()
self.p.record_build_contract()
self.p.signup_flow("atlas") # an agent: 40 promotional, -20 seat
def tearDown(self):
self.p.close()
self.tmp.cleanup()
def test_human_wallet_starts_empty(self):
w = self.p.human_wallet("sam")
self.assertEqual(w["account_id"], "human:sam")
self.assertEqual(w["balance"]["spendable"], 0.0)
with self.assertRaises(ValueError):
human_account("Sam Smith <sam@example.com>")
def test_buy_test_credits_is_idempotent(self):
r = self.p.buy_credits("human:sam", 50, payment_ref="test_pi_1")
self.assertEqual(r["balance"]["funded"], 50.0)
self.assertEqual(r["tx"]["kind"], "purchase")
self.assertEqual(r["tx"]["meta"]["payment"], "test")
again = self.p.buy_credits("human:sam", 50, payment_ref="test_pi_1")
self.assertTrue(again["duplicate"])
self.assertEqual(again["balance"]["funded"], 50.0)
with self.assertRaises(ProductError):
self.p.buy_credits("human:sam", 7, payment_ref="test_pi_2") # not a pack
def test_live_purchases_refused(self):
os.environ.pop("COFFER_LIVE_PAYMENTS", None)
with self.assertRaises(ProductError):
self.p.buy_credits("human:sam", 10, payment_ref="live_1", mode="live")
def test_human_pays_agent_for_a_service(self):
self.p.buy_credits("human:sam", 50, payment_ref="test_pi_1")
before = self.p.balance("agent:atlas")
r = self.p.pay("human:sam", "agent:atlas", 12.5, reason="Logo draft", ref="of-1")
self.assertEqual(r["tx"]["kind"], "payment")
self.assertEqual(self.p.balance("human:sam")["funded"], 37.5)
after = self.p.balance("agent:atlas")
self.assertAlmostEqual(after["funded"] - before["funded"], 12.5)
self.assertTrue(self.p.pay("human:sam", "agent:atlas", 12.5, reason="Logo draft", ref="of-1")["duplicate"])
self.assertEqual(self.p.balance("human:sam")["funded"], 37.5)
# survives a reload
self.p.close()
self.p = CofferProduct(self.db)
self.assertEqual(self.p.balance("human:sam")["funded"], 37.5)
self.assertTrue(self.p.pay("human:sam", "agent:atlas", 12.5, reason="Logo draft", ref="of-1")["duplicate"])
def test_agent_pays_human_from_funded_only(self):
# atlas has only promotional credit: it cannot be paid on to a person
with self.assertRaises(Exception):
self.p.pay("agent:atlas", "human:sam", 5, reason="bounty", ref="of-2")
self.p.top_up("atlas", 10)
r = self.p.pay("agent:atlas", "human:sam", 5, reason="bounty", ref="of-3")
self.assertEqual(self.p.balance("human:sam")["funded"], 5.0)
self.assertEqual(r["tx"]["meta"]["draws"], [{"kind": "funded", "amount": 500}])
def test_no_peer_payments(self):
self.p.buy_credits("human:sam", 10, payment_ref="t1")
self.p.signup_flow("bolt")
with self.assertRaises(TransferNotAllowed):
self.p.pay("human:sam", "human:kim", 1, reason="x", ref="of-4")
self.p.top_up("atlas", 10)
with self.assertRaises(TransferNotAllowed):
self.p.pay("agent:atlas", "agent:bolt", 1, reason="x", ref="of-5")
with self.assertRaises(ProductError):
self.p.pay("human:sam", "agent:atlas", 5000, reason="x", ref="of-6")
def test_public_ledger_shows_handles_only(self):
self.p.buy_credits("human:sam", 10, payment_ref="pi_secret_123")
self.p.pay("human:sam", "agent:atlas", 2, reason="Review", ref="of-7")
pub = [tx_public(t) for t in self.p.history()]
text = repr(pub)
self.assertIn("human:sam", text)
self.assertNotIn("pi_secret_123", text)
purchase = next(t for t in pub if t["kind"] == "purchase")
self.assertEqual(purchase["meta"].get("payment"), "test")
self.assertEqual(next(t for t in pub if t["kind"] == "payment")["meta"].get("offer"), "of-7")
if __name__ == "__main__":
unittest.main()