RRRosmontis
34089d0042
Include utility modules, shared channels, prompts, read-only skills, wardrobe memory and regression tests. Externalize legacy credentials and exclude private runtime state.
395 lines
18 KiB
Python
395 lines
18 KiB
Python
"""Persistent, per-target notification events with command/chat rendering.
|
|
|
|
Producers store structured facts and a deterministic legacy rendering. Delivery
|
|
state is independent from LLM rendering: a model outage must never lose an event.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Awaitable, Callable, Optional
|
|
|
|
import identity
|
|
from channel import get_channel, send_message
|
|
|
|
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
DB_PATH = os.environ.get(
|
|
"AGENT_DB_PATH", os.path.join(BASE_DIR, "data", "agent.db")
|
|
)
|
|
DELIVERY_LEASE_SECONDS = 300.0
|
|
|
|
|
|
@dataclass
|
|
class Event:
|
|
kind: str
|
|
dedupe_key: str
|
|
legacy_text: str
|
|
uid: Optional[int] = None
|
|
payload: dict[str, Any] = field(default_factory=dict)
|
|
channel: Optional[str] = None
|
|
msg_type: Optional[str] = None
|
|
target_id: Optional[int] = None
|
|
created_at: float = field(default_factory=time.time)
|
|
|
|
|
|
Renderer = Callable[[Event, str, int], Awaitable[str]]
|
|
Acknowledger = Callable[[Event], None]
|
|
|
|
|
|
class NotificationCenter:
|
|
def __init__(self, db_path: str = DB_PATH):
|
|
self.db_path = db_path
|
|
self._lock = threading.RLock()
|
|
self._renderer: Optional[Renderer] = None
|
|
self._acknowledgers: dict[str, Acknowledger] = {}
|
|
self._active_conversations: dict[int, int] = {}
|
|
os.makedirs(os.path.dirname(self.db_path), exist_ok=True)
|
|
self._init_db()
|
|
|
|
def _connect(self):
|
|
con = sqlite3.connect(self.db_path, timeout=30)
|
|
con.row_factory = sqlite3.Row
|
|
con.execute("PRAGMA journal_mode=WAL")
|
|
return con
|
|
|
|
def _init_db(self):
|
|
with self._lock, self._connect() as con:
|
|
con.executescript(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS notification_events(
|
|
id TEXT PRIMARY KEY,
|
|
kind TEXT NOT NULL,
|
|
dedupe_key TEXT NOT NULL UNIQUE,
|
|
uid INTEGER,
|
|
channel TEXT,
|
|
msg_type TEXT,
|
|
target_id INTEGER,
|
|
legacy_text TEXT NOT NULL,
|
|
payload TEXT NOT NULL,
|
|
created_at REAL NOT NULL,
|
|
status TEXT NOT NULL DEFAULT 'pending',
|
|
history_written INTEGER NOT NULL DEFAULT 0
|
|
);
|
|
CREATE TABLE IF NOT EXISTS notification_deliveries(
|
|
event_id TEXT NOT NULL,
|
|
channel TEXT NOT NULL,
|
|
msg_type TEXT NOT NULL,
|
|
target_id INTEGER NOT NULL,
|
|
mode TEXT NOT NULL,
|
|
content TEXT,
|
|
status TEXT NOT NULL DEFAULT 'pending',
|
|
attempts INTEGER NOT NULL DEFAULT 0,
|
|
next_attempt_at REAL NOT NULL DEFAULT 0,
|
|
last_error TEXT NOT NULL DEFAULT '',
|
|
claim_owner TEXT,
|
|
claim_until REAL,
|
|
PRIMARY KEY(event_id, channel, msg_type, target_id)
|
|
);
|
|
"""
|
|
)
|
|
# Keep databases created before owner/lease delivery claims usable.
|
|
delivery_columns = {
|
|
row["name"] for row in con.execute("PRAGMA table_info(notification_deliveries)")
|
|
}
|
|
for name, declaration in (
|
|
("claim_owner", "TEXT"), ("claim_until", "REAL")
|
|
):
|
|
if name not in delivery_columns:
|
|
con.execute(
|
|
f"ALTER TABLE notification_deliveries ADD COLUMN {name} {declaration}"
|
|
)
|
|
|
|
def configure(self, renderer: Optional[Renderer] = None):
|
|
self._renderer = renderer
|
|
|
|
def begin_conversation(self, uid: int):
|
|
with self._lock:
|
|
uid = int(uid)
|
|
self._active_conversations[uid] = self._active_conversations.get(uid, 0) + 1
|
|
|
|
def end_conversation(self, uid: int):
|
|
with self._lock:
|
|
uid = int(uid)
|
|
count = self._active_conversations.get(uid, 0) - 1
|
|
if count > 0:
|
|
self._active_conversations[uid] = count
|
|
else:
|
|
self._active_conversations.pop(uid, None)
|
|
|
|
def register_acknowledger(self, kind: str, callback: Acknowledger):
|
|
self._acknowledgers[kind] = callback
|
|
|
|
@staticmethod
|
|
def _row_event(row: sqlite3.Row) -> Event:
|
|
return Event(
|
|
kind=row["kind"], dedupe_key=row["dedupe_key"], uid=row["uid"],
|
|
channel=row["channel"], msg_type=row["msg_type"],
|
|
target_id=row["target_id"], legacy_text=row["legacy_text"],
|
|
payload=json.loads(row["payload"] or "{}"), created_at=row["created_at"],
|
|
)
|
|
|
|
@staticmethod
|
|
def _targets(event: Event) -> list[tuple[str, str, int]]:
|
|
if event.uid is not None:
|
|
return [
|
|
(name, "private", int(target))
|
|
for name, target in identity.push_targets(event.uid)
|
|
]
|
|
if event.channel and event.msg_type and event.target_id is not None:
|
|
return [(event.channel, event.msg_type, int(event.target_id))]
|
|
return []
|
|
|
|
def publish(self, event: Event) -> Optional[str]:
|
|
"""Persist an event and all currently enabled targets.
|
|
|
|
The event ID is returned only when at least one delivery is accepted. A
|
|
repeated key still refreshes the target set, so an event first published
|
|
without a binding can be accepted by a later retry.
|
|
"""
|
|
event_id = uuid.uuid4().hex
|
|
with self._lock, self._connect() as con:
|
|
con.execute(
|
|
"""INSERT INTO notification_events
|
|
(id,kind,dedupe_key,uid,channel,msg_type,target_id,legacy_text,payload,created_at)
|
|
VALUES (?,?,?,?,?,?,?,?,?,?)
|
|
ON CONFLICT(dedupe_key) DO NOTHING""",
|
|
(event_id, event.kind, event.dedupe_key, event.uid, event.channel,
|
|
event.msg_type, event.target_id, event.legacy_text,
|
|
json.dumps(event.payload, ensure_ascii=False), event.created_at),
|
|
)
|
|
row = con.execute(
|
|
"SELECT * FROM notification_events WHERE dedupe_key=?",
|
|
(event.dedupe_key,),
|
|
).fetchone()
|
|
event_id = row["id"]
|
|
stored_event = self._row_event(row)
|
|
targets = self._targets(stored_event)
|
|
for name, msg_type, target in targets:
|
|
con.execute(
|
|
"""INSERT INTO notification_deliveries
|
|
(event_id,channel,msg_type,target_id,mode) VALUES (?,?,?,?,?)
|
|
ON CONFLICT(event_id,channel,msg_type,target_id) DO UPDATE SET
|
|
status='pending',next_attempt_at=0,last_error='',
|
|
claim_owner=NULL,claim_until=NULL
|
|
WHERE notification_deliveries.status='suppressed'""",
|
|
(event_id, name, msg_type, target,
|
|
self._mode(name, stored_event, target)),
|
|
)
|
|
return event_id if targets else None
|
|
|
|
@staticmethod
|
|
def _mode(channel: str, event: Event, target_id: int) -> str:
|
|
try:
|
|
from agent.modes import ModeStore
|
|
uid = event.uid if event.msg_type != "group" else None
|
|
group_id = target_id if event.msg_type == "group" else None
|
|
return ModeStore().get_mode(channel, uid, group_id)
|
|
except Exception:
|
|
return "command"
|
|
|
|
async def _render(self, event: Event, mode: str, target: int) -> str:
|
|
if mode != "chat":
|
|
return event.legacy_text
|
|
if self._renderer is not None:
|
|
try:
|
|
text = (await self._renderer(event, mode, target)).strip()
|
|
if text:
|
|
return text
|
|
except Exception as exc:
|
|
print(f"[notifications] renderer failed {event.kind}: {type(exc).__name__}: {exc}")
|
|
return event.legacy_text
|
|
|
|
def _claim_delivery(self, row: sqlite3.Row, owner: str, now: float) -> bool:
|
|
"""Atomically acquire one due delivery, including an expired prior lease."""
|
|
with self._lock, self._connect() as con:
|
|
cursor = con.execute(
|
|
"""UPDATE notification_deliveries
|
|
SET claim_owner=?,claim_until=?
|
|
WHERE event_id=? AND channel=? AND msg_type=? AND target_id=?
|
|
AND status='pending' AND next_attempt_at<=?
|
|
AND (claim_until IS NULL OR claim_until<=?)""",
|
|
(owner, time.time() + DELIVERY_LEASE_SECONDS,
|
|
row["event_id"], row["channel"], row["msg_type"], row["target_id"],
|
|
now, now),
|
|
)
|
|
return cursor.rowcount == 1
|
|
|
|
async def deliver_due(self, limit: int = 20) -> int:
|
|
"""Claim and deliver a bounded due batch. Returns successful target count."""
|
|
now = time.time()
|
|
with self._lock, self._connect() as con:
|
|
# Do not LIMIT before active-conversation filtering: a deferred head
|
|
# row must not hide later rows that can make progress in this batch.
|
|
rows = con.execute(
|
|
"""SELECT d.*,e.kind,e.dedupe_key,e.uid,e.channel AS event_channel,
|
|
e.msg_type AS event_msg_type,e.target_id AS event_target,
|
|
e.legacy_text,e.payload,e.created_at,e.history_written
|
|
FROM notification_deliveries d
|
|
JOIN notification_events e ON e.id=d.event_id
|
|
WHERE d.status='pending' AND d.next_attempt_at<=?
|
|
AND (d.claim_until IS NULL OR d.claim_until<=?)
|
|
ORDER BY e.created_at,d.event_id""",
|
|
(now, now),
|
|
).fetchall()
|
|
delivered = 0
|
|
claimed = 0
|
|
for row in rows:
|
|
if claimed >= int(limit):
|
|
break
|
|
event = self._row_event(row)
|
|
channel_name, msg_type, target = row["channel"], row["msg_type"], row["target_id"]
|
|
mode = row["mode"]
|
|
if mode == "chat" and row["uid"] is not None:
|
|
with self._lock:
|
|
if self._active_conversations.get(int(row["uid"]), 0) > 0:
|
|
continue
|
|
|
|
owner = uuid.uuid4().hex
|
|
if not self._claim_delivery(row, owner, now):
|
|
continue
|
|
claimed += 1
|
|
ok = False
|
|
try:
|
|
# Push bindings/switches may change after publish. A disabled
|
|
# private target is final for this delivery unless republished.
|
|
if event.uid is not None:
|
|
enabled = any(
|
|
name == channel_name and int(target_id) == int(target)
|
|
for name, target_id in identity.push_targets(event.uid)
|
|
)
|
|
if not enabled:
|
|
with self._lock, self._connect() as con:
|
|
con.execute(
|
|
"""UPDATE notification_deliveries
|
|
SET status='suppressed',last_error='push target disabled',
|
|
claim_owner=NULL,claim_until=NULL
|
|
WHERE event_id=? AND channel=? AND msg_type=? AND target_id=?
|
|
AND status='pending' AND claim_owner=?""",
|
|
(row["event_id"], channel_name, msg_type, target, owner),
|
|
)
|
|
con.execute(
|
|
"""UPDATE notification_events SET status='sent'
|
|
WHERE id=? AND NOT EXISTS(
|
|
SELECT 1 FROM notification_deliveries
|
|
WHERE event_id=? AND status='pending')""",
|
|
(row["event_id"], row["event_id"]),
|
|
)
|
|
continue
|
|
|
|
content = row["content"]
|
|
if not content:
|
|
content = await self._render(event, mode, target)
|
|
with self._lock, self._connect() as con:
|
|
cursor = con.execute(
|
|
"""UPDATE notification_deliveries SET content=?
|
|
WHERE event_id=? AND channel=? AND msg_type=? AND target_id=?
|
|
AND status='pending' AND claim_owner=?""",
|
|
(content, row["event_id"], channel_name, msg_type, target, owner),
|
|
)
|
|
if cursor.rowcount != 1:
|
|
continue
|
|
channel = get_channel(channel_name)
|
|
error = ""
|
|
try:
|
|
if channel is not None:
|
|
ok = await send_message(channel, msg_type, target, content, mirror=False)
|
|
if not ok:
|
|
error = "channel unavailable or send failed"
|
|
except Exception as exc:
|
|
error = f"{type(exc).__name__}: {exc}"[:300]
|
|
with self._lock, self._connect() as con:
|
|
if ok:
|
|
cursor = con.execute(
|
|
"""UPDATE notification_deliveries
|
|
SET status='sent',last_error='',claim_owner=NULL,claim_until=NULL
|
|
WHERE event_id=? AND channel=? AND msg_type=? AND target_id=?
|
|
AND status='pending' AND claim_owner=?""",
|
|
(row["event_id"], channel_name, msg_type, target, owner),
|
|
)
|
|
if cursor.rowcount != 1:
|
|
continue
|
|
con.execute(
|
|
"""UPDATE notification_events SET status='sent'
|
|
WHERE id=? AND NOT EXISTS(
|
|
SELECT 1 FROM notification_deliveries
|
|
WHERE event_id=? AND status='pending')""",
|
|
(row["event_id"], row["event_id"]),
|
|
)
|
|
else:
|
|
attempts = int(row["attempts"]) + 1
|
|
delay = min(3600, 2 ** min(attempts, 10))
|
|
con.execute(
|
|
"""UPDATE notification_deliveries
|
|
SET attempts=?,next_attempt_at=?,last_error=?,
|
|
claim_owner=NULL,claim_until=NULL
|
|
WHERE event_id=? AND channel=? AND msg_type=? AND target_id=?
|
|
AND status='pending' AND claim_owner=?""",
|
|
(attempts, time.time() + delay, error, row["event_id"],
|
|
channel_name, msg_type, target, owner),
|
|
)
|
|
except Exception as exc:
|
|
# Rendering/database/target lookup failures must not strand a
|
|
# claim until its lease expires when we can release it safely.
|
|
attempts = int(row["attempts"]) + 1
|
|
delay = min(3600, 2 ** min(attempts, 10))
|
|
with self._lock, self._connect() as con:
|
|
con.execute(
|
|
"""UPDATE notification_deliveries
|
|
SET attempts=?,next_attempt_at=?,last_error=?,
|
|
claim_owner=NULL,claim_until=NULL
|
|
WHERE event_id=? AND channel=? AND msg_type=? AND target_id=?
|
|
AND status='pending' AND claim_owner=?""",
|
|
(attempts, time.time() + delay,
|
|
f"{type(exc).__name__}: {exc}"[:300], row["event_id"],
|
|
channel_name, msg_type, target, owner),
|
|
)
|
|
continue
|
|
if ok:
|
|
delivered += 1
|
|
acknowledger = self._acknowledgers.get(event.kind)
|
|
if acknowledger:
|
|
try:
|
|
acknowledger(event)
|
|
except Exception as exc:
|
|
print(f"[notifications] acknowledgement failed {event.kind}: {exc}")
|
|
if mode == "chat":
|
|
self._write_history_once(row["event_id"], event, content)
|
|
return delivered
|
|
|
|
def _write_history_once(self, event_id: str, event: Event, content: str):
|
|
if event.uid is None:
|
|
return
|
|
with self._lock, self._connect() as con:
|
|
row = con.execute(
|
|
"SELECT history_written FROM notification_events WHERE id=?", (event_id,)
|
|
).fetchone()
|
|
if not row or row["history_written"]:
|
|
return
|
|
ref = str(event.payload.get("mid") or event.payload.get("key") or event.dedupe_key)
|
|
try:
|
|
from chat import engine
|
|
engine.append_history(event.uid, "you", f"[event:{event.kind}:{ref}] {content}")
|
|
except Exception as exc:
|
|
print(f"[notifications] history append failed: {exc}")
|
|
return
|
|
con.execute(
|
|
"UPDATE notification_events SET history_written=1 WHERE id=?", (event_id,)
|
|
)
|
|
|
|
async def worker(self, interval: float = 2.0):
|
|
while True:
|
|
try:
|
|
await self.deliver_due()
|
|
except Exception as exc:
|
|
print(f"[notifications] delivery worker: {type(exc).__name__}: {exc}")
|
|
await asyncio.sleep(interval)
|
|
|
|
|
|
center = NotificationCenter()
|