feat(fase1): fail-closed expert realize gate (F1)

This commit is contained in:
Kjell Tore Guttormsen 2026-07-07 08:00:23 +02:00
commit 9720acb18c
2 changed files with 167 additions and 0 deletions

View file

@ -17,10 +17,19 @@ from __future__ import annotations
import hashlib
import json
from decimal import ROUND_HALF_UP, Decimal
from pathlib import Path
from pydantic import BaseModel, Field
from portfolio_optimiser.verdicts import _APPROVED_DECISIONS, ProposalFeatures, Verdict
class RealizationRefused(RuntimeError):
"""Fail-closed gate (SC5, mirrors ``verdicts.PromotionRefused``): a non-approved verdict was
offered for realization. Nothing is written only human/persona-approved savings enter the
ledger, never raw agent output (self-contamination)."""
class LedgerEntry(BaseModel):
"""One realized saving, linked to the approving verdict."""
@ -151,3 +160,57 @@ class SavingsLedger(BaseModel):
raise FileNotFoundError(f"savings ledger not found: {path!r}")
rows = json.loads(p.read_text(encoding="utf-8"))
return cls(entries=[LedgerEntry(**row) for row in rows])
def realize(
ledger: SavingsLedger,
features: ProposalFeatures,
verdict: Verdict,
*,
project_id: str,
dimension: str,
approver: str,
experiment: str,
timestamp: str,
) -> LedgerEntry:
"""Realize an APPROVED candidate into ``ledger`` and return the entry (SC5).
FAIL-CLOSED: a verdict whose ``decision`` is not an approval raises ``RealizationRefused`` and
writes NOTHING only human/persona-approved savings enter the ledger (mirrors
``promote_verdict``). The approval set is the PROMOTION set ``{approved,
approved_with_adjustment}``, NOT the run-path binary ``FeedbackContract`` (H6).
Deliberately NOT wired into ``run_project`` (role split C3): the system READS context; the
expert/persona realizes out of band mirroring how ``promote_verdict`` is never called in the
run path (self-contamination guard).
``project_id`` and ``dimension`` are required keywords: a ``LedgerEntry`` is scoped to a project
and a dimension, and neither ``features`` nor ``verdict`` carries them.
NOK->øre conversion happens HERE and only here, via ``Decimal`` to avoid binary-float error:
``12345.67`` NOK -> ``1234567`` øre exactly (a raw ``float * 100`` would drift to ...66.9999).
``timestamp`` is a required keyword (no wall-clock default), so the entry is deterministic."""
if verdict.decision not in _APPROVED_DECISIONS:
raise RealizationRefused(
f"refusing to realize a non-approved verdict (decision={verdict.decision!r}); "
"only human/persona-approved savings enter the ledger (SC5)"
)
amount_ore = int(
(Decimal(str(features.claimed_saving_nok)) * 100).quantize(
Decimal("1"), rounding=ROUND_HALF_UP
)
)
entry = LedgerEntry(
project_id=project_id,
dimension=dimension,
candidate_identity=_candidate_identity(
affected_codes=features.affected_codes,
measure_type=features.measure_type,
amount_ore=amount_ore,
),
amount_ore=amount_ore,
verdict_id=verdict.id,
provenance=stamp(approver=approver, experiment=experiment, timestamp=timestamp),
)
ledger.add_realized(entry)
return entry

View file

@ -15,12 +15,17 @@ Pattern: ``tests/test_step8_promotion_loadbearing.py:77`` (fail-closed gate trio
from __future__ import annotations
import pytest
from portfolio_optimiser.ledger import (
LedgerEntry,
RealizationRefused,
SavingsLedger,
_candidate_identity,
realize,
stamp,
)
from portfolio_optimiser.verdicts import ProposalFeatures, Verdict
_TS = "2026-07-06T00:00:00Z"
_CODES = frozenset({"ENERGI-TOTAL-EL"})
@ -76,3 +81,102 @@ def test_exact_full_key_duplicate_is_not_stored_twice() -> None:
assert ledger.add_realized(_entry("energi", amount_ore=5000)) is True
assert ledger.add_realized(_entry("energi", amount_ore=5000)) is False # exact duplicate
assert ledger.portfolio_total() == 5000
# --- Step 6: fail-closed realize gate + deterministic øre conversion (SC5/SC8) --------------------
def _features(nok: float = 5000.0) -> ProposalFeatures:
return ProposalFeatures(affected_codes=_CODES, measure_type=_MEASURE, claimed_saving_nok=nok)
def _verdict(decision: str) -> Verdict:
return Verdict(
id="v-abc", proposal_features=_features(), decision=decision, rationale="expert prose"
)
def test_realize_fail_closed_refuses_non_approved() -> None:
"""LOAD-BEARING (SC5): a non-approved verdict — OTHERWISE fully valid (correct amount, valid
provenance, so detaching the gate WOULD add it) raises ``RealizationRefused`` and the ledger is
unchanged. RED the moment the ``raise`` detaches: the raw entry then appears (self-contamination)."""
ledger = SavingsLedger()
with pytest.raises(RealizationRefused):
realize(
ledger,
_features(),
_verdict("rejected"),
project_id="P1",
dimension="energi",
approver="ekspert",
experiment="sim",
timestamp=_TS,
)
assert ledger.entries == [] # nothing written
assert ledger.portfolio_total() == 0
def test_realize_approved_reaches_ledger() -> None:
"""CAUSALITY: a HITL-approved verdict reaches the ledger (5000 NOK -> 500000 øre)."""
ledger = SavingsLedger()
entry = realize(
ledger,
_features(),
_verdict("approved"),
project_id="P1",
dimension="energi",
approver="ekspert",
experiment="sim",
timestamp=_TS,
)
assert entry in ledger.entries
assert entry.verdict_id == "v-abc"
assert ledger.portfolio_total() == 500000
def test_realize_accepts_approved_with_adjustment() -> None:
"""The gate uses the PROMOTION set: ``approved_with_adjustment`` is admitted (not the binary
run-path FeedbackContract H6)."""
ledger = SavingsLedger()
realize(
ledger,
_features(),
_verdict("approved_with_adjustment"),
project_id="P1",
dimension="energi",
approver="ekspert",
experiment="sim",
timestamp=_TS,
)
assert ledger.portfolio_total() == 500000
def test_realize_requires_timestamp_keyword() -> None:
"""SC8: ``timestamp`` is a required keyword — no wall-clock default. Omitting it raises TypeError."""
ledger = SavingsLedger()
with pytest.raises(TypeError):
realize(
ledger,
_features(),
_verdict("approved"),
project_id="P1",
dimension="energi",
approver="ekspert",
experiment="sim",
) # no timestamp
def test_realize_ore_conversion_is_exact() -> None:
"""øre boundary: 12345.67 NOK -> 1234567 øre exactly, no binary-float ...66.9999 drift."""
ledger = SavingsLedger()
entry = realize(
ledger,
_features(12345.67),
_verdict("approved"),
project_id="P1",
dimension="energi",
approver="ekspert",
experiment="sim",
timestamp=_TS,
)
assert entry.amount_ore == 1234567