diff --git a/.claude/rules/stack-and-commands.md b/.claude/rules/stack-and-commands.md index 2f25d11..d9ac00a 100644 --- a/.claude/rules/stack-and-commands.md +++ b/.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) diff --git a/README.md b/README.md index 8a70347..5a1e45f 100644 --- a/README.md +++ b/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 diff --git a/src/fin_hub/cli.py b/src/fin_hub/cli.py index 44d48ab..a411eb2 100644 --- a/src/fin_hub/cli.py +++ b/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", diff --git a/src/fin_hub/schemas/session_tokens.py b/src/fin_hub/schemas/session_tokens.py new file mode 100644 index 0000000..23a0699 --- /dev/null +++ b/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 diff --git a/src/fin_hub/services/exchange.py b/src/fin_hub/services/exchange.py index 1219a6f..90e5077 100644 --- a/src/fin_hub/services/exchange.py +++ b/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) diff --git a/src/fin_hub/services/ledger.py b/src/fin_hub/services/ledger.py index 9057517..8cdca2e 100644 --- a/src/fin_hub/services/ledger.py +++ b/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)")} diff --git a/src/fin_hub/services/session_tokens.py b/src/fin_hub/services/session_tokens.py new file mode 100644 index 0000000..65f2d47 --- /dev/null +++ b/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", +] diff --git a/tests/fixtures/session-tokens-2026-08.json b/tests/fixtures/session-tokens-2026-08.json new file mode 100644 index 0000000..0efb059 --- /dev/null +++ b/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 + } + ] +} diff --git a/tests/test_session_tokens.py b/tests/test_session_tokens.py new file mode 100644 index 0000000..9f9f6c3 --- /dev/null +++ b/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, + ) diff --git a/workplans/FIN-WP-0007-ai-plan-token-effectiveness.md b/workplans/FIN-WP-0007-ai-plan-token-effectiveness.md index 999a1e7..6b407eb 100644 --- a/workplans/FIN-WP-0007-ai-plan-token-effectiveness.md +++ b/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