Spaces:
Running
Running
Deploy Aureon OS (deploy/selfhost stack: nginx edge + fusion API x2 + console + HNC organ + redis + postgres; paper mode) (part 3)
1ccf652 verified | """ | |
| ApprovalQueue — the director's desk: Aureon prepares the big plays, Gary decides. | |
| ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ | |
| Aureon runs the reversible day-to-day itself. But the big, irreversible plays — a | |
| live trade, moving money, submitting a grant, a reply that leaves the building — | |
| belong to the director. This is where they wait for him. | |
| The organism **prepares** each such move and `propose()`s it here; Gary reviews it | |
| (console, watch, or by email reply) and `decide()`s approve or reject. That decision | |
| is **recorded** — it is the human gate — and it is *all* this module does. It never | |
| executes the move. By construction there is **no consumer** of the queue that fires a | |
| live trade / payment / filing / outbound email: approving only flips the item's status | |
| to ``approved``, leaving the actual irreversible execution to Gary's deliberate hand | |
| (or a live executor he arms himself, separately and explicitly). Aureon runs full | |
| speed up to the edge of consequence; the human blesses the step across it. | |
| Backed by an append-only JSONL log (`state/approvals.jsonl`) via ``bus_trace`` — | |
| atomic, bounded, multi-writer-safe, never-raises. Current state is the fold of the | |
| append log (latest status per id). Guarded throughout. | |
| Gary Leckey · Aureon Institute | |
| """ | |
| from __future__ import annotations | |
| import logging | |
| import os | |
| import time | |
| import uuid | |
| from typing import Any | |
| logger = logging.getLogger("aureon.core.approval_queue") | |
| _TRACE = "approvals" | |
| # When this many proposals already await the director, the organism stops piling on | |
| # more — it senses its own backlog and waits for the desk to clear (backpressure). | |
| _DEFAULT_MAX_PENDING = 5 | |
| _KINDS = {"trade", "email", "payment", "filing", "grant", "deal", "other"} | |
| _OPEN = "pending" | |
| _DECISIONS = {"approve": "approved", "reject": "rejected"} | |
| def _new_id() -> str: | |
| return uuid.uuid4().hex[:8] | |
| class ApprovalQueue: | |
| """Prepare → propose → (human) decide → record. Never executes.""" | |
| def __init__(self, *, cap: int = 500) -> None: | |
| self._cap = cap | |
| # ── propose (Aureon prepares a big play; nothing executes) ────────────── | |
| def propose(self, kind: str, summary: str, params: dict[str, Any] | None = None, | |
| prepared_by: str = "soul", *, risk: str = "high") -> str | None: | |
| """Enqueue one PENDING proposal for the human. Returns its id, or None if a | |
| matching pending item already exists (dedup) or on any error. Never executes.""" | |
| try: | |
| k = kind if kind in _KINDS else "other" | |
| text = str(summary or "").strip()[:400] or "(no summary)" | |
| # dedup against open items of the same kind + summary | |
| for item in self.pending(): | |
| if item.get("kind") == k and item.get("summary") == text: | |
| return None | |
| item_id = _new_id() | |
| self._append({ | |
| "event": "proposed", "id": item_id, "kind": k, "summary": text, | |
| "params": dict(params or {}), "prepared_by": str(prepared_by), | |
| "risk": str(risk), "requires_human": True, "status": _OPEN, | |
| "note": "", "created_at": time.time(), "decided_at": None, | |
| }) | |
| return item_id | |
| except Exception as exc: # noqa: BLE001 — a desk that jams never crashes the organism | |
| logger.debug("propose skipped: %s", exc) | |
| return None | |
| # ── decide (the human gate; records only, never executes) ─────────────── | |
| def decide(self, item_id: str, decision: str, approver: str = "gary", | |
| note: str = "", *, | |
| auth_context: dict[str, Any] | None = None) -> dict[str, Any] | None: | |
| """Record the human's approve/reject. This is the gate — it does NOT execute | |
| the move. Returns the updated item, or None if unknown/already decided.""" | |
| try: | |
| status = _DECISIONS.get(str(decision).strip().lower()) | |
| if status is None: | |
| return None | |
| item = self.get(item_id) | |
| if item is None or item.get("status") != _OPEN: | |
| return None | |
| auth = dict(auth_context or {}) | |
| identity_kind = str(auth.get("identity_kind") or "").strip() | |
| authn_method = str(auth.get("authn_method") or "").strip() | |
| approval_auth = ( | |
| { | |
| "authenticated": True, | |
| "identity_kind": identity_kind, | |
| "authn_method": authn_method, | |
| } | |
| if identity_kind and authn_method else { | |
| "authenticated": False, | |
| "identity_kind": "", | |
| "authn_method": "", | |
| } | |
| ) | |
| self._append({ | |
| "event": "decided", "id": item_id, "kind": item.get("kind"), | |
| "summary": item.get("summary"), "params": item.get("params"), | |
| "prepared_by": item.get("prepared_by"), "risk": item.get("risk"), | |
| "requires_human": True, "status": status, "note": str(note)[:400], | |
| "approver": str(approver), "created_at": item.get("created_at"), | |
| "decided_at": time.time(), "approval_auth": approval_auth, | |
| }) | |
| return self.get(item_id) | |
| except Exception as exc: # noqa: BLE001 | |
| logger.debug("decide skipped: %s", exc) | |
| return None | |
| # ── read (fold the append log to latest-per-id) ───────────────────────── | |
| def _fold(self) -> dict[str, dict[str, Any]]: | |
| from aureon.core.bus_trace import read_trace | |
| latest: dict[str, dict[str, Any]] = {} | |
| for row in read_trace(_TRACE, limit=self._cap): | |
| rid = row.get("id") | |
| if rid: | |
| latest[str(rid)] = row | |
| return latest | |
| def get(self, item_id: str) -> dict[str, Any] | None: | |
| return self._fold().get(str(item_id)) | |
| def pending(self) -> list[dict[str, Any]]: | |
| items = [r for r in self._fold().values() if r.get("status") == _OPEN] | |
| return sorted(items, key=lambda r: r.get("created_at") or 0.0) | |
| def recent(self, limit: int = 20) -> list[dict[str, Any]]: | |
| items = sorted(self._fold().values(), key=lambda r: r.get("created_at") or 0.0, reverse=True) | |
| return items[: max(1, limit)] | |
| # ── the organism senses its own desk (backpressure) ──────────────────── | |
| def _max_pending(self) -> int: | |
| try: | |
| return max(1, int(os.environ.get("AUREON_APPROVAL_MAX_PENDING", "") or _DEFAULT_MAX_PENDING)) | |
| except (TypeError, ValueError): | |
| return _DEFAULT_MAX_PENDING | |
| def backlog(self) -> dict[str, Any]: | |
| """How much the director's desk is holding — the organism's read of being | |
| blocked on the human. ``blocked`` is the backpressure signal.""" | |
| pend = self.pending() | |
| now = time.time() | |
| oldest = min((float(r.get("created_at") or now) for r in pend), default=now) | |
| cap = self._max_pending() | |
| return {"pending_count": len(pend), "oldest_age_s": round(now - oldest, 1), | |
| "max_pending": cap, "blocked": len(pend) >= cap} | |
| def is_backpressured(self) -> bool: | |
| """True when the desk is full — prepare no more until the director clears it.""" | |
| try: | |
| return bool(self.backlog()["blocked"]) | |
| except Exception: # noqa: BLE001 — never block the organism on a read error | |
| return False | |
| def summary(self) -> dict[str, Any]: | |
| folded = list(self._fold().values()) | |
| counts: dict[str, int] = {} | |
| for r in folded: | |
| s = str(r.get("status") or "?") | |
| counts[s] = counts.get(s, 0) + 1 | |
| return {"pending": self.pending(), "recent": self.recent(20), | |
| "counts": counts, "total": len(folded), "backlog": self.backlog(), | |
| "trust": self.trust(), | |
| "note": "records the human decision; never executes the live move"} | |
| # ── the director's trust (how he has decided — a signal, never a lever) ─ | |
| def trust(self) -> dict[str, Any]: | |
| """How the director has been deciding: approved vs rejected over all decided | |
| items. ``approve_ratio`` is ``None`` until he has decided at least one — the | |
| organism must never fabricate a trust level it has not earned. This RECORDS a | |
| felt signal; it authorizes nothing.""" | |
| try: | |
| approved = rejected = 0 | |
| for r in self._fold().values(): | |
| st = r.get("status") | |
| if st == "approved": | |
| approved += 1 | |
| elif st == "rejected": | |
| rejected += 1 | |
| decided = approved + rejected | |
| return {"approved": approved, "rejected": rejected, "decided": decided, | |
| "approve_ratio": (approved / decided) if decided else None} | |
| except Exception as exc: # noqa: BLE001 — a read error is never a trust level | |
| logger.debug("trust read skipped: %s", exc) | |
| return {"approved": 0, "rejected": 0, "decided": 0, "approve_ratio": None} | |
| def _append(self, payload: dict[str, Any]) -> None: | |
| from aureon.core.bus_trace import append_trace | |
| append_trace(_TRACE, payload, cap=self._cap) | |
| _queue: ApprovalQueue | None = None | |
| def get_approval_queue() -> ApprovalQueue: | |
| """Process-global approval-queue singleton (the director's desk).""" | |
| global _queue | |
| if _queue is None: | |
| _queue = ApprovalQueue() | |
| return _queue | |
| __all__ = ["ApprovalQueue", "get_approval_queue"] | |