bvandeusen
9cd0de3883
start_briefing_scheduler was called from before_serving (event loop thread) and used run_coroutine_threadsafe(...).result() which blocks the calling thread waiting for the coroutine to complete — but since the calling thread IS the event loop, the coroutine could never run, causing a 10s timeout and zero jobs scheduled. Fix: make start_briefing_scheduler async and await _get_briefing_enabled_users() directly. Also use asyncio.create_task for the catch-up rather than run_coroutine_threadsafe. The background thread jobs (_run_user_slot_sync) continue to use run_coroutine_threadsafe correctly since they run on the APScheduler thread, not the event loop thread. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
337 lines
13 KiB
Python
337 lines
13 KiB
Python
"""
|
|
APScheduler-based briefing scheduler — per-user, timezone-aware.
|
|
|
|
Each enabled user gets 4 individual CronTrigger jobs keyed to their IANA
|
|
timezone (stored in briefing_config.timezone). Changing the config via the
|
|
settings UI calls update_user_schedule() which live-patches the scheduler
|
|
without a restart.
|
|
|
|
Uses a background thread scheduler (not async) because APScheduler 3.x's
|
|
AsyncIOScheduler has known issues with Quart/hypercorn. Jobs are async
|
|
functions wrapped with asyncio.run_coroutine_threadsafe().
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
from datetime import date, datetime, time, timedelta
|
|
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
|
|
|
from apscheduler.schedulers.background import BackgroundScheduler
|
|
from apscheduler.triggers.cron import CronTrigger
|
|
from sqlalchemy import select
|
|
|
|
from fabledassistant.models import async_session
|
|
from fabledassistant.models.setting import Setting
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_scheduler: BackgroundScheduler | None = None
|
|
_loop: asyncio.AbstractEventLoop | None = None
|
|
|
|
# Slot definitions: (name, hour, minute) — local time in the user's timezone
|
|
SLOTS = [
|
|
("compilation", 4, 0),
|
|
("morning", 8, 0),
|
|
("midday", 12, 0),
|
|
("afternoon", 16, 0),
|
|
]
|
|
|
|
|
|
# ── Helpers ───────────────────────────────────────────────────────────────────
|
|
|
|
def _resolve_timezone(tz_str: str) -> str:
|
|
"""Validate and return an IANA timezone string, falling back to UTC."""
|
|
if not tz_str:
|
|
return "UTC"
|
|
try:
|
|
ZoneInfo(tz_str)
|
|
return tz_str
|
|
except (ZoneInfoNotFoundError, KeyError):
|
|
logger.warning("Invalid timezone %r in briefing config, falling back to UTC", tz_str)
|
|
return "UTC"
|
|
|
|
|
|
async def _get_briefing_enabled_users() -> list[tuple[int, str]]:
|
|
"""Return [(user_id, iana_timezone)] for all users with briefing enabled."""
|
|
import json
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
select(Setting).where(Setting.key == "briefing_config")
|
|
)
|
|
rows = list(result.scalars().all())
|
|
|
|
enabled = []
|
|
for row in rows:
|
|
try:
|
|
config = json.loads(row.value) if row.value else {}
|
|
if config.get("enabled"):
|
|
tz = _resolve_timezone(config.get("timezone", "UTC"))
|
|
enabled.append((row.user_id, tz))
|
|
except Exception:
|
|
pass
|
|
return enabled
|
|
|
|
|
|
def _job_id(user_id: int, slot: str) -> str:
|
|
return f"briefing_{slot}_user_{user_id}"
|
|
|
|
|
|
def _add_user_jobs(user_id: int, tz: str) -> None:
|
|
"""Add (or replace) all 4 slot jobs for a user in their timezone."""
|
|
if _scheduler is None or _loop is None:
|
|
return
|
|
for slot_name, hour, minute in SLOTS:
|
|
_scheduler.add_job(
|
|
_run_user_slot_sync,
|
|
CronTrigger(hour=hour, minute=minute, timezone=tz),
|
|
args=[user_id, slot_name],
|
|
id=_job_id(user_id, slot_name),
|
|
replace_existing=True,
|
|
misfire_grace_time=3600,
|
|
)
|
|
logger.info("Scheduled briefing jobs for user %d in timezone %s", user_id, tz)
|
|
|
|
|
|
def _remove_user_jobs(user_id: int) -> None:
|
|
"""Remove all slot jobs for a user."""
|
|
if _scheduler is None:
|
|
return
|
|
for slot_name, _, _ in SLOTS:
|
|
jid = _job_id(user_id, slot_name)
|
|
if _scheduler.get_job(jid):
|
|
_scheduler.remove_job(jid)
|
|
logger.info("Removed briefing jobs for user %d", user_id)
|
|
|
|
|
|
# ── Public API ────────────────────────────────────────────────────────────────
|
|
|
|
def update_user_schedule(user_id: int, config: dict) -> None:
|
|
"""
|
|
Called when a user saves their briefing config via the settings UI.
|
|
Live-patches the scheduler — no restart required.
|
|
"""
|
|
if config.get("enabled"):
|
|
tz = _resolve_timezone(config.get("timezone", "UTC"))
|
|
_add_user_jobs(user_id, tz)
|
|
else:
|
|
_remove_user_jobs(user_id)
|
|
|
|
|
|
# ── Job execution ─────────────────────────────────────────────────────────────
|
|
|
|
async def _run_slot_for_user(user_id: int, slot: str) -> None:
|
|
"""Execute one slot job for one user."""
|
|
from fabledassistant.services.briefing_conversations import (
|
|
get_or_create_today_conversation, post_message
|
|
)
|
|
from fabledassistant.services.briefing_pipeline import run_compilation, run_slot_injection
|
|
from fabledassistant.services.settings import get_setting
|
|
from fabledassistant.config import Config
|
|
|
|
model = await get_setting(user_id, "default_model", Config.OLLAMA_MODEL)
|
|
|
|
if slot == "compilation":
|
|
# Refresh external data first
|
|
try:
|
|
import json
|
|
from fabledassistant.services.rss import refresh_all_feeds
|
|
config_raw = await get_setting(user_id, "briefing_config", "{}")
|
|
config = json.loads(config_raw) if isinstance(config_raw, str) else {}
|
|
await refresh_all_feeds(user_id)
|
|
from fabledassistant.services import weather as wx
|
|
for key, loc in config.get("locations", {}).items():
|
|
if loc.get("lat") and loc.get("lon"):
|
|
await wx.refresh_location_cache(
|
|
user_id=user_id,
|
|
location_key=key,
|
|
location_label=loc.get("label", key),
|
|
lat=loc["lat"],
|
|
lon=loc["lon"],
|
|
)
|
|
except Exception:
|
|
logger.warning("Pre-compilation refresh failed for user %d", user_id, exc_info=True)
|
|
|
|
await _run_profile_closeout(user_id, model)
|
|
conv = await get_or_create_today_conversation(user_id, model)
|
|
text = await run_compilation(user_id, slot, model)
|
|
if text:
|
|
await post_message(conv.id, "assistant", text)
|
|
|
|
else:
|
|
conv = await get_or_create_today_conversation(user_id, model)
|
|
text = await run_slot_injection(user_id, slot, model)
|
|
if text:
|
|
await post_message(conv.id, "user", f"[{slot.title()} briefing update]")
|
|
await post_message(conv.id, "assistant", text)
|
|
|
|
try:
|
|
from fabledassistant.services.push import send_push_notification
|
|
slot_labels = {
|
|
"compilation": "Morning briefing ready",
|
|
"morning": "Office briefing ready",
|
|
"midday": "Midday check-in",
|
|
"afternoon": "End of day wrap-up",
|
|
}
|
|
await send_push_notification(
|
|
user_id=user_id,
|
|
title="Briefing",
|
|
body=slot_labels.get(slot, "Briefing update"),
|
|
url="/briefing",
|
|
)
|
|
except Exception:
|
|
logger.debug("Push notification failed for briefing slot %s", slot, exc_info=True)
|
|
|
|
logger.info("Briefing slot '%s' completed for user %d", slot, user_id)
|
|
|
|
|
|
def _run_user_slot_sync(user_id: int, slot: str) -> None:
|
|
"""Synchronous wrapper called by APScheduler's background thread."""
|
|
if _loop is None:
|
|
logger.error("No event loop available for briefing slot %s user %d", slot, user_id)
|
|
return
|
|
future = asyncio.run_coroutine_threadsafe(_run_slot_for_user(user_id, slot), _loop)
|
|
try:
|
|
future.result(timeout=600)
|
|
except Exception:
|
|
logger.exception("Briefing slot '%s' failed for user %d", slot, user_id)
|
|
|
|
|
|
async def _run_profile_closeout(user_id: int, model: str) -> None:
|
|
"""
|
|
Read yesterday's briefing conversation, extract preference observations,
|
|
and append them to the briefing profile note.
|
|
"""
|
|
from fabledassistant.services.briefing_profile import append_observations
|
|
from fabledassistant.services.briefing_pipeline import _llm_synthesise
|
|
from fabledassistant.models.conversation import Conversation, Message
|
|
|
|
yesterday = (date.today() - timedelta(days=1))
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
select(Conversation).where(
|
|
Conversation.user_id == user_id,
|
|
Conversation.conversation_type == "briefing",
|
|
Conversation.briefing_date == yesterday,
|
|
)
|
|
)
|
|
conv = result.scalars().first()
|
|
if not conv:
|
|
return
|
|
msgs_result = await session.execute(
|
|
select(Message).where(Message.conversation_id == conv.id).order_by(Message.created_at)
|
|
)
|
|
messages = list(msgs_result.scalars().all())
|
|
|
|
if len(messages) < 2:
|
|
return
|
|
|
|
transcript = "\n".join(
|
|
f"{m.role.upper()}: {m.content[:500]}" for m in messages[-20:]
|
|
)
|
|
system = (
|
|
"You are reviewing a day's briefing conversation to extract preference observations. "
|
|
"Identify any patterns, preferences, or schedule facts the user revealed. "
|
|
"Write 2-5 short bullet points. Be specific and factual. "
|
|
"Example: '- User skipped news about finance', '- Prefers weather for home location first'. "
|
|
"If nothing notable, output only: (nothing to note)"
|
|
)
|
|
observations = await _llm_synthesise(system, transcript, model)
|
|
if observations and "(nothing to note)" not in observations.lower():
|
|
await append_observations(user_id, observations)
|
|
|
|
|
|
# ── Startup / catchup ─────────────────────────────────────────────────────────
|
|
|
|
async def _catchup_missed_slots(loop: asyncio.AbstractEventLoop) -> None:
|
|
"""
|
|
On startup, fire any slot that was missed in the last 24 hours
|
|
(one catch-up per slot per user, evaluated in the user's local timezone).
|
|
"""
|
|
users = await _get_briefing_enabled_users()
|
|
for user_id, tz in users:
|
|
user_tz = ZoneInfo(tz)
|
|
now_local = datetime.now(user_tz)
|
|
today_local = now_local.date()
|
|
|
|
for slot_name, hour, minute in SLOTS:
|
|
slot_local = datetime.combine(today_local, time(hour, minute), tzinfo=user_tz)
|
|
if slot_local > now_local:
|
|
continue # Not yet due
|
|
age = (now_local - slot_local).total_seconds()
|
|
if age > 86400:
|
|
continue # More than 24h ago — skip
|
|
|
|
# Check if today's conversation already has a message from after slot time
|
|
async with async_session() as session:
|
|
from fabledassistant.models.conversation import Conversation, Message
|
|
result = await session.execute(
|
|
select(Conversation).where(
|
|
Conversation.user_id == user_id,
|
|
Conversation.conversation_type == "briefing",
|
|
Conversation.briefing_date == today_local,
|
|
)
|
|
)
|
|
conv = result.scalars().first()
|
|
if conv:
|
|
# Convert slot_local to UTC for DB comparison (stored as UTC)
|
|
slot_utc = slot_local.astimezone(ZoneInfo("UTC"))
|
|
msgs = await session.execute(
|
|
select(Message).where(
|
|
Message.conversation_id == conv.id,
|
|
Message.created_at >= slot_utc,
|
|
).limit(1)
|
|
)
|
|
if msgs.scalars().first():
|
|
continue # Already covered
|
|
|
|
logger.info(
|
|
"Catching up missed briefing slot '%s' for user %d (tz: %s)",
|
|
slot_name, user_id, tz,
|
|
)
|
|
try:
|
|
await _run_slot_for_user(user_id, slot_name)
|
|
except Exception:
|
|
logger.exception(
|
|
"Catch-up for slot '%s' user %d failed", slot_name, user_id
|
|
)
|
|
|
|
|
|
async def start_briefing_scheduler(loop: asyncio.AbstractEventLoop) -> None:
|
|
"""
|
|
Start the APScheduler background scheduler with per-user timezone-aware jobs.
|
|
Must be awaited from the app's before_serving hook (async context).
|
|
"""
|
|
global _scheduler, _loop
|
|
if _scheduler is not None:
|
|
return
|
|
|
|
_loop = loop
|
|
_scheduler = BackgroundScheduler(timezone="UTC")
|
|
|
|
# Await directly — we're already on the event loop, so run_coroutine_threadsafe
|
|
# would deadlock (it blocks the calling thread, which IS the event loop thread).
|
|
try:
|
|
users = await _get_briefing_enabled_users()
|
|
except Exception:
|
|
logger.exception("Failed to load briefing users at startup")
|
|
users = []
|
|
|
|
for user_id, tz in users:
|
|
_add_user_jobs(user_id, tz)
|
|
|
|
_scheduler.start()
|
|
logger.info(
|
|
"Briefing scheduler started with %d user(s) across %d job(s)",
|
|
len(users), len(users) * len(SLOTS),
|
|
)
|
|
|
|
asyncio.create_task(_catchup_missed_slots(loop))
|
|
|
|
|
|
def stop_briefing_scheduler() -> None:
|
|
global _scheduler, _loop
|
|
if _scheduler:
|
|
_scheduler.shutdown(wait=False)
|
|
_scheduler = None
|
|
_loop = None
|