91 lines
3.2 KiB
Python
91 lines
3.2 KiB
Python
"""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
|