""" 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 [], } # ── LLM 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_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 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) -> str: """ Lighter update for 8am/12pm/4pm — gathers fresh data and produces a slot-specific update prompt. Returns the text to inject as a new user→assistant exchange. """ if model is None: model = await get_setting(user_id, "default_model", Config.OLLAMA_MODEL) 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." ) return await _llm_synthesise( system, _unified_user_prompt(internal_data, external_data, slot, temp_unit), model, )