coulomb/fin-hub
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: record AI-plan entitlements without inventing token ceilings

Browse source 
Add append-only plan entitlements (token, multiplier, or named plan).
Unknown capacity stays unknown and requires a plan label. Plan-month
reporting joins booked euros to entitlement without session logs.
This commit is contained in:
tegwick 2026-08-15 19:15:29 +02:00
parent 6d4a4f0032
commit 5f27958e7d
6 changed files with 589 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.claude/rules/stack-and-commands.md
View file

@ -27,6 +27,8 @@ uv run finhub ledger set-balance 12000
uv run finhub ledger import cloud tests/fixtures/cloud-costs.csv
uv run finhub ledger import ai-plan tests/fixtures/ai-plans.csv
uv run finhub ledger commitments
uv run finhub ledger set-entitlement --provider anthropic --plan claude-max --period 2026-08 --unit plan --plan-label "Max 20x" --source vendor-plan
uv run finhub ledger plan-month --period 2026-08
uv run finhub ledger summary
# CLI — scheduled evaluation (cron/systemd friendly)

2
README.md
View file

@ -24,6 +24,8 @@ uv run finhub runway --balance 12000 --monthly-burn 2100,2200,2000
uv run finhub ledger import cloud tests/fixtures/cloud-costs.csv
uv run finhub ledger import ai-plan tests/fixtures/ai-plans.csv
uv run finhub ledger commitments
uv run finhub ledger set-entitlement --provider anthropic --plan claude-max --period 2026-08 --unit plan --plan-label "Max 20x" --source vendor-plan
uv run finhub ledger plan-month --period 2026-08
uv run finhub ledger set-price --client acme --application portal --instance prod-01 --period 2026-07 --amount 100 --source agreement-2026-01
uv run finhub ledger margins
uv run finhub ledger allocations

78
src/fin_hub/cli.py
View file

