# ============================================================================= # Projet : AIR — Accounting Intermediate Representation # Auteur : Simon-Pierre Boucher # Contact : contact@spboucher.ai # Fichier : syscalls.py # Description : The agent SDK — accounting syscalls; agents NEVER touch the ledger directly. # ============================================================================= """AIR syscalls (CLAUDE.md §3.5). AI agents do not get the ledger, the compiler internals, or the files — they get THIS interface and nothing else: CreateEconomicEvent | Validate | Compile | Post | Reverse ClosePeriod | GenerateReport (Merge/Reconcile arrive in Phase 6) Guarantees: - every call — successful or failed — appends one record to the hash-chained audit log (who, what, params digest, outcome); - drafts are staged in the AIR home and only reach the books through the deterministic compiler (Post); - posting into a closed period is refused with a compiler-grade diagnostic; - corrections are reversals (Reverse), never edits. """ from __future__ import annotations import hashlib import json from datetime import datetime, timezone from pathlib import Path from typing import Any, Callable from aic.compiler import compile_document from aic.diagnostics import Diagnostic from alsl.loader import load_policy_set from core.events import AirDocument, EconomicEvent from core.journal import CompiledJournal from kernel.audit import AuditLog from kernel.ledger import Ledger from kernel.reporting import REPORTS from kernel.workspace import Workspace _AUTHOR = "Simon-Pierre Boucher " class SyscallError(RuntimeError): """A syscall was refused; the refusal is in the audit log.""" class AirKernel: """One agent's handle on the books. All access is syscalls.""" def __init__( self, home: str | Path | Workspace, actor: str, policies: str | None = None, clock: Callable[[], datetime] | None = None, ): self.workspace = home if isinstance(home, Workspace) else Workspace.open(home) self.actor = actor policies_path = policies or self.workspace.default_policies() if not policies_path: raise SyscallError("no policy set configured for this AIR home") self.policies = load_policy_set(policies_path) self.audit = AuditLog(path=self.workspace.root / "audit.jsonl") self._ledger = Ledger(path=self.workspace.ledger_path) # private: no syscall exposes it raw self._clock = clock or (lambda: datetime.now(timezone.utc)) self._drafts_dir = self.workspace.root / "drafts" self._drafts_dir.mkdir(parents=True, exist_ok=True) # -- audit plumbing ------------------------------------------------------------ def _audited(self, syscall: str, params: dict[str, Any], fn: Callable[[], Any]) -> Any: try: result = fn() except Exception as exc: # failures are audited too — an agent's denied action is evidence self.audit.append(self.actor, syscall, params, "error", detail=str(exc).splitlines()[0], at=self._clock().isoformat()) raise self.audit.append(self.actor, syscall, params, "ok", at=self._clock().isoformat()) return result # -- drafts --------------------------------------------------------------------- def _draft_paths(self) -> list[Path]: return sorted(self._drafts_dir.glob("*.json")) def _load_drafts(self) -> AirDocument: events = [json.loads(p.read_text(encoding="utf-8"))["event"] for p in self._draft_paths()] if not events: raise SyscallError("no draft events staged; call create_economic_event first") return AirDocument.model_validate({"events": events}) # -- syscalls --------------------------------------------------------------------- def create_economic_event(self, event_data: dict[str, Any]) -> str: """Stage a draft economic event. Validated against the AIR schema now; it reaches the books only via post().""" def run() -> str: event = EconomicEvent.model_validate(event_data) payload = {"_author": _AUTHOR, "staged_by": self.actor, "event": json.loads(event.model_dump_json(exclude_none=True))} (self._drafts_dir / f"{event.id}.json").write_text( json.dumps(payload, indent=2), encoding="utf-8") return event.id return self._audited( "CreateEconomicEvent", {"event_id": event_data.get("id"), "type": event_data.get("type")}, run, ) def validate(self) -> list[Diagnostic]: """Compile the staged drafts without posting; returns diagnostics.""" def run() -> list[Diagnostic]: document = self._load_drafts() _, diagnostics = compile_document(document, self.policies) return diagnostics return self._audited("Validate", {"drafts": len(self._draft_paths())}, run) def compile(self) -> CompiledJournal: """Deterministically compile the staged drafts (no posting).""" def run() -> CompiledJournal: document = self._load_drafts() journal, _ = compile_document(document, self.policies) return journal return self._audited("Compile", {"drafts": len(self._draft_paths())}, run) def post(self, optimize: bool = False) -> dict[str, Any]: """Compile the drafts and append the entries to the books. Refuses events dated in a closed period. Idempotent: the key is the content hash of the staged drafts. optimize=True applies the Phase 6 passes (duplicate detection, netting, fusion) before posting. """ def run() -> dict[str, Any]: document = self._load_drafts() self._refuse_closed_periods(document) journal, _ = compile_document(document, self.policies, optimize=optimize) key = hashlib.sha256( document.model_dump_json().encode("utf-8")).hexdigest()[:16] appended = self._ledger.post_journal(journal, key) # archive the posted drafts; the staging area empties archive = self.workspace.root / "documents" archive.mkdir(exist_ok=True) for path in self._draft_paths(): path.rename(archive / f"posted_{key}_{path.name}") return {"entries": len(journal.entries), "appended": appended, "idempotency_key": key, "entry_ids": [e.id for e in journal.entries]} return self._audited( "Merge" if optimize else "Post", {"drafts": len(self._draft_paths()), "optimize": optimize}, run, ) def merge(self) -> dict[str, Any]: """Post the drafts with optimization: duplicates flagged, refunds netted against their sales, identical payments fused into batches.""" return self.post(optimize=True) def reverse(self, entry_id: str) -> str: """Append the exact contra of a posted entry. Never deletes.""" def run() -> str: try: contra = self._ledger.reverse_entry( entry_id, f"rev:{self.actor}:{entry_id}") except KeyError as exc: raise SyscallError(str(exc)) from None return contra.id return self._audited("Reverse", {"entry_id": entry_id}, run) def close_period(self, period: str) -> None: """Close an accounting period ("YYYY-MM"); later posts into it are refused.""" def run() -> None: if len(period) != 7 or period[4] != "-": raise SyscallError(f"invalid period '{period}': use YYYY-MM") meta = self.workspace.meta closed = sorted(set(meta.get("closed_periods", []) or []) | {period}) meta["closed_periods"] = closed (self.workspace.root / "meta.json").write_text( json.dumps(meta, indent=2), encoding="utf-8") return self._audited("ClosePeriod", {"period": period}, run) def reconcile(self, statement_path: str | Path, cash_account: str = "1000", tolerance_days: int = 3) -> dict[str, Any]: """Match a bank statement (camt.053 / MT940 / CSV) against the books' cash movements. Returns the summary; the full report is in `render`.""" def run() -> dict[str, Any]: from kernel.bank_formats import parse_statement from kernel.reconcile import cash_movements, reconcile transactions = parse_statement(Path(statement_path)) movements = cash_movements(self._ledger, cash_account) result = reconcile(transactions, movements, tolerance_days=tolerance_days) summary: dict[str, Any] = dict(result.summary()) summary["clean"] = result.is_clean summary["render"] = result.render() return summary return self._audited( "Reconcile", {"statement": Path(statement_path).name, "cash_account": cash_account}, run, ) def generate_report(self, name: str, fmt: str = "text") -> str: def run() -> str: if name not in REPORTS: raise SyscallError( f"unknown report '{name}'; available: {sorted(REPORTS)}") return REPORTS[name](self._ledger, fmt) return self._audited("GenerateReport", {"report": name, "format": fmt}, run) # -- helpers -------------------------------------------------------------------- def _refuse_closed_periods(self, document: AirDocument) -> None: closed = set(self.workspace.meta.get("closed_periods", []) or []) for event in document.events: period = event.date.strftime("%Y-%m") if period in closed: raise SyscallError( f"error[AIR-E700]: event '{event.id}' is dated {event.date} " f"but period {period} is closed\n" f" help: date the correction in an open period and reference " f"the original via related_event" )