feat: publish unblended AI-plan effectiveness series

Report booked cost, entitlement, token coverage, and work as three
series: work per measured token, work per measured+estimated token,
and work per euro. Months below 50% coverage are unfit for token
trends. There is no blended efficiency number.

.claude/rules/stack-and-commands.md

@@ -33,6 +33,7 @@ uv run finhub ledger ingest-session-tokens tests/fixtures/session-tokens-2026-08
 uv run finhub ledger session-tokens --period 2026-08
 uv run finhub ledger allocate-plan --fact FACT --method measured_token_share
 uv run finhub ledger plan-allocations
+uv run finhub ledger effectiveness --period 2026-08
 uv run finhub ledger summary
 
 # CLI — scheduled evaluation (cron/systemd friendly)

README.md

@@ -30,6 +30,7 @@ uv run finhub ledger ingest-session-tokens tests/fixtures/session-tokens-2026-08
 uv run finhub ledger session-tokens --period 2026-08
 uv run finhub ledger allocate-plan --fact FACT --method measured_token_share
 uv run finhub ledger plan-allocations
+uv run finhub ledger effectiveness --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

docs/ai-plan-token-effectiveness.md

@@ -106,8 +106,11 @@ State Hub `measurement_kind` maps as:
 | no events | `unknown` residual, not a zero observation |
 
 A derived overlay (implied €/token, work per token) is only as strong
-as its weakest input. A month whose coverage is incomplete is
+as its weakest input. Coverage is measured / (measured + estimated).
-`insufficient` for trend comparison, not interpolated.
+A month whose coverage is below `0.50` is `unfit` for token-series
+trend comparison, not interpolated. The effectiveness report publishes
+three series — work per measured token, work per measured+estimated
+token, and work per euro — and never a blended efficiency number.
 
 ## What fin-hub will not do
 

src/fin_hub/app.py

@@ -6,7 +6,7 @@ from fastapi import FastAPI
 from pydantic import BaseModel
 
 from fin_hub import __version__
-from fin_hub.routers import create_runway_router
+from fin_hub.routers import create_effectiveness_router, create_runway_router
 
 
 class HealthResponse(BaseModel):
@@ -19,7 +19,7 @@ def create_app() -> FastAPI:
     app = FastAPI(
         title="Fin Hub API",
         version=__version__,
-        description="Read surfaces for runway projection and monthly burn.",
+        description="Read surfaces for runway projection, monthly burn, and AI-plan effectiveness.",
     )
 
     @app.get("/healthz", response_model=HealthResponse, tags=["system"])
@@ -27,6 +27,7 @@ def create_app() -> FastAPI:
         return HealthResponse(service="fin-hub", status="ok", version=__version__)
 
     app.include_router(create_runway_router())
+    app.include_router(create_effectiveness_router())
     return app
 
 

src/fin_hub/cli.py

@@ -18,6 +18,7 @@ from fin_hub.ingest.hosteurope import parse_hosteurope_csv
 from fin_hub.services.alerts import evaluate_budget_alerts
 from fin_hub.services.allocation import shared_cost_allocations
 from fin_hub.services.billing import build_billing_basis
+from fin_hub.services.effectiveness import effectiveness_report
 from fin_hub.services.evaluate import evaluate_runway
 from fin_hub.services.reporting_allocation import (
     allocate_ai_plan_report,
@@ -254,6 +255,15 @@ def _cmd_ledger_plan_allocations(args: argparse.Namespace) -> int:
     return 0
 
 
+def _cmd_ledger_effectiveness(args: argparse.Namespace) -> int:
+    rows = effectiveness_report(
+        ledger_path=_ledger_path(args),
+        period_month=args.period,
+    )
+    print(json.dumps([row.as_dict() for row in rows], indent=2))
+    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))
@@ -476,6 +486,14 @@ def build_parser() -> argparse.ArgumentParser:
     ledger_plan_allocations.add_argument("--ledger", help="Ledger database path")
     ledger_plan_allocations.set_defaults(func=_cmd_ledger_plan_allocations)
 
