From e3c1e97cfa432d742a6adf3d9fe822357186fb67 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 25 Mar 2026 10:36:52 -0400 Subject: [PATCH] feat(briefing): add task change detection helpers and task_id to _gather_internal Co-Authored-By: Claude Sonnet 4.6 --- .../services/briefing_pipeline.py | 88 ++++++++++++++++++- tests/test_briefing_pipeline.py | 36 ++++++++ 2 files changed, 123 insertions(+), 1 deletion(-) diff --git a/src/fabledassistant/services/briefing_pipeline.py b/src/fabledassistant/services/briefing_pipeline.py index 92435a8..3306327 100644 --- a/src/fabledassistant/services/briefing_pipeline.py +++ b/src/fabledassistant/services/briefing_pipeline.py @@ -5,11 +5,14 @@ Slot names: 'compilation' (4am), 'morning' (8am), 'midday' (12pm), 'afternoon' ( """ import asyncio +import hashlib import logging -from datetime import date +from datetime import date, datetime, timezone import httpx +from fabledassistant.models import async_session + from fabledassistant.config import Config from fabledassistant.services.settings import get_setting @@ -38,6 +41,86 @@ def format_task(task: dict) -> str: 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: @@ -49,10 +132,12 @@ async def _gather_internal(user_id: int) -> dict: today = date.today().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) all_tasks = [ { + "task_id": t.id, "title": t.title, "status": t.status, "due_date": t.due_date.isoformat() if t.due_date else None, @@ -105,6 +190,7 @@ async def _gather_internal(user_id: int) -> dict: "high_priority": high_priority, "calendar_events": calendar_events, "active_projects": projects_summary, + "all_tasks_raw": all_tasks, } diff --git a/tests/test_briefing_pipeline.py b/tests/test_briefing_pipeline.py index f642b6b..c1f725f 100644 --- a/tests/test_briefing_pipeline.py +++ b/tests/test_briefing_pipeline.py @@ -29,6 +29,42 @@ def test_slot_greeting(): assert slot_greeting("midday") != slot_greeting("afternoon") +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 + assert h == compute_task_hash(task) + 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 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 + + @pytest.mark.asyncio async def test_post_message_accepts_metadata(): """post_message should accept an optional metadata dict and store it."""