fix(soft-delete): filter trashed rows across read/write paths

Drift-audit Group 3 (soft-delete lifecycle gaps). Trashed rows were
leaking into reads and being mutated/resurrected by writes:

- update SELECTs now exclude trashed rows: update_milestone,
  update_project, update_event, and get_milestone_in_project (the latter
  backs all four milestone routes). Mutating a trashed row silently
  persisted and reappeared on restore.
- MCP get_recent (notes/projects/events) and list_tags now filter
  deleted_at IS NULL, so trashed items stop surfacing in the agent's
  bootstrap context and tag counts.
- convert_task_to_note clears recurrence_rule + recurrence_next_spawn_at
  so a demoted note can't spawn children via the (now-live) sweep.
- caldav pull skips locally-trashed events (by caldav_uid) instead of
  resurrecting them via update or creating a duplicate live copy.
- trash _cascade now stamps the FULL sub-task subtree (iterative descent),
  not just direct children, so deeply nested sub-tasks restore as one
  batch. Test updated for the new descent query.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/fabledassistant/mcp/tools/recent.py

@@ -38,7 +38,8 @@ async def get_recent(days: int = 7, limit: int = 25) -> dict:
     items: list[dict] = []
     async with async_session() as session:
         notes = (await session.execute(
-            select(Note).where(Note.user_id == uid, Note.updated_at >= since)
+            select(Note).where(Note.user_id == uid, Note.updated_at >= since,
+                               Note.deleted_at.is_(None))
             .order_by(Note.updated_at.desc()).limit(limit)
         )).scalars().all()
         for n in notes:
@@ -50,7 +51,8 @@ async def get_recent(days: int = 7, limit: int = 25) -> dict:
             })
         projects = (await session.execute(
             select(Project).where(Project.user_id == uid,
-                                  Project.updated_at >= since)
+                                  Project.updated_at >= since,
+                                  Project.deleted_at.is_(None))
             .order_by(Project.updated_at.desc()).limit(limit)
         )).scalars().all()
         for p in projects:
@@ -62,7 +64,8 @@ async def get_recent(days: int = 7, limit: int = 25) -> dict:
             })
         events = (await session.execute(
             select(Event).where(Event.user_id == uid,
-                                Event.updated_at >= since)
+                                Event.updated_at >= since,
+                                Event.deleted_at.is_(None))
             .order_by(Event.updated_at.desc()).limit(limit)
         )).scalars().all()
         for e in events:

src/fabledassistant/mcp/tools/tags.py

@@ -39,7 +39,7 @@ async def list_tags(limit: int = 50) -> dict:
     limit = max(1, min(limit, 200))
     async with async_session() as session:
         result = await session.execute(
-            select(Note.tags).where(Note.user_id == uid)
+            select(Note.tags).where(Note.user_id == uid, Note.deleted_at.is_(None))
         )
         tag_lists = [row[0] for row in result.all()]
     counts = _aggregate_tag_counts(tag_lists)

src/fabledassistant/services/caldav_sync.py

@@ -125,7 +125,7 @@ async def sync_user_events(user_id: int) -> dict:
         logger.warning("CalDAV pull sync failed for user %d", user_id, exc_info=True)
         return {"error": "CalDAV fetch failed"}
 
-    created = updated = unchanged = 0
+    created = updated = unchanged = skipped = 0
 
     async with async_session() as session:
         for ev in remote_events:
@@ -149,6 +149,14 @@ async def sync_user_events(user_id: int) -> dict:
             )
             existing = result.scalar_one_or_none()
 
