feat: APScheduler briefing slots, catch-up logic, profile close-out

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

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(

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