FabledScribe/src/fabledassistant/services/briefing_scheduler.py

"""
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, timezone
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),
]

# Weekly review runs Sunday at 6pm by default
WEEKLY_REVIEW_DAY = "sun"   # APScheduler day_of_week format
WEEKLY_REVIEW_HOUR = 18
WEEKLY_REVIEW_MINUTE = 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, dict]]:
    """Return [(user_id, iana_timezone, config)] for all users with briefing enabled."""
    import json
    async with async_session() as session:
        result = await session.execute(
            select(Setting).where(Setting.key.in_(["briefing_config", "user_timezone"]))
        )
        rows = list(result.scalars().all())

    by_user: dict[int, dict[str, str]] = {}
    for row in rows:
        by_user.setdefault(row.user_id, {})[row.key] = row.value or ""

    enabled = []
    for user_id, settings in by_user.items():
        try:
            config = json.loads(settings.get("briefing_config", "{}") or "{}")
            if config.get("enabled"):
                tz_str = settings.get("user_timezone") or config.get("timezone", "UTC")
                tz = _resolve_timezone(tz_str)
                enabled.append((user_id, tz, config))
        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, config: dict | None = None) -> None:
    """Add (or replace) slot jobs for a user, skipping disabled slots."""
    if _scheduler is None or _loop is None:
        return
    enabled_slots = (config or {}).get("slots", {})
    for slot_name, hour, minute in SLOTS:
        jid = _job_id(user_id, slot_name)
        # compilation always runs; other slots default to True if not in config
        slot_on = enabled_slots.get(slot_name, True)
        if not slot_on:
            if _scheduler.get_job(jid):
                _scheduler.remove_job(jid)
            continue
        _scheduler.add_job(
            _run_user_slot_sync,
            CronTrigger(hour=hour, minute=minute, timezone=tz),
            args=[user_id, slot_name],
            id=jid,
            replace_existing=True,
            misfire_grace_time=3600,
        )
    # Weekly review job — runs once per week
    weekly_jid = _job_id(user_id, "weekly_review")
    weekly_on = enabled_slots.get("weekly_review", True)
    if weekly_on:
        _scheduler.add_job(
            _run_user_slot_sync,
            CronTrigger(
                day_of_week=WEEKLY_REVIEW_DAY,
                hour=WEEKLY_REVIEW_HOUR,
                minute=WEEKLY_REVIEW_MINUTE,
                timezone=tz,
            ),
            args=[user_id, "weekly_review"],
            id=weekly_jid,
            replace_existing=True,
            misfire_grace_time=7200,
        )
    elif _scheduler.get_job(weekly_jid):
        _scheduler.remove_job(weekly_jid)

    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)
    weekly_jid = _job_id(user_id, "weekly_review")
    if _scheduler.get_job(weekly_jid):
        _scheduler.remove_job(weekly_jid)
    logger.info("Removed briefing jobs for user %d", user_id)


# ── Public API ────────────────────────────────────────────────────────────────

def update_user_schedule(user_id: int, config: dict, tz_override: str | None = None) -> None:
    """
    Called when a user saves their briefing config via the settings UI.
    Live-patches the scheduler — no restart required.
    tz_override takes priority over any timezone in config.
    """
    if config.get("enabled"):
        tz_str = tz_override or config.get("timezone", "UTC")
        tz = _resolve_timezone(tz_str)
        _add_user_jobs(user_id, tz, config)
    else:
        _remove_user_jobs(user_id)


# ── Job execution ─────────────────────────────────────────────────────────────

async def _auto_pause_stale_projects(user_id: int) -> list[str]:
    """Pause active projects with no note/task activity in 14+ days. Returns paused project titles."""
    from sqlalchemy import select as _sel, func as _func
    from fabledassistant.models.project import Project
    from fabledassistant.models.note import Note
    from fabledassistant.models import async_session as _session

    paused: list[str] = []
    threshold = datetime.now(timezone.utc) - timedelta(days=14)
    try:
        async with _session() as session:
            projects = (await session.execute(
                _sel(Project).where(Project.user_id == user_id, Project.status == "active")
            )).scalars().all()
            for p in projects:
                latest = (await session.execute(
                    _sel(_func.max(Note.updated_at)).where(Note.project_id == p.id)
                )).scalar()
                if latest is None or latest < threshold:
                    p.status = "paused"
                    paused.append(p.title or "Untitled")
            if paused:
                await session.commit()
                logger.info("Auto-paused %d stale projects for user %d: %s", len(paused), user_id, paused)
    except Exception:
        logger.debug("Auto-pause check failed for user %d", user_id, exc_info=True)
    return paused


async def _persist_agentic_messages(
    conv_id: int,
    slot: str,
    metadata: dict | None,
) -> None:
    """Persist the intermediate turns from an agentic briefing run.

    ``metadata["agentic_messages"]`` is the full message list the agent
    generated — system prompt, user trigger, assistant tool-call turns,
    tool-role results, and the final assistant prose.

    To stay compatible with the existing chat loader in ``routes/chat.py``,
    tool results are folded back into the parent assistant message's
    ``tool_calls[i]["result"]`` field rather than being persisted as
    separate ``role="tool"`` rows. This matches how regular chat
    persists agentic turns, so the follow-up chat endpoint can rehydrate
    the tool sequence using its existing logic.

    Persists everything except the system prompt (implicit in the chat
    pipeline) and the final assistant prose (the caller posts that
    separately with the user-facing metadata block). The synthetic user
    trigger message is persisted so Ollama sees a user→assistant→user
    sequence rather than an orphaned assistant reply — it's tagged as
    intermediate so the UI can hide it.

    Legacy (non-agentic) briefings have no ``agentic_messages`` and this
    function is a no-op.
    """
    from fabledassistant.services.briefing_conversations import post_message

    if not metadata:
        return
    agentic_messages = metadata.get("agentic_messages") or []
    if not agentic_messages:
        return

    # Drop the system prompt (index 0) and the final assistant prose
    # (last item). The caller posts the final prose as its own message.
    middle = agentic_messages[1:-1]

    # Walk the middle sequence, pairing each assistant tool-call turn
    # with the tool-role results that immediately follow it.
    i = 0
    n = len(middle)
    while i < n:
        m = middle[i]
        role = m.get("role", "")
        content = m.get("content", "") or ""
        tag = {"briefing_slot": slot, "briefing_intermediate": True}

        if role == "assistant" and m.get("tool_calls"):
            # Collect subsequent tool-role results, matching them
            # positionally onto this assistant's tool_calls. Normalise
            # each entry to the flat storage format the chat loader
            # expects: {"function": <name>, "arguments": <args>,
            # "result": <result>, "status": "success"|"error"}.
            raw_tool_calls = list(m["tool_calls"])
            flat_tool_calls: list[dict] = []
            result_idx = 0
            j = i + 1

            import json as _json
            for raw_tc in raw_tool_calls:
                fn = raw_tc.get("function") or {}
                name = fn.get("name") if isinstance(fn, dict) else str(fn)
                arguments = fn.get("arguments") if isinstance(fn, dict) else {}
                if isinstance(arguments, str):
                    try:
                        arguments = _json.loads(arguments)
                    except Exception:
                        arguments = {}

                # Pair up with the next available tool-role message
                parsed_result: object = {}
                status = "success"
                if j < n and middle[j].get("role") == "tool":
                    tool_content = middle[j].get("content", "") or ""
                    try:
                        parsed_result = _json.loads(tool_content)
                    except Exception:
                        parsed_result = tool_content
                    if isinstance(parsed_result, dict) and parsed_result.get("success") is False:
                        status = "error"
                    j += 1
                    result_idx += 1

                flat_tool_calls.append({
                    "function": name,
                    "arguments": arguments,
                    "result": parsed_result,
                    "status": status,
                })

            try:
                await post_message(
                    conv_id, "assistant", content,
                    metadata=tag,
                    tool_calls=flat_tool_calls,
                )
            except Exception:
                logger.warning(
                    "Failed to persist agentic assistant turn for conv %d slot %s",
                    conv_id, slot, exc_info=True,
                )
            i = j  # skip the tool results we just folded in
            continue

        if role == "tool":
            # Unpaired tool result — shouldn't normally happen, but be
            # defensive and persist it as an assistant-visible note so we
            # don't lose the receipt entirely.
            i += 1
            continue

        if role == "user":
            # Skip the synthetic user trigger ("Generate my morning briefing…").
            # Persisting it would recreate the exact "[Midday briefing update]"
            # problem PR 2 is designed to eliminate: fake user messages
            # cluttering chat history. The LLM can follow an all-assistant
            # sequence just fine since the chat endpoint injects the real
            # system prompt on follow-up.
            i += 1
            continue

        # assistant without tool_calls — persist as-is (rare intermediate)
        try:
            await post_message(conv_id, role, content, metadata=tag)
        except Exception:
            logger.warning(
                "Failed to persist agentic %s message for conv %d slot %s",
                role, conv_id, slot, exc_info=True,
            )
        i += 1


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

    # Morning slot: skip if today is not a configured work day
    if slot == "morning":
        from fabledassistant.services.user_profile import get_profile
        tz_str = await get_setting(user_id, "user_timezone") or "UTC"
        try:
            user_tz = ZoneInfo(tz_str)
        except Exception:
            user_tz = ZoneInfo("UTC")
        today_abbr = datetime.now(user_tz).strftime("%a")  # 'Mon', 'Tue', …
        profile = await get_profile(user_id)
        work_days = (profile.work_schedule or {}).get("days", ["Mon", "Tue", "Wed", "Thu", "Fri"])
        if today_abbr not in work_days:
            logger.info(
                "Skipping morning slot for user %d — %s not a configured work day",
                user_id, today_abbr,
            )
            return

    model = await get_setting(user_id, "default_model", Config.OLLAMA_MODEL)

    if slot == "compilation":
        # Auto-pause stale projects before compiling the briefing
        await _auto_pause_stale_projects(user_id)

        # 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, metadata = await run_compilation(user_id, slot, model)
        if text:
            # Persist the agentic tool-call sequence as its own messages
            # so follow-up chat can see the receipts. Each intermediate
            # message is tagged with briefing_slot so the chat context
            # loader can decide whether to include them in history.
            await _persist_agentic_messages(conv.id, slot, metadata)
            final_meta = {k: v for k, v in metadata.items() if k != "agentic_messages"}
            final_meta["briefing_slot"] = slot
            await post_message(conv.id, "assistant", text, metadata=final_meta)

    else:
        conv = await get_or_create_today_conversation(user_id, model)
        text, slot_metadata = await run_slot_injection(user_id, slot, model)
        if text:
            # No more synthetic "[Midday briefing update]" user-role
            # messages. Slot updates are plain assistant messages tagged
            # with briefing_slot so the chat endpoint can filter them
            # from the LLM's view of history on follow-ups (they remain
            # visible in the UI).
            await _persist_agentic_messages(conv.id, slot, slot_metadata)
            final_meta = {k: v for k, v in slot_metadata.items() if k != "agentic_messages"}
            final_meta["briefing_slot"] = slot
            await post_message(conv.id, "assistant", text, metadata=final_meta)

    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.user_profile import append_observations
    from fabledassistant.services.llm import generate_completion
    from fabledassistant.services.tz import user_today
    from fabledassistant.models.conversation import Conversation, Message

    # User-local "yesterday" so closeout always targets the day that just
    # ended in the user's timezone, regardless of container TZ.
    yesterday = (await user_today(user_id)) - 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)"
    )
    try:
        observations = (await generate_completion(
            [
                {"role": "system", "content": system},
                {"role": "user", "content": transcript},
            ],
            model,
        )).strip()
    except Exception:
        logger.warning("Profile closeout synthesis failed for user %d", user_id, exc_info=True)
        observations = ""
    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, _config 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, config in users:
        _add_user_jobs(user_id, tz, config)

    from fabledassistant.services.recurrence import spawn_recurring_tasks as _spawn_recurring

    def _run_recurrence_spawn() -> None:
        future = asyncio.run_coroutine_threadsafe(_spawn_recurring(), _loop)
        try:
            count = future.result(timeout=300)
            logger.info("Recurrence spawn: %d task(s) created", count)
        except Exception as exc:
            logger.error("Recurrence spawn failed: %s", exc)

    _scheduler.add_job(
        _run_recurrence_spawn,
        CronTrigger(hour=0, minute=0, timezone="UTC"),
        id="recurrence_daily",
        replace_existing=True,
    )

    def _run_kokoro_update_check() -> None:
        from fabledassistant.services.tts import check_for_kokoro_updates
        future = asyncio.run_coroutine_threadsafe(check_for_kokoro_updates(), _loop)
        try:
            future.result(timeout=300)
        except Exception as exc:
            logger.error("Kokoro update check failed: %s", exc)

    _scheduler.add_job(
        _run_kokoro_update_check,
        CronTrigger(hour=3, minute=0, timezone="UTC"),
        id="kokoro_update_check_daily",
        replace_existing=True,
    )

    _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