feat(journal): auto-scheduler for curator (Phase 2)

The curator now runs automatically every 15 minutes against any
journal conversation that has user messages newer than its last
curator run. Manual triggers from Phase 1b still work and now also
stamp the timestamp so the scheduler doesn't double-process.

Migration 0048:
- conversations.last_curator_run_at (timestamptz, nullable).
- Partial index ix_conversations_journal_last_curator on the column
  filtered to conversation_type='journal'. The scheduler's candidate
  query is "journal AND (NULL OR stale)" so an index narrowed to
  journal rows is the right shape — index size stays small even on
  instances with many non-journal conversations.

models/conversation.py:
- New `last_curator_run_at` column on Conversation. DateTime imported.

services/curator_scheduler.py (new):
- IntervalTrigger every 15 min via BackgroundScheduler (same pattern
  as journal_scheduler.py).
- _candidate_conversations(): SELECT journal conversations where the
  newest user message is newer than last_curator_run_at (or NULL).
  Capped at 20 per sweep so a backlog after downtime doesn't stall
  the scheduler.
- _sweep() processes candidates sequentially under an asyncio.Lock
  so overlapping ticks can't double-fire on the same conversation.
  Failed runs leave the timestamp alone — natural retry on next sweep.
- start_/stop_curator_scheduler() wired into app.py boot/shutdown.

routes/journal.py:
- Manual /api/journal/curator/run/<conv_id> stamps last_curator_run_at
  on success. Errors don't stamp so the scheduler retries.

What's still pending:
- Phase 3: feedback loop (curator summary into chat context). Currently
  the curator's summary lives in the run result but doesn't reach the
  chat model.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

alembic/versions/0048_conversation_last_curator_run.py

@@ -0,0 +1,47 @@
+"""conversations.last_curator_run_at — tracks scheduler progress per-conversation
+
+Revision ID: 0048
+Revises: 0047
+Create Date: 2026-05-22
+
+Phase 2 of the conversation+curator architecture (Fable #172). The
+scheduler runs every 15 minutes and processes any journal conversation
+with messages newer than its `last_curator_run_at` timestamp. NULL
+means "never run; process all of today on first sweep."
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "0048"
+down_revision = "0047"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "conversations",
+        sa.Column(
+            "last_curator_run_at",
+            sa.DateTime(timezone=True),
+            nullable=True,
+        ),
+    )
+    # Indexed because the scheduler's selection query is
+    # WHERE conversation_type='journal' AND (last_curator_run_at IS NULL OR ...)
+    # which benefits from a partial index narrowed to journal rows.
+    op.create_index(
+        "ix_conversations_journal_last_curator",
+        "conversations",
+        ["last_curator_run_at"],
+        postgresql_where=sa.text("conversation_type = 'journal'"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(
+        "ix_conversations_journal_last_curator",
+        table_name="conversations",
+    )
+    op.drop_column("conversations", "last_curator_run_at")

src/fabledassistant/app.py

@@ -337,6 +337,10 @@ def create_app() -> Quart:
         )
         start_version_pinning_scheduler(asyncio.get_running_loop())
 
+        # Start curator scheduler (15-min sweep of journal conversations)
+        from fabledassistant.services.curator_scheduler import start_curator_scheduler
+        start_curator_scheduler(asyncio.get_running_loop())
+
         # Voice model loading (enabled via Admin → Config in the UI, or VOICE_ENABLED env var)
         from fabledassistant.services.stt import load_stt_model
         from fabledassistant.services.tts import load_tts_model
@@ -353,6 +357,8 @@ def create_app() -> Quart:
             stop_version_pinning_scheduler,
         )
         stop_version_pinning_scheduler()
+        from fabledassistant.services.curator_scheduler import stop_curator_scheduler
+        stop_curator_scheduler()
 
     @app.route("/")
     async def serve_index():

src/fabledassistant/models/conversation.py

@@ -1,6 +1,6 @@
 import datetime
 
-from sqlalchemy import Date, ForeignKey, Index, Integer, Text
+from sqlalchemy import Date, DateTime, ForeignKey, Index, Integer, Text
 from sqlalchemy import inspect
 from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.orm import Mapped, mapped_column, relationship
@@ -25,6 +25,12 @@ class Conversation(Base, TimestampMixin):
     day_date: Mapped[datetime.date | None] = mapped_column(Date, nullable=True)
     # NULL = orphan notes only; -1 = all notes; positive int = specific project
     rag_project_id: Mapped[int | None] = mapped_column(Integer, nullable=True, default=None)
+    # Curator scheduler bookkeeping (Phase 2, Fable #172). NULL = never run;
+    # scheduler processes journal conversations where this is NULL OR older
+    # than the most recent user message. See services/curator_scheduler.py.
+    last_curator_run_at: Mapped[datetime.datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True, default=None
+    )
 
     messages: Mapped[list["Message"]] = relationship(
         back_populates="conversation",

src/fabledassistant/routes/journal.py

@@ -201,6 +201,22 @@ async def trigger_curator_run(conv_id: int):
 
     from fabledassistant.services.curator import run_curator_for_conversation
     result = await run_curator_for_conversation(conv_id)
+
+    # Stamp last_curator_run_at on success so the scheduler doesn't
+    # immediately re-process the same conversation on its next sweep.
+    # Errored runs intentionally leave the timestamp alone so the
+    # scheduler retries them.
+    if not result.error:
+        import datetime as _dt
+        from sqlalchemy import update as _update
+        async with _async_session() as _sess:
+            await _sess.execute(
+                _update(_Conversation)
+                .where(_Conversation.id == conv_id)
+                .values(last_curator_run_at=_dt.datetime.now(_dt.timezone.utc))
+            )
+            await _sess.commit()
+
     return jsonify(result.to_dict())
 
 

src/fabledassistant/services/curator_scheduler.py

@@ -0,0 +1,175 @@
+"""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[int]:
+    """Return conversation IDs that have user messages newer than the
+    last curator run. Limited to journal type, capped at the sweep limit.
+    """
+    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)
+            .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] for row in rows.all()]
+
+
+async def _stamp_last_run(conv_id: int) -> None:
+    """Bump last_curator_run_at to now() after a successful pass."""
+    async with async_session() as session:
+        await session.execute(
+            update(Conversation)
+            .where(Conversation.id == conv_id)
+            .values(last_curator_run_at=datetime.datetime.now(timezone.utc))
+        )
+        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 in candidates:
+            try:
+                result = await run_curator_for_conversation(conv_id)
+                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)
+            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")