feat(consolidation): debounced gate for task body consolidation

New services/consolidation.py module with maybe_consolidate() — the
debounced trigger gate. Two reasons:

- log_added: gated by DEFAULT_LOG_THRESHOLD (3) counted since the task's
  consolidated_at timestamp.
- task_closed: bypasses the count gate; fires whenever status flips to
  done/cancelled.

Both reasons gated by the auto_consolidate_tasks user setting (default
on). Per-task asyncio.Lock prevents two simultaneous passes for the same
task. consolidate_task is a stub here — full implementation in the next
commit.

src/fabledassistant/services/consolidation.py

@@ -0,0 +1,86 @@
+"""Background task-body consolidation pipeline.
+
+Reads a task's description (user goal) + work logs (chronological) and writes
+a 1-3 paragraph summary into Note.body via the background model. Triggered by
+log accumulation (debounced), status transitions to terminal states, and a
+manual API endpoint.
+
+Design: docs/superpowers/specs/2026-05-13-task-as-durable-record-design.md
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+from collections import defaultdict
+from datetime import datetime, timezone
+
+from sqlalchemy import func, select
+
+from fabledassistant.models import async_session
+from fabledassistant.models.note import Note
+from fabledassistant.models.task_log import TaskLog
+
+logger = logging.getLogger(__name__)
+
+# Trigger thresholds. Tunable as constants; could be promoted to env vars if
+# the defaults prove wrong in practice.
+DEFAULT_LOG_THRESHOLD = 3
+MAX_LOGS_FOR_PROMPT = 50
+MAX_PROMPT_INPUT_CHARS = 8000
+
+# Per-task asyncio locks to prevent two simultaneous consolidations of the
+# same task. Single-process; no cross-process coordination needed.
+_locks: dict[int, asyncio.Lock] = defaultdict(asyncio.Lock)
+
+
+async def _logs_since_last_consolidation(user_id: int, task_id: int) -> int:
+    """Count work logs that arrived after the most recent consolidation pass.
+
+    Returns 0 if the task doesn't exist. Returns the total log count when
+    consolidated_at is NULL (i.e. never consolidated).
+    """
+    async with async_session() as session:
+        task = (
+            await session.execute(
+                select(Note).where(Note.id == task_id, Note.user_id == user_id)
+            )
+        ).scalars().first()
+        if task is None:
+            return 0
+        stmt = select(func.count(TaskLog.id)).where(
+            TaskLog.task_id == task_id, TaskLog.user_id == user_id,
+        )
+        if task.consolidated_at is not None:
+            stmt = stmt.where(TaskLog.created_at > task.consolidated_at)
+        return int((await session.execute(stmt)).scalar() or 0)
+
+
+async def _auto_consolidate_enabled(user_id: int) -> bool:
+    """User-level setting; default true. Manual endpoint bypasses this."""
+    from fabledassistant.services.settings import get_setting
+    val = await get_setting(user_id, "auto_consolidate_tasks", "true")
+    return str(val).lower() in ("true", "1", "yes")
+
+
+async def maybe_consolidate(user_id: int, task_id: int, *, reason: str) -> None:
+    """Debounced gate. Decides whether to schedule a consolidate_task pass.
+
+    reason='log_added'   — gated by log count >= DEFAULT_LOG_THRESHOLD
+    reason='task_closed' — proceeds unconditionally (subject to setting)
+    """
+    if not await _auto_consolidate_enabled(user_id):
+        return
+    if reason == "log_added":
+        n = await _logs_since_last_consolidation(user_id, task_id)
+        if n < DEFAULT_LOG_THRESHOLD:
+            return
+    elif reason != "task_closed":
+        logger.warning("maybe_consolidate: unknown reason %r", reason)
+        return
+    # Fire-and-forget; consolidate_task handles its own errors.
+    asyncio.create_task(consolidate_task(user_id, task_id))
+
+
+async def consolidate_task(user_id: int, task_id: int) -> None:
+    """Stub — full implementation in the next task in the plan."""
+    raise NotImplementedError