feat(task-plane): add Owner and GPT notification outbox

active-runtime/app/autonomy_notification_plane.py

@@ -0,0 +1,89 @@
+"""Durable Owner/GPT notification outbox with retry and acknowledgement."""
+from __future__ import annotations
+
+import json
+import sqlite3
+import time
+import uuid
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Callable, Mapping
+
+
+@dataclass(frozen=True)
+class OutboxEvent:
+    event_id: str
+    recipient: str
+    kind: str
+    payload: Mapping[str, Any]
+    attempts: int
+    available_at: float
+    acknowledged: bool
+
+
+class DurableOutbox:
+    RECIPIENTS = ("owner", "gpt")
+
+    def __init__(self, path: str | Path) -> None:
+        self.path = str(path)
+        self.db = sqlite3.connect(self.path)
+        self.db.execute("PRAGMA journal_mode=WAL")
+        self.db.execute("CREATE TABLE IF NOT EXISTS outbox (event_id TEXT PRIMARY KEY, recipient TEXT NOT NULL, kind TEXT NOT NULL, payload TEXT NOT NULL, attempts INTEGER NOT NULL DEFAULT 0, available_at REAL NOT NULL, delivered_at REAL, acknowledged INTEGER NOT NULL DEFAULT 0, last_error TEXT)")
+        self.db.commit()
+
+    def enqueue(self, kind: str, payload: Mapping[str, Any], *, event_id: str | None = None, recipients: tuple[str, ...] = RECIPIENTS) -> list[str]:
+        ids: list[str] = []
+        for recipient in recipients:
+            if recipient not in self.RECIPIENTS:
+                raise ValueError("recipient is outside the governed notification plane")
+            eid = event_id or str(uuid.uuid4())
+            if event_id is not None and len(recipients) > 1:
+                eid = f"{event_id}:{recipient}"
+            self.db.execute("INSERT OR IGNORE INTO outbox(event_id,recipient,kind,payload,available_at) VALUES(?,?,?,?,?)", (eid, recipient, kind, json.dumps(dict(payload), sort_keys=True), time.time()))
+            ids.append(eid)
+        self.db.commit()
+        return ids
+
+    def due(self, *, now: float | None = None, limit: int = 20) -> list[OutboxEvent]:
+        now = time.time() if now is None else now
+        rows = self.db.execute("SELECT event_id,recipient,kind,payload,attempts,available_at,acknowledged FROM outbox WHERE delivered_at IS NULL AND available_at<=? ORDER BY available_at,event_id LIMIT ?", (now, limit)).fetchall()
+        return [OutboxEvent(r[0], r[1], r[2], json.loads(r[3]), r[4], r[5], bool(r[6])) for r in rows]
+
+    def delivered(self, event_id: str, *, now: float | None = None) -> None:
+        self.db.execute("UPDATE outbox SET delivered_at=? WHERE event_id=?", (time.time() if now is None else now, event_id))
+        self.db.commit()
+
+    def failed(self, event_id: str, error: str, *, now: float | None = None, max_backoff: float = 300.0) -> None:
+        now = time.time() if now is None else now
+        row = self.db.execute("SELECT attempts FROM outbox WHERE event_id=?", (event_id,)).fetchone()
+        attempts = int(row[0]) + 1 if row else 1
+        delay = min(max_backoff, 2 ** min(attempts, 8))
+        self.db.execute("UPDATE outbox SET attempts=?,available_at=?,last_error=? WHERE event_id=?", (attempts, now + delay, error[:200], event_id))
+        self.db.commit()
+
+    def acknowledge(self, event_id: str) -> None:
+        self.db.execute("UPDATE outbox SET acknowledged=1 WHERE event_id=?", (event_id,))
+        self.db.commit()
+
+    def close(self) -> None:
+        self.db.close()
+
+
+class NotificationPlane:
+    def __init__(self, outbox: DurableOutbox) -> None:
+        self.outbox = outbox
+
+    def emit(self, kind: str, payload: Mapping[str, Any], *, event_id: str | None = None) -> list[str]:
+        return self.outbox.enqueue(kind, payload, event_id=event_id)
+
+    def deliver_once(self, send: Callable[[OutboxEvent], None], *, now: float | None = None) -> list[str]:
+        delivered: list[str] = []
+        for event in self.outbox.due(now=now):
+            try:
+                send(event)
+            except Exception as exc:
+                self.outbox.failed(event.event_id, type(exc).__name__, now=now)
+            else:
+                self.outbox.delivered(event.event_id, now=now)
+                delivered.append(event.event_id)
+        return delivered