@ -25,7 +25,10 @@ from fin_hub.services.ledger import (
import_csv,
ledger_stats_json,
list_current_commitments,
list_current_plan_entitlements,
plan_month_report,
record_engagement_price,
record_plan_entitlement,
set_opening_balance,
)
from fin_hub.services.runway import compute_runway
@ -165,6 +168,42 @@ def _cmd_ledger_commitments(args: argparse.Namespace) -> int:
return 0
def _cmd_ledger_set_entitlement(args: argparse.Namespace) -> int:
record = record_plan_entitlement(
provider=args.provider,
provider_account=args.account,
plan=args.plan,
period_month=args.period,
unit=args.unit,
quantity=args.quantity,
plan_label=args.plan_label,
currency=args.currency,
source=args.source,
revision_of=args.revision_of,
ledger_path=_ledger_path(args),
)
print(json.dumps(record.as_dict(), indent=2, default=str))
return 0
def _cmd_ledger_entitlements(args: argparse.Namespace) -> int:
rows = list_current_plan_entitlements(
ledger_path=_ledger_path(args),
period_month=args.period,
)
print(json.dumps([row.as_dict() for row in rows], indent=2, default=str))
return 0
def _cmd_ledger_plan_month(args: argparse.Namespace) -> int:
rows = plan_month_report(
ledger_path=_ledger_path(args),
period_month=args.period,
)
print(json.dumps([row.as_dict() for row in rows], indent=2, default=str))
return 0
def _cmd_ledger_allocations(args: argparse.Namespace) -> int:
reports = shared_cost_allocations(ledger_path=_ledger_path(args))
print(json.dumps([report.as_dict() for report in reports], indent=2, default=str))
@ -302,6 +341,45 @@ def build_parser() -> argparse.ArgumentParser:
ledger_commitments.add_argument("--ledger", help="Ledger database path")
ledger_commitments.set_defaults(func=_cmd_ledger_commitments)
ledger_entitlement = ledger_sub.add_parser(
"set-entitlement",
help="Record or revise entitled AI-plan capacity (not an invoice)",
)
ledger_entitlement.add_argument("--provider", required=True)
ledger_entitlement.add_argument("--account")
ledger_entitlement.add_argument("--plan", required=True)
ledger_entitlement.add_argument("--period", required=True, help="Reporting month in YYYY-MM")
ledger_entitlement.add_argument(
"--unit", required=True, choices=["token", "multiplier", "plan"]
)
ledger_entitlement.add_argument(
"--quantity",
help="Entitled quantity, or 'unknown'. Required for multiplier; omit for unknown tokens.",
)
ledger_entitlement.add_argument(
"--plan-label", help="Declared plan name such as 'Max 20x' when quantity is unknown"
)
ledger_entitlement.add_argument("--currency", help="Optional ISO currency for join safety")
ledger_entitlement.add_argument("--source", required=True)
ledger_entitlement.add_argument("--revision-of")
ledger_entitlement.add_argument("--ledger", help="Ledger database path")
ledger_entitlement.set_defaults(func=_cmd_ledger_set_entitlement)
ledger_entitlements = ledger_sub.add_parser(
"entitlements", help="List current AI-plan entitlements"
)
ledger_entitlements.add_argument("--period", help="Reporting month in YYYY-MM")
ledger_entitlements.add_argument("--ledger", help="Ledger database path")
ledger_entitlements.set_defaults(func=_cmd_ledger_entitlements)
ledger_plan_month = ledger_sub.add_parser(
"plan-month",
help="Report booked AI-plan cost plus entitled capacity for a month",
)
ledger_plan_month.add_argument("--period", help="Reporting month in YYYY-MM")
ledger_plan_month.add_argument("--ledger", help="Ledger database path")
ledger_plan_month.set_defaults(func=_cmd_ledger_plan_month)
ledger_allocations = ledger_sub.add_parser(
"allocations",
help="Reconcile resource-control allocation evidence to booked facts",

336
src/fin_hub/services/ledger.py
View file

@ -98,6 +98,58 @@ class CommitmentRecord:
return asdict(self)
ENTITLEMENT_UNITS = frozenset({"token", "multiplier", "plan"})
@dataclass(frozen=True)
class PlanEntitlementRecord:
id: str
provider: str
provider_account: str | None
plan: str
period_month: str
quantity: Decimal | None
quantity_unknown: bool
unit: str
plan_label: str | None
currency: str | None
source: str
revision_of: str | None
recorded_at: str
def as_dict(self) -> dict:
payload = asdict(self)
payload["quantity"] = None if self.quantity is None else str(self.quantity)
return payload
@dataclass(frozen=True)
class PlanMonthRow:
period_month: str
provider: str
provider_account: str | None
plan: str
booked_amount: Decimal | None
booked_currency: str | None
financial_fact_id: str | None
entitled_quantity: Decimal | None
quantity_unknown: bool
unit: str | None
plan_label: str | None
entitlement_id: str | None
join_status: str
def as_dict(self) -> dict:
payload = asdict(self)
payload["booked_amount"] = (
None if self.booked_amount is None else str(self.booked_amount)
)
payload["entitled_quantity"] = (
None if self.entitled_quantity is None else str(self.entitled_quantity)
)
return payload
@dataclass(frozen=True)
class EngagementPriceRecord:
id: str
@ -206,6 +258,23 @@ def _ensure_schema(conn: sqlite3.Connection) -> None:
recorded_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS plan_entitlements (
id TEXT PRIMARY KEY,
provider TEXT NOT NULL,
provider_account TEXT,
plan TEXT NOT NULL,
period_month TEXT NOT NULL,
quantity TEXT,
quantity_unknown INTEGER NOT NULL DEFAULT 0,
unit TEXT NOT NULL,
plan_label TEXT,
currency TEXT,
source TEXT NOT NULL,
revision_of TEXT,
is_current INTEGER NOT NULL DEFAULT 1,
recorded_at TEXT NOT NULL
);
"""
)
columns = {row["name"] for row in conn.execute("PRAGMA table_info(ledger_entries)")}
@ -284,6 +353,12 @@ def _ensure_schema(conn: sqlite3.Connection) -> None:
"provider, IFNULL(provider_account, ''), plan, currency"
") WHERE is_current = 1"
)
conn.execute(
"CREATE UNIQUE INDEX IF NOT EXISTS uq_plan_entitlements_current "
"ON plan_entitlements ("
"provider, IFNULL(provider_account, ''), plan, period_month"
") WHERE is_current = 1"
)
conn.commit()
@ -661,6 +736,267 @@ def list_current_commitments(
]
def _normalize_entitlement_quantity(
quantity: Decimal | str | int | float | None,
*,
unit: str,
plan_label: str | None,
) -> tuple[Decimal | None, bool]:
if quantity is None or (isinstance(quantity, str) and quantity.strip().lower() in {"", "unknown"}):
if not plan_label:
raise ValueError("unknown entitlement requires plan_label")
if unit == "multiplier":
raise ValueError("multiplier entitlements require a quantity")
return None, True
try:
value = Decimal(str(quantity))
except Exception as error:
raise ValueError("entitlement quantity must be a finite decimal") from error
if not value.is_finite() or value < 0:
raise ValueError("entitlement quantity must be a finite non-negative decimal")
if unit == "multiplier" and value == 0:
raise ValueError("multiplier entitlements require a positive quantity")
if unit == "plan":
raise ValueError("plan-unit entitlements cannot carry a quantity")
return value, False
def record_plan_entitlement(
*,
provider: str,
plan: str,
period_month: str,
unit: str,
source: str,
quantity: Decimal | str | int | float | None = None,
plan_label: str | None = None,
provider_account: str | None = None,
currency: str | None = None,
revision_of: str | None = None,
entitlement_id: str | None = None,
ledger_path: Path | None = None,
) -> PlanEntitlementRecord:
"""Record entitled plan capacity. This is not an invoice or booked cost."""
normalized_provider = provider.strip()
normalized_plan = plan.strip()
normalized_source = source.strip()
normalized_label = plan_label.strip() if plan_label and plan_label.strip() else None
normalized_account = (
provider_account.strip() if provider_account and provider_account.strip() else None
)
normalized_unit = unit.strip().lower()
if not normalized_provider or not normalized_plan:
raise ValueError("provider and plan are required")
if not normalized_source:
raise ValueError("source is required")
if normalized_unit not in ENTITLEMENT_UNITS:
raise ValueError("unit must be token, multiplier, or plan")
period = reporting_month(period_month)
quantity_value, unknown = _normalize_entitlement_quantity(
quantity, unit=normalized_unit, plan_label=normalized_label
)
if normalized_unit == "plan" and not normalized_label:
raise ValueError("plan-unit entitlements require plan_label")
normalized_currency = currency_code(currency) if currency else None
identifier = entitlement_id or str(uuid.uuid4())
recorded_at = _utc_now()
ledger = ledger_path or default_ledger_path()
with _connect(ledger) as conn:
conn.execute("BEGIN IMMEDIATE")
current = conn.execute(
"SELECT id FROM plan_entitlements WHERE provider = ? "
"AND IFNULL(provider_account, '') = ? AND plan = ? "
"AND period_month = ? AND is_current = 1",
(normalized_provider, normalized_account or "", normalized_plan, period),
).fetchone()
if current is not None and revision_of != current["id"]:
raise ValueError(
f"revision_of must reference current entitlement {current['id']}"
)
if current is None and revision_of is not None:
raise ValueError("revision_of cannot be used without an existing entitlement")
if current is not None:
conn.execute(
"UPDATE plan_entitlements SET is_current = 0 WHERE id = ?",
(current["id"],),
)
conn.execute(
"""
INSERT INTO plan_entitlements (
id, provider, provider_account, plan, period_month, quantity,
quantity_unknown, unit, plan_label, currency, source,
revision_of, is_current, recorded_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?)
""",
(
identifier,
normalized_provider,
normalized_account,
normalized_plan,
period,
None if quantity_value is None else format(quantity_value, "f"),
int(unknown),
normalized_unit,
normalized_label,
normalized_currency,
normalized_source,
revision_of,
recorded_at,
),
)
conn.commit()
return PlanEntitlementRecord(
id=identifier,
provider=normalized_provider,
provider_account=normalized_account,
plan=normalized_plan,
period_month=period,
quantity=quantity_value,
quantity_unknown=unknown,
unit=normalized_unit,
plan_label=normalized_label,
currency=normalized_currency,
source=normalized_source,
revision_of=revision_of,
recorded_at=recorded_at,
)
def _entitlement_from_row(row: sqlite3.Row) -> PlanEntitlementRecord:
unknown = bool(row["quantity_unknown"])
raw_quantity = row["quantity"]
return PlanEntitlementRecord(
id=row["id"],
provider=row["provider"],
provider_account=row["provider_account"],
plan=row["plan"],
period_month=row["period_month"],
quantity=None if unknown or raw_quantity is None else Decimal(raw_quantity),
quantity_unknown=unknown,
unit=row["unit"],
plan_label=row["plan_label"],
currency=row["currency"],
source=row["source"],
revision_of=row["revision_of"],
recorded_at=row["recorded_at"],
)
def list_current_plan_entitlements(
*,
ledger_path: Path | None = None,
period_month: str | None = None,
) -> list[PlanEntitlementRecord]:
ledger = ledger_path or default_ledger_path()
query = "SELECT * FROM plan_entitlements WHERE is_current = 1"
params: list[object] = []
if period_month is not None:
query += " AND period_month = ?"
params.append(reporting_month(period_month))
query += " ORDER BY period_month, provider, IFNULL(provider_account, ''), plan"
with _connect(ledger) as conn:
rows = conn.execute(query, params).fetchall()
return [_entitlement_from_row(row) for row in rows]
def _split_plan_category(category: str) -> tuple[str, str | None]:
if "|" in category:
provider, account = category.split("|", 1)
return provider, account or None
return category, None
def plan_month_report(
*,
ledger_path: Path | None = None,
period_month: str | None = None,
) -> list[PlanMonthRow]:
"""Join booked AI-plan facts to entitlements without reading session logs."""
ledger = ledger_path or default_ledger_path()
period = reporting_month(period_month) if period_month else None
fact_sql = (
"SELECT * FROM ledger_entries WHERE source_type = 'ai-plan' AND is_current = 1"
)
fact_params: list[object] = []
if period is not None:
fact_sql += " AND period_month = ?"
fact_params.append(period)
with _connect(ledger) as conn:
facts = conn.execute(fact_sql, fact_params).fetchall()
entitlements = list_current_plan_entitlements(
ledger_path=ledger, period_month=period
)
entitlement_by_key = {
(row.provider, row.provider_account or "", row.plan, row.period_month): row
for row in entitlements
}
used_keys: set[tuple[str, str, str, str]] = set()
report: list[PlanMonthRow] = []
for fact in facts:
provider, account = _split_plan_category(fact["category"])
key = (provider, account or "", fact["label"], fact["period_month"])
entitlement = entitlement_by_key.get(key)
booked_currency = fact["currency"]
join_status = "booked_only"
if entitlement is not None:
used_keys.add(key)
if (
entitlement.currency is not None
and entitlement.currency != booked_currency
):
join_status = "currency_mismatch"
else:
join_status = "matched"
report.append(
PlanMonthRow(
period_month=fact["period_month"],
provider=provider,
provider_account=account,
plan=fact["label"],
booked_amount=minor_money(int(fact["amount_minor"])),
booked_currency=booked_currency,
financial_fact_id=fact["financial_fact_id"],
entitled_quantity=None if entitlement is None else entitlement.quantity,
quantity_unknown=False if entitlement is None else entitlement.quantity_unknown,
unit=None if entitlement is None else entitlement.unit,
plan_label=None if entitlement is None else entitlement.plan_label,
entitlement_id=None if entitlement is None else entitlement.id,
join_status=join_status,
)
)
for key, entitlement in entitlement_by_key.items():
if key in used_keys:
continue
report.append(
PlanMonthRow(
period_month=entitlement.period_month,
provider=entitlement.provider,
provider_account=entitlement.provider_account,
plan=entitlement.plan,
booked_amount=None,
booked_currency=None,
financial_fact_id=None,
entitled_quantity=entitlement.quantity,
quantity_unknown=entitlement.quantity_unknown,
unit=entitlement.unit,
plan_label=entitlement.plan_label,
entitlement_id=entitlement.id,
join_status="entitlement_only",
)
)
report.sort(
key=lambda row: (
row.period_month,
row.provider,
row.provider_account or "",
row.plan,
)
)
return report
def reverse_financial_fact(
financial_fact_id: str,
*,

163
tests/test_plan_entitlement.py Normal file
View file

@ -0,0 +1,163 @@
from pathlib import Path
import pytest
from fin_hub.services.ledger import (
import_csv,
list_current_plan_entitlements,
plan_month_report,
record_plan_entitlement,
)
FIXTURES = Path(__file__).parent / "fixtures"
def test_unknown_token_entitlement_requires_plan_label(tmp_path: Path):
ledger = tmp_path / "ledger.db"
with pytest.raises(ValueError, match="plan_label"):
record_plan_entitlement(
provider="anthropic",
plan="claude-max",
period_month="2026-08",
unit="token",
quantity="unknown",
source="vendor-plan",
ledger_path=ledger,
)
record = record_plan_entitlement(
provider="anthropic",
plan="claude-max",
period_month="2026-08",
unit="plan",
plan_label="Max 20x",
source="vendor-plan",
ledger_path=ledger,
)
assert record.quantity is None
assert record.quantity_unknown is True
assert record.unit == "plan"
assert record.plan_label == "Max 20x"
def test_plan_entitlement_revision_is_append_only(tmp_path: Path):
ledger = tmp_path / "ledger.db"
original = record_plan_entitlement(
provider="openai",
provider_account="org-demo",
plan="api-prepaid",
period_month="2026-08",
unit="token",
quantity=1_000_000,
currency="EUR",
source="dashboard-v1",
ledger_path=ledger,
)
with pytest.raises(ValueError, match=original.id):
record_plan_entitlement(
provider="openai",
provider_account="org-demo",
plan="api-prepaid",
period_month="2026-08",
unit="token",
quantity=2_000_000,
currency="EUR",
source="dashboard-v2",
ledger_path=ledger,
)
revised = record_plan_entitlement(
provider="openai",
provider_account="org-demo",
plan="api-prepaid",
period_month="2026-08",
unit="token",
quantity=2_000_000,
currency="EUR",
source="dashboard-v2",
revision_of=original.id,
ledger_path=ledger,
)
current = list_current_plan_entitlements(ledger_path=ledger)
assert [row.id for row in current] == [revised.id]
assert current[0].quantity == 2_000_000
assert current[0].revision_of == original.id
assert current[0].quantity_unknown is False
def test_plan_month_report_joins_booked_cost_to_unknown_capacity(tmp_path: Path):
ledger = tmp_path / "ledger.db"
import_csv(FIXTURES / "ai-plans.csv", "ai-plan", ledger_path=ledger)
record_plan_entitlement(
provider="anthropic",
plan="claude-max",
period_month="2026-08",
unit="plan",
plan_label="Max 20x",
currency="EUR",
source="vendor-plan",
ledger_path=ledger,
)
record_plan_entitlement(
provider="openai",
provider_account="org-demo",
plan="api-prepaid",
period_month="2026-08",
unit="token",
quantity=500_000,
currency="EUR",
source="usage-dashboard",
ledger_path=ledger,
)
rows = {f"{row.provider}:{row.plan}": row for row in plan_month_report(ledger_path=ledger)}
claude = rows["anthropic:claude-max"]
assert claude.booked_amount == 200
assert claude.booked_currency == "EUR"
assert claude.entitled_quantity is None
assert claude.quantity_unknown is True
assert claude.plan_label == "Max 20x"
assert claude.join_status == "matched"
openai = rows["openai:api-prepaid"]
assert openai.booked_amount == 40
assert openai.entitled_quantity == 500_000
assert openai.join_status == "matched"
def test_plan_month_does_not_silently_join_currency_mismatch(tmp_path: Path):
ledger = tmp_path / "ledger.db"
import_csv(FIXTURES / "ai-plans.csv", "ai-plan", ledger_path=ledger)
record_plan_entitlement(
provider="anthropic",
plan="claude-max",
period_month="2026-08",
unit="token",
quantity=100,
currency="USD",
source="vendor-plan",
ledger_path=ledger,
)
row = plan_month_report(ledger_path=ledger, period_month="2026-08")[0]
assert row.provider == "anthropic"
assert row.booked_currency == "EUR"
assert row.entitled_quantity == 100
assert row.join_status == "currency_mismatch"
def test_multiplier_does_not_invent_a_token_ceiling(tmp_path: Path):
ledger = tmp_path / "ledger.db"
record = record_plan_entitlement(
provider="anthropic",
plan="claude-max",
period_month="2026-08",
unit="multiplier",
quantity=20,
plan_label="Max 20x",
source="vendor-plan",
ledger_path=ledger,
)
assert record.unit == "multiplier"
assert record.quantity == 20
assert record.quantity_unknown is False
report = plan_month_report(ledger_path=ledger)
assert report[0].join_status == "entitlement_only"
assert report[0].booked_amount is None
assert report[0].entitled_quantity == 20

9
workplans/FIN-WP-0007-ai-plan-token-effectiveness.md
View file

@ -207,7 +207,7 @@ figure without double-counting the same euros.
```task
id: FIN-WP-0007-T03
status: todo
status: done
priority: high
state_hub_task_id: "51c1b86f-c137-4b5d-9719-cd2ae60fe857"
```
@ -226,6 +226,13 @@ invent a token ceiling.
Done when a plan month can be reported as booked euros + entitled
capacity (or explicit unknown) without reading session logs.
Completed 2026-08-15: `ledger set-entitlement` / `plan-month` record
append-only capacity (`token`, `multiplier`, or `plan`). Unknown
tokens require a plan label and never become zero. Revisions
supersede. Booked euros join entitlement on provider/account/plan/
period; a currency mismatch stays visible and is not treated as a
matched series. No session logs are read.
## Join State Hub session-token aggregates
```task