rex/services.py
Rama master · réplica de lectura del repositorio original.
"""Service initialization for scheduler, email, calendar, messaging, and notifications."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any
from rex.calendar_service import CalendarService
from rex.email_service import EmailService
from rex.notification import EscalationManager, Notifier
from rex.openclaw.event_bus import EventBus, get_event_bus
from rex.scheduler import Scheduler, set_scheduler
@dataclass
class AppServices:
event_bus: EventBus
scheduler: Scheduler
email: EmailService
calendar: CalendarService
# OPENCLAW-REPLACE: sms field retained for API compatibility; None until OpenClaw messaging is wired
sms: Any
notifier: Notifier
escalation_manager: EscalationManager
_SERVICES: AppServices | None = None
def initialize_services(
*,
storage_path: Path | str | None = None,
email_mock_path: Path | str | None = None,
calendar_mock_path: Path | str | None = None,
sms_mock_path: Path | str | None = None,
notifications_path: Path | str | None = None,
now_func: Callable[[], datetime] | None = None,
) -> AppServices:
"""Initialize core services and register default scheduler jobs."""
global _SERVICES
if _SERVICES is not None:
return _SERVICES
event_bus = get_event_bus()
scheduler = Scheduler(storage_path=storage_path, now_func=now_func) # type: ignore[arg-type]
set_scheduler(scheduler)
email = EmailService(event_bus, mock_data_path=email_mock_path) # type: ignore[arg-type]
calendar = CalendarService(event_bus, mock_data_path=calendar_mock_path)
# OPENCLAW-REPLACE: SMS service stub — pending migration to OpenClaw messaging
sms: Any = None # sms_mock_path retained in signature for API compatibility
# Initialize notification system
if notifications_path:
notifications_path = (
Path(notifications_path) if isinstance(notifications_path, str) else notifications_path
)
escalation_manager = EscalationManager()
notifier = Notifier(
storage_path=notifications_path, # type: ignore[arg-type]
escalation_manager=escalation_manager,
)
# Register scheduler handlers
scheduler.register_handler("email_triage", lambda job: _run_email_triage(email))
scheduler.register_handler("calendar_sync", lambda job: _run_calendar_sync(calendar))
scheduler.register_handler("flush_digests", lambda job: notifier.flush_digests()) # type: ignore[arg-type]
scheduler.register_handler(
"check_escalations", lambda job: _check_escalations(notifier, escalation_manager)
)
_register_default_jobs(scheduler)
# Subscribe notifier to event bus events
notifier.setup_event_subscriptions()
_SERVICES = AppServices(
event_bus=event_bus,
scheduler=scheduler,
email=email,
calendar=calendar,
sms=sms,
notifier=notifier,
escalation_manager=escalation_manager,
)
return _SERVICES
def reset_services() -> None:
"""Reset cached services (primarily for tests)."""
global _SERVICES
_SERVICES = None
def _email_triage_owners() -> list[str]:
"""Return the owners scheduled email triage runs for.
Owners are users with explicit email account assignments (plus the
``default`` profile for legacy unassigned accounts). When nothing is
configured, the legacy job runs only as the distinct ``default`` profile
— stub-only, never another user's real mail.
"""
try:
from rex.email_accounts import DEFAULT_PROFILE, EmailAccountResolver
owners = EmailAccountResolver.load().configured_user_ids()
return owners or [DEFAULT_PROFILE]
except Exception as exc:
import logging
logging.getLogger(__name__).warning("Failed to resolve email triage owners: %s", exc)
return []
def _run_email_triage(email: EmailService) -> None:
"""Run scheduled email triage per owner, in isolated owner contexts.
One owner's failure is logged and never falls through to (or exposes)
another owner's account.
"""
import logging
logger = logging.getLogger(__name__)
for owner in _email_triage_owners():
try:
email.triage_unread(user_id=owner)
except Exception as exc:
logger.warning("Scheduled email triage failed for user %r: %s", owner, exc)
def _calendar_sync_owners() -> list[str]:
"""Return the owners scheduled calendar sync runs for.
Owners are users with explicit calendar account assignments (plus the
``default`` profile for legacy/unassigned accounts). When nothing is
configured, the legacy job runs only as the distinct ``default`` profile
— stub-only, never another user's real calendar.
"""
try:
from rex.calendar_accounts import DEFAULT_PROFILE, CalendarAccountResolver
owners = CalendarAccountResolver.load().configured_user_ids()
return owners or [DEFAULT_PROFILE]
except Exception as exc:
import logging
logging.getLogger(__name__).warning("Failed to resolve calendar sync owners: %s", exc)
return []
def _run_calendar_sync(calendar: CalendarService) -> None:
"""Run scheduled calendar sync per owner, in isolated owner contexts.
One owner's failure is logged and never falls through to (or exposes)
another owner's account.
"""
import logging
logger = logging.getLogger(__name__)
for owner in _calendar_sync_owners():
try:
calendar.refresh_upcoming(user_id=owner)
except Exception as exc:
logger.warning("Scheduled calendar sync failed for user %r: %s", owner, exc)
def _check_escalations(notifier: Notifier, escalation_manager: EscalationManager) -> None:
"""Check for notifications that need escalation and resend them."""
to_escalate = escalation_manager.check_escalations()
for notif_id, next_channel in to_escalate:
rule = escalation_manager.pending_escalations.get(notif_id)
if not rule:
continue
try:
notifier.send_to_channel(next_channel, rule.notification)
except Exception as exc:
import logging
logger = logging.getLogger(__name__)
logger.warning(
"Failed to escalate notification %s to %s: %s",
notif_id,
next_channel,
exc,
)
def _register_default_jobs(scheduler: Scheduler) -> None:
defaults = [
("Email triage", 900, "email_triage"),
("Calendar sync", 1800, "calendar_sync"),
("Flush notification digests", 3600, "flush_digests"),
("Check notification escalations", 300, "check_escalations"),
]
for name, interval, handler in defaults:
if scheduler.find_job_by_name(name) is None:
scheduler.add_job(name, interval, handler)
# Register retention cleanup jobs (config-driven; skipped safely if config
# is unavailable or stores are disabled).
_try_register_retention_jobs(scheduler)
def _try_register_retention_jobs(scheduler: Scheduler) -> None:
"""Attempt to register retention cleanup jobs from config.
Failures are logged at DEBUG level and do not prevent startup.
"""
import logging
_log = logging.getLogger(__name__)
try:
from rex.config_manager import load_config
from rex.retention import setup_retention_jobs
raw_config = load_config()
results = setup_retention_jobs(raw_config, scheduler)
if any(results.values()):
_log.info("Retention cleanup jobs registered: %s", results)
else:
_log.debug("No retention cleanup jobs registered (check config)")
except Exception as exc:
_log.debug("Retention cleanup jobs setup skipped: %s", exc)
__all__ = ["AppServices", "initialize_services", "reset_services"]