mirror of
https://github.com/agent0ai/agent-zero.git
synced 2026-08-04 05:29:52 +00:00
5c1d50f928
Capture each notification payload together with its GUID and update cursor under the notification manager lock. This prevents concurrent notify_user events from advancing the WebUI cursor without delivering the corresponding toast.\n\nAdd regression coverage for notification cursor continuity and update the helper contracts.
250 lines
8.6 KiB
Python
250 lines
8.6 KiB
Python
from dataclasses import dataclass
|
|
import uuid
|
|
import threading
|
|
from datetime import datetime, timedelta
|
|
from enum import Enum
|
|
|
|
from helpers.localization import Localization
|
|
|
|
|
|
class NotificationType(Enum):
|
|
INFO = "info"
|
|
SUCCESS = "success"
|
|
WARNING = "warning"
|
|
ERROR = "error"
|
|
PROGRESS = "progress"
|
|
|
|
|
|
class NotificationPriority(Enum):
|
|
NORMAL = 10
|
|
HIGH = 20
|
|
|
|
|
|
@dataclass
|
|
class NotificationItem:
|
|
manager: "NotificationManager"
|
|
no: int
|
|
type: NotificationType
|
|
priority: NotificationPriority
|
|
title: str
|
|
message: str
|
|
detail: str # HTML content for expandable details
|
|
timestamp: datetime
|
|
display_time: int = 3 # Display duration in seconds, default 3 seconds
|
|
read: bool = False
|
|
id: str = ""
|
|
group: str = "" # Group identifier for grouping related notifications
|
|
|
|
def __post_init__(self):
|
|
if not self.id:
|
|
self.id = str(uuid.uuid4())
|
|
# Ensure type is always NotificationType
|
|
if isinstance(self.type, str):
|
|
self.type = NotificationType(self.type)
|
|
|
|
def mark_read(self):
|
|
self.read = True
|
|
self.manager.update_item(self.no, read=True)
|
|
|
|
def output(self):
|
|
return {
|
|
"no": self.no,
|
|
"id": self.id,
|
|
"type": self.type.value if isinstance(self.type, NotificationType) else self.type,
|
|
"priority": self.priority.value if isinstance(self.priority, NotificationPriority) else self.priority,
|
|
"title": self.title,
|
|
"message": self.message,
|
|
"detail": self.detail,
|
|
"timestamp": Localization.get().serialize_datetime(self.timestamp),
|
|
"display_time": self.display_time,
|
|
"read": self.read,
|
|
"group": self.group,
|
|
}
|
|
|
|
|
|
class NotificationManager:
|
|
def __init__(self, max_notifications: int = 100):
|
|
self._lock = threading.RLock()
|
|
self.guid: str = str(uuid.uuid4())
|
|
self.updates: list[int] = []
|
|
self.notifications: list[NotificationItem] = []
|
|
self.max_notifications = max_notifications
|
|
|
|
@staticmethod
|
|
def send_notification(
|
|
type: NotificationType,
|
|
priority: NotificationPriority,
|
|
message: str,
|
|
title: str = "",
|
|
detail: str = "",
|
|
display_time: int = 3,
|
|
group: str = "",
|
|
id: str = "",
|
|
) -> NotificationItem:
|
|
from agent import AgentContext
|
|
return AgentContext.get_notification_manager().add_notification(
|
|
type, priority, message, title, detail, display_time, group, id
|
|
)
|
|
|
|
def add_notification(
|
|
self,
|
|
type: NotificationType,
|
|
priority: NotificationPriority,
|
|
message: str,
|
|
title: str = "",
|
|
detail: str = "",
|
|
display_time: int = 3,
|
|
group: str = "",
|
|
id: str = "",
|
|
) -> NotificationItem:
|
|
with self._lock:
|
|
existing = None
|
|
if id:
|
|
existing = next((n for n in self.notifications if n.id == id), None)
|
|
|
|
if existing:
|
|
existing.type = NotificationType(type)
|
|
existing.priority = NotificationPriority(priority)
|
|
existing.title = title
|
|
existing.message = message
|
|
existing.detail = detail
|
|
existing.timestamp = Localization.get().now()
|
|
existing.display_time = display_time
|
|
existing.group = group
|
|
existing.read = False
|
|
self.updates.append(existing.no)
|
|
item = existing
|
|
else:
|
|
# Create notification item
|
|
item = NotificationItem(
|
|
manager=self,
|
|
no=len(self.notifications),
|
|
type=NotificationType(type),
|
|
priority=NotificationPriority(priority),
|
|
title=title,
|
|
message=message,
|
|
detail=detail,
|
|
timestamp=Localization.get().now(),
|
|
display_time=display_time,
|
|
id=id,
|
|
group=group,
|
|
)
|
|
|
|
self.notifications.append(item)
|
|
self.updates.append(item.no)
|
|
self._enforce_limit()
|
|
|
|
from helpers.state_monitor_integration import mark_dirty_all
|
|
mark_dirty_all(reason="notification.NotificationManager.add_notification")
|
|
return item
|
|
|
|
def _enforce_limit(self):
|
|
with self._lock:
|
|
if len(self.notifications) > self.max_notifications:
|
|
# Remove oldest notifications
|
|
to_remove = len(self.notifications) - self.max_notifications
|
|
self.notifications = self.notifications[to_remove:]
|
|
# Adjust notification numbers
|
|
for i, notification in enumerate(self.notifications):
|
|
notification.no = i
|
|
# Adjust updates list
|
|
self.updates = [no - to_remove for no in self.updates if no >= to_remove]
|
|
|
|
def get_recent_notifications(self, seconds: int = 30) -> list[NotificationItem]:
|
|
cutoff = Localization.get().now() - timedelta(seconds=seconds)
|
|
with self._lock:
|
|
return [n for n in self.notifications if n.timestamp >= cutoff]
|
|
|
|
def output(self, start: int | None = None, end: int | None = None) -> list[dict]:
|
|
return self.output_with_state(start, end)[0]
|
|
|
|
def output_with_state(
|
|
self, start: int | None = None, end: int | None = None
|
|
) -> tuple[list[dict], str, int]:
|
|
with self._lock:
|
|
if start is None:
|
|
start = 0
|
|
if end is None:
|
|
end = len(self.updates)
|
|
updates = self.updates[start:end]
|
|
out = []
|
|
seen = set()
|
|
for update in updates:
|
|
if update not in seen and update < len(self.notifications):
|
|
out.append(self.notifications[update].output())
|
|
seen.add(update)
|
|
return out, self.guid, len(self.updates)
|
|
|
|
def output_all(self) -> list[dict]:
|
|
with self._lock:
|
|
notifications = list(self.notifications)
|
|
return [n.output() for n in notifications]
|
|
|
|
def mark_read_by_ids(self, notification_ids: list[str]) -> int:
|
|
ids = {nid for nid in notification_ids if isinstance(nid, str) and nid.strip()}
|
|
if not ids:
|
|
return 0
|
|
|
|
changed_nos: list[int] = []
|
|
with self._lock:
|
|
for notification in self.notifications:
|
|
if notification.id in ids and not notification.read:
|
|
notification.read = True
|
|
changed_nos.append(notification.no)
|
|
if changed_nos:
|
|
self.updates.extend(changed_nos)
|
|
|
|
if not changed_nos:
|
|
return 0
|
|
|
|
from helpers.state_monitor_integration import mark_dirty_all
|
|
mark_dirty_all(reason="notification.NotificationManager.mark_read_by_ids")
|
|
return len(changed_nos)
|
|
|
|
def update_item(self, no: int, **kwargs) -> None:
|
|
self._update_item(no, **kwargs)
|
|
|
|
def _update_item(self, no: int, **kwargs):
|
|
changed = False
|
|
with self._lock:
|
|
if no < len(self.notifications):
|
|
item = self.notifications[no]
|
|
for key, value in kwargs.items():
|
|
if hasattr(item, key):
|
|
setattr(item, key, value)
|
|
self.updates.append(no)
|
|
changed = True
|
|
|
|
if not changed:
|
|
return
|
|
|
|
from helpers.state_monitor_integration import mark_dirty_all
|
|
mark_dirty_all(reason="notification.NotificationManager._update_item")
|
|
|
|
def mark_all_read(self):
|
|
changed_nos: list[int] = []
|
|
with self._lock:
|
|
for notification in self.notifications:
|
|
if not notification.read:
|
|
notification.read = True
|
|
changed_nos.append(notification.no)
|
|
if changed_nos:
|
|
self.updates.extend(changed_nos)
|
|
|
|
if not changed_nos:
|
|
return
|
|
|
|
from helpers.state_monitor_integration import mark_dirty_all
|
|
mark_dirty_all(reason="notification.NotificationManager.mark_all_read")
|
|
|
|
def clear_all(self):
|
|
with self._lock:
|
|
self.notifications = []
|
|
self.updates = []
|
|
self.guid = str(uuid.uuid4())
|
|
from helpers.state_monitor_integration import mark_dirty_all
|
|
mark_dirty_all(reason="notification.NotificationManager.clear_all")
|
|
|
|
def get_notifications_by_type(self, type: NotificationType) -> list[NotificationItem]:
|
|
with self._lock:
|
|
return [n for n in self.notifications if n.type == type]
|