spb/air Public MIT

1# =============================================================================2# Projet : AIR — Accounting Intermediate Representation3# Auteur : Simon-Pierre Boucher4# Contact : contact@spboucher.ai5# Fichier : provenance.py6# Description : Immutable provenance graph — accounting SSA: every amount has one traceable origin.7# =============================================================================8"""Provenance graph: the accounting analogue of SSA form.910In LLVM's SSA form every value is defined exactly once and every use points11back to its unique definition. AIR transposes this to money: every amount that12appears anywhere in a compilation (an invoice line, a tax amount, a converted13FX amount, a posted journal line) is a *node* defined exactly once, and every14derived amount points to the node(s) it was computed from, together with the15operation that produced it (e.g. "tax:GST@0.05", "fx:USD->CAD@1.3500").1617The graph is append-only and nodes are immutable: corrections never rewrite18history, they add new nodes (mirroring reversal entries in the ledger).19"""20from __future__ import annotations2122import itertools23from dataclasses import dataclass, field24from typing import Iterator2526from core.money import Money272829@dataclass(frozen=True, slots=True)30class ProvenanceNode:31    """A single, immutable definition of an amount."""3233    id: str34    kind: str                       # e.g. "invoice_line", "tax", "fx", "journal_line"35    operation: str                  # human/machine readable derivation, e.g. "tax:GST@0.05"36    amount: Money37    inputs: tuple[str, ...] = ()    # ids of the nodes this amount was derived from38    source_ref: str | None = None   # external anchor: event id, document URI, ...394041class ProvenanceError(KeyError):42    """Raised when a node id is unknown or redefined (SSA violation)."""434445@dataclass46class ProvenanceGraph:47    """Append-only DAG of amount definitions."""4849    _nodes: dict[str, ProvenanceNode] = field(default_factory=dict)50    _counter: itertools.count = field(default_factory=itertools.count)5152    def define(53        self,54        kind: str,55        operation: str,56        amount: Money,57        inputs: tuple[str, ...] = (),58        source_ref: str | None = None,59    ) -> ProvenanceNode:60        """Define a new amount (exactly once — ids are generated, never reused)."""61        for parent in inputs:62            if parent not in self._nodes:63                raise ProvenanceError(f"Unknown provenance input: {parent}")64        node = ProvenanceNode(65            id=f"prov_{next(self._counter):06d}",66            kind=kind,67            operation=operation,68            amount=amount,69            inputs=inputs,70            source_ref=source_ref,71        )72        self._nodes[node.id] = node73        return node7475    def get(self, node_id: str) -> ProvenanceNode:76        try:77            return self._nodes[node_id]78        except KeyError:79            raise ProvenanceError(f"Unknown provenance node: {node_id}") from None8081    def trace(self, node_id: str) -> list[ProvenanceNode]:82        """Full ancestry of a node (the node first, origins last), depth-first."""83        seen: set[str] = set()84        out: list[ProvenanceNode] = []8586        def walk(nid: str) -> None:87            if nid in seen:88                return89            seen.add(nid)90            node = self.get(nid)91            out.append(node)92            for parent in node.inputs:93                walk(parent)9495        walk(node_id)96        return out9798    def __len__(self) -> int:99        return len(self._nodes)100101    def __iter__(self) -> Iterator[ProvenanceNode]:102        return iter(self._nodes.values())103104    def to_dicts(self) -> list[dict[str, object]]:105        """Serialize for audit export (amounts as strings, never floats)."""106        return [107            {108                "id": n.id,109                "kind": n.kind,110                "operation": n.operation,111                "amount": str(n.amount.amount),112                "currency": n.amount.currency,113                "inputs": list(n.inputs),114                "source_ref": n.source_ref,115            }116            for n in self._nodes.values()117        ]118