FabledScribe/src/fabledassistant/services/curator_scheduler.py

"""APScheduler entry that runs the journal curator periodically.

Phase 2 of the conversation+curator architecture (Fable #172). Every
`CURATOR_INTERVAL_MIN` minutes, scans every journal conversation that
has user messages newer than its `last_curator_run_at` timestamp and
runs `curator.run_curator_for_conversation` against it.

Design notes:
- One global job, not per-user. The scheduler is system-wide because
  the curator runs are bounded (one short Ollama call per conversation)
  and the cost of "scan all journal conversations" is tiny next to the
  cost of an LLM call.
- Idempotent. If no journal conversation has new messages, the job
  does nothing. If a curator run fails, the timestamp is NOT advanced
  so the next sweep retries.
- Bounded concurrency. The job processes conversations sequentially —
  the Ollama server is a single bottleneck and parallelism would
  contend on KV cache slots. With OLLAMA_MAX_LOADED_MODELS=2 (chat +
  curator separate models), the curator can run undisturbed by chat.
- Skipped when the curator has nothing meaningful to do: only
  considers conversations whose newest user message is newer than
  `last_curator_run_at`. Read-only sweeps are common and cheap.

The scheduler is started in app.py alongside the existing journal_scheduler.
"""
from __future__ import annotations

import asyncio
import datetime
import logging
from datetime import timezone

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.interval import IntervalTrigger
from sqlalchemy import func, select, update

from fabledassistant.models import async_session
from fabledassistant.models.conversation import Conversation, Message
from fabledassistant.services.curator import run_curator_for_conversation

logger = logging.getLogger(__name__)

# 15 minutes between sweeps. Adjustable later (settings or env var) if
# the cadence turns out wrong in practice. Tested in conversation: long
# enough that small-talk doesn't generate constant curator runs, short
# enough that captures appear before the user forgets what they said.
CURATOR_INTERVAL_MIN = 15

# Hard cap on conversations processed per sweep — if there's a backlog
# (e.g. service was down for a day), we don't want one sweep to take an
# hour. Anything past this cap rolls to the next sweep.
_MAX_CONVERSATIONS_PER_SWEEP = 20

_scheduler: BackgroundScheduler | None = None
_loop: asyncio.AbstractEventLoop | None = None
_running_lock = asyncio.Lock()  # prevents overlapping sweeps if one runs long


async def _candidate_conversations() -> list[tuple[int, datetime.datetime | None]]:
    """Return (conv_id, last_curator_run_at) for journal conversations that
    have user messages newer than the last curator run. Capped at the
    sweep limit.

    Returning the timestamp alongside the id lets `_sweep` pass it to the
    curator as the `since` cutoff, so each sweep only sees messages added
    since the previous successful pass — no re-extracting already-captured
    beats.
    """
    async with async_session() as session:
        # Per-conversation: newest USER message timestamp, vs last_curator_run_at.
        latest_user_msg = (
            select(
                Message.conversation_id.label("conv_id"),
                func.max(Message.created_at).label("latest_at"),
            )
            .where(Message.role == "user")
            .group_by(Message.conversation_id)
            .subquery()
        )
        stmt = (
            select(Conversation.id, Conversation.last_curator_run_at)
            .join(latest_user_msg, latest_user_msg.c.conv_id == Conversation.id)
            .where(Conversation.conversation_type == "journal")
            .where(
                (Conversation.last_curator_run_at.is_(None))
                | (latest_user_msg.c.latest_at > Conversation.last_curator_run_at)
            )
            .order_by(latest_user_msg.c.latest_at.asc())
            .limit(_MAX_CONVERSATIONS_PER_SWEEP)
        )
        rows = await session.execute(stmt)
        return [(row[0], row[1]) for row in rows.all()]


async def _stamp_last_run(conv_id: int, summary: str | None = None) -> None:
    """Bump last_curator_run_at to now() after a successful pass.

    When `summary` is non-empty, also persists it to `curator_summary`
    so the chat model picks it up on subsequent turns (Phase 3 feedback
    loop). An empty summary means the curator had nothing meaningful
    to capture — we still bump the timestamp (the run succeeded), but
    leave the existing summary in place rather than clobbering useful
    context with "".
    """
    values: dict = {"last_curator_run_at": datetime.datetime.now(timezone.utc)}
    if summary:
        values["curator_summary"] = summary.strip()[:240]
    async with async_session() as session:
        await session.execute(
            update(Conversation).where(Conversation.id == conv_id).values(**values)
        )
        await session.commit()


async def _sweep() -> None:
    """Process all candidate journal conversations.

    Wrapped in a lock so that if a sweep is still running when the next
    interval fires, the new one waits / is skipped rather than running
    concurrently. Ollama doesn't love parallel requests on the same model.
    """
    if _running_lock.locked():
        logger.info("Curator sweep already in progress; skipping this tick")
        return
    async with _running_lock:
        try:
            candidates = await _candidate_conversations()
        except Exception:
            logger.exception("Curator candidate query failed")
            return
        if not candidates:
            logger.debug("Curator sweep: no journal conversations need processing")
            return
        logger.info(
            "Curator sweep: %d journal conversation(s) to process",
            len(candidates),
        )
        for conv_id, last_run_at in candidates:
            try:
                # Pass last_run_at as the `since` cutoff so the curator only
                # examines messages added since the previous successful pass.
                # Without this, every sweep re-loads the curator's default
                # 24h window — and the LLM re-extracts beats from messages
                # that were already captured, producing duplicate moments.
                result = await run_curator_for_conversation(
                    conv_id, since=last_run_at,
                )
                if result.error:
                    logger.warning(
                        "Curator run errored for conv %d (will retry next sweep): %s",
                        conv_id, result.error,
                    )
                    continue
                await _stamp_last_run(conv_id, summary=result.summary)
            except Exception:
                logger.exception(
                    "Curator sweep crashed on conv %d (will retry next sweep)",
                    conv_id,
                )


def _sweep_threadsafe() -> None:
    """BackgroundScheduler runs jobs on its own thread; bridge to the asyncio loop."""
    if _loop is None:
        logger.warning("Curator scheduler tick but no asyncio loop registered")
        return
    asyncio.run_coroutine_threadsafe(_sweep(), _loop)


def start_curator_scheduler(loop: asyncio.AbstractEventLoop) -> None:
    global _scheduler, _loop
    if _scheduler is not None:
        return
    _loop = loop
    _scheduler = BackgroundScheduler()
    _scheduler.add_job(
        _sweep_threadsafe,
        trigger=IntervalTrigger(minutes=CURATOR_INTERVAL_MIN),
        id="curator_sweep",
        replace_existing=True,
        # Don't fire the very first tick immediately — let the app finish
        # boot, then start the cadence. APScheduler computes next_run
        # from now + interval by default.
    )
    _scheduler.start()
    logger.info(
        "Curator scheduler started (interval=%d min, max %d conversations per sweep)",
        CURATOR_INTERVAL_MIN, _MAX_CONVERSATIONS_PER_SWEEP,
    )


def stop_curator_scheduler() -> None:
    global _scheduler
    if _scheduler is not None:
        _scheduler.shutdown(wait=False)
        _scheduler = None
        logger.info("Curator scheduler stopped")