spb/air Public MIT
AIR — The Language of Accounting.
Python 100%
1# =============================================================================2# Projet : AIR — Accounting Intermediate Representation3# Auteur : Simon-Pierre Boucher4# Contact : contact@spboucher.ai5# Fichier : syscalls.py6# Description : The agent SDK — accounting syscalls; agents NEVER touch the ledger directly.7# =============================================================================8"""AIR syscalls (CLAUDE.md §3.5).910AI agents do not get the ledger, the compiler internals, or the files —11they get THIS interface and nothing else:1213 CreateEconomicEvent | Validate | Compile | Post | Reverse14 ClosePeriod | GenerateReport (Merge/Reconcile arrive in Phase 6)1516Guarantees:17- every call — successful or failed — appends one record to the hash-chained18 audit log (who, what, params digest, outcome);19- drafts are staged in the AIR home and only reach the books through the20 deterministic compiler (Post);21- posting into a closed period is refused with a compiler-grade diagnostic;22- corrections are reversals (Reverse), never edits.23"""24from __future__ import annotations2526import hashlib27import json28from datetime import datetime, timezone29from pathlib import Path30from typing import Any, Callable3132from aic.compiler import compile_document33from aic.diagnostics import Diagnostic34from alsl.loader import load_policy_set35from core.events import AirDocument, EconomicEvent36from core.journal import CompiledJournal37from kernel.audit import AuditLog38from kernel.ledger import Ledger39from kernel.reporting import REPORTS40from kernel.workspace import Workspace4142_AUTHOR = "Simon-Pierre Boucher <contact@spboucher.ai>"434445class SyscallError(RuntimeError):46 """A syscall was refused; the refusal is in the audit log."""474849class AirKernel:50 """One agent's handle on the books. All access is syscalls."""5152 def __init__(53 self,54 home: str | Path | Workspace,55 actor: str,56 policies: str | None = None,57 clock: Callable[[], datetime] | None = None,58 ):59 self.workspace = home if isinstance(home, Workspace) else Workspace.open(home)60 self.actor = actor61 policies_path = policies or self.workspace.default_policies()62 if not policies_path:63 raise SyscallError("no policy set configured for this AIR home")64 self.policies = load_policy_set(policies_path)65 self.audit = AuditLog(path=self.workspace.root / "audit.jsonl")66 self._ledger = Ledger(path=self.workspace.ledger_path) # private: no syscall exposes it raw67 self._clock = clock or (lambda: datetime.now(timezone.utc))68 self._drafts_dir = self.workspace.root / "drafts"69 self._drafts_dir.mkdir(parents=True, exist_ok=True)7071 # -- audit plumbing ------------------------------------------------------------72 def _audited(self, syscall: str, params: dict[str, Any],73 fn: Callable[[], Any]) -> Any:74 try:75 result = fn()76 except Exception as exc:77 # failures are audited too — an agent's denied action is evidence78 self.audit.append(self.actor, syscall, params, "error",79 detail=str(exc).splitlines()[0],80 at=self._clock().isoformat())81 raise82 self.audit.append(self.actor, syscall, params, "ok",83 at=self._clock().isoformat())84 return result8586 # -- drafts ---------------------------------------------------------------------87 def _draft_paths(self) -> list[Path]:88 return sorted(self._drafts_dir.glob("*.json"))8990 def _load_drafts(self) -> AirDocument:91 events = [json.loads(p.read_text(encoding="utf-8"))["event"]92 for p in self._draft_paths()]93 if not events:94 raise SyscallError("no draft events staged; call create_economic_event first")95 return AirDocument.model_validate({"events": events})9697 # -- syscalls ---------------------------------------------------------------------98 def create_economic_event(self, event_data: dict[str, Any]) -> str:99 """Stage a draft economic event. Validated against the AIR schema now;100 it reaches the books only via post()."""101 def run() -> str:102 event = EconomicEvent.model_validate(event_data)103 payload = {"_author": _AUTHOR, "staged_by": self.actor,104 "event": json.loads(event.model_dump_json(exclude_none=True))}105 (self._drafts_dir / f"{event.id}.json").write_text(106 json.dumps(payload, indent=2), encoding="utf-8")107 return event.id108 return self._audited(109 "CreateEconomicEvent",110 {"event_id": event_data.get("id"), "type": event_data.get("type")},111 run,112 )113114 def validate(self) -> list[Diagnostic]:115 """Compile the staged drafts without posting; returns diagnostics."""116 def run() -> list[Diagnostic]:117 document = self._load_drafts()118 _, diagnostics = compile_document(document, self.policies)119 return diagnostics120 return self._audited("Validate", {"drafts": len(self._draft_paths())}, run)121122 def compile(self) -> CompiledJournal:123 """Deterministically compile the staged drafts (no posting)."""124 def run() -> CompiledJournal:125 document = self._load_drafts()126 journal, _ = compile_document(document, self.policies)127 return journal128 return self._audited("Compile", {"drafts": len(self._draft_paths())}, run)129130 def post(self, optimize: bool = False) -> dict[str, Any]:131 """Compile the drafts and append the entries to the books.132133 Refuses events dated in a closed period. Idempotent: the key is the134 content hash of the staged drafts. optimize=True applies the Phase 6135 passes (duplicate detection, netting, fusion) before posting.136 """137 def run() -> dict[str, Any]:138 document = self._load_drafts()139 self._refuse_closed_periods(document)140 journal, _ = compile_document(document, self.policies,141 optimize=optimize)142 key = hashlib.sha256(143 document.model_dump_json().encode("utf-8")).hexdigest()[:16]144 appended = self._ledger.post_journal(journal, key)145 # archive the posted drafts; the staging area empties146 archive = self.workspace.root / "documents"147 archive.mkdir(exist_ok=True)148 for path in self._draft_paths():149 path.rename(archive / f"posted_{key}_{path.name}")150 return {"entries": len(journal.entries), "appended": appended,151 "idempotency_key": key,152 "entry_ids": [e.id for e in journal.entries]}153 return self._audited(154 "Merge" if optimize else "Post",155 {"drafts": len(self._draft_paths()), "optimize": optimize},156 run,157 )158159 def merge(self) -> dict[str, Any]:160 """Post the drafts with optimization: duplicates flagged, refunds161 netted against their sales, identical payments fused into batches."""162 return self.post(optimize=True)163164 def reverse(self, entry_id: str) -> str:165 """Append the exact contra of a posted entry. Never deletes."""166 def run() -> str:167 try:168 contra = self._ledger.reverse_entry(169 entry_id, f"rev:{self.actor}:{entry_id}")170 except KeyError as exc:171 raise SyscallError(str(exc)) from None172 return contra.id173 return self._audited("Reverse", {"entry_id": entry_id}, run)174175 def close_period(self, period: str) -> None:176 """Close an accounting period ("YYYY-MM"); later posts into it are refused."""177 def run() -> None:178 if len(period) != 7 or period[4] != "-":179 raise SyscallError(f"invalid period '{period}': use YYYY-MM")180 meta = self.workspace.meta181 closed = sorted(set(meta.get("closed_periods", []) or []) | {period})182 meta["closed_periods"] = closed183 (self.workspace.root / "meta.json").write_text(184 json.dumps(meta, indent=2), encoding="utf-8")185 return self._audited("ClosePeriod", {"period": period}, run)186187 def reconcile(self, statement_path: str | Path,188 cash_account: str = "1000",189 tolerance_days: int = 3) -> dict[str, Any]:190 """Match a bank statement (camt.053 / MT940 / CSV) against the books'191 cash movements. Returns the summary; the full report is in `render`."""192 def run() -> dict[str, Any]:193 from kernel.bank_formats import parse_statement194 from kernel.reconcile import cash_movements, reconcile195196 transactions = parse_statement(Path(statement_path))197 movements = cash_movements(self._ledger, cash_account)198 result = reconcile(transactions, movements,199 tolerance_days=tolerance_days)200 summary: dict[str, Any] = dict(result.summary())201 summary["clean"] = result.is_clean202 summary["render"] = result.render()203 return summary204 return self._audited(205 "Reconcile",206 {"statement": Path(statement_path).name,207 "cash_account": cash_account},208 run,209 )210211 def generate_report(self, name: str, fmt: str = "text") -> str:212 def run() -> str:213 if name not in REPORTS:214 raise SyscallError(215 f"unknown report '{name}'; available: {sorted(REPORTS)}")216 return REPORTS[name](self._ledger, fmt)217 return self._audited("GenerateReport", {"report": name, "format": fmt}, run)218219 # -- helpers --------------------------------------------------------------------220 def _refuse_closed_periods(self, document: AirDocument) -> None:221 closed = set(self.workspace.meta.get("closed_periods", []) or [])222 for event in document.events:223 period = event.date.strftime("%Y-%m")224 if period in closed:225 raise SyscallError(226 f"error[AIR-E700]: event '{event.id}' is dated {event.date} "227 f"but period {period} is closed\n"228 f" help: date the correction in an open period and reference "229 f"the original via related_event"230 )231