diff --git a/src/fabledassistant/services/consolidation.py b/src/fabledassistant/services/consolidation.py
new file mode 100644
index 0000000..183f066
--- /dev/null
+++ b/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
diff --git a/tests/test_consolidation.py b/tests/test_consolidation.py
new file mode 100644
index 0000000..fe4766f
--- /dev/null
+++ b/tests/test_consolidation.py
@@ -0,0 +1,108 @@
+"""Tests for the task-body consolidation pipeline (gate + full pass).
+
+Design: docs/superpowers/specs/2026-05-13-task-as-durable-record-design.md
+"""
+from unittest.mock import AsyncMock, patch
+
+
+# ── Gate logic ───────────────────────────────────────────────────────────────
+
+
+async def test_maybe_consolidate_below_threshold_does_not_fire():
+    """log_added with only 2 logs since last pass → no consolidation."""
+    from fabledassistant.services import consolidation
+
+    with patch.object(
+        consolidation, "_logs_since_last_consolidation",
+        new=AsyncMock(return_value=2),
+    ), patch.object(
+        consolidation, "consolidate_task", new=AsyncMock(),
+    ) as mock_consolidate, patch.object(
+        consolidation, "_auto_consolidate_enabled", new=AsyncMock(return_value=True),
+    ):
+        await consolidation.maybe_consolidate(
+            user_id=1, task_id=42, reason="log_added",
+        )
+
+    mock_consolidate.assert_not_awaited()
+
+
+async def test_maybe_consolidate_at_threshold_fires():
+    """log_added with N logs since last pass → consolidation scheduled."""
+    from fabledassistant.services import consolidation
+
+    # asyncio.create_task is the actual scheduler; replace it so the test
+    # doesn't need an event-loop background task to settle.
+    with patch.object(
+        consolidation, "_logs_since_last_consolidation",
+        new=AsyncMock(return_value=3),
+    ), patch.object(
+        consolidation, "_auto_consolidate_enabled", new=AsyncMock(return_value=True),
+    ), patch(
+        "fabledassistant.services.consolidation.asyncio.create_task",
+    ) as mock_create_task:
+        await consolidation.maybe_consolidate(
+            user_id=1, task_id=42, reason="log_added",
+        )
+
+    mock_create_task.assert_called_once()
+
+
+async def test_maybe_consolidate_task_closed_always_fires():
+    """task_closed bypasses the log-count gate."""
+    from fabledassistant.services import consolidation
+
+    with patch.object(
+        consolidation, "_logs_since_last_consolidation",
+        new=AsyncMock(return_value=0),
+    ), patch.object(
+        consolidation, "_auto_consolidate_enabled", new=AsyncMock(return_value=True),
+    ), patch(
+        "fabledassistant.services.consolidation.asyncio.create_task",
+    ) as mock_create_task:
+        await consolidation.maybe_consolidate(
+            user_id=1, task_id=42, reason="task_closed",
+        )
+
+    mock_create_task.assert_called_once()
+
+
+async def test_maybe_consolidate_setting_off_blocks_both_reasons():
+    """auto_consolidate_tasks=false → neither trigger fires."""
+    from fabledassistant.services import consolidation
+
+    with patch.object(
+        consolidation, "_logs_since_last_consolidation",
+        new=AsyncMock(return_value=5),
+    ), patch.object(
+        consolidation, "_auto_consolidate_enabled", new=AsyncMock(return_value=False),
+    ), patch(
+        "fabledassistant.services.consolidation.asyncio.create_task",
+    ) as mock_create_task:
+        await consolidation.maybe_consolidate(
+            user_id=1, task_id=42, reason="log_added",
+        )
+        await consolidation.maybe_consolidate(
+            user_id=1, task_id=42, reason="task_closed",
+        )
+
+    mock_create_task.assert_not_called()
+
+
+async def test_maybe_consolidate_unknown_reason_is_noop():
+    """Unknown reasons get logged and skipped — not raised."""
+    from fabledassistant.services import consolidation
+
+    with patch.object(
+        consolidation, "_logs_since_last_consolidation",
+        new=AsyncMock(return_value=99),
+    ), patch.object(
+        consolidation, "_auto_consolidate_enabled", new=AsyncMock(return_value=True),
+    ), patch(
+        "fabledassistant.services.consolidation.asyncio.create_task",
+    ) as mock_create_task:
+        await consolidation.maybe_consolidate(
+            user_id=1, task_id=42, reason="some_other_reason",
+        )
+
+    mock_create_task.assert_not_called()