# Briefing Service Improvements — Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** Replace the briefing's repetitive, prose-only output with a pre-filtered, structured pipeline that tracks task changes, classifies RSS topics, gates stale weather, and surfaces news source links with per-story reactions. **Architecture:** Pre-processing stage runs before gather — task change detection diffs against a DB snapshot, RSS filtering applies topic preferences and reaction weights, weather is gated at 24h. Structured card metadata is stored on `Message.metadata` (JSONB) at write time; the frontend reads it on load to render `WeatherCard.vue` and reaction buttons without any SSE changes. **Tech Stack:** Python 3.12 + SQLAlchemy 2.0 async, Quart, httpx (Open-Meteo `current_weather=true` + `past_days=1`), Vue 3 + TypeScript, existing `TagInput.vue`, FastMCP (fable-mcp). **Spec:** `docs/specs/2026-03-25-briefing-improvements-design.md` --- ## File Map ### New backend files - `alembic/versions/0028_add_briefing_improvements.py` - `src/fabledassistant/services/rss_classifier.py` - `src/fabledassistant/services/briefing_preferences.py` ### Modified backend files - `src/fabledassistant/models/conversation.py` — add `Message.metadata` JSONB column + `to_dict()` - `src/fabledassistant/models/rss_feed.py` — add `RssItem.topics`, `RssItem.classified_at` - `src/fabledassistant/services/briefing_conversations.py` — `post_message(... metadata=None)` - `src/fabledassistant/services/weather.py` — add `current_weather=true`, `past_days=1`, `parse_weather_card_data()` - `src/fabledassistant/services/rss.py` — trigger classification after storing new items - `src/fabledassistant/services/briefing_pipeline.py` — pre-processing stage, task snapshot helpers, return `(text, metadata)` from `run_compilation` - `src/fabledassistant/routes/briefing.py` — unpack tuple from `run_compilation`; add reaction endpoints ### New frontend files - `frontend/src/components/WeatherCard.vue` ### Modified frontend files - `frontend/src/api/client.ts` — add `postRssReaction()`, `deleteRssReaction()` - `frontend/src/views/BriefingView.vue` — render `WeatherCard`, reaction buttons from metadata - `frontend/src/views/SettingsView.vue` — add "News Preferences" subsection ### New MCP files - `fable-mcp/fable_mcp/tools/briefing.py` ### Modified MCP files - `fable-mcp/fable_mcp/server.py` — register briefing tools ### Test files - `tests/test_briefing_pipeline.py` — extend with new tests - `tests/test_weather_service.py` — extend with card parser tests - `tests/test_rss_service.py` — extend with classifier + preferences tests --- ## Task 1: Migration 0028 **Files:** - Create: `alembic/versions/0028_add_briefing_improvements.py` - [ ] **Step 1: Create the migration file** ```python """Add briefing improvements: rss_items topics/classified_at, messages metadata, rss_item_reactions, briefing_task_snapshot.""" from alembic import op revision = "0028" down_revision = "0027" def upgrade() -> None: op.execute(""" ALTER TABLE rss_items ADD COLUMN IF NOT EXISTS topics TEXT[] DEFAULT '{}', ADD COLUMN IF NOT EXISTS classified_at TIMESTAMPTZ """) op.execute(""" ALTER TABLE messages ADD COLUMN IF NOT EXISTS metadata JSONB """) op.execute(""" CREATE TABLE IF NOT EXISTS rss_item_reactions ( id SERIAL PRIMARY KEY, user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, rss_item_id INTEGER NOT NULL REFERENCES rss_items(id) ON DELETE CASCADE, reaction TEXT NOT NULL CHECK (reaction IN ('up', 'down')), created_at TIMESTAMPTZ DEFAULT NOW(), UNIQUE (user_id, rss_item_id) ) """) op.execute( "CREATE INDEX IF NOT EXISTS ix_rss_item_reactions_user_id " "ON rss_item_reactions(user_id)" ) op.execute(""" CREATE TABLE IF NOT EXISTS briefing_task_snapshot ( id SERIAL PRIMARY KEY, user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, task_id INTEGER NOT NULL REFERENCES notes(id) ON DELETE CASCADE, snapshot_hash TEXT NOT NULL, last_briefed TIMESTAMPTZ DEFAULT NOW(), UNIQUE (user_id, task_id) ) """) op.execute( "CREATE INDEX IF NOT EXISTS ix_briefing_task_snapshot_user_id " "ON briefing_task_snapshot(user_id)" ) def downgrade() -> None: op.execute("DROP TABLE IF EXISTS briefing_task_snapshot") op.execute("DROP TABLE IF EXISTS rss_item_reactions") op.execute("ALTER TABLE messages DROP COLUMN IF EXISTS metadata") op.execute("ALTER TABLE rss_items DROP COLUMN IF EXISTS classified_at") op.execute("ALTER TABLE rss_items DROP COLUMN IF EXISTS topics") ``` - [ ] **Step 2: Apply migration inside the running container** ```bash docker compose exec app alembic upgrade head ``` Expected: `Running upgrade 0027 -> 0028, Add briefing improvements` - [ ] **Step 3: Commit** ```bash git add alembic/versions/0028_add_briefing_improvements.py git commit -m "feat(briefing): add migration 0028 — briefing improvements schema" ``` --- ## Task 2: Update SQLAlchemy Models **Files:** - Modify: `src/fabledassistant/models/conversation.py` - Modify: `src/fabledassistant/models/rss_feed.py` - [ ] **Step 1: Add `metadata` column to `Message` in `models/conversation.py`** In the `Message` class, after the existing `tool_calls` column, add: ```python metadata: Mapped[dict | None] = mapped_column(JSONB, nullable=True) ``` In `Message.to_dict()`, add `"metadata": self.metadata` to the returned dict. - [ ] **Step 2: Add `topics` and `classified_at` columns to `RssItem` in `models/rss_feed.py`** Add imports at the top (if not already present): ```python from sqlalchemy import ARRAY, DateTime, Text # ARRAY is new from sqlalchemy.dialects.postgresql import ARRAY as PG_ARRAY ``` Actually SQLAlchemy's native ARRAY works for PostgreSQL. Use: ```python from sqlalchemy import ARRAY, Text ``` In the `RssItem` class, after `fetched_at`, add: ```python topics: Mapped[list[str]] = mapped_column( ARRAY(Text), nullable=False, default=list, server_default="{}" ) classified_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True ) ``` In `RssItem.to_dict()`, add: ```python "topics": self.topics or [], "classified_at": self.classified_at.isoformat() if self.classified_at else None, ``` - [ ] **Step 3: Rebuild the container to pick up model changes** ```bash docker compose up --build -d app ``` - [ ] **Step 4: Write a quick smoke test (no DB needed)** In `tests/test_briefing_models.py`, add: ```python def test_message_metadata_field_exists(): from fabledassistant.models.conversation import Message m = Message(conversation_id=1, role="assistant", content="hi", metadata={"foo": 1}) assert m.metadata == {"foo": 1} d = m.to_dict() assert "metadata" in d def test_rss_item_topics_field_exists(): from fabledassistant.models.rss_feed import RssItem item = RssItem(feed_id=1, guid="x", title="Test", topics=["technology"]) assert item.topics == ["technology"] d = item.to_dict() assert "topics" in d assert "classified_at" in d ``` - [ ] **Step 5: Run tests** ```bash docker compose exec app /opt/venv/bin/pytest tests/test_briefing_models.py -v ``` Expected: both new tests PASS. - [ ] **Step 6: Commit** ```bash git add src/fabledassistant/models/conversation.py src/fabledassistant/models/rss_feed.py tests/test_briefing_models.py git commit -m "feat(briefing): add Message.metadata and RssItem.topics/classified_at columns" ``` --- ## Task 3: Extend `post_message()` **Files:** - Modify: `src/fabledassistant/services/briefing_conversations.py` - [ ] **Step 1: Write the failing test** In `tests/test_briefing_pipeline.py`, add: ```python @pytest.mark.asyncio async def test_post_message_accepts_metadata(): """post_message should accept an optional metadata dict and store it.""" from unittest.mock import AsyncMock, patch, MagicMock mock_msg = MagicMock() mock_msg.id = 1 with patch("fabledassistant.services.briefing_conversations.async_session") as mock_session_cls: mock_session = AsyncMock() mock_session.__aenter__ = AsyncMock(return_value=mock_session) mock_session.__aexit__ = AsyncMock(return_value=False) mock_session.add = MagicMock() mock_session.get = AsyncMock(return_value=None) mock_session.commit = AsyncMock() mock_session.refresh = AsyncMock() mock_session_cls.return_value = mock_session from fabledassistant.services.briefing_conversations import post_message import importlib import fabledassistant.services.briefing_conversations as bc importlib.reload(bc) # Should not raise TypeError await bc.post_message(1, "assistant", "text", metadata={"rss_item_ids": [1, 2]}) ``` - [ ] **Step 2: Run test to verify it fails** ```bash docker compose exec app /opt/venv/bin/pytest tests/test_briefing_pipeline.py::test_post_message_accepts_metadata -v ``` Expected: FAIL — `TypeError: post_message() got an unexpected keyword argument 'metadata'` - [ ] **Step 3: Update `post_message()` signature** In `services/briefing_conversations.py`, change: ```python async def post_message(conversation_id: int, role: str, content: str) -> Message: """Append a message to a briefing conversation.""" async with async_session() as session: msg = Message( conversation_id=conversation_id, role=role, content=content, status="complete", ) ``` To: ```python async def post_message( conversation_id: int, role: str, content: str, metadata: dict | None = None, ) -> Message: """Append a message to a briefing conversation.""" async with async_session() as session: msg = Message( conversation_id=conversation_id, role=role, content=content, status="complete", metadata=metadata, ) ``` - [ ] **Step 4: Run test to verify it passes** ```bash docker compose exec app /opt/venv/bin/pytest tests/test_briefing_pipeline.py::test_post_message_accepts_metadata -v ``` Expected: PASS. - [ ] **Step 5: Commit** ```bash git add src/fabledassistant/services/briefing_conversations.py tests/test_briefing_pipeline.py git commit -m "feat(briefing): extend post_message() with optional metadata parameter" ``` --- ## Task 4: Weather Service — Card Data Parser **Files:** - Modify: `src/fabledassistant/services/weather.py` - Test: `tests/test_weather_service.py` - [ ] **Step 1: Write failing tests** Open `tests/test_weather_service.py` and add: ```python def test_parse_weather_card_data_returns_none_when_stale(): """Should return None when cache is older than 24 hours.""" from datetime import datetime, timezone, timedelta from unittest.mock import MagicMock from fabledassistant.services.weather import parse_weather_card_data cache = MagicMock() cache.fetched_at = datetime.now(timezone.utc) - timedelta(hours=25) cache.forecast_json = {} assert parse_weather_card_data(cache) is None def test_parse_weather_card_data_returns_card_schema(): """Should return the correct metadata.weather schema for fresh cache.""" from datetime import datetime, timezone, timedelta, date from unittest.mock import MagicMock from fabledassistant.services.weather import parse_weather_card_data today = date.today().isoformat() yesterday = (date.today() - timedelta(days=1)).isoformat() tomorrow = (date.today() + timedelta(days=1)).isoformat() cache = MagicMock() cache.fetched_at = datetime.now(timezone.utc) cache.location_label = "Berlin, DE" cache.forecast_json = { "current_weather": {"temperature": 12.0, "weathercode": 2}, "daily": { "time": [yesterday, today, tomorrow], "temperature_2m_max": [14.0, 16.0, 18.0], "temperature_2m_min": [9.0, 8.0, 10.0], "precipitation_sum": [0.0, 0.0, 1.5], "weathercode": [1, 2, 61], "windspeed_10m_max": [10.0, 12.0, 8.0], } } card = parse_weather_card_data(cache) assert card is not None assert card["current_temp"] == 12 assert card["condition"] == "Partly cloudy" assert card["today_high"] == 16 assert card["today_low"] == 8 assert card["yesterday_high"] == 14 assert card["yesterday_low"] == 9 assert len(card["forecast"]) >= 1 assert card["location"] == "Berlin, DE" ``` - [ ] **Step 2: Run tests to verify they fail** ```bash docker compose exec app /opt/venv/bin/pytest tests/test_weather_service.py -v -k "parse_weather_card" ``` Expected: FAIL — `ImportError: cannot import name 'parse_weather_card_data'` - [ ] **Step 3: Update `_fetch_open_meteo` to request current weather + past day** In `services/weather.py`, update `_fetch_open_meteo`: ```python async def _fetch_open_meteo(lat: float, lon: float) -> dict: """Fetch 7-day forecast from Open-Meteo with current conditions and yesterday's data.""" async with httpx.AsyncClient(timeout=15.0) as client: resp = await client.get(OPEN_METEO_URL, params={ "latitude": lat, "longitude": lon, "daily": OPEN_METEO_DAILY, "current_weather": "true", "past_days": 1, "timezone": "auto", "forecast_days": 7, }) resp.raise_for_status() return resp.json() ``` - [ ] **Step 4: Add `parse_weather_card_data()` to `services/weather.py`** Add after `detect_changes()`: ```python def parse_weather_card_data( cache_row, temp_unit: str = "C", ) -> dict | None: """ Parse a WeatherCache row into the metadata.weather card schema. Returns None if the cache is stale (older than 24 hours). """ from datetime import date, timedelta if cache_row is None or cache_row.fetched_at is None: return None age_seconds = (datetime.now(timezone.utc) - cache_row.fetched_at).total_seconds() if age_seconds > 86400: return None raw = cache_row.forecast_json or {} current_weather = raw.get("current_weather", {}) days = parse_forecast(raw) today_str = date.today().isoformat() yesterday_str = (date.today() - timedelta(days=1)).isoformat() today_day = next((d for d in days if d["date"] == today_str), None) yesterday_day = next((d for d in days if d["date"] == yesterday_str), None) future_days = [d for d in days if d["date"] > today_str][:5] def to_temp(c: float) -> int: if temp_unit == "F": return round(c * 9 / 5 + 32) return round(c) def day_label(date_str: str) -> str: from datetime import date as _date try: return _date.fromisoformat(date_str).strftime("%a") except ValueError: return date_str return { "location": getattr(cache_row, "location_label", ""), "fetched_at": cache_row.fetched_at.isoformat(), "current_temp": to_temp(current_weather.get("temperature", 0)), "condition": describe_weathercode(current_weather.get("weathercode", 0)), "today_high": to_temp(today_day["temp_max"]) if today_day else None, "today_low": to_temp(today_day["temp_min"]) if today_day else None, "yesterday_high": to_temp(yesterday_day["temp_max"]) if yesterday_day else None, "yesterday_low": to_temp(yesterday_day["temp_min"]) if yesterday_day else None, "forecast": [ { "day": day_label(d["date"]), "condition": d["description"], "high": to_temp(d["temp_max"]), "low": to_temp(d["temp_min"]), } for d in future_days ], } ``` - [ ] **Step 5: Run tests** ```bash docker compose exec app /opt/venv/bin/pytest tests/test_weather_service.py -v -k "parse_weather_card" ``` Expected: both PASS. - [ ] **Step 6: Commit** ```bash git add src/fabledassistant/services/weather.py tests/test_weather_service.py git commit -m "feat(briefing): add past_days/current_weather to Open-Meteo fetch; add parse_weather_card_data()" ``` --- ## Task 5: RSS Classifier Service **Files:** - Create: `src/fabledassistant/services/rss_classifier.py` - Test: `tests/test_rss_service.py` - [ ] **Step 1: Write failing test** In `tests/test_rss_service.py`, add: ```python @pytest.mark.asyncio async def test_classify_items_batch_returns_topic_map(): """classify_items_batch should return a dict mapping item_id to topic list.""" from unittest.mock import AsyncMock, patch fake_response = '{"1": ["technology", "ai"], "2": ["politics"]}' with patch( "fabledassistant.services.rss_classifier._llm_classify", new_callable=AsyncMock, return_value=fake_response, ): from fabledassistant.services import rss_classifier import importlib; importlib.reload(rss_classifier) items = [ {"id": 1, "title": "OpenAI releases GPT-5", "content": "..."}, {"id": 2, "title": "EU passes new law", "content": "..."}, ] result = await rss_classifier.classify_items_batch(items, user_include_topics=[]) assert result[1] == ["technology", "ai"] assert result[2] == ["politics"] @pytest.mark.asyncio async def test_classify_items_batch_handles_llm_failure(): """classify_items_batch should return empty lists on LLM error.""" from unittest.mock import AsyncMock, patch with patch( "fabledassistant.services.rss_classifier._llm_classify", new_callable=AsyncMock, side_effect=Exception("LLM unavailable"), ): from fabledassistant.services import rss_classifier import importlib; importlib.reload(rss_classifier) items = [{"id": 5, "title": "Some news", "content": ""}] result = await rss_classifier.classify_items_batch(items, user_include_topics=[]) assert result == {} # Empty on failure — items stay unclassified ``` - [ ] **Step 2: Run tests to confirm FAIL** ```bash docker compose exec app /opt/venv/bin/pytest tests/test_rss_service.py -v -k "classify" ``` Expected: FAIL — module not found. - [ ] **Step 3: Create `services/rss_classifier.py`** ```python """ RSS item topic classifier. Classifies RSS items into topic tags using a fast non-streaming LLM call. Called from rss.py after new items are stored — fire-and-forget. """ import json import logging from datetime import datetime, timezone import httpx from fabledassistant.config import Config logger = logging.getLogger(__name__) STANDARD_TOPICS = [ "technology", "science", "politics", "business", "health", "environment", "local", "entertainment", "sports", "other", ] _CLASSIFY_PROMPT = """\ Classify each news item into 1-3 topics. Use only topics from this list: {vocab}. Return ONLY a JSON object mapping item_id (as string) to a list of topics. Example: {{"1": ["technology", "ai"], "2": ["politics"]}} Items: {items_block}""" async def _llm_classify(prompt: str, model: str) -> str: """Make a fast non-streaming LLM call and return the raw text response.""" payload = { "model": model, "messages": [{"role": "user", "content": prompt}], "stream": False, "options": {"num_ctx": 2048, "temperature": 0.0}, } async with httpx.AsyncClient(timeout=30.0) as client: resp = await client.post(f"{Config.OLLAMA_URL}/api/chat", json=payload) resp.raise_for_status() return resp.json().get("message", {}).get("content", "") async def classify_items_batch( items: list[dict], user_include_topics: list[str], model: str | None = None, ) -> dict[int, list[str]]: """ Classify a batch of RSS items into topic tags. Args: items: list of dicts with 'id', 'title', 'content' user_include_topics: extra topics from user preferences to add to vocabulary model: Ollama model name; defaults to Config.OLLAMA_MODEL Returns: dict mapping item_id (int) -> list of topic strings. Items not returned had classification fail; callers should leave classified_at=NULL. """ if not items: return {} if model is None: model = Config.OLLAMA_MODEL vocab = STANDARD_TOPICS + [t for t in user_include_topics if t not in STANDARD_TOPICS] items_block = "\n".join( f"[{item['id']}] {item['title']} — {item.get('content', '')[:300]}" for item in items ) prompt = _CLASSIFY_PROMPT.format(vocab=", ".join(vocab), items_block=items_block) try: raw = await _llm_classify(prompt, model) # Extract JSON from response (LLM may wrap it in markdown) raw = raw.strip() if raw.startswith("```"): raw = raw.split("```")[1] if raw.startswith("json"): raw = raw[4:] parsed = json.loads(raw) return {int(k): v for k, v in parsed.items() if isinstance(v, list)} except Exception: logger.warning("RSS classification failed", exc_info=True) return {} async def classify_and_store( item_ids: list[int], user_id: int, ) -> None: """ Classify unclassified RSS items and write results to DB. Called as a fire-and-forget task from rss.py. """ from sqlalchemy import select from fabledassistant.models import async_session from fabledassistant.models.rss_feed import RssItem from fabledassistant.services.settings import get_setting if not item_ids: return # Load the items async with async_session() as session: result = await session.execute( select(RssItem).where(RssItem.id.in_(item_ids)) ) items = list(result.scalars().all()) if not items: return # Get user's include topics to extend vocabulary raw_include = await get_setting(user_id, "briefing_include_topics", "[]") try: include_topics = json.loads(raw_include) if isinstance(raw_include, str) else [] except Exception: include_topics = [] model = await get_setting(user_id, "default_model", Config.OLLAMA_MODEL) # Classify in batches of 10 batch_size = 10 all_results: dict[int, list[str]] = {} for i in range(0, len(items), batch_size): batch = items[i: i + batch_size] batch_dicts = [{"id": it.id, "title": it.title, "content": it.content} for it in batch] results = await classify_items_batch(batch_dicts, include_topics, model=model) all_results.update(results) # Write back to DB now = datetime.now(timezone.utc) async with async_session() as session: for item in items: item_db = await session.get(RssItem, item.id) if item_db is None: continue topics = all_results.get(item.id) if topics is not None: item_db.topics = topics item_db.classified_at = now await session.commit() ``` - [ ] **Step 4: Run tests** ```bash docker compose exec app /opt/venv/bin/pytest tests/test_rss_service.py -v -k "classify" ``` Expected: both PASS. - [ ] **Step 5: Commit** ```bash git add src/fabledassistant/services/rss_classifier.py tests/test_rss_service.py git commit -m "feat(briefing): add rss_classifier service for LLM-based topic tagging" ``` --- ## Task 6: Briefing Preferences Service > **Dependency:** Task 2 Step 2 must be complete before this task — `score_and_filter_items` expects items to have a `topics` key in their dicts, which comes from the `RssItem.topics` column added in Task 2. Run Task 2 first. **Files:** - Create: `src/fabledassistant/services/briefing_preferences.py` - Test: `tests/test_rss_service.py` - [ ] **Step 1: Write failing tests** In `tests/test_rss_service.py`, add: ```python def test_score_rss_items_excludes_blacklisted_topics(): """Items with excluded topics should be removed.""" from fabledassistant.services.briefing_preferences import score_and_filter_items items = [ {"id": 1, "title": "Tech news", "topics": ["technology"], "published_at": "2026-03-25T08:00:00"}, {"id": 2, "title": "Sports score", "topics": ["sports"], "published_at": "2026-03-25T08:00:00"}, ] result = score_and_filter_items( items, include_topics=["technology"], exclude_topics=["sports"], topic_scores={}, max_items=10, ) ids = [r["id"] for r in result] assert 1 in ids assert 2 not in ids def test_score_rss_items_boosts_included_topics(): """Items matching include_topics should rank higher than neutral items.""" from fabledassistant.services.briefing_preferences import score_and_filter_items items = [ {"id": 1, "title": "Random news", "topics": ["other"], "published_at": "2026-03-25T07:00:00"}, {"id": 2, "title": "Tech news", "topics": ["technology"], "published_at": "2026-03-25T06:00:00"}, ] result = score_and_filter_items( items, include_topics=["technology"], exclude_topics=[], topic_scores={}, max_items=10, ) # Tech item should be ranked first despite being older assert result[0]["id"] == 2 def test_score_rss_items_no_preferences_returns_all(): """With no preferences, all items should be returned sorted by recency.""" from fabledassistant.services.briefing_preferences import score_and_filter_items items = [ {"id": 1, "title": "A", "topics": [], "published_at": "2026-03-24T10:00:00"}, {"id": 2, "title": "B", "topics": [], "published_at": "2026-03-25T10:00:00"}, ] result = score_and_filter_items(items, [], [], {}, max_items=10) assert result[0]["id"] == 2 # Newer first ``` - [ ] **Step 2: Run tests to confirm FAIL** ```bash docker compose exec app /opt/venv/bin/pytest tests/test_rss_service.py -v -k "score_rss" ``` Expected: FAIL — module not found. - [ ] **Step 3: Create `services/briefing_preferences.py`** ```python """ Briefing preferences: load topic settings, aggregate reaction scores, filter and rank RSS items for briefing inclusion. """ import json import logging from datetime import datetime, timezone from sqlalchemy import func, select from fabledassistant.models import async_session logger = logging.getLogger(__name__) async def load_topic_preferences(user_id: int) -> tuple[list[str], list[str]]: """ Return (include_topics, exclude_topics) from user settings. """ from fabledassistant.services.settings import get_setting raw_include = await get_setting(user_id, "briefing_include_topics", "[]") raw_exclude = await get_setting(user_id, "briefing_exclude_topics", "[]") def _parse(raw) -> list[str]: try: val = json.loads(raw) if isinstance(raw, str) else raw return [str(t) for t in val] if isinstance(val, list) else [] except Exception: return [] return _parse(raw_include), _parse(raw_exclude) async def load_topic_reaction_scores(user_id: int) -> dict[str, float]: """ Aggregate per-topic reaction scores from the last 30 days. Returns a dict of topic -> net_score (positive = liked, negative = disliked). Uses rss_item_reactions joined to rss_items.topics. """ from fabledassistant.models.rss_feed import RssItem try: async with async_session() as session: # Raw SQL is simpler here due to ARRAY unnest result = await session.execute( __import__("sqlalchemy", fromlist=["text"]).text(""" SELECT unnest(i.topics) AS topic, SUM(CASE r.reaction WHEN 'up' THEN 1 ELSE -1 END) AS score FROM rss_item_reactions r JOIN rss_items i ON i.id = r.rss_item_id WHERE r.user_id = :uid AND r.created_at > NOW() - INTERVAL '30 days' GROUP BY topic """).bindparams(uid=user_id) ) return {row.topic: float(row.score) for row in result} except Exception: logger.warning("Failed to load topic reaction scores", exc_info=True) return {} def score_and_filter_items( items: list[dict], include_topics: list[str], exclude_topics: list[str], topic_scores: dict[str, float], max_items: int = 10, ) -> list[dict]: """ Score, filter, and rank RSS items for briefing inclusion. Scoring: - Hard-exclude: any item tagged with an excluded topic is removed. - Base score: 0.0 - +2.0 per topic that appears in include_topics - +1.0 / -1.0 per topic based on reaction score (clamped per topic) - Tiebreak: newer published_at wins Returns up to max_items items, highest score first. Items with classified_at=None (unclassified) pass through with score=0. """ include_set = set(include_topics) exclude_set = set(exclude_topics) scored = [] for item in items: item_topics = item.get("topics") or [] # Hard exclude if exclude_set and any(t in exclude_set for t in item_topics): continue score = 0.0 for topic in item_topics: if topic in include_set: score += 2.0 if topic in topic_scores: score += max(-1.0, min(1.0, topic_scores[topic])) # Parse published_at for tiebreak pub_str = item.get("published_at") or "" try: pub_ts = datetime.fromisoformat(pub_str).timestamp() if pub_str else 0.0 except ValueError: pub_ts = 0.0 scored.append((score, pub_ts, item)) # Sort: highest score first, then newest first scored.sort(key=lambda x: (x[0], x[1]), reverse=True) return [item for _, _, item in scored[:max_items]] ``` - [ ] **Step 4: Run tests** ```bash docker compose exec app /opt/venv/bin/pytest tests/test_rss_service.py -v -k "score_rss" ``` Expected: all PASS. - [ ] **Step 5: Commit** ```bash git add src/fabledassistant/services/briefing_preferences.py tests/test_rss_service.py git commit -m "feat(briefing): add briefing_preferences service for RSS scoring and filtering" ``` --- ## Task 7: Task Change Detection **Files:** - Modify: `src/fabledassistant/services/briefing_pipeline.py` - Test: `tests/test_briefing_pipeline.py` - [ ] **Step 1: Write failing tests** In `tests/test_briefing_pipeline.py`, add: ```python def test_compute_task_snapshot_hash(): """compute_task_hash should return a stable SHA-256 hex string.""" from fabledassistant.services.briefing_pipeline import compute_task_hash task = {"status": "todo", "priority": "high", "due_date": "2026-03-25", "title": "Write spec"} h = compute_task_hash(task) assert len(h) == 64 # SHA-256 hex # Same inputs produce same hash assert h == compute_task_hash(task) # Different status produces different hash assert h != compute_task_hash({**task, "status": "done"}) @pytest.mark.asyncio async def test_split_changed_tasks_all_new(): """split_changed_tasks should return all tasks as changed when no snapshot exists.""" from unittest.mock import AsyncMock, patch, MagicMock from fabledassistant.services.briefing_pipeline import split_changed_tasks tasks = [ {"task_id": 1, "title": "A", "status": "todo", "priority": "none", "due_date": None}, ] with patch( "fabledassistant.services.briefing_pipeline.async_session" ) as mock_cls: mock_session = AsyncMock() mock_session.__aenter__ = AsyncMock(return_value=mock_session) mock_session.__aexit__ = AsyncMock(return_value=False) mock_session.execute = AsyncMock(return_value=MagicMock( scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[]))) )) mock_cls.return_value = mock_session changed, unchanged_count = await split_changed_tasks(user_id=1, tasks=tasks) assert len(changed) == 1 assert unchanged_count == 0 ``` - [ ] **Step 2: Run tests to verify FAIL** ```bash docker compose exec app /opt/venv/bin/pytest tests/test_briefing_pipeline.py -v -k "task_snapshot or task_hash or split_changed" ``` Expected: FAIL. - [ ] **Step 3: Add helpers to `briefing_pipeline.py`** Add these imports at the top of `briefing_pipeline.py`: ```python import hashlib from datetime import datetime, timezone ``` Add these functions after `format_task()`: ```python 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 select, text from fabledassistant.models import async_session 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 from fabledassistant.models import async_session 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() ``` - [ ] **Step 4: Update `_gather_internal` to include `task_id`** In `_gather_internal`, change the task serialisation from: ```python all_tasks = [ { "title": t.title, "status": t.status, "due_date": t.due_date.isoformat() if t.due_date else None, "priority": t.priority, } for t in all_task_objs ] ``` To: ```python 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, } for t in all_task_objs ] ``` Also add `"all_tasks_raw": all_tasks` to the dict that `_gather_internal` returns, alongside the existing keys (`overdue_tasks`, `due_today`, etc.). This is the key `run_compilation` will use for snapshot diffing. - [ ] **Step 5: Run tests** ```bash docker compose exec app /opt/venv/bin/pytest tests/test_briefing_pipeline.py -v -k "task_snapshot or task_hash or split_changed" ``` Expected: all PASS. - [ ] **Step 6: Commit** ```bash git add src/fabledassistant/services/briefing_pipeline.py tests/test_briefing_pipeline.py git commit -m "feat(briefing): add task change detection helpers and task_id to _gather_internal" ``` --- ## Task 8: Wire Pre-processing into `run_compilation` **Files:** - Modify: `src/fabledassistant/services/briefing_pipeline.py` - Modify: `src/fabledassistant/routes/briefing.py` - Modify: `src/fabledassistant/services/briefing_scheduler.py` (read the file first to find all `run_compilation` calls) - [ ] **Step 1: Read the briefing scheduler to find callers** ```bash docker compose exec app grep -n "run_compilation\|post_message" src/fabledassistant/services/briefing_scheduler.py ``` Note every line number that calls `run_compilation` or `post_message` — you must update all of them in step 4. - [ ] **Step 2: Update `_external_system_prompt` and `_external_user_prompt` for news cards** In `_external_system_prompt()`, replace the current string with: ```python def _external_system_prompt() -> str: return ( "You are a briefing assistant for external information. Your job is to present " "selected news items and summarise any remaining RSS content. " "IMPORTANT: Weather is handled separately — do NOT include any weather section.\n\n" "Format each news item EXACTLY as:\n" "**[Headline text](source_url)**\n" "*Outlet Name · Day Month*\n" "One or two sentence summary.\n\n" "Present news items in the EXACT ORDER they are provided. Do not reorder them. " "After the news cards, add a brief paragraph for any remaining context." ) ``` - [ ] **Step 3: Rewrite `run_compilation` to add pre-processing and return `(text, metadata)`** Replace the existing `run_compilation` function entirely: ```python 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.briefing_profile import get_profile_body 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 profile_body, temp_unit = await asyncio.gather( get_profile_body(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) from fabledassistant.services.weather import get_cached_weather_rows # Parallel raw gather — include weather rows in the same gather to avoid a second 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), ) # Task change detection — uses the raw task dicts added in Task 7 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")] # Weather staleness gate — parse_weather_card_data 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 ────────────────────────────────────────────────────── # Rebuild internal_data with changed tasks only internal_data_filtered = dict(internal_data) internal_data_filtered["unchanged_task_count"] = unchanged_count # Replace task lists with only changed tasks (formatted) today = internal_data["date"] changed_formatted = [format_task(t) for t in changed_tasks] internal_data_filtered["overdue_tasks"] = [ f for f in changed_formatted if any(t.get("due_date") and t["due_date"] < today and t.get("status") != "done" for t in changed_tasks if format_task(t) == f) ] # Simplified: pass all changed tasks, let the LLM sort by urgency internal_data_filtered["changed_tasks"] = changed_formatted # Build filtered external data (no weather — card handles it) external_data_filtered = { "rss_items": filtered_rss, "weather": [], # Suppressed — handled by WeatherCard } internal_text, external_text = await asyncio.gather( _llm_synthesise( _internal_system_prompt(profile_body), _internal_user_prompt(internal_data_filtered, slot), model, ), _llm_synthesise( _external_system_prompt(), _external_user_prompt(external_data_filtered, slot, temp_unit), model, ), ) # ── Post-processing ──────────────────────────────────────────────────── # Upsert task snapshots so next run can diff await upsert_task_snapshots(user_id, all_tasks) # Build metadata metadata: dict = {"rss_item_ids": rss_item_ids, "weather": weather_card} # Build output text if not internal_text and not external_text: logger.warning( "Briefing compilation produced no content for user %d slot %s", user_id, slot ) return "", metadata greeting = slot_greeting(slot) parts = [f"**{greeting} — {today}**", ""] if internal_text: parts += ["## Your Day", "", internal_text, ""] if external_text: parts += ["## The World", "", external_text] return "\n".join(parts).strip(), metadata ``` > **Note:** `_gather_internal` needs to also return the raw task objects (with `task_id`) for snapshot diffing. See step below. - [ ] **Step 4: Add `get_cached_weather_rows()` to `services/weather.py`** The new pipeline needs the raw `WeatherCache` ORM rows (not the processed dicts) so it can call `parse_weather_card_data()`. Add this function to `weather.py`: ```python async def get_cached_weather_rows(user_id: int) -> list: """Return raw WeatherCache ORM rows for a user (for card parsing).""" async with async_session() as session: result = await session.execute( select(WeatherCache).where(WeatherCache.user_id == user_id) ) return list(result.scalars().all()) ``` - [ ] **Step 5: Update `_internal_user_prompt` to handle changed tasks** The prompt currently references `overdue_tasks`, `due_today`, `high_priority`. Add an `unchanged_task_count` line: ```python def _internal_user_prompt(data: dict, slot: str) -> str: lines = [f"Briefing slot: {slot}", f"Date: {data['date']}", ""] if data.get("unchanged_task_count", 0) > 0: lines.append( f"({data['unchanged_task_count']} tasks are unchanged since the last briefing " "— acknowledge briefly, do not list them.)" ) lines.append("") # ... rest unchanged ``` - [ ] **Step 6: Update `routes/briefing.py` trigger endpoint** In `manual_trigger()`, unpack the tuple: ```python text, metadata = await run_compilation(g.user.id, slot, model) msg = await post_message(conv.id, "assistant", text, metadata=metadata) ``` - [ ] **Step 7: Update `services/briefing_scheduler.py`** Find all calls to `run_compilation` (from step 1) and unpack the returned tuple, passing `metadata` to `post_message`. The pattern will be the same as step 6. - [ ] **Step 8: Run the full test suite** ```bash make test ``` Expected: all existing tests still PASS (no regressions). - [ ] **Step 9: Commit** ```bash git add src/fabledassistant/services/briefing_pipeline.py \ src/fabledassistant/services/weather.py \ src/fabledassistant/routes/briefing.py \ src/fabledassistant/services/briefing_scheduler.py git commit -m "feat(briefing): wire pre-processing pipeline; run_compilation now returns (text, metadata)" ``` --- ## Task 9: Trigger RSS Classification from `rss.py` **Files:** - Modify: `src/fabledassistant/services/rss.py` - [ ] **Step 1: Find the `fetch_and_cache_feed` function and add classification trigger** After `await _prune_old_items(feed_id)` (the last line of `fetch_and_cache_feed`), add: ```python # Queue classification for newly stored items if any were added if new_count > 0 and new_item_ids: import asyncio as _asyncio from fabledassistant.services.rss_classifier import classify_and_store # Fire-and-forget — don't await, don't block feed fetch _asyncio.create_task(classify_and_store(new_item_ids, _feed_user_id)) return new_count ``` To make this work, you need to: 1. Collect `new_item_ids` during the loop (after `session.commit()`, refresh the items to get their DB-assigned IDs, or collect IDs inline) 2. Expose `_feed_user_id` by fetching it from the `RssFeed` row Update `fetch_and_cache_feed` to capture new item IDs and the feed's `user_id`: ```python async def fetch_and_cache_feed(feed_id: int, url: str) -> int: # ... existing fetch/parse code ... new_count = 0 new_item_ids: list[int] = [] feed_user_id: int | None = None async with async_session() as session: for entry in parsed.entries: # ... existing upsert code ... item = RssItem(feed_id=feed_id, **item_data) session.add(item) new_count += 1 feed_row = await session.get(RssFeed, feed_id) if feed_row: feed_row.last_fetched_at = datetime.now(timezone.utc) feed_user_id = feed_row.user_id if not feed_row.title and parsed.feed.get("title"): feed_row.title = parsed.feed.title[:200] await session.commit() # Collect IDs of newly inserted items after commit. # We query classified_at IS NULL (not just the items inserted above) because # classification is best-effort and may have failed on previous fetches. # Re-queuing all unclassified items for this feed on each fetch is intentional: # it provides automatic retry without a separate retry loop. The classifier # only writes to items it successfully classifies, so already-classified items # are not re-processed (they have classified_at set). if new_count > 0: result = await session.execute( select(RssItem.id).where( RssItem.feed_id == feed_id, RssItem.classified_at.is_(None), ) ) new_item_ids = list(result.scalars().all()) await _prune_old_items(feed_id) if new_count > 0 and new_item_ids and feed_user_id is not None: import asyncio as _asyncio from fabledassistant.services.rss_classifier import classify_and_store _asyncio.create_task(classify_and_store(new_item_ids, feed_user_id)) return new_count ``` - [ ] **Step 2: Run the test suite** ```bash make test ``` Expected: all PASS (no regressions — classification is fire-and-forget so existing tests are unaffected). - [ ] **Step 3: Commit** ```bash git add src/fabledassistant/services/rss.py git commit -m "feat(briefing): trigger RSS classification after new items are stored" ``` --- ## Task 10: Reaction Endpoints **Files:** - Modify: `src/fabledassistant/routes/briefing.py` - [ ] **Step 1: Add reaction endpoints at the bottom of `routes/briefing.py`** ```python # ── RSS Reactions ───────────────────────────────────────────────────────────── @briefing_bp.route("/rss-reactions", methods=["POST"]) @_REQUIRE async def upsert_rss_reaction(): """Upsert a 👍/👎 reaction on an RSS item. Same reaction toggles off; opposite flips.""" data = await request.get_json() rss_item_id = data.get("rss_item_id") reaction = data.get("reaction") if not rss_item_id or reaction not in ("up", "down"): return jsonify({"error": "rss_item_id and reaction ('up'|'down') required"}), 400 from sqlalchemy import text as _text async with async_session() as session: # Ownership check: verify item belongs to a feed owned by this user result = await session.execute( _text(""" SELECT i.id FROM rss_items i JOIN rss_feeds f ON f.id = i.feed_id WHERE i.id = :item_id AND f.user_id = :uid """).bindparams(item_id=rss_item_id, uid=g.user.id) ) if result.first() is None: return jsonify({"error": "Not found"}), 404 # Check existing reaction existing = await session.execute( _text(""" SELECT id, reaction FROM rss_item_reactions WHERE user_id = :uid AND rss_item_id = :item_id """).bindparams(uid=g.user.id, item_id=rss_item_id) ) row = existing.first() if row is None: # Insert await session.execute( _text(""" INSERT INTO rss_item_reactions (user_id, rss_item_id, reaction) VALUES (:uid, :item_id, :reaction) """).bindparams(uid=g.user.id, item_id=rss_item_id, reaction=reaction) ) action = "created" elif row.reaction == reaction: # Toggle off await session.execute( _text(""" DELETE FROM rss_item_reactions WHERE user_id = :uid AND rss_item_id = :item_id """).bindparams(uid=g.user.id, item_id=rss_item_id) ) action = "removed" else: # Flip await session.execute( _text(""" UPDATE rss_item_reactions SET reaction = :reaction WHERE user_id = :uid AND rss_item_id = :item_id """).bindparams(reaction=reaction, uid=g.user.id, item_id=rss_item_id) ) action = "updated" await session.commit() return jsonify({"ok": True, "action": action}) @briefing_bp.route("/rss-reactions/", methods=["DELETE"]) @_REQUIRE async def delete_rss_reaction(item_id: int): """Explicitly remove a reaction (useful for MCP/external API callers).""" from sqlalchemy import text as _text async with async_session() as session: await session.execute( _text(""" DELETE FROM rss_item_reactions WHERE user_id = :uid AND rss_item_id = :item_id """).bindparams(uid=g.user.id, item_id=item_id) ) await session.commit() return jsonify({"ok": True}) ``` - [ ] **Step 2: Run the test suite** ```bash make test ``` Expected: all PASS. - [ ] **Step 3: Commit** ```bash git add src/fabledassistant/routes/briefing.py git commit -m "feat(briefing): add POST/DELETE /api/briefing/rss-reactions endpoints" ``` --- ## Task 11: Frontend API Helpers **Files:** - Modify: `frontend/src/api/client.ts` - [ ] **Step 1: Add the two new API helpers to `client.ts`** Find the section at the end of `client.ts` where other helpers are defined and add: ```typescript export async function postRssReaction( rssItemId: number, reaction: 'up' | 'down' ): Promise<{ ok: boolean; action: string }> { return apiPost('/api/briefing/rss-reactions', { rss_item_id: rssItemId, reaction }); } export async function deleteRssReaction(rssItemId: number): Promise<{ ok: boolean }> { return apiDelete(`/api/briefing/rss-reactions/${rssItemId}`); } ``` - [ ] **Step 2: Run TypeScript check** ```bash make typecheck ``` Expected: no errors. - [ ] **Step 3: Commit** ```bash git add frontend/src/api/client.ts git commit -m "feat(briefing): add postRssReaction and deleteRssReaction API helpers" ``` --- ## Task 12: WeatherCard.vue **Files:** - Create: `frontend/src/components/WeatherCard.vue` - [ ] **Step 1: Create the component** ```vue ``` - [ ] **Step 2: TypeScript check** ```bash make typecheck ``` Expected: no errors. - [ ] **Step 3: Commit** ```bash git add frontend/src/components/WeatherCard.vue git commit -m "feat(briefing): add WeatherCard.vue component" ``` --- ## Task 13: BriefingView.vue — Metadata Integration **Files:** - Modify: `frontend/src/views/BriefingView.vue` - [ ] **Step 1: Read the current `BriefingView.vue` to understand its structure** ```bash docker compose exec app cat frontend/src/views/BriefingView.vue | head -100 ``` Or use the Read tool. Understand: how messages are loaded, how they are rendered, where to insert the WeatherCard. - [ ] **Step 2: Add WeatherCard import and type definitions** At the top of `