bvandeusen/FabledScribe
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 50 Wiki Activity

feat: per-user timezone-aware briefing scheduler

Browse Source 
Briefing slots (4am/8am/12pm/4pm) now fire in each user's local timezone
rather than UTC, so the schedule matches the user's actual day.

Backend:
- briefing_scheduler: replaced 4 global UTC cron jobs with per-user
  CronTrigger jobs keyed to the user's IANA timezone; update_user_schedule()
  live-patches the scheduler when config is saved (no restart needed)
- Catchup logic evaluates missed slots in the user's local timezone
- put_config route calls update_user_schedule() after saving

Frontend:
- New Timezone section in Briefing settings: text input pre-filled from
  browser (Intl.DateTimeFormat) with a Detect button for re-detection
- Slot times shown as fixed local times (4:00 am etc.) with the configured
  timezone displayed beneath, replacing the old UTC-conversion display
- timezone field added to BriefingConfig type and default

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Bryan Van Deusen
2026-03-12 08:34:31 -04:00
parent 6e57ce4555
commit 5ea3bb5aff
4 changed files with 193 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/fabledassistant/routes/briefing.py
+3
View File
@@ -44,6 +44,9 @@ async def get_config():
async def put_config():
data = await request.get_json()
await set_settings_batch(g.user.id, {"briefing_config": json.dumps(data)})
# Live-patch the scheduler so the new timezone takes effect immediately.
from fabledassistant.services.briefing_scheduler import update_user_schedule
update_user_schedule(g.user.id, data)
return jsonify({"ok": True})
src/fabledassistant/services/briefing_scheduler.py
+140 -69
View File
@@ -1,14 +1,20 @@
"""
APScheduler-based briefing scheduler.
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().
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 datetime import date, datetime, time, timedelta
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
@@ -20,8 +26,9 @@ from fabledassistant.models.setting import Setting
logger = logging.getLogger(__name__)
_scheduler: BackgroundScheduler | None = None
_loop: asyncio.AbstractEventLoop | None = None
# Slot definitions: (name, hour, minute)
# Slot definitions: (name, hour, minute) — local time in the user's timezone
SLOTS = [
("compilation", 4, 0),
("morning", 8, 0),
@@ -30,8 +37,22 @@ SLOTS = [
]
# ── 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]]:
"""Return [(user_id, model)] for users with briefing enabled."""
"""Return [(user_id, iana_timezone)] for all users with briefing enabled."""
import json
async with async_session() as session:
result = await session.execute(
@@ -44,12 +65,60 @@ async def _get_briefing_enabled_users() -> list[tuple[int, str]]:
try:
config = json.loads(row.value) if row.value else {}
if config.get("enabled"):
enabled.append((row.user_id, "")) # model resolved per-user at runtime
tz = _resolve_timezone(config.get("timezone", "UTC"))
enabled.append((row.user_id, tz))
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) -> None:
"""Add (or replace) all 4 slot jobs for a user in their timezone."""
if _scheduler is None or _loop is None:
return
for slot_name, hour, minute in SLOTS:
_scheduler.add_job(
_run_user_slot_sync,
CronTrigger(hour=hour, minute=minute, timezone=tz),
args=[user_id, slot_name],
id=_job_id(user_id, slot_name),
replace_existing=True,
misfire_grace_time=3600,
)
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)
logger.info("Removed briefing jobs for user %d", user_id)
# ── Public API ────────────────────────────────────────────────────────────────
def update_user_schedule(user_id: int, config: dict) -> None:
"""
Called when a user saves their briefing config via the settings UI.
Live-patches the scheduler — no restart required.
"""
if config.get("enabled"):
tz = _resolve_timezone(config.get("timezone", "UTC"))
_add_user_jobs(user_id, tz)
else:
_remove_user_jobs(user_id)
# ── Job execution ─────────────────────────────────────────────────────────────
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 (
@@ -64,12 +133,11 @@ async def _run_slot_for_user(user_id: int, slot: str) -> None:
if slot == "compilation":
# Refresh external data first
try:
from fabledassistant.services.rss import refresh_all_feeds
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)
# Refresh weather for configured locations
from fabledassistant.services import weather as wx
for key, loc in config.get("locations", {}).items():
if loc.get("lat") and loc.get("lon"):
@@ -83,25 +151,19 @@ async def _run_slot_for_user(user_id: int, slot: str) -> None:
except Exception:
logger.warning("Pre-compilation refresh failed for user %d", user_id, exc_info=True)
# Run previous day's profile close-out
await _run_profile_closeout(user_id, model)
# Create today's conversation and post opening message
conv = await get_or_create_today_conversation(user_id, model)
text = await run_compilation(user_id, slot, model)
if text:
await post_message(conv.id, "assistant", text)
else:
# Inject slot update into today's conversation
conv = await get_or_create_today_conversation(user_id, model)
text = await run_slot_injection(user_id, slot, model)
if text:
# Post as a system-injected user prompt + assistant response pair
await post_message(conv.id, "user", f"[{slot.title()} briefing update]")
await post_message(conv.id, "assistant", text)
# Send push notification
try:
from fabledassistant.services.push import send_push_notification
slot_labels = {
@@ -122,10 +184,22 @@ async def _run_slot_for_user(user_id: int, slot: str) -> None:
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, ask the LLM to extract preference
observations, and append them to the briefing profile note.
Read yesterday's briefing conversation, extract preference observations,
and append them to the briefing profile note.
"""
from fabledassistant.services.briefing_profile import append_observations
from fabledassistant.services.briefing_pipeline import _llm_synthesise
@@ -149,7 +223,7 @@ async def _run_profile_closeout(user_id: int, model: str) -> None:
messages = list(msgs_result.scalars().all())
if len(messages) < 2:
return # Nothing interesting to learn from
return
transcript = "\n".join(
f"{m.role.upper()}: {m.content[:500]}" for m in messages[-20:]
@@ -166,100 +240,97 @@ async def _run_profile_closeout(user_id: int, model: str) -> None:
await append_observations(user_id, observations)
def _run_slot_sync(slot: str, loop: asyncio.AbstractEventLoop) -> None:
"""Synchronous wrapper called by APScheduler's background thread."""
async def _job():
users = await _get_briefing_enabled_users()
for user_id, _ in users:
try:
await _run_slot_for_user(user_id, slot)
except Exception:
logger.exception("Briefing slot '%s' failed for user %d", slot, user_id)
future = asyncio.run_coroutine_threadsafe(_job(), loop)
try:
future.result(timeout=600) # 10 min max per slot run
except Exception:
logger.exception("Briefing slot '%s' job failed", slot)
# ── Startup / catchup ─────────────────────────────────────────────────────────
async def _catchup_missed_slots(loop: asyncio.AbstractEventLoop) -> None:
"""
On startup, check if any slot was missed within the last 24 hours.
Fire it once if so. Never backfill more than one slot per slot-name.
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).
"""
now = datetime.now(timezone.utc)
today = now.date()
users = await _get_briefing_enabled_users()
for user_id, tz 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_time = datetime.combine(today, time(hour, minute), tzinfo=timezone.utc)
if slot_time > now:
continue # Hasn't happened yet today
age = (now - slot_time).total_seconds()
if age > 86400:
continue # More than 24h ago — skip
# Check if we already have a message for this slot today
# Simple heuristic: if today's briefing conversation has messages posted
# after the slot time, consider it covered
users = await _get_briefing_enabled_users()
for user_id, _ in users:
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,
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_time,
Message.created_at >= slot_utc,
).limit(1)
)
if msgs.scalars().first():
continue # Already covered
# Fire the missed slot
logger.info("Catching up missed briefing slot '%s' for user %d", slot_name, user_id)
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)
logger.exception(
"Catch-up for slot '%s' user %d failed", slot_name, user_id
)
def start_briefing_scheduler(loop: asyncio.AbstractEventLoop) -> None:
"""
Start the APScheduler background scheduler.
Start the APScheduler background scheduler with per-user timezone-aware jobs.
Must be called from the app's before_serving hook with the running event loop.
"""
global _scheduler
global _scheduler, _loop
if _scheduler is not None:
return
_loop = loop
_scheduler = BackgroundScheduler(timezone="UTC")
for slot_name, hour, minute in SLOTS:
_scheduler.add_job(
_run_slot_sync,
CronTrigger(hour=hour, minute=minute),
args=[slot_name, loop],
id=f"briefing_{slot_name}",
replace_existing=True,
misfire_grace_time=3600, # Fire up to 1h late rather than skip
)
# Schedule jobs synchronously: run the async query in the provided loop
future = asyncio.run_coroutine_threadsafe(_get_briefing_enabled_users(), loop)
try:
users = future.result(timeout=10)
except Exception:
logger.exception("Failed to load briefing users at startup")
users = []
for user_id, tz in users:
_add_user_jobs(user_id, tz)
_scheduler.start()
logger.info("Briefing scheduler started")
logger.info(
"Briefing scheduler started with %d user(s) across %d job(s)",
len(users), len(users) * len(SLOTS),
)
# Catch up missed slots in the background
asyncio.run_coroutine_threadsafe(_catchup_missed_slots(loop), loop)
def stop_briefing_scheduler() -> None:
global _scheduler
global _scheduler, _loop
if _scheduler:
_scheduler.shutdown(wait=False)
_scheduler = None
_loop = None