fix(chat,curator): unstick chat from silent generation crashes; curator only sees new messages
Two related reliability fixes. 1. routes/chat.py — guard run_generation against uncaught exceptions. run_generation is launched with asyncio.create_task(); any exception raised inside the coroutine is silently swallowed by the event loop, the buffer stays in GenerationState.RUNNING forever, and every subsequent POST /api/chat/conversations/<id>/messages returns 409 'Generation already in progress' — locking the user out of the chat with no log trail. Observed in dev 2026-05-22: assistant message 768 created at 20:36:59 with status=generating, stayed in that state for an hour+, and four follow-up message attempts returned 409 instantly. The generation task hung before any internal log line could fire, so the only diagnostic was the 409 responses themselves. Wrap run_generation in _run_generation_guarded() that catches exceptions, logs with full traceback, transitions the buffer to ERRORED, emits a final 'done' SSE event so any active stream client closes cleanly, and marks the assistant message status=error in the DB. After this, a stuck conversation recovers on its own the next time the user sends a message — no manual DB poke needed. 2. services/curator_scheduler.py — pass last_curator_run_at as 'since' to the curator so each sweep only sees messages added after the previous successful pass. Previously the scheduler called run_curator_for_conversation(conv_id) with no 'since' argument, so the curator defaulted to its 24h lookback window. Within an active journal session that meant every 15-min sweep re-extracted beats from messages already captured on prior sweeps — producing duplicate moments. _candidate_conversations() now returns (conv_id, last_curator_run_at) tuples; _sweep() threads the timestamp through. First-run case (last_curator_run_at IS NULL) falls back to the curator's default 24h window, which is what we want — process recent backlog on first contact, then only deltas after. Manual trigger path (POST /api/journal/curator/run/<conv_id>) is intentionally NOT changed; it still passes since=None so the 24h re-sweep behaviour is preserved for ad-hoc 'reprocess today' clicks from the UI. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -56,9 +56,15 @@ _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 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.
|
||||
@@ -72,7 +78,7 @@ async def _candidate_conversations() -> list[int]:
|
||||
.subquery()
|
||||
)
|
||||
stmt = (
|
||||
select(Conversation.id)
|
||||
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(
|
||||
@@ -83,7 +89,7 @@ async def _candidate_conversations() -> list[int]:
|
||||
.limit(_MAX_CONVERSATIONS_PER_SWEEP)
|
||||
)
|
||||
rows = await session.execute(stmt)
|
||||
return [row[0] for row in rows.all()]
|
||||
return [(row[0], row[1]) for row in rows.all()]
|
||||
|
||||
|
||||
async def _stamp_last_run(conv_id: int, summary: str | None = None) -> None:
|
||||
@@ -129,9 +135,16 @@ async def _sweep() -> None:
|
||||
"Curator sweep: %d journal conversation(s) to process",
|
||||
len(candidates),
|
||||
)
|
||||
for conv_id in candidates:
|
||||
for conv_id, last_run_at in candidates:
|
||||
try:
|
||||
result = await run_curator_for_conversation(conv_id)
|
||||
# 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",
|
||||
|
||||
Reference in New Issue
Block a user