diff --git a/src/fabledassistant/routes/briefing.py b/src/fabledassistant/routes/briefing.py index d3bbfd5..9514537 100644 --- a/src/fabledassistant/routes/briefing.py +++ b/src/fabledassistant/routes/briefing.py @@ -308,14 +308,52 @@ async def manual_trigger(): logger.warning("Pre-trigger refresh failed for user %d", g.user.id, exc_info=True) from fabledassistant.services.briefing_pipeline import run_compilation + from fabledassistant.services.briefing_scheduler import _persist_agentic_messages model = await get_setting(g.user.id, "default_model", "") conv = await get_or_create_today_conversation(g.user.id, model) text, metadata = await run_compilation(g.user.id, slot, model) - msg = await post_message(conv.id, "assistant", text, metadata=metadata) + await _persist_agentic_messages(conv.id, slot, metadata) + final_meta = {k: v for k, v in metadata.items() if k != "agentic_messages"} + final_meta["briefing_slot"] = slot + msg = await post_message(conv.id, "assistant", text, metadata=final_meta) return jsonify({"conversation_id": conv.id, "message_id": msg.id, "slot": slot}) +@briefing_bp.route("/reset-today", methods=["POST"]) +@_REQUIRE +async def reset_today_briefing(): + """Delete all messages in today's briefing conversation. + + The conversation row itself is kept so its id stays stable for any + open UI sessions. Intended for "wipe and start fresh" workflows + driven from the MCP when iterating on prompts. Pair with + ``POST /api/briefing/trigger`` to immediately regenerate. + """ + from datetime import date as _date + from sqlalchemy import delete as _delete + + today = _date.today() + async with async_session() as session: + result = await session.execute( + select(Conversation).where( + Conversation.user_id == g.user.id, + Conversation.conversation_type == "briefing", + Conversation.briefing_date == today, + ) + ) + conv = result.scalars().first() + if conv is None: + return jsonify({"deleted": 0, "conversation_id": None}) + deleted_result = await session.execute( + _delete(Message).where(Message.conversation_id == conv.id) + ) + await session.commit() + deleted = deleted_result.rowcount or 0 + + return jsonify({"deleted": deleted, "conversation_id": conv.id}) + + # ── RSS Reactions ────────────────────────────────────────────────────────────── @briefing_bp.route("/rss-reactions", methods=["POST"]) diff --git a/src/fabledassistant/services/briefing_pipeline.py b/src/fabledassistant/services/briefing_pipeline.py index 7b81c7c..26e4206 100644 --- a/src/fabledassistant/services/briefing_pipeline.py +++ b/src/fabledassistant/services/briefing_pipeline.py @@ -1,19 +1,14 @@ """ -Briefing pipeline: parallel data gather + two-lane LLM synthesis. +Briefing pipeline: agentic tool-use loop + UI metadata gather. Slot names: 'compilation' (4am), 'morning' (8am), 'midday' (12pm), 'afternoon' (4pm) """ import asyncio -import hashlib import logging -from datetime import datetime, timezone +from datetime import datetime from zoneinfo import ZoneInfo, ZoneInfoNotFoundError -import httpx - -from fabledassistant.models import async_session - from fabledassistant.config import Config from fabledassistant.services.settings import get_setting @@ -22,421 +17,6 @@ logger = logging.getLogger(__name__) SLOT_NAMES = ("compilation", "morning", "midday", "afternoon") - -def format_task(task: dict, today: str | None = None) -> str: - parts = [task.get("title", "Untitled")] - if task.get("status"): - parts.append(f"[{task['status'].replace('_', ' ').title()}]") - if task.get("due_date"): - parts.append(f"due {task['due_date']}") - if today and task["due_date"] < today and task.get("status") not in ("done", "cancelled"): - from datetime import date - try: - due = date.fromisoformat(task["due_date"]) - td = date.fromisoformat(today) - days = (td - due).days - parts.append(f"({days} day{'s' if days != 1 else ''} overdue)") - except ValueError: - pass - if task.get("priority") and task["priority"] not in ("none", ""): - parts.append(f"priority: {task['priority']}") - if task.get("project_title"): - parts.append(f"project: {task['project_title']}") - return " — ".join(parts) - - -def compute_task_hash(task: dict) -> str: - """Stable SHA-256 of the task's key change-detectable fields.""" - key = "|".join([ - str(task.get("status") or ""), - str(task.get("priority") or ""), - str(task.get("due_date") or ""), - str(task.get("title") or ""), - ]) - return hashlib.sha256(key.encode()).hexdigest() - - -async def split_changed_tasks( - user_id: int, - tasks: list[dict], -) -> tuple[list[dict], int]: - """ - Compare tasks against the briefing_task_snapshot table. - Returns (changed_tasks, unchanged_count). - changed_tasks includes new tasks (no snapshot row) and tasks whose hash differs. - """ - from sqlalchemy import text - - if not tasks: - return [], 0 - - task_ids = [t["task_id"] for t in tasks if t.get("task_id")] - - async with async_session() as session: - result = await session.execute( - text(""" - SELECT task_id, snapshot_hash - FROM briefing_task_snapshot - WHERE user_id = :uid AND task_id = ANY(:ids) - """).bindparams(uid=user_id, ids=task_ids) - ) - snapshots = {row.task_id: row.snapshot_hash for row in result} - - changed = [] - unchanged_count = 0 - for task in tasks: - current_hash = compute_task_hash(task) - stored_hash = snapshots.get(task.get("task_id")) - if stored_hash is None or stored_hash != current_hash: - changed.append(task) - else: - unchanged_count += 1 - - return changed, unchanged_count - - -async def upsert_task_snapshots(user_id: int, tasks: list[dict]) -> None: - """Upsert snapshot hashes for all tasks included in this briefing.""" - from sqlalchemy import text - - if not tasks: - return - - now = datetime.now(timezone.utc) - async with async_session() as session: - for task in tasks: - task_id = task.get("task_id") - if not task_id: - continue - await session.execute( - text(""" - INSERT INTO briefing_task_snapshot (user_id, task_id, snapshot_hash, last_briefed) - VALUES (:uid, :tid, :hash, :now) - ON CONFLICT (user_id, task_id) - DO UPDATE SET snapshot_hash = EXCLUDED.snapshot_hash, - last_briefed = EXCLUDED.last_briefed - """).bindparams( - uid=user_id, - tid=task_id, - hash=compute_task_hash(task), - now=now, - ) - ) - await session.commit() - - -# ── Internal data gather ────────────────────────────────────────────────────── - -async def _gather_internal(user_id: int) -> dict: - """Collect tasks, calendar events, and project data.""" - from fabledassistant.services.notes import list_notes - from fabledassistant.services.projects import list_projects - from fabledassistant.services.caldav import is_caldav_configured, list_events - - tz_name = await get_setting(user_id, "user_timezone") or "UTC" - try: - user_tz = ZoneInfo(tz_name) - except ZoneInfoNotFoundError: - user_tz = ZoneInfo("UTC") - - today = datetime.now(user_tz).date().isoformat() - - # Tasks: overdue, due today, high priority in-progress - all_tasks: list[dict] = [] - try: - all_task_objs, _total = await list_notes(user_id, is_task=True, limit=100) - # Build project title lookup for enrichment - project_titles: dict[int, str] = {} - try: - for p in await list_projects(user_id): - project_titles[p.id] = p.title or "Untitled" - except Exception: - pass - all_tasks = [ - { - "task_id": t.id, - "title": t.title, - "status": t.status, - "due_date": t.due_date.isoformat() if t.due_date else None, - "priority": t.priority, - "project_title": project_titles.get(t.project_id) if t.project_id else None, - } - for t in all_task_objs - ] - overdue = [ - format_task(t, today) for t in all_tasks - if t.get("due_date") and t["due_date"] < today and t.get("status") not in ("done", "cancelled") - ] - due_today = [ - format_task(t, today) for t in all_tasks - if t.get("due_date") == today and t.get("status") not in ("done", "cancelled") - ] - high_priority = [ - format_task(t, today) for t in all_tasks - if t.get("priority") == "high" and t.get("status") not in ("done", "cancelled") and - format_task(t, today) not in overdue and format_task(t, today) not in due_today - ][:5] - except Exception: - logger.warning("Failed to gather tasks for briefing", exc_info=True) - overdue, due_today, high_priority = [], [], [] - - # Calendar events today — internal store - calendar_events: list[str] = [] - internal_events: list[dict] = [] - try: - from fabledassistant.services.events import list_events as list_internal_events - today_date = datetime.now(user_tz).date() - day_start = datetime(today_date.year, today_date.month, today_date.day, 0, 0, 0, tzinfo=user_tz) - day_end = datetime(today_date.year, today_date.month, today_date.day, 23, 59, 59, tzinfo=user_tz) - internal_events = await list_internal_events( - user_id=user_id, date_from=day_start, date_to=day_end - ) - for e in internal_events: - if e.get("all_day"): - time_str = "all day" - elif e.get("start_dt"): - from datetime import datetime as _dt - start = e["start_dt"] - if isinstance(start, str): - start = _dt.fromisoformat(start) - local_dt = start.astimezone(user_tz) if start.tzinfo else start.replace(tzinfo=timezone.utc).astimezone(user_tz) - time_str = local_dt.strftime("%-I:%M %p") - else: - time_str = "unknown time" - calendar_events.append(f"{e.get('title', 'Event')} at {time_str}") - except Exception: - logger.warning("Failed to gather internal calendar events for briefing", exc_info=True) - # Also pull CalDAV events (deduped) - try: - if await is_caldav_configured(user_id): - caldav_evs = await list_events(user_id, start=today, end=today) - for e in (caldav_evs or []): - summary = f"{e.get('summary', 'Event')} at {e.get('dtstart', 'unknown time')}" - if summary not in calendar_events: - calendar_events.append(summary) - except Exception: - logger.warning("Failed to gather CalDAV calendar events for briefing", exc_info=True) - - # Weather × calendar cross-reference - weather_warnings: list[str] = [] - outdoor_keywords = {"park", "field", "patio", "garden", "outdoor", "trail", "beach", - "pool", "stadium", "yard", "lawn", "hike", "walk", "run", "bike", - "soccer", "baseball", "football", "tennis", "golf", "playground"} - try: - import json as _json - raw_config = await get_setting(user_id, "briefing_config", "{}") - config = _json.loads(raw_config) if isinstance(raw_config, str) else (raw_config or {}) - locations = config.get("locations", {}) - loc = locations.get("home") or locations.get("work") - if loc and loc.get("lat") and loc.get("lon") and internal_events: - from fabledassistant.services.weather import fetch_hourly_precip - hourly_precip = await fetch_hourly_precip(loc["lat"], loc["lon"]) - if hourly_precip: - for e in internal_events: - if e.get("all_day"): - continue - # Check if event title or location contains outdoor keywords - event_text = f"{e.get('title', '')} {e.get('location', '')}".lower() - if not any(kw in event_text for kw in outdoor_keywords): - continue - # Get the event's start hour - start = e.get("start_dt") - if not start: - continue - from datetime import datetime as _dt - if isinstance(start, str): - start = _dt.fromisoformat(start) - local_start = start.astimezone(user_tz) if start.tzinfo else start.replace(tzinfo=timezone.utc).astimezone(user_tz) - # Check precip probability around the event time (hour before through event) - hour_key = local_start.strftime("%Y-%m-%dT%H:00") - precip = hourly_precip.get(hour_key, 0) - if precip >= 30: - time_str = local_start.strftime("%-I:%M %p") - weather_warnings.append( - f"Rain likely ({precip}%) around {time_str} when '{e.get('title', 'Event')}' " - f"is scheduled — consider an umbrella or backup plan" - ) - except Exception: - logger.debug("Weather cross-reference failed", exc_info=True) - - # Recent activity — what was the user working on yesterday? - recent_activity: list[str] = [] - stale_projects: list[str] = [] - try: - from sqlalchemy import select as _sel, func as _func - from fabledassistant.models.note import Note - from fabledassistant.models.project import Project - - yesterday = datetime.now(user_tz) - __import__('datetime').timedelta(days=1) - async with async_session() as session: - # Notes/tasks updated in last 24h, grouped by project - result = await session.execute( - _sel(Project.title, _func.count(Note.id)) - .join(Note, Note.project_id == Project.id) - .where(Note.user_id == user_id) - .where(Note.updated_at >= yesterday) - .group_by(Project.title) - .order_by(_func.count(Note.id).desc()) - .limit(3) - ) - for proj_title, count in result.all(): - recent_activity.append(f"{proj_title}: {count} item{'s' if count != 1 else ''} updated") - - # Stale projects — active projects with no note/task activity in 7+ days - week_ago = datetime.now(user_tz) - __import__('datetime').timedelta(days=7) - active_projects_q = await session.execute( - _sel(Project).where(Project.user_id == user_id, Project.status == "active") - ) - for p in active_projects_q.scalars().all(): - latest = await session.execute( - _sel(_func.max(Note.updated_at)) - .where(Note.project_id == p.id, Note.user_id == user_id) - ) - last_activity = latest.scalar() - if last_activity is None or last_activity < week_ago: - stale_projects.append(p.title or "Untitled") - except Exception: - logger.debug("Failed to gather recent activity for briefing", exc_info=True) - - # Projects: active projects - projects_summary = [] - try: - projects = await list_projects(user_id) - for p in projects[:5]: - projects_summary.append(p.title or "Untitled project") - except Exception: - logger.warning("Failed to gather projects for briefing", exc_info=True) - - # Build prioritized "Your Day" focus list - # Priority: overdue > due today > high priority in-progress > other in-progress - focus_tasks: list[str] = [] - seen_titles: set[str] = set() - for t in sorted(all_tasks, key=lambda x: ( - 0 if (x.get("due_date") and x["due_date"] < today and x.get("status") not in ("done", "cancelled")) else - 1 if (x.get("due_date") == today and x.get("status") not in ("done", "cancelled")) else - 2 if (x.get("priority") == "high" and x.get("status") not in ("done", "cancelled")) else - 3 if (x.get("status") == "in_progress") else 4 - )): - if t.get("status") in ("done", "cancelled"): - continue - formatted = format_task(t, today) - if formatted not in seen_titles: - seen_titles.add(formatted) - focus_tasks.append(formatted) - if len(focus_tasks) >= 5: - break - - # Calendar gap analysis — extract event times for LLM - event_time_blocks: list[str] = [] - for e in internal_events: - try: - if e.get("all_day"): - continue - start = e.get("start_dt") - end = e.get("end_dt") - if not start: - continue - from datetime import datetime as _dt - if isinstance(start, str): - start = _dt.fromisoformat(start) - if isinstance(end, str): - end = _dt.fromisoformat(end) - s_local = start.astimezone(user_tz) if start.tzinfo else start.replace(tzinfo=timezone.utc).astimezone(user_tz) - s_str = s_local.strftime("%-I:%M %p") - if end: - e_local = end.astimezone(user_tz) if end.tzinfo else end.replace(tzinfo=timezone.utc).astimezone(user_tz) - e_str = e_local.strftime("%-I:%M %p") - event_time_blocks.append(f"{e.get('title', 'Event')}: {s_str}–{e_str}") - else: - event_time_blocks.append(f"{e.get('title', 'Event')}: {s_str}–?") - except Exception: - pass - - return { - "date": today, - "overdue_tasks": overdue, - "due_today": due_today, - "high_priority": high_priority, - "calendar_events": calendar_events, - "active_projects": projects_summary, - "all_tasks_raw": all_tasks, - "focus_tasks": focus_tasks, - "event_time_blocks": event_time_blocks, - "weather_warnings": weather_warnings, - "recent_activity": recent_activity, - "stale_projects": stale_projects, - } - - -async def _gather_weekly_review(user_id: int) -> dict: - """Collect 7-day activity summary for the weekly review.""" - from fabledassistant.services.notes import list_notes - from fabledassistant.services.projects import list_projects - from fabledassistant.services.events import list_events as list_internal_events - - tz_name = await get_setting(user_id, "user_timezone") or "UTC" - try: - user_tz = ZoneInfo(tz_name) - except ZoneInfoNotFoundError: - user_tz = ZoneInfo("UTC") - - now = datetime.now(user_tz) - week_ago = now - __import__('datetime').timedelta(days=7) - next_week = now + __import__('datetime').timedelta(days=7) - - # Tasks completed this week - completed_tasks = [] - overdue_tasks = [] - try: - all_task_objs, _ = await list_notes(user_id, is_task=True, limit=200) - for t in all_task_objs: - if t.status == "done" and t.completed_at and t.completed_at >= week_ago: - completed_tasks.append(t.title or "Untitled") - elif t.due_date and t.due_date.isoformat() < now.date().isoformat() and t.status not in ("done", "cancelled"): - overdue_tasks.append(t.title or "Untitled") - except Exception: - logger.debug("Weekly review: failed to gather tasks", exc_info=True) - - # Notes created this week - notes_created = 0 - try: - all_notes, _ = await list_notes(user_id, is_task=False, limit=200) - notes_created = sum(1 for n in all_notes if n.created_at and n.created_at >= week_ago) - except Exception: - logger.debug("Weekly review: failed to count notes", exc_info=True) - - # Project activity - active_projects = [] - try: - projects = await list_projects(user_id) - active_projects = [p.title for p in projects if p.status == "active"][:5] - except Exception: - pass - - # Events next week - upcoming_events = [] - try: - next_events = await list_internal_events( - user_id=user_id, - date_from=now, - date_to=next_week, - ) - for e in next_events[:8]: - upcoming_events.append(f"{e.get('title', 'Event')} — {e.get('start_dt', '')[:10]}") - except Exception: - pass - - return { - "completed_count": len(completed_tasks), - "completed_highlights": completed_tasks[:5], - "overdue_count": len(overdue_tasks), - "overdue_items": overdue_tasks[:5], - "notes_created": notes_created, - "active_projects": active_projects, - "upcoming_events": upcoming_events, - } - - # ── External data gather ────────────────────────────────────────────────────── async def _gather_external(user_id: int) -> dict: @@ -687,265 +267,6 @@ async def run_agentic_briefing( return final_text, messages -# ── Legacy one-shot synthesis ───────────────────────────────────────────────── - -async def _llm_synthesise(system_prompt: str, user_prompt: str, model: str, num_ctx: int = 4096) -> str: - """Single non-streaming LLM call. Returns the assistant's response text.""" - payload = { - "model": model, - "messages": [ - {"role": "system", "content": system_prompt}, - {"role": "user", "content": user_prompt}, - ], - "stream": False, - "options": {"num_ctx": num_ctx, "temperature": 0.4}, - } - try: - async with httpx.AsyncClient(timeout=120.0) as client: - resp = await client.post(f"{Config.OLLAMA_URL}/api/chat", json=payload) - resp.raise_for_status() - data = resp.json() - return data.get("message", {}).get("content", "").strip() - except Exception: - logger.warning("LLM synthesis failed", exc_info=True) - return "" - - -def _unified_system_prompt(profile_body: str, slot: str = "compilation") -> str: - base = ( - "You are a personal assistant. " - "Speak naturally and conversationally — as if talking to the user, not writing a report. " - "Use no markdown: no headers, no bullet points, no bold, no lists. Write in flowing prose. " - "Be warm, concise, and human. " - "The user can reply to act on your suggestions via tool calls.\n\n" - ) - - if slot == "compilation": - base += ( - "This is the MORNING BRIEFING — the full daily overview. " - "Weave together what matters today: mention the weather, note calendar events and tasks, " - "briefly reference one or two noteworthy news themes, and suggest a daily focus. " - "Be actionable: for overdue or due-today tasks, suggest a concrete next step. " - "Aim for 4 to 8 sentences.\n\n" - ) - elif slot in ("morning", "midday"): - base += ( - "This is a MIDDAY CHECK-IN — brief progress nudge, not a full briefing. " - "Focus on what's been done and what hasn't been touched yet today. " - "If weather conditions have changed, mention it briefly. " - "Keep it to 2-3 sentences. Don't repeat news or the full task list.\n\n" - ) - elif slot == "afternoon": - base += ( - "This is the AFTERNOON WRAP-UP — help the user close out the day. " - "Summarize what was accomplished today. Flag anything still overdue. " - "Ask if there's anything to capture before tomorrow — notes, thoughts, follow-ups. " - "Keep it to 2-4 sentences. Reflective and encouraging tone.\n\n" - ) - elif slot == "weekly_review": - base += ( - "This is the WEEKLY REVIEW — a reflective recap of the past 7 days. " - "Summarize accomplishments, flag lingering overdue items, note how many notes were created, " - "and preview the upcoming week's key events. " - "Be reflective and encouraging — celebrate progress, gently flag what's stuck. " - "Aim for 5 to 8 sentences. This wraps up a chapter.\n\n" - ) - - if profile_body: - base += f"User profile:\n{profile_body}\n" - return base - - -def _unified_user_prompt(internal_data: dict, external_data: dict, slot: str, temp_unit: str = "C") -> str: - lines = [f"Date: {internal_data['date']}", f"Slot: {slot}", ""] - - # Weekly review — separate data path - if slot == "weekly_review": - weekly = internal_data.get("weekly_review") or {} - lines.append(f"COMPLETED THIS WEEK: {weekly.get('completed_count', 0)} tasks") - if weekly.get("completed_highlights"): - lines.extend(f" - {t}" for t in weekly["completed_highlights"]) - lines.append("") - if weekly.get("overdue_count", 0) > 0: - lines.append(f"STILL OVERDUE: {weekly['overdue_count']} task(s)") - lines.extend(f" - {t}" for t in weekly.get("overdue_items", [])) - lines.append("") - lines.append(f"NOTES CREATED: {weekly.get('notes_created', 0)}") - lines.append("") - if weekly.get("active_projects"): - lines.append(f"ACTIVE PROJECTS: {', '.join(weekly['active_projects'])}") - lines.append("") - if weekly.get("upcoming_events"): - lines.append("UPCOMING WEEK:") - lines.extend(f" - {e}" for e in weekly["upcoming_events"]) - lines.append("") - return "\n".join(lines) - - # For check-in slots, build a slimmer prompt - if slot in ("morning", "midday"): - # Just tasks status + any weather changes - if internal_data.get("due_today"): - lines.append("TASKS DUE TODAY (note which are still untouched):") - lines.extend(f" - {t}" for t in internal_data["due_today"]) - lines.append("") - if internal_data.get("overdue_tasks"): - lines.append(f"STILL OVERDUE: {len(internal_data['overdue_tasks'])} task(s)") - lines.append("") - if internal_data.get("calendar_events"): - lines.append("REMAINING EVENTS TODAY:") - lines.extend(f" - {e}" for e in internal_data["calendar_events"]) - lines.append("") - return "\n".join(lines) - - if slot == "afternoon": - if internal_data.get("due_today"): - lines.append("TASKS THAT WERE DUE TODAY:") - lines.extend(f" - {t}" for t in internal_data["due_today"]) - lines.append("") - if internal_data.get("overdue_tasks"): - overdue = internal_data["overdue_tasks"] - lines.append(f"OVERDUE ({len(overdue)} remaining):") - lines.extend(f" - {t}" for t in overdue[:3]) - lines.append("") - lines.append("Ask the user if there's anything to capture before tomorrow — a note, a thought, a follow-up task.") - return "\n".join(lines) - - # ── Full compilation prompt below ────────────────────────────────────────── - - # Weather (brief — card handles detail) - weather = external_data.get("weather") or [] - if weather: - loc = weather[0] - days = loc.get("days") or [] - if days: - d = days[0] - t_min = _format_temp(d["temp_min"], temp_unit) - t_max = _format_temp(d["temp_max"], temp_unit) - unit_sym = f"°{temp_unit}" - lines.append( - f"WEATHER: {loc['location_label']} — {d['description']}, " - f"{t_min}–{t_max}{unit_sym}" - ) - lines.append("") - - # Today's calendar events - if internal_data.get("calendar_events"): - lines.append("TODAY'S EVENTS:") - lines.extend(f" - {e}" for e in internal_data["calendar_events"]) - lines.append("") - - # Weather warnings for outdoor events - if internal_data.get("weather_warnings"): - lines.append("WEATHER ALERTS (mention these naturally when discussing affected events):") - lines.extend(f" ⚠ {w}" for w in internal_data["weather_warnings"]) - lines.append("") - - # Tasks due today - if internal_data.get("due_today"): - lines.append("DUE TODAY (suggest next steps — offer to mark in progress or help prioritize):") - lines.extend(f" - {t}" for t in internal_data["due_today"]) - lines.append("") - - # Overdue tasks - if internal_data.get("overdue_tasks"): - overdue = internal_data["overdue_tasks"] - lines.append(f"OVERDUE ({len(overdue)} task{'s' if len(overdue) != 1 else ''}) — suggest rescheduling or breaking down:") - lines.extend(f" - {t}" for t in overdue[:5]) - if len(overdue) > 5: - lines.append(f" (and {len(overdue) - 5} more)") - lines.append("") - - # High priority in-progress - if internal_data.get("high_priority"): - lines.append("HIGH PRIORITY:") - lines.extend(f" - {t}" for t in internal_data["high_priority"]) - lines.append("") - - # Your Day — prioritized focus + calendar gaps (compilation slot only) - if internal_data.get("focus_tasks") and slot == "compilation": - lines.append("YOUR DAY — top tasks to focus on (mention these as a suggested plan):") - for i, t in enumerate(internal_data["focus_tasks"], 1): - lines.append(f" {i}. {t}") - if internal_data.get("event_time_blocks"): - lines.append("") - lines.append(" Scheduled time blocks:") - lines.extend(f" {b}" for b in internal_data["event_time_blocks"]) - lines.append(" (Identify free blocks between events for focused work.)") - lines.append("") - - # Project continuity — what was the user working on? - if internal_data.get("recent_activity") and slot == "compilation": - lines.append("YESTERDAY'S ACTIVITY (mention what the user was working on to help them pick up):") - lines.extend(f" - {a}" for a in internal_data["recent_activity"]) - lines.append("") - - if internal_data.get("stale_projects") and slot == "compilation": - lines.append("STALE PROJECTS (no activity in 7+ days — gently flag):") - lines.extend(f" - {p}" for p in internal_data["stale_projects"]) - lines.append("") - - # News — clustered by topic, ranked by preference score - rss = external_data.get("rss_items") or [] - topic_scores = external_data.get("topic_scores") or {} - if rss: - # Group articles by their primary topic - clusters: dict[str, list[dict]] = {} - uncategorized: list[dict] = [] - for item in rss: - topics = item.get("topics") or [] - if topics: - primary = topics[0] - clusters.setdefault(primary, []).append(item) - else: - uncategorized.append(item) - - # Score each cluster by aggregate preference (positive = user likes this topic) - def cluster_score(topic: str, items: list[dict]) -> float: - pref = topic_scores.get(topic, 0.0) - return pref * 2.0 + len(items) # preference-weighted, size as tiebreak - - # Filter out clusters where the user has a strongly negative preference - filtered_clusters = [ - (topic, items) for topic, items in clusters.items() - if topic_scores.get(topic, 0.0) >= -2.0 - ] - - # Sort by preference-weighted score - sorted_clusters = sorted(filtered_clusters, key=lambda x: cluster_score(x[0], x[1]), reverse=True) - - lines.append("NEWS THEMES (synthesize 1-2 themes into your briefing; mention article count per theme):") - for i, (topic, items) in enumerate(sorted_clusters[:4]): - count = len(items) - titles = [it.get("title", "") for it in items[:3]] - sources = list({it.get("feed_title") or it.get("source") or "News" for it in items}) - pref_indicator = "" - pref = topic_scores.get(topic, 0.0) - if pref >= 2.0: - pref_indicator = " ★" # user's preferred topic - lines.append(f" [{topic.title()}]{pref_indicator} ({count} article{'s' if count != 1 else ''} from {', '.join(sources[:2])})") - for t in titles: - lines.append(f" • {t}") - # Include excerpt for preferred clusters or the top cluster - if pref >= 1.0 or i == 0: - excerpt = (items[0].get("content") or items[0].get("snippet") or "")[:400].strip() - if excerpt: - lines.append(f" > {excerpt}") - if uncategorized: - lines.append(f" [Other] ({len(uncategorized)} uncategorized article{'s' if len(uncategorized) != 1 else ''})") - lines.append("") - - return "\n".join(lines) - - -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}" - - - - # ── Main entry point ─────────────────────────────────────────────────────────── async def _get_temp_unit(user_id: int) -> str: @@ -966,14 +287,16 @@ async def run_compilation( model: str | None = None, ) -> tuple[str, dict]: """ - Run the full two-lane briefing pipeline for a user and slot. - Returns (briefing_text, metadata_dict) where metadata contains - weather card data and rss_item_ids for frontend rendering. + Run the agentic briefing loop and gather UI metadata (RSS + weather). + + Returns ``(briefing_text, metadata)`` where metadata contains + ``rss_item_ids``, ``rss_items``, ``weather`` for frontend rendering, + and ``agentic_messages`` (the full tool-call sequence) for the + scheduler to persist as separate conversation rows. """ if model is None: model = await get_setting(user_id, "default_model", Config.OLLAMA_MODEL) - from fabledassistant.services.user_profile import build_profile_context from fabledassistant.services.briefing_preferences import ( load_topic_preferences, load_topic_reaction_scores, @@ -981,31 +304,15 @@ async def run_compilation( ) from fabledassistant.services.weather import parse_weather_card_data, get_cached_weather_rows - profile_context, temp_unit = await asyncio.gather( - build_profile_context(user_id), - _get_temp_unit(user_id), - ) - - # ── Pre-processing ────────────────────────────────────────────────────────── include_topics, exclude_topics = await load_topic_preferences(user_id) topic_scores = await load_topic_reaction_scores(user_id) - # Parallel raw gather — weather rows fetched in same gather to avoid extra DB round-trip - internal_data, external_data, weather_rows = await asyncio.gather( - _gather_internal(user_id), + external_data, weather_rows, temp_unit = await asyncio.gather( _gather_external(user_id), get_cached_weather_rows(user_id), + _get_temp_unit(user_id), ) - # Weekly review: gather 7-day summary data - if slot == "weekly_review": - internal_data["weekly_review"] = await _gather_weekly_review(user_id) - - # Task change detection - all_tasks = internal_data.get("all_tasks_raw", []) - changed_tasks, unchanged_count = await split_changed_tasks(user_id, all_tasks) - - # RSS filtering raw_rss = external_data.get("rss_items") or [] filtered_rss = score_and_filter_items( raw_rss, @@ -1028,48 +335,17 @@ async def run_compilation( if item.get("id") ] - # Weather staleness gate — returns None if data is >24h old weather_card = parse_weather_card_data(weather_rows[0], temp_unit) if weather_rows else None - # ── LLM Synthesis ────────────────────────────────────────────────────────── - # Build filtered internal data with only changed tasks - internal_data_filtered = dict(internal_data) - internal_data_filtered["unchanged_task_count"] = unchanged_count - internal_data_filtered["changed_tasks"] = [format_task(t) for t in changed_tasks] + briefing_text, agentic_messages = await run_agentic_briefing( + user_id, slot, model, conv_id=None, + ) - # Build filtered external data (suppress weather prose — card handles it) - external_data_filtered = { - "rss_items": filtered_rss, - "weather": [], - "topic_scores": topic_scores, + metadata: dict = { + "rss_item_ids": rss_item_ids, + "rss_items": rss_items_meta, + "weather": weather_card, } - - briefing_mode = await get_setting(user_id, "briefing_mode", "legacy") - agentic_messages: list[dict] = [] - briefing_text = "" - - if briefing_mode == "agentic": - briefing_text, agentic_messages = await run_agentic_briefing( - user_id, slot, model, conv_id=None, - ) - if not briefing_text: - logger.warning( - "Agentic briefing returned empty for user %d slot %s — falling back to legacy path", - user_id, slot, - ) - - if not briefing_text: - briefing_text = await _llm_synthesise( - _unified_system_prompt(profile_context, slot), - _unified_user_prompt(internal_data_filtered, external_data_filtered, slot, temp_unit), - model, - num_ctx=8192, - ) - - # ── Post-processing ───────────────────────────────────────────────────────── - await upsert_task_snapshots(user_id, all_tasks) - - metadata: dict = {"rss_item_ids": rss_item_ids, "rss_items": rss_items_meta, "weather": weather_card} if agentic_messages: metadata["agentic_messages"] = agentic_messages @@ -1086,50 +362,18 @@ async def run_slot_injection( model: str | None = None, ) -> tuple[str, dict]: """ - Lighter update for 8am/12pm/4pm — generates a slot-specific check-in. + Lighter check-in update for 8am/12pm/4pm slots. - Honors the ``briefing_mode`` setting just like ``run_compilation``: - agentic mode routes through the tool-use loop, legacy mode runs the - one-shot synthesis path. Falls back to legacy automatically if - agentic returns empty. - - Returns ``(text, metadata)`` where metadata may contain - ``agentic_messages`` (the full tool-call sequence) if the agentic - path was used. + Runs the agentic loop with the slot-specific prompt. Returns + ``(text, metadata)`` where metadata contains ``agentic_messages`` + for the scheduler to persist. """ if model is None: model = await get_setting(user_id, "default_model", Config.OLLAMA_MODEL) - briefing_mode = await get_setting(user_id, "briefing_mode", "legacy") - agentic_messages: list[dict] = [] - text = "" - - if briefing_mode == "agentic": - text, agentic_messages = await run_agentic_briefing( - user_id, slot, model, conv_id=None, - ) - if not text: - logger.warning( - "Agentic slot injection returned empty for user %d slot %s — falling back to legacy", - user_id, slot, - ) - - if not text: - internal_data, external_data, temp_unit = await asyncio.gather( - _gather_internal(user_id), - _gather_external(user_id), - _get_temp_unit(user_id), - ) - system = ( - f"You are a personal assistant giving a brief {slot} check-in. " - "The user already had their morning briefing — focus only on what's changed or newly relevant. " - "Speak naturally in 2-3 sentences, no markdown formatting, no headers or bullet points." - ) - text = await _llm_synthesise( - system, - _unified_user_prompt(internal_data, external_data, slot, temp_unit), - model, - ) + text, agentic_messages = await run_agentic_briefing( + user_id, slot, model, conv_id=None, + ) metadata: dict = {} if agentic_messages: diff --git a/src/fabledassistant/services/briefing_scheduler.py b/src/fabledassistant/services/briefing_scheduler.py index e2b766f..f5b8134 100644 --- a/src/fabledassistant/services/briefing_scheduler.py +++ b/src/fabledassistant/services/briefing_scheduler.py @@ -444,7 +444,7 @@ async def _run_profile_closeout(user_id: int, model: str) -> None: and append them to the briefing profile note. """ from fabledassistant.services.user_profile import append_observations - from fabledassistant.services.briefing_pipeline import _llm_synthesise + from fabledassistant.services.llm import generate_completion from fabledassistant.models.conversation import Conversation, Message yesterday = (date.today() - timedelta(days=1)) @@ -477,7 +477,17 @@ async def _run_profile_closeout(user_id: int, model: str) -> None: "Example: '- User skipped news about finance', '- Prefers weather for home location first'. " "If nothing notable, output only: (nothing to note)" ) - observations = await _llm_synthesise(system, transcript, model) + try: + observations = (await generate_completion( + [ + {"role": "system", "content": system}, + {"role": "user", "content": transcript}, + ], + model, + )).strip() + except Exception: + logger.warning("Profile closeout synthesis failed for user %d", user_id, exc_info=True) + observations = "" if observations and "(nothing to note)" not in observations.lower(): await append_observations(user_id, observations) diff --git a/src/fabledassistant/services/user_profile.py b/src/fabledassistant/services/user_profile.py index 14d65f8..bdb4fdb 100644 --- a/src/fabledassistant/services/user_profile.py +++ b/src/fabledassistant/services/user_profile.py @@ -114,7 +114,7 @@ async def _consolidate_observations(user_id: int) -> str: learned_summary paragraph. Prunes raw entries older than 30 days afterwards. """ from fabledassistant.config import Config - from fabledassistant.services.briefing_pipeline import _llm_synthesise + from fabledassistant.services.llm import generate_completion from fabledassistant.services.settings import get_setting async with async_session() as session: @@ -145,7 +145,17 @@ async def _consolidate_observations(user_id: int) -> str: user_prompt += f"New observations:\n{obs_text}" model = await get_setting(user_id, "default_model", Config.OLLAMA_MODEL) - new_summary = await _llm_synthesise(system, user_prompt, model) + try: + new_summary = (await generate_completion( + [ + {"role": "system", "content": system}, + {"role": "user", "content": user_prompt}, + ], + model, + )).strip() + except Exception: + logger.warning("Observation consolidation failed for user %d", user_id, exc_info=True) + new_summary = "" if new_summary: cutoff = (date.today() - timedelta(days=30)).isoformat()