Merge pull request 'dev to main: improvements and bug fixes' (#4) from dev into main
Reviewed-on: bvandeusen/FabledAssistant#4
This commit was merged in pull request #4.
This commit is contained in:
@@ -328,6 +328,8 @@ export interface BriefingConfig {
|
||||
work_days: number[];
|
||||
slots: BriefingSlots;
|
||||
notifications: boolean;
|
||||
temp_unit: 'C' | 'F';
|
||||
timezone: string;
|
||||
}
|
||||
|
||||
export interface BriefingFeed {
|
||||
@@ -360,6 +362,8 @@ const DEFAULT_BRIEFING_CONFIG: BriefingConfig = {
|
||||
work_days: [1, 2, 3, 4, 5],
|
||||
slots: { compilation: true, morning: true, midday: false, afternoon: false },
|
||||
notifications: true,
|
||||
temp_unit: 'C',
|
||||
timezone: '',
|
||||
};
|
||||
|
||||
export async function getBriefingConfig(): Promise<BriefingConfig> {
|
||||
|
||||
@@ -19,6 +19,8 @@ const config = reactive<BriefingConfig>({
|
||||
work_days: [1, 2, 3, 4, 5],
|
||||
slots: { compilation: true, morning: true, midday: false, afternoon: false },
|
||||
notifications: true,
|
||||
temp_unit: 'C',
|
||||
timezone: '',
|
||||
})
|
||||
|
||||
// Step 2 — locations
|
||||
|
||||
+28
-17
@@ -109,6 +109,28 @@ export const useChatStore = defineStore("chat", () => {
|
||||
}
|
||||
}
|
||||
|
||||
// Drain the next queued message for a conversation, if conditions are met.
|
||||
// Called both at stream-end and after fetchConversation, so orphaned queue
|
||||
// messages (e.g. from a navigation away mid-stream) are picked up on return.
|
||||
function _tryDrainQueue(convId: number) {
|
||||
const queue = convQueues.value[convId];
|
||||
if (!queue?.length) return;
|
||||
if (isStreamingConv(convId)) return; // stream-end will drain naturally
|
||||
if (currentConversation.value?.id !== convId) return; // not our conversation
|
||||
const next = queue.shift()!;
|
||||
_saveQueue(convId);
|
||||
setTimeout(() => sendMessage(
|
||||
next.content,
|
||||
next.contextNoteId,
|
||||
next.includeNoteIds,
|
||||
next.think,
|
||||
next.contextNoteTitle,
|
||||
next.excludeNoteIds,
|
||||
next.ragProjectId,
|
||||
next.workspaceProjectId,
|
||||
), 0);
|
||||
}
|
||||
|
||||
function clearQueue() {
|
||||
const id = currentConversation.value?.id;
|
||||
if (id) {
|
||||
@@ -158,6 +180,9 @@ export const useChatStore = defineStore("chat", () => {
|
||||
`/api/chat/conversations/${id}`
|
||||
);
|
||||
_loadQueue(id);
|
||||
// Drain any messages that were queued but never sent because the user
|
||||
// navigated away before the previous stream finished.
|
||||
_tryDrainQueue(id);
|
||||
} catch (e) {
|
||||
useToastStore().show("Failed to load conversation", "error");
|
||||
throw e;
|
||||
@@ -431,23 +456,9 @@ export const useChatStore = defineStore("chat", () => {
|
||||
s.pendingTool = null;
|
||||
}
|
||||
|
||||
// Process next queued message, if any.
|
||||
// Use setTimeout so this frame resolves before the next send begins.
|
||||
const queue = convQueues.value[convId];
|
||||
if (queue?.length && currentConversation.value?.id === convId) {
|
||||
const next = queue.shift()!;
|
||||
_saveQueue(convId);
|
||||
setTimeout(() => sendMessage(
|
||||
next.content,
|
||||
next.contextNoteId,
|
||||
next.includeNoteIds,
|
||||
next.think,
|
||||
next.contextNoteTitle,
|
||||
next.excludeNoteIds,
|
||||
next.ragProjectId,
|
||||
next.workspaceProjectId,
|
||||
), 0);
|
||||
}
|
||||
// Process next queued message if this is still the active conversation.
|
||||
// If the user has navigated away, _tryDrainQueue will fire on fetchConversation.
|
||||
_tryDrainQueue(convId);
|
||||
}
|
||||
|
||||
async function reconnectIfGenerating(convId: number): Promise<void> {
|
||||
|
||||
@@ -126,6 +126,8 @@ const briefingConfig = ref<BriefingConfig>({
|
||||
work_days: [1, 2, 3, 4, 5],
|
||||
slots: { compilation: true, morning: true, midday: false, afternoon: false },
|
||||
notifications: true,
|
||||
temp_unit: 'C',
|
||||
timezone: '',
|
||||
});
|
||||
const briefingFeeds = ref<BriefingFeed[]>([]);
|
||||
const briefingSaving = ref(false);
|
||||
@@ -138,6 +140,10 @@ const addingFeed = ref(false);
|
||||
async function loadBriefingTab() {
|
||||
briefingConfig.value = await getBriefingConfig();
|
||||
briefingFeeds.value = await getBriefingFeeds();
|
||||
// Auto-populate timezone from browser if not already stored.
|
||||
if (!briefingConfig.value.timezone) {
|
||||
briefingConfig.value.timezone = Intl.DateTimeFormat().resolvedOptions().timeZone;
|
||||
}
|
||||
}
|
||||
|
||||
async function geocodeLocation(key: 'home' | 'work') {
|
||||
@@ -173,6 +179,7 @@ function toggleWorkDay(day: number) {
|
||||
days.sort();
|
||||
}
|
||||
|
||||
|
||||
async function saveBriefingSettings() {
|
||||
briefingSaving.value = true;
|
||||
briefingSaved.value = false;
|
||||
@@ -1316,6 +1323,22 @@ function formatUserDate(iso: string): string {
|
||||
</label>
|
||||
<p class="field-hint">Look up weather for locations in today's calendar events.</p>
|
||||
</div>
|
||||
|
||||
<div class="field" style="margin-top: 1rem">
|
||||
<label class="field-label">Temperature unit</label>
|
||||
<div class="briefing-unit-toggle">
|
||||
<button
|
||||
type="button"
|
||||
:class="['briefing-unit-btn', { active: briefingConfig.temp_unit === 'C' }]"
|
||||
@click="briefingConfig.temp_unit = 'C'"
|
||||
>°C</button>
|
||||
<button
|
||||
type="button"
|
||||
:class="['briefing-unit-btn', { active: briefingConfig.temp_unit === 'F' }]"
|
||||
@click="briefingConfig.temp_unit = 'F'"
|
||||
>°F</button>
|
||||
</div>
|
||||
</div>
|
||||
</section>
|
||||
|
||||
<!-- Work schedule -->
|
||||
@@ -1333,13 +1356,42 @@ function formatUserDate(iso: string): string {
|
||||
</div>
|
||||
</section>
|
||||
|
||||
<!-- Timezone -->
|
||||
<section class="settings-section full-width">
|
||||
<h2>Timezone</h2>
|
||||
<p class="section-desc">
|
||||
Briefing slots fire at the times below in this timezone.
|
||||
Auto-detected from your browser — override if the server should use a different zone.
|
||||
</p>
|
||||
<div class="briefing-timezone-row">
|
||||
<input
|
||||
type="text"
|
||||
class="input"
|
||||
v-model="briefingConfig.timezone"
|
||||
placeholder="e.g. America/New_York"
|
||||
style="flex: 1"
|
||||
/>
|
||||
<button
|
||||
type="button"
|
||||
class="btn-secondary"
|
||||
@click="briefingConfig.timezone = Intl.DateTimeFormat().resolvedOptions().timeZone"
|
||||
>Detect</button>
|
||||
</div>
|
||||
<p class="field-hint">
|
||||
Use an
|
||||
<a href="https://en.wikipedia.org/wiki/List_of_tz_database_time_zones" target="_blank" rel="noopener">IANA timezone name</a>
|
||||
(e.g. <code>Europe/London</code>, <code>America/Chicago</code>).
|
||||
Your browser reports: <strong>{{ Intl.DateTimeFormat().resolvedOptions().timeZone }}</strong>
|
||||
</p>
|
||||
</section>
|
||||
|
||||
<!-- Slots -->
|
||||
<section class="settings-section full-width">
|
||||
<h2>Scheduled Slots</h2>
|
||||
<p class="section-desc">Each active slot will post an update into your briefing conversation.</p>
|
||||
<p class="section-desc">Each active slot posts an update into your briefing conversation at the listed local time.</p>
|
||||
<div class="briefing-slot-list">
|
||||
<div
|
||||
v-for="[key, label, time] in ([
|
||||
v-for="[key, label, localTime] in ([
|
||||
['compilation', 'Morning briefing', '4:00 am'],
|
||||
['morning', 'Office check-in', '8:00 am'],
|
||||
['midday', 'Midday update', '12:00 pm'],
|
||||
@@ -1350,13 +1402,16 @@ function formatUserDate(iso: string): string {
|
||||
>
|
||||
<div class="briefing-slot-info">
|
||||
<span class="briefing-slot-label">{{ label }}</span>
|
||||
<span class="briefing-slot-time">{{ time }}</span>
|
||||
<span class="briefing-slot-time">{{ localTime }}</span>
|
||||
</div>
|
||||
<label>
|
||||
<input type="checkbox" v-model="briefingConfig.slots[key]" />
|
||||
</label>
|
||||
</div>
|
||||
</div>
|
||||
<p v-if="briefingConfig.timezone" class="field-hint" style="margin-top: 0.5rem">
|
||||
Firing in timezone: <strong>{{ briefingConfig.timezone }}</strong>
|
||||
</p>
|
||||
</section>
|
||||
|
||||
<!-- RSS Feeds -->
|
||||
@@ -2697,6 +2752,38 @@ function formatUserDate(iso: string): string {
|
||||
flex-wrap: wrap;
|
||||
margin-top: 0.5rem;
|
||||
}
|
||||
.briefing-timezone-row {
|
||||
display: flex;
|
||||
gap: 0.5rem;
|
||||
align-items: center;
|
||||
margin-bottom: 0.5rem;
|
||||
}
|
||||
.briefing-unit-toggle {
|
||||
display: flex;
|
||||
gap: 0;
|
||||
border: 1px solid var(--color-border);
|
||||
border-radius: 8px;
|
||||
overflow: hidden;
|
||||
width: fit-content;
|
||||
margin-top: 0.25rem;
|
||||
}
|
||||
.briefing-unit-btn {
|
||||
padding: 0.35rem 1rem;
|
||||
background: var(--color-bg-card);
|
||||
color: var(--color-text-muted);
|
||||
font-size: 0.85rem;
|
||||
cursor: pointer;
|
||||
border: none;
|
||||
font-family: inherit;
|
||||
transition: all 0.15s;
|
||||
}
|
||||
.briefing-unit-btn:first-child {
|
||||
border-right: 1px solid var(--color-border);
|
||||
}
|
||||
.briefing-unit-btn.active {
|
||||
background: var(--color-primary);
|
||||
color: #fff;
|
||||
}
|
||||
.briefing-day-btn {
|
||||
padding: 0.35rem 0.65rem;
|
||||
border: 1px solid var(--color-border);
|
||||
|
||||
@@ -80,7 +80,7 @@ watch(
|
||||
activeNoteId.value = tc.result.data.id as number;
|
||||
}
|
||||
if (
|
||||
["create_task", "update_task"].includes(tc.function) &&
|
||||
["create_task", "update_task", "create_milestone", "update_milestone"].includes(tc.function) &&
|
||||
tc.status === "success"
|
||||
) {
|
||||
taskPanelRef.value?.reload();
|
||||
@@ -115,6 +115,7 @@ function scrollToBottom() {
|
||||
}
|
||||
|
||||
watch(() => chatStore.streamingContent, scrollToBottom);
|
||||
watch(() => chatStore.currentConversation?.messages.length, scrollToBottom);
|
||||
|
||||
function togglePanel(panel: keyof typeof panelOpen.value) {
|
||||
const open = panelOpen.value;
|
||||
|
||||
@@ -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})
|
||||
|
||||
|
||||
|
||||
@@ -192,16 +192,26 @@ def _internal_user_prompt(data: dict, slot: str) -> str:
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def _external_user_prompt(data: dict, slot: str) -> str:
|
||||
def _format_temp(value: float, unit: str) -> str:
|
||||
"""Convert Celsius to the requested unit and format as an integer string."""
|
||||
if unit == "F":
|
||||
return f"{value * 9 / 5 + 32:.0f}"
|
||||
return f"{value:.0f}"
|
||||
|
||||
|
||||
def _external_user_prompt(data: dict, slot: str, temp_unit: str = "C") -> str:
|
||||
unit_sym = f"°{temp_unit}"
|
||||
lines = [f"Briefing slot: {slot}", ""]
|
||||
if data["weather"]:
|
||||
lines.append("WEATHER:")
|
||||
for loc in data["weather"]:
|
||||
lines.append(f" {loc['location_label']}:")
|
||||
for day in loc["days"][:3]:
|
||||
t_min = _format_temp(day["temp_min"], temp_unit)
|
||||
t_max = _format_temp(day["temp_max"], temp_unit)
|
||||
lines.append(
|
||||
f" {day['date']}: {day['description']}, "
|
||||
f"{day['temp_min']}–{day['temp_max']}°C, {day['precip_mm']}mm rain"
|
||||
f"{t_min}–{t_max}{unit_sym}, {day['precip_mm']}mm rain"
|
||||
)
|
||||
if loc["changes_since_last_fetch"]:
|
||||
lines.append(" FORECAST CHANGES:")
|
||||
@@ -218,6 +228,18 @@ def _external_user_prompt(data: dict, slot: str) -> str:
|
||||
|
||||
# ── Main entry point ───────────────────────────────────────────────────────────
|
||||
|
||||
async def _get_temp_unit(user_id: int) -> str:
|
||||
"""Read the user's preferred temperature unit from briefing_config ('C' or 'F')."""
|
||||
import json
|
||||
raw = await get_setting(user_id, "briefing_config", "{}")
|
||||
try:
|
||||
config = json.loads(raw) if isinstance(raw, str) else (raw or {})
|
||||
unit = config.get("temp_unit", "C")
|
||||
return unit if unit in ("C", "F") else "C"
|
||||
except Exception:
|
||||
return "C"
|
||||
|
||||
|
||||
async def run_compilation(user_id: int, slot: str, model: str | None = None) -> str:
|
||||
"""
|
||||
Run the full two-lane briefing pipeline for a user and slot.
|
||||
@@ -227,7 +249,10 @@ async def run_compilation(user_id: int, slot: str, model: str | None = None) ->
|
||||
model = await get_setting(user_id, "default_model", Config.OLLAMA_MODEL)
|
||||
|
||||
from fabledassistant.services.briefing_profile import get_profile_body
|
||||
profile_body = await get_profile_body(user_id)
|
||||
profile_body, temp_unit = await asyncio.gather(
|
||||
get_profile_body(user_id),
|
||||
_get_temp_unit(user_id),
|
||||
)
|
||||
|
||||
# Parallel gather
|
||||
internal_data, external_data = await asyncio.gather(
|
||||
@@ -244,7 +269,7 @@ async def run_compilation(user_id: int, slot: str, model: str | None = None) ->
|
||||
),
|
||||
_llm_synthesise(
|
||||
_external_system_prompt(),
|
||||
_external_user_prompt(external_data, slot),
|
||||
_external_user_prompt(external_data, slot, temp_unit),
|
||||
model,
|
||||
),
|
||||
)
|
||||
@@ -268,9 +293,10 @@ async def run_slot_injection(user_id: int, slot: str, model: str | None = None)
|
||||
if model is None:
|
||||
model = await get_setting(user_id, "default_model", Config.OLLAMA_MODEL)
|
||||
|
||||
internal_data, external_data = await asyncio.gather(
|
||||
internal_data, external_data, temp_unit = await asyncio.gather(
|
||||
_gather_internal(user_id),
|
||||
_gather_external(user_id),
|
||||
_get_temp_unit(user_id),
|
||||
)
|
||||
|
||||
system = (
|
||||
@@ -281,6 +307,6 @@ async def run_slot_injection(user_id: int, slot: str, model: str | None = None)
|
||||
f"Slot: {slot}\n\n"
|
||||
+ _internal_user_prompt(internal_data, slot)
|
||||
+ "\n\n"
|
||||
+ _external_user_prompt(external_data, slot)
|
||||
+ _external_user_prompt(external_data, slot, temp_unit)
|
||||
)
|
||||
return await _llm_synthesise(system, user_prompt, model)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -436,6 +436,24 @@ _CORE_TOOLS = [
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "update_milestone",
|
||||
"description": "Update the title, description, or status of an existing milestone.",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"project": {"type": "string", "description": "Project title the milestone belongs to"},
|
||||
"milestone": {"type": "string", "description": "Current milestone title to look up"},
|
||||
"title": {"type": "string", "description": "New title (omit to keep current)"},
|
||||
"description": {"type": "string", "description": "New description (omit to keep current)"},
|
||||
"status": {"type": "string", "enum": ["active", "completed", "cancelled"], "description": "New status (omit to keep current)"},
|
||||
},
|
||||
"required": ["project", "milestone"],
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
"type": "function",
|
||||
"function": {
|
||||
@@ -1426,6 +1444,35 @@ async def execute_tool(user_id: int, tool_name: str, arguments: dict) -> dict:
|
||||
)
|
||||
return {"success": True, "type": "milestone", "data": ms.to_dict()}
|
||||
|
||||
elif tool_name == "update_milestone":
|
||||
from fabledassistant.services.projects import get_project_by_title as _gpbt, list_projects as _lp
|
||||
from fabledassistant.services.milestones import get_milestone_by_title as _gmbt, update_milestone as _um
|
||||
project_name = arguments.get("project", "")
|
||||
milestone_name = arguments.get("milestone", "")
|
||||
if not project_name or not milestone_name:
|
||||
return {"success": False, "error": "Both project and milestone are required"}
|
||||
proj = await _gpbt(user_id, project_name)
|
||||
if proj is None:
|
||||
all_p = await _lp(user_id)
|
||||
matches = [p for p in all_p if project_name.lower() in p.title.lower()]
|
||||
proj = matches[0] if matches else None
|
||||
if proj is None:
|
||||
return {"success": False, "error": f"Project '{project_name}' not found"}
|
||||
ms = await _gmbt(user_id, proj.id, milestone_name)
|
||||
if ms is None:
|
||||
return {"success": False, "error": f"Milestone '{milestone_name}' not found in project '{proj.title}'. Use list_milestones to see available milestones."}
|
||||
fields: dict[str, object] = {}
|
||||
if "title" in arguments:
|
||||
fields["title"] = arguments["title"]
|
||||
if "description" in arguments:
|
||||
fields["description"] = arguments["description"]
|
||||
if "status" in arguments:
|
||||
fields["status"] = arguments["status"]
|
||||
if not fields:
|
||||
return {"success": False, "error": "No fields to update — provide at least one of: title, description, status"}
|
||||
updated = await _um(user_id, ms.id, **fields)
|
||||
return {"success": True, "type": "milestone", "data": updated.to_dict()}
|
||||
|
||||
elif tool_name == "list_milestones":
|
||||
from fabledassistant.services.projects import get_project_by_title as _gpbt, list_projects as _lp
|
||||
from fabledassistant.services.milestones import get_project_milestone_summary
|
||||
|
||||
Reference in New Issue
Block a user