portfolio-optimiser-claude/src/portfolio_optimiser_claude/ledger.py

"""The savings ledger (parity rows 2-3) — typed, fail-closed, dimension-free sum.

Realized savings enter the book ONLY through the expert gate: ``realize``
requires an explicit APPROVED expert verdict (the §4.1 binary run-path shape),
a named expert identity, and an explicit timestamp (no wall-clock default —
the same determinism rule as promotion §6). Entries are keyed DIMENSION-FREE
(project + candidate identity + amount; the dimension label is annotation
only), so the same realized saving surfaced via two dimensions lands in one
first-write-wins slot (§4.2-style idempotence) and is never double-counted.
Persistence is deterministic JSON: sort_keys, indent 2, LF, trailing newline.

Ledger semantics are a STACK-LOCAL contract mirrored from the MAF plan's
capability description — never from MAF code; detail semantics may diverge
(shareability of the format is a proposed decision point in the brief).
"""

from __future__ import annotations

import hashlib
import json
from collections.abc import Iterable
from pathlib import Path

from pydantic import BaseModel, ConfigDict, Field

from portfolio_optimiser_claude.contracts import FeedbackContract


class LedgerGateError(ValueError):
    """A realization was refused at the expert gate — nothing enters the book."""


class LedgerEntry(BaseModel):
    """One realized saving: dimension-free key, provenance-stamped (expert + timestamp)."""

    model_config = ConfigDict(frozen=True)

    key: str = Field(min_length=1)
    project: str = Field(min_length=1)
    measure_type: str = Field(min_length=1)
    affected_codes: tuple[str, ...]
    amount_nok: float = Field(gt=0, allow_inf_nan=False)
    expert: str = Field(min_length=1)
    timestamp: str = Field(min_length=1)
    dimension: str | None = None


class _LedgerFile(BaseModel):
    """The persisted ledger shape — schema-validated on load, fail-fast (§10-style)."""

    entries: list[LedgerEntry]


def _mint_entry_key(
    project: str, measure_type: str, affected_codes: tuple[str, ...], amount_nok: float
) -> str:
    # DIMENSION-FREE by construction: the dimension label never participates,
    # so two dimensions surfacing the same realized saving mint the same key.
    canonical = json.dumps(
        {
            "affected_codes": sorted(affected_codes),
            "amount_nok": amount_nok,
            "measure_type": measure_type,
            "project": project,
        },
        sort_keys=True,
        separators=(",", ":"),
    )
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:16]


class SavingsLedger:
    """The typed book of realized savings — FIRST-write-wins per dimension-free key."""

    def __init__(self) -> None:
        self._entries: dict[str, LedgerEntry] = {}

    def __len__(self) -> int:
        return len(self._entries)

    def realize(
        self,
        *,
        project: str,
        measure_type: str,
        affected_codes: Iterable[str],
        amount_nok: float,
        verdict: FeedbackContract,
        expert: str,
        timestamp: str,
        dimension: str | None = None,
    ) -> LedgerEntry:
        """Enter one realized saving — fail-closed on anything short of expert approval.

        ``expert`` and ``timestamp`` are explicit REQUIRED arguments (no
        wall-clock default, §6-style determinism). Re-realizing the same
        dimension-free key returns the FIRST entry unchanged (idempotent).
        """
        if verdict.decision != "approved":
            raise LedgerGateError(
                f"realization refused: verdict decision {verdict.decision!r} is not "
                "'approved' — only expert-approved savings enter the book (fail-closed)"
            )
        if not expert.strip():
            raise LedgerGateError(
                "realization refused: expert identity is blank — realized savings "
                "require a named expert (fail-closed)"
            )
        codes = tuple(sorted(affected_codes))
        entry = LedgerEntry(
            key=_mint_entry_key(project, measure_type, codes, amount_nok),
            project=project,
            measure_type=measure_type,
            affected_codes=codes,
            amount_nok=amount_nok,
            expert=expert,
            timestamp=timestamp,
            dimension=dimension,
        )
        return self._entries.setdefault(entry.key, entry)

    def entries(self) -> list[LedgerEntry]:
        """All entries, ordered by key (deterministic)."""
        return sorted(self._entries.values(), key=lambda entry: entry.key)

    def total_realized_nok(self) -> float:
        """The dimension-free sum: one addend per distinct key, never double-counted."""
        return sum(entry.amount_nok for entry in self._entries.values())

    def to_json(self) -> str:
        """Deterministic JSON: sort_keys, indent 2, trailing newline."""
        payload = {"entries": [entry.model_dump() for entry in self.entries()]}
        return json.dumps(payload, sort_keys=True, indent=2, ensure_ascii=False) + "\n"

    def save(self, path: Path) -> None:
        """Persist deterministically (LF only) — identical books yield identical bytes."""
        path.write_text(self.to_json(), encoding="utf-8", newline="\n")

    @classmethod
    def load(cls, path: Path) -> SavingsLedger:
        """Load a persisted ledger, schema-validated fail-fast (§10-style).

        ONE failure type for the whole boundary: unparsable bytes, a non-object
        top level (an array or a bare string would otherwise surface as a raw
        ``TypeError`` from ``**``) and a wrong-shaped object all raise
        ``ValueError``. The normalization lives HERE, at the public entrance,
        not in one caller — a book is refused the same way whoever opens it.
        """
        payload = json.loads(path.read_text(encoding="utf-8"))
        if not isinstance(payload, dict):
            raise ValueError(
                f"ledger {path} is not a JSON object: top level is "
                f"{type(payload).__name__} — a book is an object with 'entries'"
            )
        parsed = _LedgerFile(**payload)
        ledger = cls()
        for entry in parsed.entries:
            ledger._entries.setdefault(entry.key, entry)
        return ledger