spb/air Public MIT

1# =============================================================================2# Projet : AIR — Accounting Intermediate Representation3# Auteur : Simon-Pierre Boucher4# Contact : contact@spboucher.ai5# Fichier : approval.py6# Description : Human approval queue — file-backed inbox inside the AIR home.7# =============================================================================8"""Approval queue: the human-in-the-loop stage.910Low-confidence or schema-invalid extractions land here instead of the ledger.11The queue lives inside the AIR home:1213    <home>/inbox/pending/<item>.json14    <home>/inbox/approved/<item>.json15    <home>/inbox/rejected/<item>.json1617Approving stamps the approver + timestamp into every event's meta (full18traceability: who let this into the books) and returns the AirDocument ready19for compilation. Nothing is ever deleted — items move between folders.20"""21from __future__ import annotations2223import json24from dataclasses import dataclass25from datetime import datetime26from pathlib import Path27from typing import Any2829from core.events import AirDocument30from ingestion.pipeline import IngestionOutcome3132_AUTHOR = "Simon-Pierre Boucher <contact@spboucher.ai>"333435@dataclass(frozen=True, slots=True)36class InboxItem:37    id: str38    status: str                      # pending | approved | rejected39    confidence: str40    reasons: list[str]41    validation_errors: list[str]42    events: list[dict[str, Any]]43    source_excerpt: str444546class ApprovalQueue:47    def __init__(self, home: str | Path):48        self.root = Path(home) / "inbox"49        for status in ("pending", "approved", "rejected"):50            (self.root / status).mkdir(parents=True, exist_ok=True)5152    # -- intake ------------------------------------------------------------------53    def submit(self, outcome: IngestionOutcome, source_text: str) -> str:54        """File a needs-review outcome into the pending inbox. Returns item id."""55        events = (56            [json.loads(e.model_dump_json(exclude_none=True))57             for e in outcome.document.events]58            if outcome.document is not None59            else outcome.extraction.events60        )61        item_id = f"inbox_{len(list(self.root.rglob('*.json'))):05d}"62        payload = {63            "_author": _AUTHOR,64            "id": item_id,65            "status": "pending",66            "confidence": str(outcome.extraction.confidence),67            "extractor": outcome.extraction.extractor,68            "reasons": outcome.reasons,69            "validation_errors": outcome.validation_errors,70            "events": events,71            "source_excerpt": source_text[:2000],72        }73        (self.root / "pending" / f"{item_id}.json").write_text(74            json.dumps(payload, indent=2, default=str), encoding="utf-8"75        )76        return item_id7778    # -- inspection ---------------------------------------------------------------79    def _load(self, status: str) -> list[InboxItem]:80        items = []81        for path in sorted((self.root / status).glob("*.json")):82            data = json.loads(path.read_text(encoding="utf-8"))83            items.append(InboxItem(84                id=data["id"], status=data["status"],85                confidence=data["confidence"],86                reasons=data.get("reasons", []),87                validation_errors=data.get("validation_errors", []),88                events=data.get("events", []),89                source_excerpt=data.get("source_excerpt", ""),90            ))91        return items9293    def pending(self) -> list[InboxItem]:94        return self._load("pending")9596    # -- decisions ----------------------------------------------------------------97    def _move(self, item_id: str, new_status: str,98              mutate: dict[str, Any]) -> dict[str, Any]:99        source = self.root / "pending" / f"{item_id}.json"100        if not source.exists():101            raise KeyError(f"no pending inbox item '{item_id}'")102        data = json.loads(source.read_text(encoding="utf-8"))103        data["status"] = new_status104        data.update(mutate)105        target = self.root / new_status / f"{item_id}.json"106        target.write_text(json.dumps(data, indent=2, default=str),107                          encoding="utf-8")108        source.unlink()109        return data110111    def approve(self, item_id: str, approver: str,112                approved_at: datetime | None = None) -> AirDocument:113        """Human approval: stamp approver + timestamp, return the document.114115        The returned document still goes through the deterministic compiler —116        approval authorizes compilation, it never writes to the ledger itself.117        """118        when = approved_at.isoformat() if approved_at else None119        data = self._move(item_id, "approved",120                          {"approver": approver, "approved_at": when})121        events = []122        for event in data["events"]:123            meta = dict(event.get("meta") or {})124            meta["approver"] = approver125            timestamps = dict(meta.get("timestamps") or {})126            if when:127                timestamps["approved"] = when128            if timestamps:129                meta["timestamps"] = timestamps130            events.append({**event, "meta": meta})131        return AirDocument.model_validate({"events": events})132133    def reject(self, item_id: str, reason: str = "") -> None:134        self._move(item_id, "rejected", {"rejection_reason": reason})135