feat(ledger): K1 — savings ledger + goal contract (parity rows 2-3)

New ledger.py: typed SavingsLedger; realize is fail-closed on an APPROVED
FeedbackContract + a named expert + an explicit timestamp (the §6 determinism
rule — no wall-clock default). The sum key is DIMENSION-FREE (the dimension
label is annotation only and never participates in the mint), so the same
realized saving surfaced via two dimensions lands in one first-write-wins
slot and is never double-counted. Deterministic JSON persistence
(sort_keys, indent 2, LF, trailing newline), schema-validated on load.

New goals.py: GoalContract (absolute target, hard/soft, fail-fast §10).
A hard goal reached raises GoalReached, a structured stop event carrying
target + observed — never a silent stop; soft flags without stopping.
The percent-goal baseline is D-E-gated: the field is reserved and
construction refuses with an explicit NotImplementedError.

Semantics are marked STACK-LOCAL in the docstrings — mirrored from the MAF
plan's capability description, never from MAF code; format shareability
stays a proposed decision point in the brief.

Two detach proofs delivered (decision gate removed -> red; dimension into
the key mint -> the double-counting test red). 400 -> 426 tests; README
synced (test count + a Value layer module block).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Kjell Tore Guttormsen 2026-07-17 04:00:14 +02:00
commit 698e8f21dd
5 changed files with 445 additions and 2 deletions

View file

@ -0,0 +1,146 @@
"""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)."""
parsed = _LedgerFile(**json.loads(path.read_text(encoding="utf-8")))
ledger = cls()
for entry in parsed.entries:
ledger._entries.setdefault(entry.key, entry)
return ledger