"""Lightweight decision ledger for NovaPlan MVP. This module provides a tiny, auditable ledger to record PlanDelta events and governance actions. It is intentionally simple but deterministic and easy to extend for anchoring to ground links in the future. """ from __future__ import annotations from typing import List from dataclasses import dataclass, field import time from .contracts import AuditLog @dataclass class LedgerEntry: """Single ledger entry representing a governance decision or delta.""" id: str contract_id: str payload: dict timestamp: float = field(default_factory=lambda: time.time()) signer: str | None = None anchor: str | None = None class Ledger: """In-memory ledger for NovaPlan MVP. Provides append and query capabilities. In a production system this would be backed by a durable store and possibly anchored to ground links. """ def __init__(self) -> None: self._entries: List[LedgerEntry] = [] def append(self, contract_id: str, payload: dict, signer: str | None = None, anchor: str | None = None) -> LedgerEntry: entry = LedgerEntry( id=f"entry-{len(self._entries)+1}", contract_id=contract_id, payload=payload, signer=signer, anchor=anchor, ) self._entries.append(entry) return entry def last(self) -> LedgerEntry | None: return self._entries[-1] if self._entries else None def entries(self) -> List[LedgerEntry]: return list(self._entries) # Convenience: append an AuditLog entry (legacy API) or 2-arg style def log(self, audit_or_entry, contract_id: str | None = None, anchor: str | None = None) -> LedgerEntry: # Legacy path: audit is an AuditLog instance and contract_id is provided via audit if isinstance(audit_or_entry, AuditLog) and contract_id is None: audit: AuditLog = audit_or_entry payload = { "entry": audit.entry, "signer": audit.signer, "timestamp": audit.timestamp, "contract_id": audit.contract_id, } return self.append(audit.contract_id, payload, signer=audit.signer) # New path: direct entry with contract_id and optional anchor if contract_id is None: raise TypeError("contract_id must be provided when calling log with (entry, contract_id, anchor=...) signature") payload = { "entry": audit_or_entry, "anchor": anchor, "timestamp": time.time(), } return self.append(contract_id, payload, signer=None, anchor=anchor) # New API: allow tagging an anchor to an entry for ground-link anchoring def log_with_anchor(self, entry: str, contract_id: str, anchor: str | None = None) -> LedgerEntry: payload = { "entry": entry, "anchor": anchor, "timestamp": time.time(), } return self.append(contract_id, payload, signer=None, anchor=anchor) def last_anchor(self) -> str | None: if not self._entries: return None # Return the anchor of the most recent entry if present return self._entries[-1].anchor