""" Briefing pipeline: parallel data gather + two-lane LLM synthesis. Slot names: 'compilation' (4am), 'morning' (8am), 'midday' (12pm), 'afternoon' (4pm) """ import asyncio import hashlib import logging from datetime import datetime, timezone 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 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: """Collect RSS items and weather.""" from fabledassistant.services.rss import get_recent_items from fabledassistant.services.weather import get_cached_weather rss_items, weather = await asyncio.gather( get_recent_items(user_id, limit=20), get_cached_weather(user_id), return_exceptions=True, ) return { "rss_items": rss_items if not isinstance(rss_items, Exception) else [], "weather": weather if not isinstance(weather, Exception) else [], } # ── Agentic briefing (tool-use loop) ────────────────────────────────────────── _BRIEFING_AGENT_MAX_ROUNDS = 8 _BRIEFING_AGENT_NUM_CTX = 8192 def _agentic_system_prompt( profile_body: str, slot: str, today_iso: str, tz_name: str, day_from_iso: str, day_to_iso: str, ) -> str: """System prompt for the agentic briefing path. Pushes the model to ground every factual claim in a tool result and to be honest when tools return nothing, rather than fabricating content to fill the narrative. Pre-computes today's window in the user's local timezone so the model can call date-sensitive tools (list_events, list_tasks filters) without having to do any timezone math itself — eliminating a whole class of "wrong day" bugs. """ tz_block = ( f"Today is {today_iso} ({tz_name}). " f"When calling list_events for today, use:\n" f" date_from = {day_from_iso}\n" f" date_to = {day_to_iso}\n" f"These are already the correct local-day boundaries — do not convert " f"them to UTC or any other timezone. For other date ranges, compute in " f"the same timezone.\n\n" ) if slot == "compilation": base = ( "You are the user's personal assistant giving their full morning briefing. " "Use the tools available to see what's actually relevant today — tasks due, " "overdue items, events on the calendar, weather, news themes, project state — " "and weave it into a warm, natural-sounding summary.\n\n" "Rules:\n" "- Call tools to see the data. Never assert facts you didn't learn from a tool.\n" "- If a tool returns nothing (no events today, no overdue tasks), say so honestly. " "Don't fabricate items to fill space.\n" "- Write flowing prose. No markdown, no headers, no bullet points.\n" "- Aim for 6 to 10 sentences. Skip topics that have nothing interesting.\n" "- Close on one or two concrete, actionable suggestions.\n\n" ) elif slot == "weekly_review": base = ( "You are the user's personal assistant delivering a weekly review. " "Use the tools available to see what was accomplished this week, what's still " "overdue, how many notes were captured, and what's coming up in the next seven days. " "Write a reflective recap that celebrates real progress and gently flags what's stuck.\n\n" "Rules:\n" "- Call tools to see the data. Never assert facts you didn't learn from a tool.\n" "- If a category is empty, say so honestly rather than inventing items.\n" "- Write flowing prose. No markdown, no bullet points.\n" "- Aim for 5 to 8 sentences. Reflective and encouraging tone.\n\n" ) else: # morning, midday, afternoon check-ins base = ( f"You are the user's personal assistant giving a brief {slot} check-in. " "Use tools to see what's changed since this morning. Focus on progress and " "what's still unaddressed.\n\n" "Rules:\n" "- Call tools to see current state. Never assert facts without tool results.\n" "- If nothing meaningful has changed, say so briefly — don't invent progress.\n" "- 3 to 5 sentences, natural prose, no markdown.\n\n" ) base = tz_block + base if profile_body: base += f"User profile (tone and preferences):\n{profile_body}\n" return base def _agentic_user_trigger(slot: str, date_str: str) -> str: """Seed user-role message that kicks off the agentic run.""" labels = { "compilation": "morning briefing", "morning": "morning check-in", "midday": "midday check-in", "afternoon": "afternoon wrap-up", "weekly_review": "weekly review", } label = labels.get(slot, f"{slot} briefing") return f"Generate my {label} for {date_str}." async def run_agentic_briefing( user_id: int, slot: str, model: str, conv_id: int | None = None, ) -> tuple[str, list[dict]]: """ Run the agentic briefing loop for a user and slot. Uses the chat pipeline's tool-use loop with a curated read-only tool subset and a slot-specific system prompt. Every fact the model states is either derived from a tool result visible in the returned message list or it's the model hallucinating — so follow-up chat in the same conversation can hold the model to what the tool results actually show. Returns ``(final_prose, message_list)`` where ``message_list`` is the full sequence including system, user trigger, tool calls, and tool results. Callers are expected to persist those intermediate turns alongside the final prose so the receipts remain in conversation history on follow-up. If the loop fails or the model returns empty prose, returns ``("", [])`` and the caller should fall back to the legacy path. """ from fabledassistant.services.llm import stream_chat_with_tools, ChatChunk # noqa: F401 from fabledassistant.services.tools import execute_tool from fabledassistant.services.briefing_tools import get_briefing_tools from fabledassistant.services.user_profile import build_profile_context profile_context = await build_profile_context(user_id) tools = await get_briefing_tools(user_id) if not tools: logger.warning( "Agentic briefing for user %d slot %s: no tools available — aborting", user_id, slot, ) return "", [] # Compute today's window in the user's local timezone so the model # receives ready-to-use ISO 8601 boundaries and never has to do its # own tz math when calling date-sensitive tools like 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") tz_name = "UTC" now_local = datetime.now(user_tz) today_iso = now_local.date().isoformat() day_start = datetime(now_local.year, now_local.month, now_local.day, 0, 0, 0, tzinfo=user_tz) day_end = datetime(now_local.year, now_local.month, now_local.day, 23, 59, 59, tzinfo=user_tz) day_from_iso = day_start.isoformat() day_to_iso = day_end.isoformat() system_prompt = _agentic_system_prompt( profile_context, slot, today_iso, tz_name, day_from_iso, day_to_iso, ) messages: list[dict] = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": _agentic_user_trigger(slot, today_iso)}, ] final_text = "" for round_idx in range(_BRIEFING_AGENT_MAX_ROUNDS): accumulated_content = "" accumulated_tool_calls: list[dict] = [] try: async for chunk in stream_chat_with_tools( messages, model, tools=tools, think=False, num_ctx=_BRIEFING_AGENT_NUM_CTX, ): if chunk.type == "content" and chunk.content: accumulated_content += chunk.content elif chunk.type == "tool_calls" and chunk.tool_calls: accumulated_tool_calls.extend(chunk.tool_calls) except Exception: logger.warning( "Agentic briefing stream failed (user %d, slot %s, round %d)", user_id, slot, round_idx, exc_info=True, ) return "", [] # Append the assistant turn (content + any tool calls) to history assistant_msg: dict = {"role": "assistant", "content": accumulated_content} if accumulated_tool_calls: assistant_msg["tool_calls"] = accumulated_tool_calls messages.append(assistant_msg) # No tool calls → the model is done if not accumulated_tool_calls: final_text = accumulated_content.strip() break # Execute each tool call and append results as tool-role messages for tc in accumulated_tool_calls: fn = tc.get("function") or {} tool_name = fn.get("name", "") arguments = fn.get("arguments") or {} if isinstance(arguments, str): try: import json as _json arguments = _json.loads(arguments) except Exception: arguments = {} try: result = await execute_tool(user_id, tool_name, arguments, conv_id=conv_id) except Exception as exc: logger.warning( "Tool %s failed during agentic briefing: %s", tool_name, exc, ) result = {"success": False, "error": str(exc)} # Serialize the result compactly for the model's context import json as _json try: result_str = _json.dumps(result, default=str)[:4000] except Exception: result_str = str(result)[:4000] messages.append({ "role": "tool", "content": result_str, "tool_name": tool_name, }) else: logger.warning( "Agentic briefing hit max rounds (%d) for user %d slot %s — using last content", _BRIEFING_AGENT_MAX_ROUNDS, user_id, slot, ) # Walk back to find the last assistant message with non-empty content for m in reversed(messages): if m.get("role") == "assistant" and m.get("content"): final_text = m["content"].strip() break 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: """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, ) -> 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. """ 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, score_and_filter_items, ) 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), _gather_external(user_id), get_cached_weather_rows(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, include_topics=include_topics, exclude_topics=exclude_topics, topic_scores=topic_scores, max_items=10, ) rss_item_ids = [item["id"] for item in filtered_rss if item.get("id")] rss_items_meta = [ { "id": item["id"], "title": item.get("title", ""), "url": item.get("url", ""), "source": item.get("feed_title", ""), "snippet": (item.get("content") or "")[:300], "published_at": item.get("published_at"), } for item in filtered_rss 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] # Build filtered external data (suppress weather prose — card handles it) external_data_filtered = { "rss_items": filtered_rss, "weather": [], "topic_scores": topic_scores, } 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 if not briefing_text: logger.warning("Briefing compilation produced no content for user %d slot %s", user_id, slot) return "", metadata return briefing_text, metadata async def run_slot_injection( user_id: int, slot: str, model: str | None = None, ) -> tuple[str, dict]: """ Lighter update for 8am/12pm/4pm — generates a slot-specific check-in. 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. """ 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, ) metadata: dict = {} if agentic_messages: metadata["agentic_messages"] = agentic_messages return text, metadata