+            if existing is not None and existing.deleted_at is not None:
+                # The user trashed this event locally. Don't resurrect it by
+                # updating, and don't create a duplicate live copy — leave it
+                # in the trash. (Propagating the delete to the remote server is
+                # tracked separately.)
+                skipped += 1
+                continue
+
             if existing is None:
                 # Create new event
                 new_ev = Event(
@@ -180,10 +188,10 @@ async def sync_user_events(user_id: int) -> dict:
         await session.commit()
 
     logger.info(
-        "CalDAV sync user %d: %d created, %d updated, %d unchanged",
-        user_id, created, updated, unchanged,
+        "CalDAV sync user %d: %d created, %d updated, %d unchanged, %d skipped (trashed)",
+        user_id, created, updated, unchanged, skipped,
     )
-    return {"created": created, "updated": updated, "unchanged": unchanged}
+    return {"created": created, "updated": updated, "unchanged": unchanged, "skipped": skipped}
 
 
 async def sync_all_users() -> None:

src/fabledassistant/services/events.py

@@ -261,7 +261,10 @@ async def update_event(user_id: int, event_id: int, **fields) -> Event | None:
     """
     async with async_session() as session:
         result = await session.execute(
-            select(Event).where(Event.id == event_id, Event.user_id == user_id)
+            select(Event).where(
+                Event.id == event_id, Event.user_id == user_id,
+                Event.deleted_at.is_(None),
+            )
         )
         event = result.scalar_one_or_none()
         if event is None:

src/fabledassistant/services/milestones.py

@@ -53,6 +53,7 @@ async def get_milestone_in_project(project_id: int, milestone_id: int) -> Milest
             select(Milestone).where(
                 Milestone.id == milestone_id,
                 Milestone.project_id == project_id,
+                Milestone.deleted_at.is_(None),
             )
         )
         return result.scalars().first()
@@ -108,7 +109,10 @@ async def list_milestones(
 async def update_milestone(user_id: int, milestone_id: int, **fields: object) -> Milestone | None:
     async with async_session() as session:
         result = await session.execute(
-            select(Milestone).where(Milestone.id == milestone_id, Milestone.user_id == user_id)
+            select(Milestone).where(
+                Milestone.id == milestone_id, Milestone.user_id == user_id,
+                Milestone.deleted_at.is_(None),
+            )
         )
         milestone = result.scalars().first()
         if milestone is None:

src/fabledassistant/services/notes.py

@@ -384,6 +384,10 @@ async def convert_task_to_note(user_id: int, note_id: int) -> Note:
         note.status = None
         note.priority = None
         note.due_date = None
+        # A plain note is not a task and must not recur — clear the rule and
+        # any armed spawn timestamp so the recurrence sweep never picks it up.
+        note.recurrence_rule = None
+        note.recurrence_next_spawn_at = None
         note.updated_at = datetime.now(timezone.utc)
         await session.commit()
         await session.refresh(note)

src/fabledassistant/services/projects.py

@@ -79,7 +79,10 @@ async def list_projects(user_id: int, status: str | None = None) -> list[Project
 async def update_project(user_id: int, project_id: int, **fields: object) -> Project | None:
     async with async_session() as session:
         result = await session.execute(
-            select(Project).where(Project.id == project_id, Project.user_id == user_id)
+            select(Project).where(
+                Project.id == project_id, Project.user_id == user_id,
+                Project.deleted_at.is_(None),
+            )
         )
         project = result.scalars().first()
         if project is None:

src/fabledassistant/services/trash.py

@@ -106,8 +106,21 @@ async def _cascade(session, user_id: int, etype: str, eid: int, batch: str, now)
         await _set(session, Note, [Note.user_id == user_id, Note.milestone_id == eid], batch, now)
         await _set(session, Milestone, [Milestone.user_id == user_id, Milestone.id == eid], batch, now)
     elif etype in ("note", "task"):
-        await _set(session, Note, [Note.user_id == user_id, Note.parent_id == eid], batch, now)  # sub-tasks
-        await _set(session, Note, [Note.user_id == user_id, Note.id == eid], batch, now)
+        # Stamp the entire sub-task subtree (not just direct children) so a
+        # deeply nested task and all its descendants trash/restore as one batch.
+        ids = [eid]
+        frontier = [eid]
+        while frontier:
+            children = (await session.execute(
+                select(Note.id).where(
+                    Note.user_id == user_id,
+                    Note.parent_id.in_(frontier),
+                    Note.deleted_at.is_(None),
+                )
+            )).scalars().all()
+            frontier = [c for c in children if c not in ids]
+            ids.extend(frontier)
+        await _set(session, Note, [Note.user_id == user_id, Note.id.in_(ids)], batch, now)
     elif etype == "event":
         await _set(session, Event, [Event.user_id == user_id, Event.id == eid], batch, now)
     elif etype == "rulebook":

tests/test_services_trash.py

@@ -21,15 +21,18 @@ def _exists_result(found=True):
 @pytest.mark.asyncio
 async def test_delete_note_returns_batch_and_commits():
     session = _make_mock_session()
-    # 1 execute for _exists_alive, then 2 for the note cascade (sub-tasks + self)
-    session.execute = AsyncMock(side_effect=[_exists_result(True), MagicMock(), MagicMock()])
+    # exists-check, then the subtree descent: one child-lookup (no children
+    # here) + one _set stamping the whole subtree.
+    no_children = MagicMock()
+    no_children.scalars.return_value.all.return_value = []
+    session.execute = AsyncMock(side_effect=[_exists_result(True), no_children, MagicMock()])
     with patch("fabledassistant.services.trash.async_session") as cls:
         cls.return_value = session
         from fabledassistant.services.trash import delete
         batch = await delete(user_id=1, entity_type="note", entity_id=5)
     assert isinstance(batch, str) and len(batch) > 0
     assert session.commit.called
-    # exists-check + 2 cascade updates
+    # exists-check + subtree-descent query + subtree _set
     assert session.execute.await_count == 3