#!/usr/bin/env python3 """routine_run_guard.py - Agents Finance template (MIT-0), v1.0, 31 October 2026. Checks the JSON file that an unattended finance routine (for example a Claude Code routine / cloud scheduled task) leaves in a pull request BEFORE a person reviews it or posts anything in the ERP. Python 3.9+, standard library only. python3 routine_run_guard.py RUN.json POLICY.json [--log guard-log.jsonl] Exit codes: 0 = PASS (ready for human review), 1 = BLOCK, 2 = unreadable input. What it checks - status is "draft" and the period is not closed - no forbidden action is reported in actions_taken - every proposed journal entry balances, uses accounts from the chart, cites evidence and has a rationale - entries at or above the approval threshold name an approver who is neither the preparer nor an agent identity (segregation of duties) - the bank reconciliation block adds up, and any unexplained difference is flagged for a person It appends one JSON line per check (time, run id, SHA-256 of the file, result) to the log, so you keep an audit trail of what was checked. Limits: actions_taken is reported by the routine itself. The real control is a read-only connector; this script is a second barrier and a record, not proof. """ import argparse import hashlib import json import re import sys from datetime import datetime, timezone from decimal import Decimal, InvalidOperation PERIOD = re.compile(r"^\d{4}-(0[1-9]|1[0-2])$") def dec(value): try: d = Decimal(str(value)) except (InvalidOperation, TypeError, ValueError): raise ValueError(f"not a number: {value!r}") if not d.is_finite(): raise ValueError(f"not a finite number: {value!r}") return d def check(run, policy): issues = [] def add(severity, code, message): issues.append({"severity": severity, "code": code, "message": message}) if run.get("status") != "draft": add("BLOCK", "NOT_DRAFT", f"status is {run.get('status')!r}; a routine may only leave drafts") period = str(run.get("period", "")) if not PERIOD.match(period): add("BLOCK", "BAD_PERIOD", f"period {period!r} is not YYYY-MM") elif period in policy.get("closed_periods", []): add("BLOCK", "PERIOD_CLOSED", f"period {period} is closed") forbidden = set(policy.get("forbidden_actions", [])) for action in run.get("actions_taken", []): if action.get("type") in forbidden: add("BLOCK", "FORBIDDEN_ACTION", f"routine reports action {action.get('type')!r} on {action.get('target')!r}") agents = set(policy.get("agent_identities", [])) threshold = dec(policy.get("approval_threshold", "0")) tol = dec(policy.get("tolerance", "0.00")) chart = {str(a) for a in policy.get("allowed_accounts", [])} for je in run.get("journal_entries", []): jid = je.get("id", "?") lines = je.get("lines", []) if not lines: add("BLOCK", "NO_LINES", f"{jid}: no lines") continue try: debit = sum((dec(l.get("debit", 0)) for l in lines), Decimal(0)) credit = sum((dec(l.get("credit", 0)) for l in lines), Decimal(0)) except ValueError as exc: add("BLOCK", "BAD_AMOUNT", f"{jid}: {exc}") continue if abs(debit - credit) > tol: add("BLOCK", "UNBALANCED", f"{jid}: debits {debit} != credits {credit}") for i, line in enumerate(lines, 1): if chart and str(line.get("account")) not in chart: add("BLOCK", "UNKNOWN_ACCOUNT", f"{jid} line {i}: account {line.get('account')!r} is not in the chart") if not je.get("evidence"): add("BLOCK", "NO_EVIDENCE", f"{jid}: no evidence reference (statement line, invoice, workpaper)") if not je.get("rationale"): add("WARN", "NO_RATIONALE", f"{jid}: no rationale") if debit >= threshold: approver, preparer = je.get("proposed_approver"), je.get("preparer") if not approver: add("BLOCK", "NO_APPROVER", f"{jid}: {debit} >= threshold {threshold} and no named approver") elif approver == preparer: add("BLOCK", "SOD_SAME_PERSON", f"{jid}: preparer and approver are both {approver!r}") elif approver in agents: add("BLOCK", "SOD_AGENT_APPROVER", f"{jid}: approver {approver!r} is an agent identity") for rec in run.get("reconciliations", []): acc = rec.get("account", "?") try: bank = dec(rec["bank_balance"]) ledger = dec(rec["ledger_balance"]) items = rec.get("reconciling_items", []) items_total = sum((dec(it.get("amount")) for it in items), Decimal(0)) declared = dec(rec.get("unexplained_difference", 0)) except (KeyError, ValueError) as exc: add("BLOCK", "BAD_RECONCILIATION", f"{acc}: {exc}") continue computed = bank - (ledger + items_total) if abs(computed - declared) > tol: add("BLOCK", "REC_ARITHMETIC", f"{acc}: unexplained difference declared {declared}, recomputed {computed}") if abs(computed) > tol: add("WARN", "REC_UNEXPLAINED", f"{acc}: {computed} unexplained; a person must investigate") for it in items: if not it.get("evidence"): add("BLOCK", "ITEM_NO_EVIDENCE", f"{acc}: reconciling item {it.get('description')!r} has no evidence") return issues def main(): parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("run") parser.add_argument("policy") parser.add_argument("--log", default="guard-log.jsonl") args = parser.parse_args() try: with open(args.run, "rb") as fh: raw = fh.read() run = json.loads(raw) with open(args.policy, encoding="utf-8") as fh: policy = json.load(fh) if not isinstance(run, dict) or not isinstance(policy, dict): raise ValueError("run and policy must be JSON objects") issues = check(run, policy) except (OSError, ValueError, TypeError, AttributeError) as exc: # JSONDecodeError is a ValueError print(f"INPUT ERROR: {exc}", file=sys.stderr) return 2 result = "BLOCK" if any(i["severity"] == "BLOCK" for i in issues) else "PASS" entry = { "checked_at": datetime.now(timezone.utc).isoformat(timespec="seconds"), "file": args.run, "run_id": run.get("run_id"), "routine": run.get("routine"), "input_sha256": hashlib.sha256(raw).hexdigest(), "result": result, "issues": issues, } with open(args.log, "a", encoding="utf-8") as fh: fh.write(json.dumps(entry, ensure_ascii=False) + "\n") print(f"{result} {args.run} run={run.get('run_id')} issues={len(issues)}") for i in issues: print(f" [{i['severity']}] {i['code']}: {i['message']}") return 1 if result == "BLOCK" else 0 if __name__ == "__main__": sys.exit(main())