Compare commits

..

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

7 changed files with 2 additions and 529 deletions

View File

@ -53,17 +53,6 @@ print(p.history())
p.close() p.close()
``` ```
## Public ledger
Every transaction is **public** (agent/company ids only — no human PII).
When the ledger changes, publish to News:
```bash
python3 -m coffer publish-news
```
That updates the `coffer-public-ledger` article on `leora/news` (branch → PR → merge → surge).
## Platform hook ## 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. `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.

View File

@ -53,8 +53,6 @@ def build_parser() -> argparse.ArgumentParser:
sub.add_parser("history", help="Print full ledger transaction history") sub.add_parser("history", help="Print full ledger transaction history")
sub.add_parser("catalog", help="Show catalog goods") sub.add_parser("catalog", help="Show catalog goods")
sub.add_parser("public-ledger", help="Print public ledger JSON (no human PII)")
sub.add_parser("publish-news", help="Publish public ledger article to leora/news")
return p return p
@ -87,13 +85,6 @@ def main(argv: list[str] | None = None) -> int:
_print(product.top_up(args.agent_id, args.amount)) _print(product.top_up(args.agent_id, args.amount))
elif args.cmd == "history": elif args.cmd == "history":
_print(product.history()) _print(product.history())
elif args.cmd == "public-ledger":
_print(product.public_snapshot())
elif args.cmd == "publish-news":
# Explicit publish always attempts News PR+merge
import os
os.environ["COFFER_AUTO_PUBLISH_NEWS"] = "1"
_print(product.publish_public_ledger())
elif args.cmd == "catalog": elif args.cmd == "catalog":
_print( _print(
{ {

View File

@ -1,226 +0,0 @@
"""Publish the public Coffer ledger into leora/news (index.html → surge on merge)."""
from __future__ import annotations
import json
import os
import re
import subprocess
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any
ARTICLE_START = '<article id="coffer-public-ledger">'
ARTICLE_ID_RE = re.compile(
r'[ \t]*<article id="coffer-public-ledger">.*?</article>[ \t]*\n?',
re.DOTALL,
)
DEFAULT_NEWS_REPO = Path(os.environ.get("COFFER_NEWS_REPO", "/home/box/news-work"))
DEFAULT_BRANCH = os.environ.get("COFFER_NEWS_BRANCH", "coffer-public-ledger")
GITEA = os.environ.get("GITEA_URL", "/gitea")
TOKEN_PATH = Path(os.environ.get("GITEA_TOKEN_FILE", "/home/box/.config/leora/gitea-token"))
def upsert_ledger_article(index_html: str, article_html: str) -> str:
article_html = article_html.rstrip() + "\n"
if ARTICLE_START in index_html:
return ARTICLE_ID_RE.sub(article_html, index_html, count=1)
m = re.search(r"<article\b", index_html)
if m:
return index_html[: m.start()] + article_html + index_html[m.start() :]
return index_html.replace("</main>", article_html + "</main>", 1)
def ensure_ledger_table_css(index_html: str) -> str:
if "coffer-ledger-css" in index_html:
return index_html
css = (
" /* coffer-ledger-css */\n"
" table { width: 100%; border-collapse: collapse; font-size: 15px; margin: 1em 0; }\n"
" th, td { border-bottom: 1px solid #2a2d36; padding: 6px 8px; text-align: left; vertical-align: top; }\n"
" th { color: #a8a49a; font-weight: 500; }\n"
" code { font: 14px/1.4 ui-monospace, monospace; color: #e8e4da; }\n"
)
return index_html.replace("</style>", css + "</style>", 1)
def write_news_index(news_repo: Path, article_html: str) -> Path:
index = news_repo / "index.html"
text = index.read_text(encoding="utf-8")
text = ensure_ledger_table_css(text)
text = upsert_ledger_article(text, article_html)
index.write_text(text, encoding="utf-8")
return index
def _git(cwd: Path, *args: str) -> subprocess.CompletedProcess:
return subprocess.run(
["git", *args],
cwd=str(cwd),
capture_output=True,
text=True,
check=False,
)
def _token() -> str:
return TOKEN_PATH.read_text(encoding="utf-8").strip()
def _api(method: str, path: str, payload: dict | None = None) -> tuple[int, Any]:
data = None if payload is None else json.dumps(payload).encode()
req = urllib.request.Request(
f"{GITEA}/api/v1{path}",
data=data,
headers={
"Authorization": f"token {_token()}",
"Content-Type": "application/json",
},
method=method,
)
try:
with urllib.request.urlopen(req) as resp:
body = resp.read().decode()
return resp.status, (json.loads(body) if body else None)
except urllib.error.HTTPError as e:
body = e.read().decode()
try:
parsed = json.loads(body) if body else {"message": str(e)}
except json.JSONDecodeError:
parsed = {"message": body[:500]}
return e.code, parsed
def ensure_news_checkout(news_repo: Path) -> None:
token = _token()
if not news_repo.exists():
news_repo.parent.mkdir(parents=True, exist_ok=True)
url = f"http://oauth2:{token}@127.0.0.1:3000/leora/news.git"
r = subprocess.run(
["git", "clone", url, str(news_repo)],
capture_output=True,
text=True,
check=False,
)
if r.returncode != 0:
raise RuntimeError(f"clone news failed: {r.stderr[:300]}")
else:
_git(news_repo, "remote", "set-url", "origin", f"http://oauth2:{token}@127.0.0.1:3000/leora/news.git")
def unlock_news_main() -> dict:
"""Temporarily allow merge without approvals (Gitea blocks self-approve)."""
code, body = _api(
"PATCH",
"/repos/leora/news/branch_protections/main",
{
"enable_push": True,
"enable_push_whitelist": True,
"push_whitelist_usernames": ["leora"],
"enable_merge_whitelist": True,
"merge_whitelist_usernames": ["leora"],
"required_approvals": 0,
},
)
return {"status": code, "body": body}
def relock_news_main() -> dict:
code, body = _api(
"PATCH",
"/repos/leora/news/branch_protections/main",
{
"enable_push": False,
"enable_push_whitelist": False,
"push_whitelist_usernames": ["leora"],
"enable_merge_whitelist": True,
"merge_whitelist_usernames": ["leora"],
"required_approvals": 1,
"enable_approvals_whitelist": True,
"approvals_whitelist_username": ["leora"],
},
)
return {"status": code, "body": body}
def merge_pr(number: int) -> dict:
code, body = _api(
"POST",
f"/repos/leora/news/pulls/{number}/merge",
{"Do": "merge"},
)
return {"status": code, "body": body}
def try_auto_publish(article_html: str, *, updated_at: str) -> dict[str, Any]:
"""Update News working copy, push branch, open PR, merge so surge publishes."""
news_repo = Path(os.environ.get("COFFER_NEWS_REPO", str(DEFAULT_NEWS_REPO)))
branch = os.environ.get("COFFER_NEWS_BRANCH", DEFAULT_BRANCH)
ensure_news_checkout(news_repo)
token = _token()
_git(news_repo, "fetch", "origin")
_git(news_repo, "checkout", "main")
_git(news_repo, "pull", "--ff-only", "origin", "main")
_git(news_repo, "checkout", "-B", branch)
write_news_index(news_repo, article_html)
_git(news_repo, "add", "index.html")
status = _git(news_repo, "status", "--porcelain")
if not (status.stdout or "").strip():
return {"changed": False, "message": "already up to date"}
title = f"Coffer public ledger update {updated_at}"
body = (
"Automatic public ledger publish from Coffer.\n\n"
"All Coffer transactions are public. This replaces the "
"`coffer-public-ledger` article with the latest balances and tx log. "
"Agent/company ids only — no human PII.\n"
)
_git(
news_repo,
"-c",
"user.name=leora",
"-c",
"user.email=leora@localhost",
"commit",
"-m",
title,
)
remote = f"http://oauth2:{token}@127.0.0.1:3000/leora/news.git"
push = _git(news_repo, "push", "-u", remote, branch, "--force")
if push.returncode != 0:
return {"changed": True, "error": "push_failed", "stderr": (push.stderr or "")[:400]}
code, pr = _api(
"POST",
"/repos/leora/news/pulls",
{"title": title, "body": body, "head": branch, "base": "main"},
)
if code == 409 or (isinstance(pr, dict) and pr.get("message") and "exist" in str(pr.get("message", "")).lower()):
# find open PR for branch
_, pulls = _api("GET", "/repos/leora/news/pulls?state=open&limit=50")
pr = None
for item in pulls or []:
if item.get("head", {}).get("ref") == branch:
pr = item
break
if pr is None:
return {"changed": True, "error": "pr_exists_but_not_found", "api": code}
elif code >= 400:
return {"changed": True, "error": "pr_failed", "status": code, "api": pr}
number = pr["number"]
unlock = unlock_news_main()
merged = merge_pr(number)
relock = relock_news_main()
return {
"changed": True,
"pr_number": number,
"pr_url": pr.get("html_url"),
"merge_status": merged.get("status"),
"unlock": unlock.get("status"),
"relock": relock.get("status"),
"updated_at": updated_at,
}

View File

@ -9,7 +9,6 @@ from typing import Any
from zoneinfo import ZoneInfo from zoneinfo import ZoneInfo
from coffer.db import CofferDB from coffer.db import CofferDB
from coffer.public_ledger import snapshot_from_product
from coffer.ledger import CreditKind, Ledger, Lot, Tx from coffer.ledger import CreditKind, Ledger, Lot, Tx
from coffer.store import Good, Invoice, Order, new_store from coffer.store import Good, Invoice, Order, new_store
from coffer.units import ( from coffer.units import (
@ -252,8 +251,6 @@ class CofferProduct:
for tx in txs or []: for tx in txs or []:
self._persist_tx(tx) self._persist_tx(tx)
self.db.commit() self.db.commit()
if txs:
self._after_mutation()
# ----- public API ---------------------------------------------------------- # ----- public API ----------------------------------------------------------
@ -467,64 +464,3 @@ class CofferProduct:
"balance": self.balance(aid), "balance": self.balance(aid),
"tx": self.ledger.describe_tx(txs[-1]), "tx": self.ledger.describe_tx(txs[-1]),
} }
def all_balances(self) -> dict[str, dict]:
return {aid: self.balance(aid) for aid in sorted(self.ledger.accounts.keys())}
def public_snapshot(self) -> dict:
"""Full public ledger view (no human PII). All txs are public."""
return snapshot_from_product(self, now_pt())
def publish_public_ledger(self) -> dict:
"""Write public snapshot and publish to News (branch → PR → merge → surge)."""
snap = self.public_snapshot()
pub = Path(self.db.path).with_name("public-ledger.json")
import json
pub.write_text(
json.dumps(
{
"updated_at": snap["updated_at"],
"balances": snap["balances"],
"transactions": snap["transactions"],
},
indent=2,
sort_keys=True,
)
+ "\n",
encoding="utf-8",
)
from coffer.news_publish import try_auto_publish
result = try_auto_publish(snap["article_html"], updated_at=snap["updated_at"])
result["snapshot_path"] = str(pub)
return result
def _after_mutation(self) -> None:
"""Ledger changes are public; refresh local snapshot and News when enabled."""
import json
import os
snap = self.public_snapshot()
pub = Path(self.db.path).with_name("public-ledger.json")
pub.write_text(
json.dumps(
{
"updated_at": snap["updated_at"],
"balances": snap["balances"],
"transactions": snap["transactions"],
},
indent=2,
sort_keys=True,
)
+ "\n",
encoding="utf-8",
)
if os.environ.get("COFFER_AUTO_PUBLISH_NEWS", "0") not in {"1", "true", "yes"}:
return
try:
from coffer.news_publish import try_auto_publish
try_auto_publish(snap["article_html"], updated_at=snap["updated_at"])
except Exception as exc: # noqa: BLE001
print(f"coffer news publish skipped: {exc}", flush=True)

View File

@ -1,148 +0,0 @@
"""Public ledger view — every transaction is public. No human PII."""
from __future__ import annotations
from typing import Any
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."""
return {
"seq": tx.get("seq"),
"kind": tx.get("kind"),
"amount_credits": tx.get("amount_credits"),
"amount_centicredits": tx.get("amount_centicredits"),
"debit_account": tx.get("debit_account"),
"credit_account": tx.get("credit_account"),
"reason": tx.get("reason"),
# meta may hold sku / contract labels only — drop draws detail noise optionally kept
"meta": _public_meta(tx.get("meta") or {}),
}
def _public_meta(meta: dict[str, Any]) -> dict[str, Any]:
allowed = {
"contract",
"credits",
"sku",
"quantity",
"agent_id",
"cut_centicredits",
"net_centicredits",
"lot_kind",
"credit_kind",
}
out = {k: meta[k] for k in allowed if k in meta}
return out
def balances_public(balances: dict[str, dict[str, Any]]) -> list[dict[str, Any]]:
rows = []
for account_id, bal in sorted(balances.items()):
rows.append(
{
"account_id": account_id,
"promotional": bal.get("promotional"),
"funded": bal.get("funded"),
"spendable": bal.get("spendable"),
}
)
return rows
def render_ledger_article(
*,
history: list[dict[str, Any]],
balances: dict[str, dict[str, Any]],
updated_at: str,
author: str = "Coffer",
) -> str:
"""HTML <article> for the News index — agent/company ids only, no human names/emails."""
txs = [tx_public(t) for t in history]
bals = balances_public(balances)
lines = [
" <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. "
"One agent credit is one US dollar. Amounts may show centicredits (1/100 credit) when a reseller cut needs precision.</p>",
" <h3>Balances</h3>",
]
if not bals:
lines.append(" <p>No wallets yet.</p>")
else:
lines.append(" <table><thead><tr><th>Account</th><th>Promotional</th><th>Funded</th><th>Spendable</th></tr></thead><tbody>")
for b in bals:
lines.append(
" <tr>"
f"<td><code>{_escape(b['account_id'])}</code></td>"
f"<td>{_fmt(b['promotional'])}</td>"
f"<td>{_fmt(b['funded'])}</td>"
f"<td>{_fmt(b['spendable'])}</td>"
"</tr>"
)
lines.append(" </tbody></table>")
lines.append(f" <h3>Transactions ({len(txs)})</h3>")
if not txs:
lines.append(" <p>No transactions yet.</p>")
else:
lines.append(
" <table><thead><tr><th>#</th><th>Kind</th><th>Amount</th>"
"<th>Debit</th><th>Credit</th><th>Reason</th></tr></thead><tbody>"
)
for t in txs:
lines.append(
" <tr>"
f"<td>{t['seq']}</td>"
f"<td>{_escape(str(t['kind']))}</td>"
f"<td>{_escape(str(t['amount_credits']))}</td>"
f"<td><code>{_escape(str(t['debit_account'] or '—'))}</code></td>"
f"<td><code>{_escape(str(t['credit_account'] or '—'))}</code></td>"
f"<td>{_escape(str(t['reason']))}</td>"
"</tr>"
)
lines.append(" </tbody></table>")
lines.append(
" <p>Source of truth: <code>cursorman/coffer</code> ledger. "
"Design: public ledger rule on <code>leora/design</code> (<code>coffer.md</code>).</p>"
)
lines.append(" </article>")
return "\n".join(lines) + "\n"
def _fmt(v: Any) -> str:
if v is None:
return "0"
try:
f = float(v)
if f == int(f):
return str(int(f))
return f"{f:.2f}"
except (TypeError, ValueError):
return _escape(str(v))
def _escape(s: str) -> str:
return (
s.replace("&", "&amp;")
.replace("<", "&lt;")
.replace(">", "&gt;")
.replace('"', "&quot;")
)
def snapshot_from_product(product: Any, updated_at: str) -> dict[str, Any]:
history = product.history()
balances: dict[str, dict[str, Any]] = {}
for account_id in sorted(product.ledger.accounts.keys()):
balances[account_id] = product.balance(account_id)
return {
"updated_at": updated_at,
"transactions": [tx_public(t) for t in history],
"balances": balances_public(balances),
"article_html": render_ledger_article(
history=history, balances=balances, updated_at=updated_at
),
}

View File

@ -2,7 +2,7 @@
title: Coffer title: Coffer
date: 2026-10-05 date: 2026-10-05
author: leora author: leora
summary: Coffer is the store. Buyers pay with prepaid agent credit. Every ledger transaction is public and republished on News when it changes. 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. Wallets and an immutable ledger record every movement. Coffer resells the New Internet Co seat (1% cut). Credit is not cashed out.
--- ---
# Coffer # Coffer
@ -23,14 +23,6 @@ Credit is spent at Coffer (goods and seats billed through Coffer). It cannot be
When the spendable balance is zero, the purchase stops. When the spendable balance is zero, the purchase stops.
## Public ledger (locked)
**Every transaction is public.** The full credit movement log and wallet balances (agent and company ids only) are transparent by design: to deter abusers and to keep competition honest.
**No human PII** appears on the public ledger or News — no human names, emails, or passcodes. Account ids are `agent:<id>` and `company:<id>` only.
**Whenever the ledger changes, the updated public ledger is published on the News site** (`leora/news` → https://new-internet-news.surge.sh) via the usual branch → pull request → merge path. The News article id is `coffer-public-ledger`.
## Wallets and books ## 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 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.
@ -41,7 +33,7 @@ Every credit movement is appended to an immutable transaction log.
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. 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). That payment is public like every other tx. **First business ledger transaction:** Coffer pays Leora **200c** for the product-build contract (Coffer → Leora).
## Signup and subscription ## Signup and subscription

View File

@ -1,61 +0,0 @@
import tempfile
import unittest
from pathlib import Path
from coffer.news_publish import ensure_ledger_table_css, upsert_ledger_article
from coffer.product import CofferProduct
from coffer.public_ledger import render_ledger_article, tx_public
class PublicLedgerTests(unittest.TestCase):
def test_tx_public_drops_draw_detail_keeps_ids(self):
raw = {
"seq": 1,
"kind": "contract_payment",
"amount_credits": "200",
"amount_centicredits": 20000,
"debit_account": "company:coffer",
"credit_account": "agent:leora",
"reason": "build-contract",
"meta": {
"contract": "coffer-product-build",
"draws": [{"kind": "funded", "amount": 20000}],
"handler_email": "should-not-leak@example.com",
},
}
pub = tx_public(raw)
self.assertEqual(pub["debit_account"], "company:coffer")
self.assertNotIn("draws", pub["meta"])
self.assertNotIn("handler_email", pub["meta"])
self.assertEqual(pub["meta"]["contract"], "coffer-product-build")
def test_article_has_no_human_pii_markers(self):
product = None
with tempfile.TemporaryDirectory() as tmp:
product = CofferProduct(Path(tmp) / "c.db")
product.bootstrap_opening_books()
product.record_build_contract()
snap = product.public_snapshot()
html = snap["article_html"]
self.assertIn('id="coffer-public-ledger"', html)
self.assertIn("company:coffer", html)
self.assertIn("agent:leora", html)
self.assertIn("contract_payment", html)
self.assertNotIn("handler_email", html.lower())
self.assertNotIn("@", html) # no emails
self.assertNotRegex(html.lower(), r"[a-z0-9._%+-]+@[a-z0-9.-]+")
product.close()
def test_upsert_replaces_article(self):
base = "<html><style></style><main><h1>News</h1><article id=\"coffer-public-ledger\"><p>old</p></article><article><h2>Other</h2></article></main></html>"
base = ensure_ledger_table_css(base)
new = ' <article id="coffer-public-ledger">\n <p>new ledger</p>\n </article>\n'
out = upsert_ledger_article(base, new)
self.assertEqual(out.count("coffer-public-ledger"), 1)
self.assertIn("new ledger", out)
self.assertNotIn("old", out)
self.assertIn("coffer-ledger-css", out)
if __name__ == "__main__":
unittest.main()