@@ -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 ×
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- %d T % 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 .
Honor s 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.
Run s 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 :
