bvandeusen
83f1676d72
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>
48 lines
1.3 KiB
Python
48 lines
1.3 KiB
Python
"""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")
|