feat(briefing): add task change detection helpers and task_id to _gather_internal

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

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,
     }
 
 

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."""