feat: ingest State Hub session tokens without booking them

Store mixed measured/estimated/superseded aggregates as session-token
evidence. Map kinds onto the evidence-basis vocabulary, keep unknown
months unknown, and refuse to post the snapshot as booked spend or
resource-control usage.

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

@@ -29,6 +29,8 @@ 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 ingest-session-tokens tests/fixtures/session-tokens-2026-08.json
+uv run finhub ledger session-tokens --period 2026-08
 uv run finhub ledger summary
 
 # CLI — scheduled evaluation (cron/systemd friendly)

README.md

@@ -26,6 +26,8 @@ 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 ingest-session-tokens tests/fixtures/session-tokens-2026-08.json
+uv run finhub ledger session-tokens --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

src/fin_hub/cli.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import argparse
 import json
 import sys
+from datetime import date
 from pathlib import Path
 
 from fin_hub.coupling.canon import emit_viability_alert
@@ -18,6 +19,11 @@ 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.evaluate import evaluate_runway
+from fin_hub.services.session_tokens import (
+    ingest_session_token_file,
+    list_current_session_token_evidence,
+    summarize_session_token_evidence,
+)
 from fin_hub.services.evidence import write_runway_evidence
 from fin_hub.services.ledger import (
     client_margin_report,
@@ -204,6 +210,27 @@ def _cmd_ledger_plan_month(args: argparse.Namespace) -> int:
     return 0
 
 
+def _cmd_ledger_ingest_session_tokens(args: argparse.Namespace) -> int:
+    record = ingest_session_token_file(Path(args.path), ledger_path=_ledger_path(args))
+    print(json.dumps(summarize_session_token_evidence(record).as_dict(), indent=2))
+    return 0
+
+
+def _cmd_ledger_session_tokens(args: argparse.Namespace) -> int:
+    period_start = date.fromisoformat(f"{args.period}-01") if args.period else None
+    records = list_current_session_token_evidence(
+        ledger_path=_ledger_path(args),
+        period_start=period_start,
+    )
+    print(
+        json.dumps(
+            [summarize_session_token_evidence(record).as_dict() for record in records],
+            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))
@@ -380,6 +407,22 @@ def build_parser() -> argparse.ArgumentParser:
     ledger_plan_month.add_argument("--ledger", help="Ledger database path")
     ledger_plan_month.set_defaults(func=_cmd_ledger_plan_month)
 
+    ledger_ingest_tokens = ledger_sub.add_parser(
+        "ingest-session-tokens",
+        help="Ingest a State Hub token-aggregate snapshot (not booked spend)",
+    )
+    ledger_ingest_tokens.add_argument("path")
+    ledger_ingest_tokens.add_argument("--ledger", help="Ledger database path")
+    ledger_ingest_tokens.set_defaults(func=_cmd_ledger_ingest_session_tokens)
+
+    ledger_session_tokens = ledger_sub.add_parser(
+        "session-tokens",
+        help="Summarize current session-token evidence by measurement basis",
+    )
+    ledger_session_tokens.add_argument("--period", help="Period start month in YYYY-MM")
+    ledger_session_tokens.add_argument("--ledger", help="Ledger database path")
+    ledger_session_tokens.set_defaults(func=_cmd_ledger_session_tokens)
+
     ledger_allocations = ledger_sub.add_parser(
         "allocations",
         help="Reconcile resource-control allocation evidence to booked facts",

src/fin_hub/schemas/session_tokens.py

@@ -0,0 +1,87 @@
+"""Session-token evidence ingested from State Hub aggregates."""
+
+from __future__ import annotations
+
+from datetime import date, datetime
+from decimal import Decimal
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
+
+MEASUREMENT_KINDS = ("measured", "allocated", "estimated", "superseded")
+EVIDENCE_BASIS = {
+    "measured": "measured",
+    "allocated": "derived",
+    "estimated": "estimated",
+    "superseded": "excluded",
+}
+
+
+class SessionTokenSlice(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    tokens_in: int = Field(ge=0)
+    tokens_out: int = Field(ge=0)
+    event_count: int = Field(ge=0)
+    confidence: Decimal | None = None
+    method: str | None = None
+
+    @field_validator("confidence")
+    @classmethod
+    def validate_confidence(cls, value: Decimal | None) -> Decimal | None:
+        if value is None:
+            return None
+        if not value.is_finite() or value < 0 or value > 1:
+            raise ValueError("confidence must be between 0 and 1")
+        return value
+
+
+class SessionTokenAssociation(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    scope: Literal["repo", "workplan", "model"]
+    scope_id: str = Field(min_length=1, max_length=256)
+    label: str | None = None
+    measurement_kind: Literal["measured", "allocated", "estimated", "superseded"]
+    tokens_in: int = Field(ge=0)
+    tokens_out: int = Field(ge=0)
+    event_count: int = Field(ge=0, default=1)
+
+
+class SessionTokenEvidence(BaseModel):
+    """State Hub session-token snapshot. Not booked spend and not usage_observation."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    schema_version: Literal["0.1"] = "0.1"
+    record_type: Literal["session_token_evidence"] = "session_token_evidence"
+    record_id: str = Field(min_length=1, max_length=256)
+    revision_of: str | None = None
+    source_system: Literal["state-hub"] = "state-hub"
+    period_start: date
+    period_end: date
+    captured_at: datetime
+    source_evidence: list[str] = Field(min_length=1)
+    totals: dict[str, SessionTokenSlice] = Field(default_factory=dict)
+    associations: list[SessionTokenAssociation] = Field(default_factory=list)
+
+    @field_validator("totals")
+    @classmethod
+    def validate_totals(cls, value: dict[str, SessionTokenSlice]) -> dict[str, SessionTokenSlice]:
+        unknown = set(value) - set(MEASUREMENT_KINDS)
+        if unknown:
+            raise ValueError(f"unknown measurement_kind: {', '.join(sorted(unknown))}")
+        return value
+
+    @model_validator(mode="after")
+    def validate_period_and_methods(self) -> "SessionTokenEvidence":
+        if self.period_end < self.period_start:
+            raise ValueError("period_end cannot precede period_start")
+        allocated = self.totals.get("allocated")
+        if (
+            allocated is not None
+            and allocated.event_count > 0
+            and not (allocated.method and allocated.method.strip())
+        ):
+            raise ValueError("allocated totals with events require method")
+        return self

src/fin_hub/services/exchange.py

@@ -232,6 +232,10 @@ def ingest_planning_evidence(
 ) -> PlanningEvidence:
     """Validate and idempotently retain planning evidence outside booked spend."""
 
+    if payload.get("record_type") == "session_token_evidence":
+        raise ValueError(
+            "session_token_evidence is not resource-control planning evidence"
+        )
     ledger = ledger_path or default_ledger_path()
     with _connect(ledger) as conn:
         _ensure_planning_schema(conn)

src/fin_hub/services/ledger.py

@@ -275,6 +275,17 @@ def _ensure_schema(conn: sqlite3.Connection) -> None:
             recorded_at TEXT NOT NULL
         );
 
+        CREATE TABLE IF NOT EXISTS session_token_evidence (
+            record_id TEXT PRIMARY KEY,
+            revision_of TEXT,
+            period_start TEXT NOT NULL,
+            period_end TEXT NOT NULL,
+            source_system TEXT NOT NULL,
+            payload_json TEXT NOT NULL,
+            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)")}

src/fin_hub/services/session_tokens.py

@@ -0,0 +1,239 @@
+"""Store State Hub session-token aggregates outside booked spend and RC usage."""
+
+from __future__ import annotations
+
+import json
+from dataclasses import asdict, dataclass
+from datetime import date
+from decimal import Decimal
+from pathlib import Path
+
+from pydantic import ValidationError
+
+from fin_hub.schemas.session_tokens import EVIDENCE_BASIS, SessionTokenEvidence
+from fin_hub.services.ledger import _connect, _utc_now, default_ledger_path
+
+
+def _ensure_session_token_schema(conn) -> None:
+    conn.execute(
+        """
+        CREATE TABLE IF NOT EXISTS session_token_evidence (
+            record_id TEXT PRIMARY KEY,
+            revision_of TEXT,
+            period_start TEXT NOT NULL,
+            period_end TEXT NOT NULL,
+            source_system TEXT NOT NULL,
+            payload_json TEXT NOT NULL,
+            is_current INTEGER NOT NULL DEFAULT 1,
+            recorded_at TEXT NOT NULL
+        )
+        """
+    )
+    conn.commit()
+
+
+def ingest_session_token_evidence(
+    payload: dict,
+    *,
+    ledger_path: Path | None = None,
+) -> SessionTokenEvidence:
+    """Retain a State Hub aggregate snapshot. Never a booked fact or RC usage row."""
+
+    if payload.get("record_type") == "usage_observation":
+        raise ValueError(
+            "usage_observation is resource-control planning evidence; "
+            "session tokens use ingest_session_token_evidence"
+        )
+    if payload.get("record_type") == "booked_cost":
+        raise ValueError("session token evidence cannot be posted as booked spend")
+    try:
+        record = SessionTokenEvidence.model_validate(payload)
+    except ValidationError as error:
+        raise ValueError(error.errors(include_url=False, include_input=False)) from error
+    ledger = ledger_path or default_ledger_path()
+    canonical = record.model_dump_json()
+    with _connect(ledger) as conn:
+        _ensure_session_token_schema(conn)
+        conn.execute("BEGIN IMMEDIATE")
+        existing = conn.execute(
+            "SELECT payload_json FROM session_token_evidence WHERE record_id = ?",
+            (record.record_id,),
+        ).fetchone()
+        if existing is not None:
+            if json.loads(existing["payload_json"]) != json.loads(canonical):
+                raise ValueError("record_id already exists with different content")
+            return record
+        if record.revision_of is not None:
+            predecessor = conn.execute(
+                "SELECT is_current FROM session_token_evidence WHERE record_id = ?",
+                (record.revision_of,),
+            ).fetchone()
+            if predecessor is None:
+                raise ValueError("revision_of must reference an existing session-token record")
+            if predecessor["is_current"] != 1:
+                raise ValueError("revision_of must reference the current record")
+            conn.execute(
+                "UPDATE session_token_evidence SET is_current = 0 WHERE record_id = ?",
+                (record.revision_of,),
+            )
+        conn.execute(
+            """
+            INSERT INTO session_token_evidence (
+                record_id, revision_of, period_start, period_end, source_system,
+                payload_json, is_current, recorded_at
+            ) VALUES (?, ?, ?, ?, ?, ?, 1, ?)
+            """,
+            (
+                record.record_id,
+                record.revision_of,
+                record.period_start.isoformat(),
+                record.period_end.isoformat(),
+                record.source_system,
+                canonical,
+                _utc_now(),
+            ),
+        )
+        conn.commit()
+    return record
+
+
+def list_current_session_token_evidence(
+    *,
+    ledger_path: Path | None = None,
+    period_start: date | None = None,
+    period_end: date | None = None,
+) -> list[SessionTokenEvidence]:
+    ledger = ledger_path or default_ledger_path()
+    query = "SELECT payload_json FROM session_token_evidence WHERE is_current = 1"
+    params: list[object] = []
+    if period_start is not None:
+        query += " AND period_start = ?"
+        params.append(period_start.isoformat())
+    if period_end is not None:
+        query += " AND period_end = ?"
+        params.append(period_end.isoformat())
+    query += " ORDER BY period_start, record_id"
+    with _connect(ledger) as conn:
+        _ensure_session_token_schema(conn)
+        rows = conn.execute(query, params).fetchall()
+    return [SessionTokenEvidence.model_validate_json(row["payload_json"]) for row in rows]
+
+
+@dataclass(frozen=True)
+class SessionTokenKindTotal:
+    basis: str
+    tokens_in: int | None
+    tokens_out: int | None
+    event_count: int
+    confidence: Decimal | None
+    method: str | None
+
+    def as_dict(self) -> dict:
+        payload = asdict(self)
+        payload["confidence"] = None if self.confidence is None else str(self.confidence)
+        return payload
+
+
+@dataclass(frozen=True)
+class SessionTokenSummary:
+    record_id: str
+    period_start: date
+    period_end: date
+    source_system: str
+    unknown_residual: bool
+    by_basis: dict[str, SessionTokenKindTotal]
+    tokens_measured: int | None
+    tokens_derived: int | None
+    tokens_estimated: int | None
+    coverage: Decimal | None
+
+    def as_dict(self) -> dict:
+        return {
+            "record_id": self.record_id,
+            "period_start": self.period_start.isoformat(),
+            "period_end": self.period_end.isoformat(),
+            "source_system": self.source_system,
+            "unknown_residual": self.unknown_residual,
+            "by_basis": {key: value.as_dict() for key, value in self.by_basis.items()},
+            "tokens_measured": self.tokens_measured,
+            "tokens_derived": self.tokens_derived,
+            "tokens_estimated": self.tokens_estimated,
+            "coverage": None if self.coverage is None else str(self.coverage),
+        }
+
+
+def _slice_total(record: SessionTokenEvidence, kind: str) -> SessionTokenKindTotal | None:
+    slice_ = record.totals.get(kind)
+    if slice_ is None or slice_.event_count == 0:
+        return None
+    return SessionTokenKindTotal(
+        basis=EVIDENCE_BASIS[kind],
+        tokens_in=slice_.tokens_in,
+        tokens_out=slice_.tokens_out,
+        event_count=slice_.event_count,
+        confidence=slice_.confidence,
+        method=slice_.method,
+    )
+
+
+def summarize_session_token_evidence(record: SessionTokenEvidence) -> SessionTokenSummary:
+    by_basis: dict[str, SessionTokenKindTotal] = {}
+    for kind in ("measured", "allocated", "estimated"):
+        total = _slice_total(record, kind)
+        if total is not None:
+            by_basis[total.basis] = total
+    measured = by_basis.get("measured")
+    derived = by_basis.get("derived")
+    estimated = by_basis.get("estimated")
+    unknown = not by_basis
+
+    def _tokens(total: SessionTokenKindTotal | None) -> int | None:
+        if total is None:
+            return None
+        return total.tokens_in + total.tokens_out
+
+    tokens_measured = _tokens(measured)
+    tokens_derived = _tokens(derived)
+    tokens_estimated = _tokens(estimated)
+    known = [
+        value
+        for value in (tokens_measured, tokens_derived, tokens_estimated)
+        if value is not None
+    ]
+    coverage = None
+    if known:
+        denominator = sum(known)
+        coverage = (
+            Decimal(tokens_measured or 0) / Decimal(denominator)
+            if denominator
+            else Decimal("0")
+        )
+    return SessionTokenSummary(
+        record_id=record.record_id,
+        period_start=record.period_start,
+        period_end=record.period_end,
+        source_system=record.source_system,
+        unknown_residual=unknown,
+        by_basis=by_basis,
+        tokens_measured=tokens_measured,
+        tokens_derived=tokens_derived,
+        tokens_estimated=tokens_estimated,
+        coverage=coverage,
+    )
+
+
+def ingest_session_token_file(path: Path, *, ledger_path: Path | None = None) -> SessionTokenEvidence:
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    if not isinstance(payload, dict):
+        raise ValueError("session token snapshot must be a JSON object")
+    return ingest_session_token_evidence(payload, ledger_path=ledger_path)
+
+
+__all__ = [
+    "SessionTokenKindTotal",
+    "SessionTokenSummary",
+    "ingest_session_token_evidence",
+    "ingest_session_token_file",
+    "list_current_session_token_evidence",
+    "summarize_session_token_evidence",
+]

tests/fixtures/session-tokens-2026-08.json

@@ -0,0 +1,60 @@
+{
+  "schema_version": "0.1",
+  "record_type": "session_token_evidence",
+  "record_id": "session-tokens:state-hub:2026-08",
+  "revision_of": null,
+  "source_system": "state-hub",
+  "period_start": "2026-08-01",
+  "period_end": "2026-08-31",
+  "captured_at": "2026-08-15T12:00:00Z",
+  "source_evidence": [
+    "state-hub:/token-events/aggregate/?month=2026-08"
+  ],
+  "totals": {
+    "measured": {
+      "tokens_in": 100,
+      "tokens_out": 50,
+      "event_count": 2,
+      "confidence": "1.0"
+    },
+    "allocated": {
+      "tokens_in": 20,
+      "tokens_out": 10,
+      "event_count": 1,
+      "confidence": "0.70",
+      "method": "state-hub-task-share"
+    },
+    "estimated": {
+      "tokens_in": 1000,
+      "tokens_out": 500,
+      "event_count": 3,
+      "confidence": "0.35"
+    },
+    "superseded": {
+      "tokens_in": 800,
+      "tokens_out": 200,
+      "event_count": 1,
+      "confidence": "0.0"
+    }
+  },
+  "associations": [
+    {
+      "scope": "repo",
+      "scope_id": "cce28433-34ce-4ab1-9ab4-4dac9222c802",
+      "label": "fin-hub",
+      "measurement_kind": "measured",
+      "tokens_in": 100,
+      "tokens_out": 50,
+      "event_count": 2
+    },
+    {
+      "scope": "workplan",
+      "scope_id": "88de5e03-a53f-4b62-a0a7-1365756da2d7",
+      "label": "FIN-WP-0007",
+      "measurement_kind": "estimated",
+      "tokens_in": 1000,
+      "tokens_out": 500,
+      "event_count": 3
+    }
+  ]
+}

tests/test_session_tokens.py

@@ -0,0 +1,121 @@
+import json
+from pathlib import Path
+
+import pytest
+
+from fin_hub.services.exchange import ingest_planning_evidence
+from fin_hub.services.ledger import import_csv, monthly_summary
+from fin_hub.services.session_tokens import (
+    ingest_session_token_evidence,
+    ingest_session_token_file,
+    list_current_session_token_evidence,
+    summarize_session_token_evidence,
+)
+
+FIXTURES = Path(__file__).parent / "fixtures"
+
+
+def test_ingest_mixed_aggregates_and_summarize(tmp_path: Path):
+    ledger = tmp_path / "ledger.db"
+    record = ingest_session_token_file(
+        FIXTURES / "session-tokens-2026-08.json", ledger_path=ledger
+    )
+    again = ingest_session_token_file(
+        FIXTURES / "session-tokens-2026-08.json", ledger_path=ledger
+    )
+    assert again.record_id == record.record_id
+    summary = summarize_session_token_evidence(record)
+    assert summary.unknown_residual is False
+    assert summary.tokens_measured == 150
+    assert summary.tokens_derived == 30
+    assert summary.tokens_estimated == 1500
+    assert "excluded" not in summary.by_basis
+    assert set(summary.by_basis) == {"measured", "derived", "estimated"}
+    assert summary.by_basis["derived"].method == "state-hub-task-share"
+    stored = list_current_session_token_evidence(ledger_path=ledger)
+    assert [item.record_id for item in stored] == [record.record_id]
+    assert monthly_summary(ledger_path=ledger) == []
+
+
+def test_empty_month_is_unknown_residual_not_zero(tmp_path: Path):
+    ledger = tmp_path / "ledger.db"
+    record = ingest_session_token_evidence(
+        {
+            "record_id": "session-tokens:state-hub:2026-07",
+            "period_start": "2026-07-01",
+            "period_end": "2026-07-31",
+            "captured_at": "2026-08-15T12:00:00Z",
+            "source_evidence": ["state-hub:/token-events/aggregate/?month=2026-07"],
+            "totals": {
+                "superseded": {
+                    "tokens_in": 9,
+                    "tokens_out": 1,
+                    "event_count": 1,
+                    "confidence": "0",
+                }
+            },
+        },
+        ledger_path=ledger,
+    )
+    summary = summarize_session_token_evidence(record)
+    assert summary.unknown_residual is True
+    assert summary.tokens_measured is None
+    assert summary.tokens_derived is None
+    assert summary.tokens_estimated is None
+    assert summary.coverage is None
+
+
+def test_allocated_events_require_a_method():
+    with pytest.raises(ValueError, match="method"):
+        ingest_session_token_evidence(
+            {
+                "record_id": "session-tokens:bad",
+                "period_start": "2026-08-01",
+                "period_end": "2026-08-31",
+                "captured_at": "2026-08-15T12:00:00Z",
+                "source_evidence": ["state-hub"],
+                "totals": {
+                    "allocated": {
+                        "tokens_in": 1,
+                        "tokens_out": 1,
+                        "event_count": 1,
+                    }
+                },
+            },
+            ledger_path=Path("/tmp/unused"),
+        )
+
+
+def test_session_tokens_cannot_be_posted_as_planning_usage(tmp_path: Path):
+    ledger = tmp_path / "ledger.db"
+    payload = json.loads((FIXTURES / "session-tokens-2026-08.json").read_text())
+    with pytest.raises(ValueError, match="not resource-control"):
+        ingest_planning_evidence(payload, ledger_path=ledger)
+    with pytest.raises(ValueError, match="usage_observation"):
+        ingest_session_token_evidence(
+            {
+                "record_type": "usage_observation",
+                "record_id": "usage:1",
+                "period_start": "2026-08-01",
+                "period_end": "2026-08-31",
+                "resource_id": "resource:invented",
+                "source_evidence": ["nope"],
+                "created_at": "2026-08-15T12:00:00Z",
+                "measures": [{"name": "tokens", "value": "1", "unit": "token"}],
+            },
+            ledger_path=ledger,
+        )
+
+
+def test_session_tokens_cannot_be_imported_as_booked_spend(tmp_path: Path):
+    ledger = tmp_path / "ledger.db"
+    with pytest.raises(ValueError, match="Unknown source type"):
+        import_csv(FIXTURES / "session-tokens-2026-08.json", "session-tokens", ledger_path=ledger)
+    with pytest.raises(ValueError, match="booked spend"):
+        ingest_session_token_evidence(
+            {
+                "record_type": "booked_cost",
+                "financial_fact_id": "fact:1",
+            },
+            ledger_path=ledger,
+        )

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

@@ -237,7 +237,7 @@ matched series. No session logs are read.
 
 ```task
 id: FIN-WP-0007-T04
-status: todo
+status: done
 priority: high
 state_hub_task_id: "29f1804f-a5ad-46b6-b9d7-8f074e31b04e"
 ```
@@ -263,6 +263,14 @@ Done when a recorded fixture of mixed measured/estimated/superseded
 aggregates stores as session-token evidence, reconciles by kind, and
 cannot be posted as booked spend or as resource-control usage.
 
+Completed 2026-08-15: `ledger ingest-session-tokens` stores a State Hub
+aggregate snapshot in `session_token_evidence`. Kinds map to
+measured/derived/estimated; superseded is excluded. Empty months are
+`unknown_residual`, not zero. The snapshot cannot be posted as
+`booked_cost` or as resource-control `usage_observation`. Allocated
+slices require a method. No session files are parsed and no
+`resource_id` is minted.
+
 ## Reporting-allocate booked plan cost across work
 
 ```task