+    ledger_effectiveness = ledger_sub.add_parser(
+        "effectiveness",
+        help="Report booked AI-plan cost against tokens and work (three series)",
+    )
+    ledger_effectiveness.add_argument("--period", help="Reporting month in YYYY-MM")
+    ledger_effectiveness.add_argument("--ledger", help="Ledger database path")
+    ledger_effectiveness.set_defaults(func=_cmd_ledger_effectiveness)
+
     ledger_allocations = ledger_sub.add_parser(
         "allocations",
         help="Reconcile resource-control allocation evidence to booked facts",

src/fin_hub/routers/__init__.py

@@ -1,3 +1,4 @@
+from fin_hub.routers.effectiveness import create_effectiveness_router
 from fin_hub.routers.runway import create_runway_router
 
-__all__ = ["create_runway_router"]
+__all__ = ["create_effectiveness_router", "create_runway_router"]

src/fin_hub/routers/effectiveness.py

@@ -0,0 +1,53 @@
+"""Effectiveness read router."""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Callable
+from pathlib import Path
+
+from fastapi import APIRouter, HTTPException, Query
+
+from fin_hub.schemas.effectiveness import EffectivenessRead
+from fin_hub.services.effectiveness import DEFAULT_COVERAGE_THRESHOLD, effectiveness_report
+from fin_hub.services.ledger import default_ledger_path
+
+LedgerPathResolver = Callable[[str | None], Path]
+
+
+def _default_ledger_path_resolver(ledger: str | None) -> Path:
+    if ledger:
+        return Path(ledger)
+    env_path = os.environ.get("FIN_HUB_LEDGER")
+    if env_path:
+        return Path(env_path)
+    return default_ledger_path()
+
+
+def create_effectiveness_router(
+    *,
+    prefix: str = "/effectiveness",
+    ledger_path_resolver: LedgerPathResolver | None = None,
+) -> APIRouter:
+    resolve_ledger = ledger_path_resolver or _default_ledger_path_resolver
+    router = APIRouter(prefix=prefix, tags=["effectiveness"])
+
+    @router.get("", response_model=list[EffectivenessRead])
+    def get_effectiveness(
+        period: str | None = Query(default=None, description="Reporting month YYYY-MM"),
+        ledger: str | None = Query(default=None, description="Ledger database path override"),
+    ) -> list[EffectivenessRead]:
+        ledger_path = resolve_ledger(ledger)
+        if not ledger_path.exists():
+            raise HTTPException(status_code=404, detail=f"Ledger not found: {ledger_path}")
+        try:
+            rows = effectiveness_report(
+                ledger_path=ledger_path,
+                period_month=period,
+                coverage_threshold=DEFAULT_COVERAGE_THRESHOLD,
+            )
+        except ValueError as exc:
+            raise HTTPException(status_code=422, detail=str(exc)) from exc
+        return [EffectivenessRead(**row.as_dict()) for row in rows]
+
+    return router

src/fin_hub/schemas/effectiveness.py

@@ -0,0 +1,34 @@
+"""Read schema for the AI-plan effectiveness surface."""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, ConfigDict
+
+
+class EffectivenessRead(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    period_month: str
+    provider: str
+    provider_account: str | None
+    plan: str
+    currency: str | None
+    booked_amount: str | None
+    entitled_quantity: str | None
+    entitled_unknown: bool
+    plan_label: str | None
+    tokens_measured: int | None
+    tokens_estimated: int | None
+    tokens_unknown: bool
+    coverage: str | None
+    coverage_threshold: str
+    trend_fit: str
+    implied_eur_per_measured_token: str | None
+    implied_eur_per_measured_estimated_token: str | None
+    implied_basis: str
+    work_count: int | None
+    work_unit: str | None
+    work_per_measured_token: str | None
+    work_per_measured_estimated_token: str | None
+    work_per_euro: str | None
+    token_join: str

src/fin_hub/schemas/session_tokens.py

@@ -36,6 +36,14 @@ class SessionTokenSlice(BaseModel):
         return value
 
 
+class SessionTokenWork(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    unit: Literal["task", "outcome"] = "task"
+    count: int = Field(ge=0)
+    source: str = Field(min_length=1)
+
+
 class SessionTokenAssociation(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
@@ -64,6 +72,7 @@ class SessionTokenEvidence(BaseModel):
     source_evidence: list[str] = Field(min_length=1)
     totals: dict[str, SessionTokenSlice] = Field(default_factory=dict)
     associations: list[SessionTokenAssociation] = Field(default_factory=list)
+    work: SessionTokenWork | None = None
 
     @field_validator("totals")
     @classmethod

src/fin_hub/services/effectiveness.py

@@ -0,0 +1,242 @@
+"""Period effectiveness of booked AI-plan spend against tokens and work."""
+
+from __future__ import annotations
+
+import calendar
+from dataclasses import asdict, dataclass
+from datetime import date
+from decimal import Decimal, ROUND_HALF_EVEN
+from pathlib import Path
+
+from fin_hub.money import reporting_month
+from fin_hub.schemas.session_tokens import SessionTokenEvidence
+from fin_hub.services.ledger import PlanMonthRow, default_ledger_path, plan_month_report
+from fin_hub.services.session_tokens import (
+    SessionTokenSummary,
+    list_current_session_token_evidence,
+    summarize_session_token_evidence,
+)
+
+DEFAULT_COVERAGE_THRESHOLD = Decimal("0.50")
+RATIO_QUANTUM = Decimal("0.000001")
+
+
+def _month_bounds(period_month: str) -> tuple[date, date]:
+    year, month = (int(part) for part in period_month.split("-"))
+    return date(year, month, 1), date(year, month, calendar.monthrange(year, month)[1])
+
+
+def _overlaps(record: SessionTokenEvidence, start: date, end: date) -> bool:
+    return record.period_start <= end and record.period_end >= start
+
+
+def _ratio(numerator: Decimal | int | None, denominator: Decimal | int | None) -> Decimal | None:
+    if numerator is None or denominator is None:
+        return None
+    denom = Decimal(denominator)
+    if denom == 0:
+        return None
+    return (Decimal(numerator) / denom).quantize(RATIO_QUANTUM, rounding=ROUND_HALF_EVEN)
+
+
+def _work_count(record: SessionTokenEvidence) -> tuple[int | None, str | None]:
+    if record.work is not None:
+        return record.work.count, record.work.unit
+    workplans = {
+        item.scope_id
+        for item in record.associations
+        if item.scope == "workplan" and item.measurement_kind != "superseded"
+    }
+    if not workplans:
+        return None, None
+    return len(workplans), "task"
+
+
+def _select_tokens(
+    period_month: str,
+    records: list[SessionTokenEvidence],
+) -> SessionTokenEvidence | None:
+    start, end = _month_bounds(period_month)
+    matches = [row for row in records if _overlaps(row, start, end)]
+    if len(matches) != 1:
+        return None
+    return matches[0]
+
+
+@dataclass(frozen=True)
+class EffectivenessRow:
+    period_month: str
+    provider: str
+    provider_account: str | None
+    plan: str
+    currency: str | None
+    booked_amount: Decimal | None
+    entitled_quantity: Decimal | None
+    entitled_unknown: bool
+    plan_label: str | None
+    tokens_measured: int | None
+    tokens_estimated: int | None
+    tokens_unknown: bool
+    coverage: Decimal | None
+    coverage_threshold: Decimal
+    trend_fit: str
+    implied_eur_per_measured_token: Decimal | None
+    implied_eur_per_measured_estimated_token: Decimal | None
+    implied_basis: str
+    work_count: int | None
+    work_unit: str | None
+    work_per_measured_token: Decimal | None
+    work_per_measured_estimated_token: Decimal | None
+    work_per_euro: Decimal | None
+    token_join: str
+
+    def as_dict(self) -> dict:
+        payload = asdict(self)
+        for key in (
+            "booked_amount",
+            "entitled_quantity",
+            "coverage",
+            "coverage_threshold",
+            "implied_eur_per_measured_token",
+            "implied_eur_per_measured_estimated_token",
+            "work_per_measured_token",
+            "work_per_measured_estimated_token",
+            "work_per_euro",
+        ):
+            value = payload[key]
+            payload[key] = None if value is None else str(value)
+        return payload
+
+
+def _row_from_plan(
+    plan: PlanMonthRow,
+    *,
+    summary: SessionTokenSummary | None,
+    work_count: int | None,
+    work_unit: str | None,
+    token_join: str,
+    threshold: Decimal,
+) -> EffectivenessRow:
+    measured = None if summary is None else summary.tokens_measured
+    estimated = None if summary is None else summary.tokens_estimated
+    unknown = True if summary is None else summary.unknown_residual
+    combined = None
+    if measured is not None or estimated is not None:
+        combined = (measured or 0) + (estimated or 0)
+        if combined == 0 and unknown:
+            combined = None
+    if measured is None and estimated is None:
+        coverage = None
+    elif estimated is None:
+        coverage = Decimal("1.00") if measured else None
+    else:
+        coverage = _ratio(measured or 0, (measured or 0) + estimated)
+    trend_fit = (
+        "comparable"
+        if coverage is not None and coverage >= threshold
+        else "unfit"
+    )
+    booked = plan.booked_amount
+    return EffectivenessRow(
+        period_month=plan.period_month,
+        provider=plan.provider,
+        provider_account=plan.provider_account,
+        plan=plan.plan,
+        currency=plan.booked_currency,
+        booked_amount=booked,
+        entitled_quantity=plan.entitled_quantity,
+        entitled_unknown=plan.quantity_unknown,
+        plan_label=plan.plan_label,
+        tokens_measured=measured,
+        tokens_estimated=estimated,
+        tokens_unknown=unknown,
+        coverage=coverage,
+        coverage_threshold=threshold,
+        trend_fit=trend_fit,
+        implied_eur_per_measured_token=_ratio(booked, measured),
+        implied_eur_per_measured_estimated_token=_ratio(booked, combined),
+        implied_basis="inferred",
+        work_count=work_count,
+        work_unit=work_unit,
+        work_per_measured_token=_ratio(work_count, measured),
+        work_per_measured_estimated_token=_ratio(work_count, combined),
+        work_per_euro=_ratio(work_count, booked),
+        token_join=token_join,
+    )
+
+
+def effectiveness_report(
+    *,
+    ledger_path: Path | None = None,
+    period_month: str | None = None,
+    coverage_threshold: Decimal = DEFAULT_COVERAGE_THRESHOLD,
+) -> list[EffectivenessRow]:
+    """Booked spend, entitlement, tokens, and work as three unblended series."""
+
+    ledger = ledger_path or default_ledger_path()
+    period = reporting_month(period_month) if period_month else None
+    plans = plan_month_report(ledger_path=ledger, period_month=period)
+    tokens = list_current_session_token_evidence(ledger_path=ledger)
+    threshold = coverage_threshold
+    rows: list[EffectivenessRow] = []
+    plans_by_period: dict[str, list[PlanMonthRow]] = {}
+    for plan in plans:
+        plans_by_period.setdefault(plan.period_month, []).append(plan)
+    for month, month_plans in plans_by_period.items():
+        booked_plans = [item for item in month_plans if item.booked_amount is not None]
+        token_record = _select_tokens(month, tokens)
+        summary = (
+            None if token_record is None else summarize_session_token_evidence(token_record)
+        )
+        work_count = None
+        work_unit = None
+        if token_record is not None:
+            work_count, work_unit = _work_count(token_record)
+        if token_record is None:
+            token_join = "missing"
+        elif len(booked_plans) == 1:
+            token_join = "period"
+        else:
+            token_join = "shared_period"
+            summary = None
+            work_count = None
+            work_unit = None
+        for plan in month_plans:
+            rows.append(
+                _row_from_plan(
+                    plan,
+                    summary=summary,
+                    work_count=work_count,
+                    work_unit=work_unit,
+                    token_join=token_join,
+                    threshold=threshold,
+                )
+            )
+    rows.sort(key=lambda row: (row.period_month, row.provider, row.plan))
+    return rows
+
+
+def cheaper_on_series(left: EffectivenessRow, right: EffectivenessRow, series: str) -> bool | None:
+    """Whether left is cheaper per unit of work on one named series.
+
+    Returns None when either month is unfit for token series, or a value is
+    missing. There is no blended series.
+    """
+
+    if series in {"measured", "measured_estimated"}:
+        if left.trend_fit != "comparable" or right.trend_fit != "comparable":
+            return None
+        attr = (
+            "work_per_measured_token"
+            if series == "measured"
+            else "work_per_measured_estimated_token"
+        )
+    elif series == "euro":
+        attr = "work_per_euro"
+    else:
+        raise ValueError("series must be measured, measured_estimated, or euro")
+    left_value = getattr(left, attr)
+    right_value = getattr(right, attr)
+    if left_value is None or right_value is None:
+        return None
+    return left_value > right_value

tests/test_effectiveness.py

@@ -0,0 +1,137 @@
+from decimal import Decimal
+from pathlib import Path
+
+from fastapi.testclient import TestClient
+
+from fin_hub.app import create_app
+from fin_hub.routers.effectiveness import create_effectiveness_router
+from fin_hub.services.effectiveness import cheaper_on_series, effectiveness_report
+from fin_hub.services.ledger import import_csv, record_plan_entitlement
+from fin_hub.services.session_tokens import ingest_session_token_evidence
+
+
+def _write_plan(path: Path, *, period: str, amount: str) -> None:
+    path.write_text(
+        "provider,plan,charge_kind,amount,currency,period_month,ongoing\n"
+        f"anthropic,claude-max,subscription,{amount},EUR,{period},true\n",
+        encoding="utf-8",
+    )
+
+
+def _token_snapshot(
+    *,
+    record_id: str,
+    period: str,
+    measured: int,
+    estimated: int,
+    work: int,
+) -> dict:
+    start, end = f"{period}-01", f"{period}-28"
+    totals = {
+        "measured": {
+            "tokens_in": measured,
+            "tokens_out": 0,
+            "event_count": 1 if measured else 0,
+            "confidence": "1.0",
+        }
+    }
+    if estimated:
+        totals["estimated"] = {
+            "tokens_in": estimated,
+            "tokens_out": 0,
+            "event_count": 1,
+            "confidence": "0.35",
+        }
+    return {
+        "record_id": record_id,
+        "period_start": start,
+        "period_end": end,
+        "captured_at": "2026-08-15T12:00:00Z",
+        "source_evidence": [f"state-hub:/token-events/aggregate/?month={period}"],
+        "totals": totals,
+        "work": {"unit": "task", "count": work, "source": "state-hub-tasks"},
+    }
+
+
+def test_low_coverage_month_is_unfit_and_has_no_blended_efficiency(tmp_path: Path):
+    ledger = tmp_path / "ledger.db"
+    high = tmp_path / "july.csv"
+    low = tmp_path / "august.csv"
+    _write_plan(high, period="2026-07", amount="200.00")
+    _write_plan(low, period="2026-08", amount="200.00")
+    import_csv(high, "ai-plan", ledger_path=ledger)
+    import_csv(low, "ai-plan", ledger_path=ledger)
+    record_plan_entitlement(
+        provider="anthropic",
+        plan="claude-max",
+        period_month="2026-07",
+        unit="plan",
+        plan_label="Max 20x",
+        source="vendor-plan",
+        ledger_path=ledger,
+    )
+    ingest_session_token_evidence(
+        _token_snapshot(
+            record_id="tokens:2026-07",
+            period="2026-07",
+            measured=200,
+            estimated=0,
+            work=2,
+        ),
+        ledger_path=ledger,
+    )
+    ingest_session_token_evidence(
+        _token_snapshot(
+            record_id="tokens:2026-08",
+            period="2026-08",
+            measured=20,
+            estimated=180,
+            work=2,
+        ),
+        ledger_path=ledger,
+    )
+
+    rows = {row.period_month: row for row in effectiveness_report(ledger_path=ledger)}
+    good = rows["2026-07"]
+    bad = rows["2026-08"]
+    assert good.trend_fit == "comparable"
+    assert bad.trend_fit == "unfit"
+    assert good.coverage == Decimal("1.00")
+    assert bad.coverage == Decimal("0.100000")
+    assert good.work_per_euro == bad.work_per_euro
+    assert good.implied_basis == "inferred"
+    payload = bad.as_dict()
+    assert "efficiency" not in payload
+    assert "blend" not in payload
+    assert cheaper_on_series(bad, good, "measured_estimated") is None
+    assert cheaper_on_series(bad, good, "measured") is None
+    assert cheaper_on_series(bad, good, "euro") is False
+
+
+def test_effectiveness_http_endpoint(tmp_path: Path):
+    ledger = tmp_path / "ledger.db"
+    source = tmp_path / "july.csv"
+    _write_plan(source, period="2026-07", amount="200.00")
+    import_csv(source, "ai-plan", ledger_path=ledger)
+    ingest_session_token_evidence(
+        _token_snapshot(
+            record_id="tokens:2026-07",
+            period="2026-07",
+            measured=200,
+            estimated=0,
+            work=2,
+        ),
+        ledger_path=ledger,
+    )
+    app = create_app()
+    app.include_router(
+        create_effectiveness_router(ledger_path_resolver=lambda _: ledger),
+        prefix="/test",
+    )
+    client = TestClient(app)
+    response = client.get("/test/effectiveness?period=2026-07")
+    assert response.status_code == 200
+    body = response.json()
+    assert body[0]["trend_fit"] == "comparable"
+    assert body[0]["work_per_euro"] == "0.010000"
+    assert "efficiency" not in body[0]

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

@@ -309,7 +309,7 @@ as `AllocationEvidence` and does not change booked totals.
 
 ```task
 id: FIN-WP-0007-T06
-status: todo
+status: done
 priority: medium
 state_hub_task_id: "cea14702-faca-4024-b6f8-1ad221b70624"
 ```
@@ -334,6 +334,12 @@ Done when two fixture months with different coverage cannot produce a
 chart in which the worse-measured month looks cheaper per unit of
 work unless that is also true on the measured-only series.
 
+Completed 2026-08-15: `ledger effectiveness` and `GET /effectiveness`
+publish the three series and mark coverage below 0.50 as `unfit`.
+Implied €/token is labelled inferred. A low-coverage month cannot be
+ranked cheaper on a token series; there is no blended efficiency
+field.
+
 ## Send canon feedback
 
 ```task