""" Briefing pipeline: agentic tool-use loop + UI metadata gather. Slot names: 'compilation' (4am), 'morning' (8am), 'midday' (12pm), 'afternoon' (4pm) """ import asyncio import logging from datetime import datetime from zoneinfo import ZoneInfo, ZoneInfoNotFoundError from fabledassistant.config import Config from fabledassistant.services.settings import get_setting logger = logging.getLogger(__name__) SLOT_NAMES = ("compilation", "morning", "midday", "afternoon") # ── External data gather ────────────────────────────────────────────────────── async def _gather_external(user_id: int) -> dict: """Collect RSS items (when enabled) and weather.""" from fabledassistant.services.weather import get_cached_weather rss_on = (await get_setting(user_id, "rss_enabled", "false")).lower() == "true" rss_items: list = [] if rss_on: from fabledassistant.services.rss import get_recent_items try: rss_items = await get_recent_items(user_id, limit=20) except Exception: pass try: weather = await get_cached_weather(user_id) except Exception: weather = [] return { "rss_items": rss_items, "weather": weather, } # ── 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. " "Weave real data from tool calls into a warm, natural-sounding summary.\n\n" "Tools to call every compilation (skip only if you already know a category is empty):\n" "- list_tasks — what's due today, overdue, or in progress\n" "- list_events — what's on the calendar today\n" "- get_weather — today's forecast\n" "- get_rss_items — recent news/blog items from the user's feeds\n" "- list_projects (optional) — active project context for narrative continuity\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, no news items), " "say so honestly. Don't fabricate items to fill space.\n" "- For news, pick one or two items worth mentioning — surface the theme, not a laundry list.\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" "When checking tasks, call list_tasks at least twice:\n" "- once with status=\"in_progress\" to see anything already being worked on " "(regardless of due date — these can quietly drag past their due dates)\n" "- once filtered by due date for what's coming up or overdue today\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, rss_override: list[dict] | 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 = {} # Default list_tasks to active statuses only so cancelled/done # items don't slip into briefing prose. The model can still # pass an explicit status filter when it wants something else. if tool_name == "list_tasks" and not arguments.get("status"): arguments["status"] = ["todo", "in_progress"] try: if tool_name == "get_rss_items" and rss_override is not None: # Use topic-scored/filtered items already computed by # the briefing pipeline rather than the raw feed dump # that execute_tool would return. Keeps the model's # view of news aligned with the user's topic prefs # and the sidebar's rss_item_ids metadata. slim = [ { "id": item.get("id"), "title": item.get("title", ""), "url": item.get("url", ""), "source": item.get("feed_title", ""), "summary": (item.get("content") or "")[:400], "published_at": item.get("published_at"), "topics": item.get("topics") or [], } for item in rss_override ] result = {"success": True, "data": {"items": slim, "count": len(slim)}} else: 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 # ── 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 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.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 include_topics, exclude_topics = await load_topic_preferences(user_id) topic_scores = await load_topic_reaction_scores(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), ) 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_cards = [ card for row in weather_rows if (card := parse_weather_card_data(row, temp_unit)) is not None ] weather_card = weather_cards[0] if weather_cards else None briefing_text, agentic_messages = await run_agentic_briefing( user_id, slot, model, conv_id=None, rss_override=filtered_rss, ) 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 check-in update for 8am/12pm/4pm slots. 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) text, agentic_messages = await run_agentic_briefing( user_id, slot, model, conv_id=None, ) metadata: dict = {} if agentic_messages: metadata["agentic_messages"] = agentic_messages return text, metadata