From d131621de6c1b27aa6930bbfc1f92fa8c401a850 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 11 Mar 2026 20:07:16 -0400 Subject: [PATCH] feat: APScheduler briefing slots, catch-up logic, profile close-out Co-Authored-By: Claude Sonnet 4.6 --- src/fabledassistant/app.py | 9 + .../services/briefing_scheduler.py | 265 ++++++++++++++++++ 2 files changed, 274 insertions(+) create mode 100644 src/fabledassistant/services/briefing_scheduler.py diff --git a/src/fabledassistant/app.py b/src/fabledassistant/app.py index 359ed0b..db17802 100644 --- a/src/fabledassistant/app.py +++ b/src/fabledassistant/app.py @@ -240,6 +240,15 @@ def create_app() -> Quart: asyncio.create_task(_delayed_backfill()) + # Start briefing scheduler + from fabledassistant.services.briefing_scheduler import start_briefing_scheduler + start_briefing_scheduler(asyncio.get_event_loop()) + + @app.after_serving + async def shutdown(): + from fabledassistant.services.briefing_scheduler import stop_briefing_scheduler + stop_briefing_scheduler() + @app.route("/") async def serve_index(): resp = await make_response( diff --git a/src/fabledassistant/services/briefing_scheduler.py b/src/fabledassistant/services/briefing_scheduler.py new file mode 100644 index 0000000..28f371c --- /dev/null +++ b/src/fabledassistant/services/briefing_scheduler.py @@ -0,0 +1,265 @@ +""" +APScheduler-based briefing scheduler. + +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, timezone + +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 + +# Slot definitions: (name, hour, minute) +SLOTS = [ + ("compilation", 4, 0), + ("morning", 8, 0), + ("midday", 12, 0), + ("afternoon", 16, 0), +] + + +async def _get_briefing_enabled_users() -> list[tuple[int, str]]: + """Return [(user_id, model)] for 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"): + enabled.append((row.user_id, "")) # model resolved per-user at runtime + except Exception: + pass + return enabled + + +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: + from fabledassistant.services.rss import refresh_all_feeds + import json + 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) + # Refresh weather for configured locations + 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) + + # Run previous day's profile close-out + await _run_profile_closeout(user_id, model) + + # Create today's conversation and post opening message + 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: + # Inject slot update into today's conversation + conv = await get_or_create_today_conversation(user_id, model) + text = await run_slot_injection(user_id, slot, model) + if text: + # Post as a system-injected user prompt + assistant response pair + await post_message(conv.id, "user", f"[{slot.title()} briefing update]") + await post_message(conv.id, "assistant", text) + + # Send push notification + 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) + + +async def _run_profile_closeout(user_id: int, model: str) -> None: + """ + Read yesterday's briefing conversation, ask the LLM to 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 # Nothing interesting to learn from + + 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) + + +def _run_slot_sync(slot: str, loop: asyncio.AbstractEventLoop) -> None: + """Synchronous wrapper called by APScheduler's background thread.""" + async def _job(): + users = await _get_briefing_enabled_users() + for user_id, _ in users: + try: + await _run_slot_for_user(user_id, slot) + except Exception: + logger.exception("Briefing slot '%s' failed for user %d", slot, user_id) + + future = asyncio.run_coroutine_threadsafe(_job(), loop) + try: + future.result(timeout=600) # 10 min max per slot run + except Exception: + logger.exception("Briefing slot '%s' job failed", slot) + + +async def _catchup_missed_slots(loop: asyncio.AbstractEventLoop) -> None: + """ + On startup, check if any slot was missed within the last 24 hours. + Fire it once if so. Never backfill more than one slot per slot-name. + """ + now = datetime.now(timezone.utc) + today = now.date() + + for slot_name, hour, minute in SLOTS: + slot_time = datetime.combine(today, time(hour, minute), tzinfo=timezone.utc) + if slot_time > now: + continue # Hasn't happened yet today + age = (now - slot_time).total_seconds() + if age > 86400: + continue # More than 24h ago — skip + # Check if we already have a message for this slot today + # Simple heuristic: if today's briefing conversation has messages posted + # after the slot time, consider it covered + users = await _get_briefing_enabled_users() + for user_id, _ in users: + 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, + ) + ) + conv = result.scalars().first() + if conv: + msgs = await session.execute( + select(Message).where( + Message.conversation_id == conv.id, + Message.created_at >= slot_time, + ).limit(1) + ) + if msgs.scalars().first(): + continue # Already covered + # Fire the missed slot + logger.info("Catching up missed briefing slot '%s' for user %d", slot_name, user_id) + 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) + + +def start_briefing_scheduler(loop: asyncio.AbstractEventLoop) -> None: + """ + Start the APScheduler background scheduler. + Must be called from the app's before_serving hook with the running event loop. + """ + global _scheduler + if _scheduler is not None: + return + + _scheduler = BackgroundScheduler(timezone="UTC") + + for slot_name, hour, minute in SLOTS: + _scheduler.add_job( + _run_slot_sync, + CronTrigger(hour=hour, minute=minute), + args=[slot_name, loop], + id=f"briefing_{slot_name}", + replace_existing=True, + misfire_grace_time=3600, # Fire up to 1h late rather than skip + ) + + _scheduler.start() + logger.info("Briefing scheduler started") + + # Catch up missed slots in the background + asyncio.run_coroutine_threadsafe(_catchup_missed_slots(loop), loop) + + +def stop_briefing_scheduler() -> None: + global _scheduler + if _scheduler: + _scheduler.shutdown(wait=False) + _scheduler = None