diff --git a/alembic/versions/0044_add_note_description_and_consolidated_at.py b/alembic/versions/0044_add_note_description_and_consolidated_at.py
new file mode 100644
index 0000000..2fe7957
--- /dev/null
+++ b/alembic/versions/0044_add_note_description_and_consolidated_at.py
@@ -0,0 +1,48 @@
+"""add note.description and note.consolidated_at
+
+Revision ID: 0044
+Revises: 0043
+Create Date: 2026-05-13
+
+Adds two columns to the ``notes`` table to support the task-as-durable-record
+design (spec 2026-05-13):
+
+- ``description``: user-stated goal / initial context. Meaningful when
+  ``is_task=true``; left NULL on knowledge notes.
+- ``consolidated_at``: timestamp of the most recent auto-summary pass. NULL
+  until the first consolidation runs.
+
+Backfill: for existing tasks we copy ``body`` into ``description`` so the
+user's original goal text is preserved when ``body`` is later overwritten by
+the auto-summary pipeline. The brief duplication window between
+``body`` and ``description`` is harmless and resolves on the first
+consolidation pass.
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "0044"
+down_revision = "0043"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column("notes", sa.Column("description", sa.Text(), nullable=True))
+    op.add_column(
+        "notes",
+        sa.Column("consolidated_at", sa.DateTime(timezone=True), nullable=True),
+    )
+    # is_task is a Python property (status IS NOT NULL); there's no DB column
+    # of that name. Backfill description from body for everything that
+    # qualifies as a task at the model layer.
+    op.execute(
+        "UPDATE notes SET description = body "
+        "WHERE status IS NOT NULL AND body IS NOT NULL"
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("notes", "consolidated_at")
+    op.drop_column("notes", "description")
diff --git a/alembic/versions/0045_add_note_version_pin_columns.py b/alembic/versions/0045_add_note_version_pin_columns.py
new file mode 100644
index 0000000..1fb5938
--- /dev/null
+++ b/alembic/versions/0045_add_note_version_pin_columns.py
@@ -0,0 +1,35 @@
+"""add pin_kind and pin_label to note_versions
+
+Revision ID: 0045
+Revises: 0044
+Create Date: 2026-05-13
+
+Two additive columns on note_versions to support tiered retention:
+
+- pin_kind: NULL (rolling autosave), 'auto' (system-declared via stability
+  scan), 'manual' (user-declared with optional commit-note label).
+- pin_label: NULL for rolling. Auto-generated for 'auto'; user-supplied
+  for 'manual' (may be NULL).
+
+No backfill — every existing row stays rolling. The auto-pin scan
+(services/version_pinning_scheduler.py, daily 03:00 UTC) will catch up on
+the first scheduled run after deploy.
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "0045"
+down_revision = "0044"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column("note_versions", sa.Column("pin_kind", sa.Text(), nullable=True))
+    op.add_column("note_versions", sa.Column("pin_label", sa.Text(), nullable=True))
+
+
+def downgrade() -> None:
+    op.drop_column("note_versions", "pin_label")
+    op.drop_column("note_versions", "pin_kind")
diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts
index 980cc66..19607c2 100644
--- a/frontend/src/api/client.ts
+++ b/frontend/src/api/client.ts
@@ -705,3 +705,32 @@ export interface ProfileObservationEntry {
 
 export const listProfileObservations = () =>
   apiGet<{ observations: ProfileObservationEntry[] }>('/api/profile/observations')
+
+
+// ── Tasks ────────────────────────────────────────────────────────────────────
+
+import type { Note as Task } from '../types/note'
+
+/** Manually trigger a consolidation pass for a task. Returns the freshly-
+ *  updated task with new body + consolidated_at. Bypasses the user's
+ *  auto_consolidate_tasks setting. */
+export const consolidateTask = (id: number) =>
+  apiPost<Task>(`/api/tasks/${id}/consolidate`, {})
+
+
+// ── Note Versions (pinning) ──────────────────────────────────────────────────
+
+import type { NoteVersion } from '../types/task'
+
+/** Mark a note version as manually pinned, optionally with a commit-note
+ *  label. Re-calling with a different label updates the label. */
+export const pinNoteVersion = (noteId: number, versionId: number, label?: string | null) =>
+  apiPost<NoteVersion>(
+    `/api/notes/${noteId}/versions/${versionId}/pin`,
+    { label: label ?? null },
+  )
+
+/** Downgrade a manually-pinned version back to rolling. Does NOT delete
+ *  the row — older rows may be FIFO-pruned by the next autosave. */
+export const unpinNoteVersion = (noteId: number, versionId: number) =>
+  apiDelete(`/api/notes/${noteId}/versions/${versionId}/pin`)
diff --git a/frontend/src/components/HistoryPanel.vue b/frontend/src/components/HistoryPanel.vue
index bd2385b..5e4cd21 100644
--- a/frontend/src/components/HistoryPanel.vue
+++ b/frontend/src/components/HistoryPanel.vue
@@ -1,6 +1,6 @@
 <script setup lang="ts">
 import { ref, computed, onMounted } from "vue";
-import { apiGet } from "@/api/client";
+import { apiGet, pinNoteVersion, unpinNoteVersion } from "@/api/client";
 import DiffView from "@/components/DiffView.vue";
 import type { DiffLine } from "@/composables/useAssist";
 
@@ -10,6 +10,8 @@ interface NoteVersion {
   title: string;
   tags: string[];
   body?: string;
+  pin_kind: "auto" | "manual" | null;
+  pin_label: string | null;
   created_at: string;
 }
 
@@ -105,6 +107,69 @@ function restore() {
   emit('restore', selectedVersion.value.body, selectedVersion.value.tags ?? []);
 }
 
+// ── Pin / unpin / edit-label state ──────────────────────────────────────────
+
+const editingLabel = ref(false);
+const draftLabel = ref("");
+const pinSaving = ref(false);
+
+function startEditLabel() {
+  if (!selectedVersion.value) return;
+  draftLabel.value = selectedVersion.value.pin_label ?? "";
+  editingLabel.value = true;
+}
+
+function cancelEditLabel() {
+  editingLabel.value = false;
+  draftLabel.value = "";
+}
+
+async function savePin() {
+  if (!selectedVersion.value || pinSaving.value) return;
+  pinSaving.value = true;
+  try {
+    const updated = await pinNoteVersion(
+      props.noteId, selectedVersion.value.id, draftLabel.value || null,
+    );
+    // Reflect server state in the local list + selected version.
+    const idx = versions.value.findIndex(v => v.id === updated.id);
+    if (idx >= 0) {
+      versions.value[idx].pin_kind = updated.pin_kind;
+      versions.value[idx].pin_label = updated.pin_label;
+    }
+    if (selectedVersion.value) {
+      selectedVersion.value.pin_kind = updated.pin_kind;
+      selectedVersion.value.pin_label = updated.pin_label;
+    }
+    editingLabel.value = false;
+  } catch {
+    // silent — leaving the form open lets the user retry
+  } finally {
+    pinSaving.value = false;
+  }
+}
+
+async function unpinSelected() {
+  if (!selectedVersion.value || pinSaving.value) return;
+  pinSaving.value = true;
+  try {
+    await unpinNoteVersion(props.noteId, selectedVersion.value.id);
+    const idx = versions.value.findIndex(v => v.id === selectedVersion.value!.id);
+    if (idx >= 0) {
+      versions.value[idx].pin_kind = null;
+      versions.value[idx].pin_label = null;
+    }
+    if (selectedVersion.value) {
+      selectedVersion.value.pin_kind = null;
+      selectedVersion.value.pin_label = null;
+    }
+  } catch {
+    // silent
+  } finally {
+    pinSaving.value = false;
+  }
+}
+
 onMounted(loadVersions);
 </script>
 
@@ -128,7 +193,25 @@ onMounted(loadVersions);
               :class="['history-item', { selected: selectedVersion?.id === v.id }]"
               @click="selectVersionItem(v)"
             >
-              <div class="history-item-title">{{ v.title || 'Untitled' }}</div>
+              <div class="history-item-title">
+                <span
+                  v-if="v.pin_kind === 'manual'"
+                  class="pin-badge pin-badge-manual"
+                  :title="v.pin_label || 'Pinned'"
+                  aria-label="manually pinned"
+                >●</span>
+                <span
+                  v-else-if="v.pin_kind === 'auto'"
+                  class="pin-badge pin-badge-auto"
+                  :title="v.pin_label || 'Auto-pinned (stable)'"
+                  aria-label="auto-pinned"
+                >◐</span>
+                {{ v.title || 'Untitled' }}
+              </div>
+              <div
+                v-if="v.pin_kind === 'manual' && v.pin_label"
+                class="history-item-label"
+              >{{ v.pin_label }}</div>
               <div class="history-item-date">{{ formatDate(v.created_at) }}</div>
             </div>
           </div>
@@ -137,7 +220,62 @@ onMounted(loadVersions);
           <div class="history-diff">
             <div v-if="loadingDetail" class="history-empty">Loading version...</div>
             <div v-else-if="!selectedVersion" class="history-empty">Select a version to compare.</div>
-            <DiffView v-else :diff="diff" />
+            <template v-else>
+              <div class="version-pin-controls">
+                <!-- Label editor -->
+                <div v-if="editingLabel" class="pin-label-form">
+                  <input
+                    v-model="draftLabel"
+                    maxlength="500"
+                    type="text"
+                    class="pin-label-input"
+                    placeholder="Optional label (e.g. 'post-network-refresh runbook')"
+                  />
+                  <button class="btn-pin-save" :disabled="pinSaving" @click="savePin">
+                    {{ pinSaving ? "Saving…" : "Save" }}
+                  </button>
+                  <button class="btn-pin-cancel" @click="cancelEditLabel">Cancel</button>
+                </div>
+
+                <!-- Default: kind-aware action row -->
+                <div v-else class="pin-actions">
+                  <span v-if="selectedVersion.pin_kind === 'manual'" class="pin-state">
+                    <span class="pin-badge pin-badge-manual">●</span>
+                    Manually pinned{{ selectedVersion.pin_label ? `: ${selectedVersion.pin_label}` : '' }}
+                  </span>
+                  <span v-else-if="selectedVersion.pin_kind === 'auto'" class="pin-state">
+                    <span class="pin-badge pin-badge-auto">◐</span>
+                    Auto-pinned{{ selectedVersion.pin_label ? `: ${selectedVersion.pin_label}` : '' }}
+                  </span>
+
+                  <button
+                    v-if="selectedVersion.pin_kind === null"
+                    class="btn-pin"
+                    :disabled="pinSaving"
+                    @click="startEditLabel"
+                  >Pin version</button>
+                  <button
+                    v-else-if="selectedVersion.pin_kind === 'manual'"
+                    class="btn-pin-edit"
+                    :disabled="pinSaving"
+                    @click="startEditLabel"
+                  >Edit label</button>
+                  <button
+                    v-else
+                    class="btn-pin"
+                    :disabled="pinSaving"
+                    @click="startEditLabel"
+                  >Pin permanently</button>
+                  <button
+                    v-if="selectedVersion.pin_kind === 'manual'"
+                    class="btn-unpin"
+                    :disabled="pinSaving"
+                    @click="unpinSelected"
+                  >Unpin</button>
+                </div>
+              </div>
+              <DiffView :diff="diff" />
+            </template>
           </div>
         </div>
 
@@ -268,4 +406,101 @@ onMounted(loadVersions);
   opacity: 0.5;
   cursor: default;
 }
+
+/* ── Pin badges + label rendering ───────────────────────────────────────── */
+.pin-badge {
+  display: inline-block;
+  width: 0.7rem;
+  text-align: center;
+  margin-right: 0.35rem;
+  font-size: 0.85em;
+  line-height: 1;
+}
+.pin-badge-manual { color: var(--color-primary, #6366f1); }
+.pin-badge-auto { color: var(--color-text-muted, rgba(255, 255, 255, 0.5)); }
+
+.history-item-label {
+  font-size: 0.72rem;
+  color: var(--color-primary, #6366f1);
+  font-style: italic;
+  margin-top: 0.15rem;
+  overflow: hidden;
+  text-overflow: ellipsis;
+  white-space: nowrap;
+}
+
+/* ── Pin controls above the diff ────────────────────────────────────────── */
+.version-pin-controls {
+  padding: 0.4rem 0.5rem 0.5rem;
+  border-bottom: 1px solid var(--color-border);
+  font-size: 0.82rem;
+}
+.pin-actions {
+  display: flex;
+  align-items: center;
+  gap: 0.5rem;
+  flex-wrap: wrap;
+}
+.pin-state {
+  font-style: italic;
+  color: var(--color-text-muted, rgba(255, 255, 255, 0.6));
+  flex: 1;
+  min-width: 0;
+  overflow: hidden;
+  text-overflow: ellipsis;
+  white-space: nowrap;
+}
+.btn-pin, .btn-pin-edit, .btn-unpin {
+  padding: 0.25rem 0.7rem;
+  font-size: 0.78rem;
+  background: transparent;
+  color: inherit;
+  border: 1px solid var(--color-border, rgba(255, 255, 255, 0.12));
+  border-radius: 999px;
+  cursor: pointer;
+}
+.btn-pin:hover:not(:disabled), .btn-pin-edit:hover:not(:disabled) {
+  background: rgba(99, 102, 241, 0.12);
+  border-color: var(--color-primary, #6366f1);
+}
+.btn-unpin:hover:not(:disabled) {
+  background: rgba(239, 68, 68, 0.10);
+  border-color: rgba(239, 68, 68, 0.5);
+}
+.pin-label-form {
+  display: flex;
+  gap: 0.4rem;
+  align-items: center;
+}
+.pin-label-input {
+  flex: 1;
+  padding: 0.3rem 0.5rem;
+  font-size: 0.85rem;
+  background: var(--color-input-bg, rgba(255, 255, 255, 0.03));
+  border: 1px solid var(--color-border, rgba(255, 255, 255, 0.12));
+  border-radius: var(--radius-sm, 4px);
+  color: inherit;
+}
+.pin-label-input:focus {
+  outline: none;
+  border-color: var(--color-primary, #6366f1);
+}
+.btn-pin-save, .btn-pin-cancel {
+  padding: 0.3rem 0.7rem;
+  font-size: 0.78rem;
+  background: transparent;
+  color: inherit;
+  border: 1px solid var(--color-border, rgba(255, 255, 255, 0.12));
+  border-radius: var(--radius-sm, 4px);
+  cursor: pointer;
+}
+.btn-pin-save:hover:not(:disabled) {
+  background: rgba(99, 102, 241, 0.12);
+  border-color: var(--color-primary, #6366f1);
+}
+.btn-pin-save:disabled, .btn-pin-cancel:disabled,
+.btn-pin:disabled, .btn-pin-edit:disabled, .btn-unpin:disabled {
+  opacity: 0.5;
+  cursor: progress;
+}
 </style>
diff --git a/frontend/src/types/note.ts b/frontend/src/types/note.ts
index 1e7c414..bceb8e2 100644
--- a/frontend/src/types/note.ts
+++ b/frontend/src/types/note.ts
@@ -6,6 +6,8 @@ export interface Note {
   id: number;
   title: string;
   body: string;
+  description: string | null;
+  consolidated_at: string | null;
   tags: string[];
   parent_id: number | null;
   parent_title?: string | null;
diff --git a/frontend/src/types/task.ts b/frontend/src/types/task.ts
index 3d838eb..b7ba893 100644
--- a/frontend/src/types/task.ts
+++ b/frontend/src/types/task.ts
@@ -31,5 +31,7 @@ export interface NoteVersion {
   title: string;
   tags: string[];
   body?: string;
+  pin_kind: "auto" | "manual" | null;
+  pin_label: string | null;
   created_at: string;
 }
diff --git a/frontend/src/views/SettingsView.vue b/frontend/src/views/SettingsView.vue
index 1fd2d78..fbf138f 100644
--- a/frontend/src/views/SettingsView.vue
+++ b/frontend/src/views/SettingsView.vue
@@ -19,6 +19,21 @@ const defaultModel = ref("");
 const userTimezone = ref("");
 const savingTimezone = ref(false);
 const timezoneSaved = ref(false);
+const autoConsolidateTasks = ref(true);
+const savingAutoConsolidate = ref(false);
+
+async function saveAutoConsolidate() {
+  savingAutoConsolidate.value = true;
+  try {
+    await apiPut('/api/settings', {
+      auto_consolidate_tasks: autoConsolidateTasks.value ? "true" : "false",
+    });
+  } catch {
+    toastStore.show('Failed to save setting', 'error');
+  } finally {
+    savingAutoConsolidate.value = false;
+  }
+}
 
 function detectTimezone() {
   userTimezone.value = Intl.DateTimeFormat().resolvedOptions().timeZone;
@@ -754,6 +769,8 @@ onMounted(async () => {
   defaultModel.value = allSettings.default_model ?? "";
   backgroundModel.value = allSettings.background_model ?? "";
   userTimezone.value = allSettings.user_timezone ?? "";
+  // Default true if unset; explicit "false" disables auto-consolidation.
+  autoConsolidateTasks.value = (allSettings.auto_consolidate_tasks ?? "true") !== "false";
   chatRetentionDays.value = allSettings.chat_retention_days !== undefined
     ? Number(allSettings.chat_retention_days)
     : 90;
@@ -1510,6 +1527,29 @@ function formatUserDate(iso: string): string {
         </div>
       </section>
 
+      <!-- Tasks -->
+      <section class="settings-section full-width">
+        <h2>Tasks</h2>
+        <p class="section-desc">
+          Task bodies are auto-summarized from accumulated work logs. The summary runs every few logs, plus on task close.
+        </p>
+        <div class="field">
+          <label class="checkbox-label">
+            <input
+              type="checkbox"
+              v-model="autoConsolidateTasks"
+              :disabled="savingAutoConsolidate"
+              @change="saveAutoConsolidate"
+            />
+            Auto-consolidate task bodies
+          </label>
+          <p class="field-hint">
+            When off, the task body is only refreshed when you click "Re-consolidate"
+            on a task. Existing summaries remain in place.
+          </p>
+        </div>
+      </section>
+
       <!-- Timezone -->
       <section class="settings-section full-width">
         <h2>Timezone</h2>
diff --git a/frontend/src/views/TaskEditorView.vue b/frontend/src/views/TaskEditorView.vue
index 1521181..db70b29 100644
--- a/frontend/src/views/TaskEditorView.vue
+++ b/frontend/src/views/TaskEditorView.vue
@@ -35,6 +35,8 @@ const toast = useToastStore();
 
 const title = ref("");
 const body = ref("");
+const description = ref("");
+const consolidatedAt = ref<string | null>(null);
 const tags = ref<string[]>([]);
 const status = ref<TaskStatus>("todo");
 const priority = ref<TaskPriority>("none");
@@ -107,6 +109,29 @@ async function toggleSubTask(sub: SubTask) {
 }
 const showPreview = ref(false);
 const sidebarOpen = ref(true);
+const reconsolidating = ref(false);
+
+// Body is machine-maintained once a consolidation pass has run. The editor
+// is gated to read-only in that state; the user can re-consolidate or rely
+// on log_work entries flowing into the next auto pass.
+const isBodyAutoMaintained = computed(() => consolidatedAt.value !== null);
+
+async function reconsolidate() {
+  if (!taskId.value || reconsolidating.value) return;
+  reconsolidating.value = true;
+  try {
+    const { consolidateTask } = await import("@/api/client");
+    const updated = await consolidateTask(taskId.value);
+    body.value = updated.body;
+    consolidatedAt.value = updated.consolidated_at ?? null;
+    savedBody = body.value;
+    toast.show("Task summary refreshed");
+  } catch {
+    toast.show("Failed to re-consolidate", "error");
+  } finally {
+    reconsolidating.value = false;
+  }
+}
 const editorRef = ref<InstanceType<typeof TiptapEditor> | null>(null);
 const titleRef = ref<HTMLInputElement | null>(null);
 const tiptapEditor = computed<Editor | null>(() => {
@@ -186,6 +211,7 @@ const { suggestedTags, appliedTags, suggestingTags, fetchTagSuggestions, applyTa
 
 let savedTitle = "";
 let savedBody = "";
+let savedDescription = "";
 let savedTags: string[] = [];
 let savedStatus: TaskStatus = "todo";
 let savedPriority: TaskPriority = "none";
@@ -198,6 +224,7 @@ function markDirty() {
   dirty.value =
     title.value !== savedTitle ||
     body.value !== savedBody ||
+    description.value !== savedDescription ||
     JSON.stringify(tags.value) !== JSON.stringify(savedTags) ||
     status.value !== savedStatus ||
     priority.value !== savedPriority ||
@@ -270,6 +297,8 @@ onMounted(async () => {
     if (store.currentTask) {
       title.value = store.currentTask.title;
       body.value = store.currentTask.body;
+      description.value = store.currentTask.description ?? "";
+      consolidatedAt.value = store.currentTask.consolidated_at ?? null;
       tags.value = [...(store.currentTask.tags || [])];
       status.value = store.currentTask.status as TaskStatus;
       priority.value = store.currentTask.priority as TaskPriority;
@@ -286,6 +315,7 @@ onMounted(async () => {
       recurrenceRule.value = noteTask.recurrence_rule ?? null;
       savedTitle = title.value;
       savedBody = body.value;
+      savedDescription = description.value;
       savedTags = [...tags.value];
       savedStatus = status.value;
       savedPriority = priority.value;
@@ -312,6 +342,7 @@ async function save() {
     const data = {
       title: title.value,
       body: body.value,
+      description: description.value,
       tags: tags.value,
       status: status.value,
       priority: priority.value,
@@ -325,6 +356,7 @@ async function save() {
       await store.updateTask(taskId.value!, data);
       savedTitle = title.value;
       savedBody = body.value;
+      savedDescription = description.value;
       savedTags = [...tags.value];
       savedStatus = status.value;
       savedPriority = priority.value;
@@ -377,6 +409,7 @@ async function doAutoSave() {
     await store.updateTask(taskId.value!, {
       title: title.value,
       body: body.value,
+      description: description.value,
       tags: tags.value,
       status: status.value,
       priority: priority.value,
@@ -388,6 +421,7 @@ async function doAutoSave() {
     } as Record<string, unknown>);
     savedTitle = title.value;
     savedBody = body.value;
+    savedDescription = description.value;
     savedTags = [...tags.value];
     savedStatus = status.value;
     savedPriority = priority.value;
@@ -431,7 +465,19 @@ useEditorGuards(dirty, save);
         @keydown.ctrl.e.prevent="tiptapEditor?.commands.focus()"
       />
 
-    </div><!-- /editor-header: title only -->
+      <div class="task-goal">
+        <label for="task-description" class="task-goal-label">Goal</label>
+        <textarea
+          id="task-description"
+          v-model="description"
+          placeholder="What are we trying to do here? (read-only context for the auto-summary)"
+          rows="2"
+          class="task-goal-input"
+          @input="markDirty"
+        ></textarea>
+      </div>
+
+    </div><!-- /editor-header: title + goal -->
 
     <!-- Two-column body: main (editor+log) | sidebar (metadata) -->
     <div class="task-body">
@@ -439,13 +485,33 @@ useEditorGuards(dirty, save);
       <!-- ── Main column ─────────────────────────────────────────── -->
       <div class="task-main" @keydown.ctrl.e.prevent="tiptapEditor?.commands.focus()">
 
-        <!-- Write / Preview tabs + toolbar sit above the editor -->
+        <!-- Auto-summary banner when consolidation has run on this task. -->
+        <div v-if="isBodyAutoMaintained" class="auto-summary-banner-editor">
+          <span class="auto-summary-icon" aria-hidden="true">✦</span>
+          Auto-summarized from work logs.
+          <button
+            type="button"
+            class="btn-reconsolidate"
+            :disabled="reconsolidating"
+            @click="reconsolidate"
+          >
+            {{ reconsolidating ? "Re-consolidating…" : "Re-consolidate" }}
+          </button>
+        </div>
+
+        <!-- Write / Preview tabs + toolbar sit above the editor.
+             Write tab hidden when body is machine-maintained — use Re-consolidate
+             or edit work logs instead. -->
         <div class="body-tabs-row">
           <div class="editor-tabs">
-            <button :class="['tab', { active: !showPreview }]" @click="showPreview = false">Write</button>
-            <button :class="['tab', { active: showPreview }]" @click="showPreview = true">Preview</button>
+            <button
+              v-if="!isBodyAutoMaintained"
+              :class="['tab', { active: !showPreview }]"
+              @click="showPreview = false"
+            >Write</button>
+            <button :class="['tab', { active: showPreview || isBodyAutoMaintained }]" @click="showPreview = true">Preview</button>
           </div>
-          <MarkdownToolbar v-show="!showPreview && assist.state.value === 'idle'" :editor="tiptapEditor" />
+          <MarkdownToolbar v-show="!showPreview && !isBodyAutoMaintained && assist.state.value === 'idle'" :editor="tiptapEditor" />
         </div>
 
         <!-- Streaming preview -->
@@ -459,9 +525,14 @@ useEditorGuards(dirty, save);
           <DiffView :diff="assist.diff.value" class="main-diff" />
         </template>
 
-        <!-- Normal: editor or preview -->
+        <!-- Normal: editor or preview. When body is machine-maintained,
+             always render the preview (read-only) — never the editor. -->
         <template v-else>
-          <div v-show="!showPreview" class="body-editor-wrap">
+          <div
+            v-if="!isBodyAutoMaintained"
+            v-show="!showPreview"
+            class="body-editor-wrap"
+          >
             <TiptapEditor
               ref="editorRef"
               :modelValue="body"
@@ -471,7 +542,11 @@ useEditorGuards(dirty, save);
               @escape="titleRef?.focus()"
             />
           </div>
-          <div v-show="showPreview" class="preview-pane prose" v-html="renderedPreview" />
+          <div
+            v-show="showPreview || isBodyAutoMaintained"
+            class="preview-pane prose"
+            v-html="renderedPreview"
+          />
         </template>
 
         <div v-if="assist.error.value" class="assist-error">{{ assist.error.value }}</div>
@@ -971,4 +1046,77 @@ useEditorGuards(dirty, save);
     overflow-y: visible;
   }
 }
+
+/* ── Goal (description) input ─────────────────────────────────────────────── */
+.task-goal {
+  display: flex;
+  flex-direction: column;
+  gap: 0.25rem;
+  margin: 0.5rem 0 0.25rem;
+}
+.task-goal-label {
+  font-family: var(--font-display, "Fraunces", serif);
+  font-style: italic;
+  font-size: 0.78rem;
+  font-weight: 500;
+  letter-spacing: 0.04em;
+  text-transform: uppercase;
+  color: var(--color-text-muted, rgba(255, 255, 255, 0.5));
+}
+.task-goal-input {
+  width: 100%;
+  resize: vertical;
+  min-height: 2.4rem;
+  padding: 0.5rem 0.6rem;
+  font: inherit;
+  font-size: 0.95rem;
+  line-height: 1.4;
+  color: var(--color-text, inherit);
+  background: var(--color-input-bg, rgba(255, 255, 255, 0.03));
+  border: 1px solid var(--color-border, rgba(255, 255, 255, 0.08));
+  border-radius: var(--radius-md, 8px);
+}
+.task-goal-input:focus {
+  outline: none;
+  border-color: var(--color-primary, #6366f1);
+}
+
+/* ── Auto-summary banner + re-consolidate button ─────────────────────────── */
+.auto-summary-banner-editor {
+  display: flex;
+  align-items: center;
+  gap: 0.6rem;
+  padding: 0.45rem 0.7rem;
+  margin-bottom: 0.5rem;
+  font-size: 0.82rem;
+  font-style: italic;
+  color: var(--color-text-muted, rgba(255, 255, 255, 0.6));
+  background: rgba(99, 102, 241, 0.06);
+  border-left: 2px solid var(--color-primary, #6366f1);
+  border-radius: var(--radius-sm, 4px);
+}
+.auto-summary-banner-editor .auto-summary-icon {
+  color: var(--color-primary, #6366f1);
+  font-style: normal;
+}
+.btn-reconsolidate {
+  margin-left: auto;
+  padding: 0.25rem 0.7rem;
+  font-size: 0.78rem;
+  font-style: normal;
+  color: inherit;
+  background: transparent;
+  border: 1px solid var(--color-border, rgba(255, 255, 255, 0.12));
+  border-radius: 999px;
+  cursor: pointer;
+  transition: background 120ms ease;
+}
+.btn-reconsolidate:hover:not(:disabled) {
+  background: rgba(99, 102, 241, 0.12);
+  border-color: var(--color-primary, #6366f1);
+}
+.btn-reconsolidate:disabled {
+  opacity: 0.5;
+  cursor: progress;
+}
 </style>
\ No newline at end of file
diff --git a/frontend/src/views/TaskViewerView.vue b/frontend/src/views/TaskViewerView.vue
index 20439d8..f9065ef 100644
--- a/frontend/src/views/TaskViewerView.vue
+++ b/frontend/src/views/TaskViewerView.vue
@@ -357,6 +357,22 @@ const subTaskProgress = computed(() => {
             @click="onTagClick"
           />
         </div>
+        <div
+          v-if="store.currentTask.description"
+          class="task-goal-display"
+        >
+          <h3 class="goal-label">Goal</h3>
+          <p class="goal-text">{{ store.currentTask.description }}</p>
+        </div>
+
+        <div
+          v-if="store.currentTask.consolidated_at"
+          class="auto-summary-banner"
+        >
+          <span class="auto-summary-icon" aria-hidden="true">✦</span>
+          Auto-summarized from work logs.
+        </div>
+
         <div
           class="body prose"
           v-html="renderedBody"
@@ -807,4 +823,42 @@ const subTaskProgress = computed(() => {
 .skel-line { height: 0.9rem; }
 .skel-line--short { width: 50%; }
 .skel-line--medium { width: 78%; }
+
+/* ── Goal block + auto-summary banner ─────────────────────────────────────── */
+.task-goal-display {
+  border-left: 2px solid var(--color-border, rgba(255, 255, 255, 0.12));
+  padding: 0.4rem 0 0.4rem 0.9rem;
+  margin: 0.75rem 0 1.25rem;
+  background: rgba(255, 255, 255, 0.02);
+}
+.goal-label {
+  font-family: var(--font-display, "Fraunces", serif);
+  font-style: italic;
+  font-size: 0.78rem;
+  font-weight: 500;
+  letter-spacing: 0.04em;
+  text-transform: uppercase;
+  color: var(--color-text-muted, rgba(255, 255, 255, 0.5));
+  margin: 0 0 0.25rem;
+}
+.goal-text {
+  margin: 0;
+  font-size: 0.95rem;
+  line-height: 1.45;
+  color: var(--color-text, inherit);
+  white-space: pre-wrap;
+}
+.auto-summary-banner {
+  display: flex;
+  align-items: center;
+  gap: 0.5rem;
+  font-size: 0.78rem;
+  font-style: italic;
+  color: var(--color-text-muted, rgba(255, 255, 255, 0.55));
+  margin: 0 0 0.75rem;
+}
+.auto-summary-icon {
+  color: var(--color-primary, #6366f1);
+  font-size: 0.85rem;
+}
 </style>
diff --git a/src/fabledassistant/app.py b/src/fabledassistant/app.py
index fce073b..761a4c4 100644
--- a/src/fabledassistant/app.py
+++ b/src/fabledassistant/app.py
@@ -331,6 +331,12 @@ def create_app() -> Quart:
         from fabledassistant.services.event_scheduler import start_event_scheduler
         start_event_scheduler(asyncio.get_running_loop())
 
+        # Start version-pinning scheduler (daily auto-pin scan at 03:00 UTC)
+        from fabledassistant.services.version_pinning_scheduler import (
+            start_version_pinning_scheduler,
+        )
+        start_version_pinning_scheduler(asyncio.get_running_loop())
+
         # Voice model loading (enabled via Admin → Config in the UI, or VOICE_ENABLED env var)
         from fabledassistant.services.stt import load_stt_model
         from fabledassistant.services.tts import load_tts_model
@@ -343,6 +349,10 @@ def create_app() -> Quart:
         stop_journal_scheduler()
         from fabledassistant.services.event_scheduler import stop_event_scheduler
         stop_event_scheduler()
+        from fabledassistant.services.version_pinning_scheduler import (
+            stop_version_pinning_scheduler,
+        )
+        stop_version_pinning_scheduler()
 
     @app.route("/")
     async def serve_index():
diff --git a/src/fabledassistant/models/note.py b/src/fabledassistant/models/note.py
index ca12310..dfce1b3 100644
--- a/src/fabledassistant/models/note.py
+++ b/src/fabledassistant/models/note.py
@@ -32,6 +32,10 @@ class Note(Base, TimestampMixin):
     )
     title: Mapped[str] = mapped_column(Text, default="")
     body: Mapped[str] = mapped_column(Text, default="")
+    description: Mapped[str | None] = mapped_column(Text, nullable=True)
+    consolidated_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
     tags: Mapped[list[str]] = mapped_column(ARRAY(Text), default=list)
     parent_id: Mapped[int | None] = mapped_column(
         Integer, ForeignKey("notes.id", ondelete="SET NULL"), nullable=True
@@ -81,6 +85,10 @@ class Note(Base, TimestampMixin):
             "id": self.id,
             "title": self.title,
             "body": self.body,
+            "description": self.description,
+            "consolidated_at": (
+                self.consolidated_at.isoformat() if self.consolidated_at else None
+            ),
             "tags": self.tags or [],
             "parent_id": self.parent_id,
             "project_id": self.project_id,
diff --git a/src/fabledassistant/models/note_version.py b/src/fabledassistant/models/note_version.py
index 93fc206..cf9a94a 100644
--- a/src/fabledassistant/models/note_version.py
+++ b/src/fabledassistant/models/note_version.py
@@ -14,6 +14,8 @@ class NoteVersion(Base, CreatedAtMixin):
     body: Mapped[str] = mapped_column(Text)
     title: Mapped[str] = mapped_column(Text, default="")
     tags: Mapped[list[str]] = mapped_column(ARRAY(Text), default=list)
+    pin_kind: Mapped[str | None] = mapped_column(Text, nullable=True)
+    pin_label: Mapped[str | None] = mapped_column(Text, nullable=True)
 
     def to_dict(self, include_body: bool = True) -> dict:
         d: dict = {
@@ -22,6 +24,8 @@ class NoteVersion(Base, CreatedAtMixin):
             "user_id": self.user_id,
             "title": self.title,
             "tags": self.tags or [],
+            "pin_kind": self.pin_kind,
+            "pin_label": self.pin_label,
             "created_at": self.created_at.isoformat(),
         }
         if include_body:
diff --git a/src/fabledassistant/routes/notes.py b/src/fabledassistant/routes/notes.py
index 7b7a5b1..d59e218 100644
--- a/src/fabledassistant/routes/notes.py
+++ b/src/fabledassistant/routes/notes.py
@@ -112,6 +112,7 @@ async def create_note_route():
             uid,
             title=data.get("title", ""),
             body=body,
+            description=data.get("description"),
             tags=tags,
             parent_id=data.get("parent_id"),
             project_id=project_id,
@@ -215,7 +216,7 @@ async def update_note_route(note_id: int):
     uid = get_current_user_id()
     data = await request.get_json()
     fields = {}
-    for key in ("title", "body", "parent_id", "project_id", "milestone_id", "status", "priority", "note_type"):
+    for key in ("title", "body", "description", "parent_id", "project_id", "milestone_id", "status", "priority", "note_type"):
         if key in data:
             fields[key] = data[key]
     if "metadata" in data:
@@ -250,7 +251,7 @@ async def patch_note_route(note_id: int):
     uid = get_current_user_id()
     data = await request.get_json()
     fields = {}
-    for key in ("title", "body", "parent_id", "project_id", "milestone_id", "status", "priority", "note_type"):
+    for key in ("title", "body", "description", "parent_id", "project_id", "milestone_id", "status", "priority", "note_type"):
         if key in data:
             fields[key] = data[key]
     if "metadata" in data:
@@ -552,6 +553,39 @@ async def get_version_route(note_id: int, version_id: int):
     return jsonify(version.to_dict(include_body=True))
 
 
+@notes_bp.route("/<int:note_id>/versions/<int:version_id>/pin", methods=["POST"])
+@login_required
+async def pin_version_route(note_id: int, version_id: int):
+    """Mark a version as manually pinned. Body: {"label": str | null}."""
+    uid = get_current_user_id()
+    data = await request.get_json() or {}
+    label = data.get("label")
+    if label is not None and not isinstance(label, str):
+        return jsonify({"error": "label must be a string or null"}), 400
+    from fabledassistant.services.version_pinning import pin_version
+    try:
+        version = await pin_version(uid, note_id, version_id, label=label)
+    except ValueError as e:
+        return jsonify({"error": str(e)}), 400
+    if version is None:
+        return not_found("Version")
+    return jsonify(version.to_dict(include_body=False))
+
+
+@notes_bp.route(
+    "/<int:note_id>/versions/<int:version_id>/pin", methods=["DELETE"],
+)
+@login_required
+async def unpin_version_route(note_id: int, version_id: int):
+    """Downgrade a manually-pinned version back to rolling."""
+    uid = get_current_user_id()
+    from fabledassistant.services.version_pinning import unpin_version
+    version = await unpin_version(uid, note_id, version_id)
+    if version is None:
+        return not_found("Version")
+    return jsonify(version.to_dict(include_body=False))
+
+
 # ── Graph route ────────────────────────────────────────────────────────────────
 
 @notes_bp.route("/graph", methods=["GET"])
diff --git a/src/fabledassistant/routes/tasks.py b/src/fabledassistant/routes/tasks.py
index ae3c974..c716914 100644
--- a/src/fabledassistant/routes/tasks.py
+++ b/src/fabledassistant/routes/tasks.py
@@ -89,7 +89,11 @@ async def list_tasks_route():
 async def create_task_route():
     uid = get_current_user_id()
     data = await request.get_json()
-    body = data.get("body", "") or data.get("description", "")
+    # Description (user-stated goal) and body (machine-maintained summary) are
+    # separate fields under the task-as-durable-record design. Don't fold one
+    # into the other.
+    body = data.get("body", "")
+    description = data.get("description")
     tags = data.get("tags", [])
 
     due_date = parse_iso_date(data.get("due_date"), "due_date")
@@ -120,6 +124,7 @@ async def create_task_route():
         uid,
         title=data.get("title", ""),
         body=body,
+        description=description,
         status=status,
         priority=priority,
         due_date=due_date,
@@ -178,11 +183,12 @@ async def update_task_route(task_id: int):
         except ValueError:
             return jsonify({"error": f"Invalid priority: {data['priority']}"}), 400
 
-    # Accept both "body" and "description" (prefer body)
+    # Body and description are distinct fields under the task-as-durable-record
+    # design. Don't alias one to the other.
     if "body" in data:
         fields["body"] = data["body"]
-    elif "description" in data:
-        fields["body"] = data["description"]
+    if "description" in data:
+        fields["description"] = data["description"]
 
     if "due_date" in data:
         if data["due_date"]:
@@ -276,3 +282,25 @@ async def delete_task_route(task_id: int):
     if not deleted:
         return not_found("Task")
     return "", 204
+
+
+@tasks_bp.route("/<int:task_id>/consolidate", methods=["POST"])
+@login_required
+async def consolidate_task_route(task_id: int):
+    """Manually trigger a consolidation pass for a task.
+
+    Bypasses the auto_consolidate_tasks setting (the user is asking
+    explicitly). Returns the task's updated state including the freshly-
+    written body and consolidated_at timestamp.
+    """
+    uid = get_current_user_id()
+    if not await can_write_note(uid, task_id):
+        return jsonify({"error": "Permission denied"}), 403
+
+    from fabledassistant.services.consolidation import consolidate_task
+    await consolidate_task(uid, task_id)
+
+    note = await get_note(uid, task_id)
+    if note is None:
+        return not_found("Task")
+    return jsonify(note.to_dict())
diff --git a/src/fabledassistant/services/backup.py b/src/fabledassistant/services/backup.py
index 4f9c88b..94ce7c8 100644
--- a/src/fabledassistant/services/backup.py
+++ b/src/fabledassistant/services/backup.py
@@ -150,6 +150,8 @@ async def export_full_backup() -> dict:
                 "title": nv.title,
                 "body": nv.body,
                 "tags": nv.tags or [],
+                "pin_kind": nv.pin_kind,
+                "pin_label": nv.pin_label,
                 "created_at": nv.created_at.isoformat(),
             }
             for nv in note_versions
@@ -314,6 +316,8 @@ async def export_user_backup(user_id: int) -> dict:
                 "title": nv.title,
                 "body": nv.body,
                 "tags": nv.tags or [],
+                "pin_kind": nv.pin_kind,
+                "pin_label": nv.pin_label,
                 "created_at": nv.created_at.isoformat(),
             }
             for nv in note_versions
@@ -605,6 +609,8 @@ async def _restore_v2(data: dict) -> dict:
                 title=nv_data.get("title", ""),
                 body=nv_data.get("body", ""),
                 tags=nv_data.get("tags", []),
+                pin_kind=nv_data.get("pin_kind"),
+                pin_label=nv_data.get("pin_label"),
                 created_at=_dt(nv_data.get("created_at")),
             )
             session.add(nv)
diff --git a/src/fabledassistant/services/consolidation.py b/src/fabledassistant/services/consolidation.py
new file mode 100644
index 0000000..ee77b5e
--- /dev/null
+++ b/src/fabledassistant/services/consolidation.py
@@ -0,0 +1,206 @@
+"""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))
+
+
+def _build_consolidation_prompt(
+    *, title: str, description: str | None, logs: list,
+) -> str:
+    """Build the LLM prompt for one consolidation pass.
+
+    Caps total log content at MAX_PROMPT_INPUT_CHARS; logs that don't fit
+    are dropped. Caller is expected to slice to the most-recent window
+    before calling.
+    """
+    log_lines: list[str] = []
+    chars = 0
+    for log in logs:
+        ts = log.created_at.isoformat() if getattr(log, "created_at", None) else "?"
+        line = f"- [{ts}] {log.content}"
+        if chars + len(line) > MAX_PROMPT_INPUT_CHARS:
+            break
+        log_lines.append(line)
+        chars += len(line)
+
+    return (
+        "You are summarizing the work done on a task. The user wrote the goal "
+        "below; do not restate it. Read the chronological work-log entries and "
+        "produce a 1-3 paragraph summary of: what was attempted, what worked, "
+        "what failed, what's current state. Use the user's voice; cite specific "
+        "commands/decisions; favor brevity over completeness. Output plain "
+        "markdown body content only — no preamble.\n\n"
+        f"TITLE: {title}\n"
+        f"GOAL (read-only context): {description or '(no goal recorded)'}\n"
+        f"WORK LOG (chronological):\n" + "\n".join(log_lines)
+    )
+
+
+async def consolidate_task(user_id: int, task_id: int) -> None:
+    """Run a consolidation pass: read description + logs, write summary to body.
+
+    Errors are logged and swallowed so the fire-and-forget caller is never
+    interrupted; on LLM failure the body and consolidated_at are left
+    untouched and the next trigger retries.
+    """
+    lock = _locks[task_id]
+    if lock.locked():
+        logger.debug(
+            "consolidate_task: skipping — already running for task %d", task_id
+        )
+        return
+    async with lock:
+        try:
+            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 or not task.status:
+                    return  # not a task, or missing
+                logs = (
+                    await session.execute(
+                        select(TaskLog)
+                        .where(
+                            TaskLog.task_id == task_id,
+                            TaskLog.user_id == user_id,
+                        )
+                        .order_by(TaskLog.created_at.asc())
+                    )
+                ).scalars().all()
+
+                if not logs:
+                    return  # nothing to summarize yet
+
+                title = task.title or ""
+                description = task.description
+                window = logs[-MAX_LOGS_FOR_PROMPT:]
+
+            prompt = _build_consolidation_prompt(
+                title=title, description=description, logs=window,
+            )
+
+            from fabledassistant.services.llm import generate_completion
+            from fabledassistant.services.settings import get_setting
+            from fabledassistant.config import Config
+
+            bg_model = await get_setting(
+                user_id, "background_model", Config.OLLAMA_BACKGROUND_MODEL,
+            )
+            summary = await generate_completion(
+                [{"role": "user", "content": prompt}],
+                model=bg_model,
+                max_tokens=800,
+                num_ctx=4096,
+            )
+            if not summary or not summary.strip():
+                return
+
+            now = datetime.now(timezone.utc)
+            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
+                task.body = summary.strip()
+                task.consolidated_at = now
+                task.updated_at = now
+                await session.commit()
+
+            from fabledassistant.services.embeddings import upsert_note_embedding
+            await upsert_note_embedding(
+                task_id, user_id, f"{title}\n{summary.strip()}".strip(),
+            )
+            logger.info(
+                "consolidate_task: refreshed task %d body (%d chars)",
+                task_id, len(summary),
+            )
+        except Exception:
+            logger.exception(
+                "consolidate_task failed for task %d", task_id,
+            )
diff --git a/src/fabledassistant/services/note_versions.py b/src/fabledassistant/services/note_versions.py
index 425c4dc..bef59fb 100644
--- a/src/fabledassistant/services/note_versions.py
+++ b/src/fabledassistant/services/note_versions.py
@@ -47,13 +47,18 @@ async def create_version(
         await session.commit()
         await session.refresh(version)
 
-        # Prune versions beyond MAX_VERSIONS
+        # Prune rolling versions beyond MAX_VERSIONS. Pinned rows
+        # (pin_kind IS NOT NULL) are excluded from both the counted
+        # bucket and the deletion candidate set, so they survive
+        # indefinitely regardless of rolling autosave volume.
         await session.execute(
             text("""
                 DELETE FROM note_versions
                 WHERE id IN (
                     SELECT id FROM note_versions
-                    WHERE note_id = :note_id AND user_id = :user_id
+                    WHERE note_id = :note_id
+                      AND user_id = :user_id
+                      AND pin_kind IS NULL
                     ORDER BY created_at DESC
                     OFFSET :max_versions
                 )
diff --git a/src/fabledassistant/services/notes.py b/src/fabledassistant/services/notes.py
index 0d46d96..4448b47 100644
--- a/src/fabledassistant/services/notes.py
+++ b/src/fabledassistant/services/notes.py
@@ -76,6 +76,7 @@ async def create_note(
     user_id: int,
     title: str = "",
     body: str = "",
+    description: str | None = None,
     tags: list[str] | None = None,
     parent_id: int | None = None,
     project_id: int | None = None,
@@ -103,6 +104,7 @@ async def create_note(
             user_id=user_id,
             title=title,
             body=body,
+            description=description,
             tags=_normalize_tags(tags or []),
             parent_id=parent_id,
             project_id=project_id,
@@ -279,6 +281,8 @@ async def update_note(user_id: int, note_id: int, **fields: object) -> Note | No
         old_body = note.body
         old_title = note.title
         old_tags = list(note.tags or [])
+        # Snapshot status to detect terminal transitions for consolidation trigger.
+        old_status = note.status
         for key, value in fields.items():
             if not hasattr(note, key):
                 continue
@@ -323,6 +327,13 @@ async def update_note(user_id: int, note_id: int, **fields: object) -> Note | No
         from fabledassistant.services.note_versions import create_version
         await create_version(user_id, note_id, old_body, old_title, old_tags)
 
+    # Trigger consolidation when a task transitions into a terminal status.
+    # Captured before mutation; the gate inside maybe_consolidate handles the
+    # auto-consolidate setting.
+    if note.status in ("done", "cancelled") and old_status != note.status:
+        from fabledassistant.services.consolidation import maybe_consolidate
+        await maybe_consolidate(user_id, note.id, reason="task_closed")
+
     if note.project_id is not None:
         await _maybe_reactivate_project(note.project_id)
         await _maybe_trigger_project_summary(user_id, note.project_id)
diff --git a/src/fabledassistant/services/tools/notes.py b/src/fabledassistant/services/tools/notes.py
index b9b3144..131ee7a 100644
--- a/src/fabledassistant/services/tools/notes.py
+++ b/src/fabledassistant/services/tools/notes.py
@@ -23,11 +23,17 @@ logger = logging.getLogger(__name__)
         "Create a new note or task. "
         "For a knowledge note, omit the status field. "
         "For an actionable task (todo, reminder, action item), set status to 'todo'. "
-        "Use this whenever the user asks to write down, save, record, or add a task/todo."
+        "Use this whenever the user asks to write down, save, record, or add a task/todo. "
+        "For standalone reusable knowledge, ALSO use this when the user explicitly asks "
+        "to save something as a note / runbook / how-to, OR when their message contains "
+        "a fenced code block or a numbered procedure (3+ steps) that's reusable beyond "
+        "a single task. For task-specific work-in-progress, use log_work instead — that "
+        "feeds the task's auto-summary."
     ),
     parameters={
         "title": {"type": "string", "description": "The title"},
-        "body": {"type": "string", "description": "Content in markdown"},
+        "body": {"type": "string", "description": "Content in markdown. NOTE: when status is set (creating a task), body is ignored — task bodies are auto-maintained from work logs. Use `description` to provide the goal/context for tasks."},
+        "description": {"type": "string", "description": "User-stated goal or initial context for a task. Read-only context for the auto-summary pipeline. Ignored when status is omitted (knowledge note)."},
         "tags": {"type": "array", "items": {"type": "string"}, "description": 'Tags (without # prefix, hyphens for multi-word: ["science-fiction", "story/idea"]). Do NOT embed #tags in the body.'},
         "project": {"type": "string", "description": "Optional project name. Only set this if the user explicitly named a project. Do NOT infer a project from the content or context."},
         "status": {"type": "string", "enum": ["todo", "in_progress", "done", "cancelled"], "description": "Set to 'todo' to create a task. Omit entirely for a knowledge note."},
@@ -43,6 +49,7 @@ logger = logging.getLogger(__name__)
 async def create_note_tool(*, user_id, arguments, **_ctx):
     title = arguments.get("title", "Untitled")
     body = arguments.get("body", "")
+    description = arguments.get("description")
     tags = arguments.get("tags", [])
     if not isinstance(title, str):
         return {"success": False, "error": "title must be a string. Call create_note once per item."}
@@ -54,6 +61,11 @@ async def create_note_tool(*, user_id, arguments, **_ctx):
     is_task = "status" in arguments and arguments["status"] is not None
     status = arguments.get("status", "todo") if is_task else None
 
+    # Task bodies are auto-maintained by the consolidation pipeline; drop any
+    # body argument arriving with a task creation so it never lands in the DB.
+    if is_task:
+        body = ""
+
     project_name = arguments.get("project")
     milestone_name = arguments.get("milestone")
     parent_task_name = arguments.get("parent_task")
@@ -80,6 +92,7 @@ async def create_note_tool(*, user_id, arguments, **_ctx):
         user_id=user_id,
         title=title,
         body=body,
+        description=description,
         tags=tags,
         status=status,
         priority=arguments.get("priority", "none") if is_task else None,
@@ -126,7 +139,8 @@ async def create_note_tool(*, user_id, arguments, **_ctx):
     description="Update an existing note or task — content, title, status, priority, or due date. Use for edits, marking tasks done, changing priority. Never use create_note for existing notes.",
     parameters={
         "query": {"type": "string", "description": "Title or keyword to find the note or task to update"},
-        "body": {"type": "string", "description": "New note content in markdown (omit if only updating task fields)"},
+        "body": {"type": "string", "description": "New note content in markdown (omit if only updating task fields). REJECTED on tasks — task bodies are auto-maintained from work logs; use `log_work` to record progress or `description` to revise the goal."},
+        "description": {"type": "string", "description": "Update the user-stated goal/context (tasks). Distinct from `body` (machine-maintained on tasks)."},
         "title": {"type": "string", "description": "Optional new title"},
         "mode": {"type": "string", "enum": ["replace", "append"], "description": "How to apply the new body: 'replace' overwrites existing content (default), 'append' adds after existing content"},
         "status": {"type": "string", "enum": ["todo", "in_progress", "done", "cancelled"], "description": "New task status. Use to mark a task done, start it, cancel it, etc."},
@@ -155,6 +169,19 @@ async def update_note_tool(*, user_id, arguments, **_ctx):
             return {"success": False, "error": f"No note found matching '{query}'."}
         note = candidates[0]
 
+    # Schema-level separation: task bodies are auto-maintained from work logs.
+    # Reject body writes here rather than silently dropping so the LLM gets
+    # nudged toward log_work / description.
+    if note.is_task and arguments.get("body"):
+        return {
+            "success": False,
+            "error": (
+                "Cannot write to `body` on a task — the body is auto-maintained "
+                "from work logs. Use the `log_work` tool to record progress, or "
+                "update `description` to revise the goal."
+            ),
+        }
+
     update_fields: dict = {}
     if new_title:
         update_fields["title"] = new_title
@@ -163,6 +190,8 @@ async def update_note_tool(*, user_id, arguments, **_ctx):
             update_fields["body"] = note.body + "\n\n" + new_body
         else:
             update_fields["body"] = new_body
+    if "description" in arguments:
+        update_fields["description"] = arguments["description"]
     if "status" in arguments:
         update_fields["status"] = arguments["status"]
     if "priority" in arguments:
diff --git a/src/fabledassistant/services/tools/tasks.py b/src/fabledassistant/services/tools/tasks.py
index c0a653b..48677ff 100644
--- a/src/fabledassistant/services/tools/tasks.py
+++ b/src/fabledassistant/services/tools/tasks.py
@@ -80,7 +80,14 @@ async def list_tasks_tool(*, user_id, arguments, **_ctx):
 
 @tool(
     name="log_work",
-    description="Add a work log entry to a task to record progress, work done, or time spent. Use this when the user says they worked on, completed, or spent time on a task.",
+    description=(
+        "Add a work log entry to a task to record progress, work done, or time spent. "
+        "Use this when the user says they worked on, completed, or spent time on a task. "
+        "Work logs feed the task's auto-summary: every few entries the task body is "
+        "rewritten from the logs by a background pass. Be specific (commands run, "
+        "decisions made, what failed vs. what worked) — the summary is only as good "
+        "as the logs."
+    ),
     parameters={
         "task": {"type": "string", "description": "Title or keyword identifying the task (required)"},
         "content": {"type": "string", "description": "Description of the work done (required)"},
@@ -113,4 +120,8 @@ async def log_work_tool(*, user_id, arguments, **_ctx):
         log = await _create_log(user_id, note.id, content, duration_minutes)
     except ValueError as e:
         return {"success": False, "error": str(e)}
+
+    from fabledassistant.services.consolidation import maybe_consolidate
+    await maybe_consolidate(user_id, note.id, reason="log_added")
+
     return {"success": True, "log": log.to_dict(), "task": note.title}
diff --git a/src/fabledassistant/services/version_pinning.py b/src/fabledassistant/services/version_pinning.py
new file mode 100644
index 0000000..159c3c7
--- /dev/null
+++ b/src/fabledassistant/services/version_pinning.py
@@ -0,0 +1,243 @@
+"""Pin/unpin and auto-pin operations on NoteVersion.
+
+The autosave-rolling system in `services/note_versions.py` continues to
+own the existing rolling-cap behavior; this module adds the manual-pin
+API and the daily auto-pin scan.
+
+Tiers (pin_kind values):
+  None     → rolling autosave; capped at MAX_VERSIONS, FIFO.
+  "auto"   → system-declared via the stability scan; capped at MAX_AUTO_PINS, FIFO.
+  "manual" → user-declared; unlimited, never pruned.
+
+Design: docs/superpowers/specs/2026-05-13-note-version-pinning-design.md
+"""
+from __future__ import annotations
+
+import datetime
+import logging
+from datetime import timezone
+
+from sqlalchemy import select, text
+
+from fabledassistant.models import async_session
+from fabledassistant.models.note_version import NoteVersion
+
+logger = logging.getLogger(__name__)
+
+MAX_AUTO_PINS = 25
+AUTO_PIN_STABILITY_DAYS = 2
+PIN_LABEL_MAX_LEN = 500
+
+
+async def pin_version(
+    user_id: int, note_id: int, version_id: int, *, label: str | None,
+) -> NoteVersion | None:
+    """Mark a version as manually pinned. Returns the updated row or None
+    if not found (or wrong user/note scope).
+
+    Acceptable on already-pinned rows (manual or auto) — promotes to
+    manual and updates the label. Labels are capped at PIN_LABEL_MAX_LEN
+    chars; longer values raise ValueError.
+    """
+    if label is not None and len(label) > PIN_LABEL_MAX_LEN:
+        raise ValueError(
+            f"pin_label too long ({len(label)} > {PIN_LABEL_MAX_LEN} chars)"
+        )
+    async with async_session() as session:
+        version = (
+            await session.execute(
+                select(NoteVersion).where(
+                    NoteVersion.id == version_id,
+                    NoteVersion.note_id == note_id,
+                    NoteVersion.user_id == user_id,
+                )
+            )
+        ).scalars().first()
+        if version is None:
+            return None
+        version.pin_kind = "manual"
+        version.pin_label = label
+        await session.commit()
+        await session.refresh(version)
+        return version
+
+
+async def unpin_version(
+    user_id: int, note_id: int, version_id: int,
+) -> NoteVersion | None:
+    """Clear pin_kind and pin_label, downgrading the row to rolling.
+
+    Does NOT delete the row. If the row is older than the rolling cap
+    depth, the next create_version call will prune it via the rolling
+    FIFO. Returns the updated row or None if not found.
+    """
+    async with async_session() as session:
+        version = (
+            await session.execute(
+                select(NoteVersion).where(
+                    NoteVersion.id == version_id,
+                    NoteVersion.note_id == note_id,
+                    NoteVersion.user_id == user_id,
+                )
+            )
+        ).scalars().first()
+        if version is None:
+            return None
+        version.pin_kind = None
+        version.pin_label = None
+        await session.commit()
+        await session.refresh(version)
+        return version
+
+
+def _format_auto_pin_label(
+    start: datetime.datetime, end: datetime.datetime | None,
+) -> str:
+    """Auto-generated label describing the stability window.
+
+    end=None → version is the latest with no successor; render as
+    "stable since {start_iso}". Otherwise render as
+    "stable {start_iso} → {end_iso}".
+    """
+    s = start.date().isoformat()
+    if end is None:
+        return f"stable since {s}"
+    return f"stable {s} → {end.date().isoformat()}"
+
+
+def _promote_stable_versions_for_note(versions_chrono: list) -> list:
+    """Mutate the input list: set pin_kind='auto' + auto-label on any
+    unpinned version whose gap to its successor (or to now, for the
+    latest) is >= AUTO_PIN_STABILITY_DAYS.
+
+    Returns the list of versions that were newly pinned (caller uses
+    this for logging / counts).
+    """
+    now = datetime.datetime.now(timezone.utc)
+    newly_pinned: list = []
+    for i, v in enumerate(versions_chrono):
+        if v.pin_kind is not None:
+            continue
+        if i + 1 < len(versions_chrono):
+            next_ts = versions_chrono[i + 1].created_at
+            is_latest = False
+        else:
+            next_ts = now
+            is_latest = True
+        v_ts = v.created_at
+        if v_ts.tzinfo is None:
+            v_ts = v_ts.replace(tzinfo=timezone.utc)
+        if next_ts.tzinfo is None:
+            next_ts = next_ts.replace(tzinfo=timezone.utc)
+        gap_days = (next_ts - v_ts).total_seconds() / 86400
+        if gap_days >= AUTO_PIN_STABILITY_DAYS:
+            v.pin_kind = "auto"
+            v.pin_label = _format_auto_pin_label(
+                v_ts, None if is_latest else next_ts,
+            )
+            newly_pinned.append(v)
+    return newly_pinned
+
+
+async def _list_user_note_ids_with_versions(user_id: int) -> list[int]:
+    """Return note_ids belonging to user_id that have at least one row in
+    note_versions. Notes with no version history are ignored."""
+    async with async_session() as session:
+        rows = (
+            await session.execute(
+                text(
+                    """
+                    SELECT DISTINCT note_id FROM note_versions
+                    WHERE user_id = :user_id
+                    """
+                ).bindparams(user_id=user_id)
+            )
+        ).all()
+    return [r[0] for r in rows]
+
+
+async def _scan_one_note(user_id: int, note_id: int) -> int:
+    """Run the promote+prune flow for one note. Returns count of newly-
+    pinned versions."""
+    async with async_session() as session:
+        versions = (
+            await session.execute(
+                select(NoteVersion)
+                .where(
+                    NoteVersion.note_id == note_id,
+                    NoteVersion.user_id == user_id,
+                )
+                .order_by(NoteVersion.created_at.asc())
+            )
+        ).scalars().all()
+        if not versions:
+            return 0
+        newly_pinned = _promote_stable_versions_for_note(list(versions))
+        if newly_pinned:
+            await session.commit()
+    if newly_pinned:
+        await prune_auto_pins(user_id=user_id, note_id=note_id)
+    return len(newly_pinned)
+
+
+async def scan_user_for_auto_pins(user_id: int) -> int:
+    """Run the scan across every versioned note for one user. Returns
+    total newly-pinned-this-pass."""
+    total = 0
+    note_ids = await _list_user_note_ids_with_versions(user_id)
+    for note_id in note_ids:
+        try:
+            total += await _scan_one_note(user_id, note_id)
+        except Exception:
+            logger.exception(
+                "auto-pin scan failed for user=%d note=%d", user_id, note_id,
+            )
+    return total
+
+
+async def scan_all_users_for_auto_pins() -> dict[int, int]:
+    """Top-level scan entrypoint. Iterates over all users and runs the
+    per-user scan. Returns {user_id: newly_pinned_count}. Per-user errors
+    are caught and logged so one user's failure doesn't stop the scan."""
+    from fabledassistant.models import User
+
+    async with async_session() as session:
+        users = (await session.execute(select(User.id))).scalars().all()
+
+    out: dict[int, int] = {}
+    for uid in users:
+        try:
+            out[uid] = await scan_user_for_auto_pins(uid)
+        except Exception:
+            logger.exception("auto-pin scan failed for user=%d", uid)
+            out[uid] = 0
+    return out
+
+
+async def prune_auto_pins(user_id: int, note_id: int) -> None:
+    """FIFO-prune the auto-pinned bucket for one note past MAX_AUTO_PINS.
+
+    Manual pins and rolling rows are untouched. Called by the scan job
+    after each note's auto-pin promotions finish.
+    """
+    async with async_session() as session:
+        await session.execute(
+            text(
+                """
+                DELETE FROM note_versions
+                WHERE id IN (
+                    SELECT id FROM note_versions
+                    WHERE note_id = :note_id
+                      AND user_id = :user_id
+                      AND pin_kind = 'auto'
+                    ORDER BY created_at DESC
+                    OFFSET :max_auto_pins
+                )
+                """
+            ).bindparams(
+                note_id=note_id,
+                user_id=user_id,
+                max_auto_pins=MAX_AUTO_PINS,
+            )
+        )
+        await session.commit()
diff --git a/src/fabledassistant/services/version_pinning_scheduler.py b/src/fabledassistant/services/version_pinning_scheduler.py
new file mode 100644
index 0000000..1b04fe1
--- /dev/null
+++ b/src/fabledassistant/services/version_pinning_scheduler.py
@@ -0,0 +1,72 @@
+"""Daily APScheduler cron for the auto-pin scan.
+
+Single global job at 03:00 UTC. Runs scan_all_users_for_auto_pins so the
+system promotes stable note versions before they get aged out of the
+rolling cap. Off-hours by design — the scan is cheap but not time-
+critical and doesn't need to interrupt regular activity.
+
+Mirrors the BackgroundScheduler + threadsafe-async-call pattern used by
+journal_scheduler.py.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from apscheduler.schedulers.background import BackgroundScheduler
+from apscheduler.triggers.cron import CronTrigger
+
+from fabledassistant.services.version_pinning import scan_all_users_for_auto_pins
+
+logger = logging.getLogger(__name__)
+
+_scheduler: BackgroundScheduler | None = None
+_loop: asyncio.AbstractEventLoop | None = None
+
+
+def _run_scan_threadsafe() -> None:
+    """APScheduler invokes this from a worker thread; bridge into the
+    asyncio loop so the scan can await its DB operations."""
+    if _loop is None:
+        logger.warning("version_pinning scheduler: no loop registered")
+        return
+
+    async def _runner():
+        try:
+            results = await scan_all_users_for_auto_pins()
+            total = sum(results.values())
+            if total > 0:
+                logger.info(
+                    "auto-pin scan: pinned %d version(s) across %d user(s)",
+                    total, len(results),
+                )
+            else:
+                logger.debug("auto-pin scan: no new pins")
+        except Exception:
+            logger.exception("auto-pin scan run failed")
+
+    asyncio.run_coroutine_threadsafe(_runner(), _loop)
+
+
+def start_version_pinning_scheduler(loop: asyncio.AbstractEventLoop) -> None:
+    global _scheduler, _loop
+    if _scheduler is not None:
+        return
+    _loop = loop
+    _scheduler = BackgroundScheduler()
+    _scheduler.add_job(
+        _run_scan_threadsafe,
+        trigger=CronTrigger(hour=3, minute=0, timezone="UTC"),
+        id="version_pinning_auto_scan",
+        replace_existing=True,
+    )
+    _scheduler.start()
+    logger.info("Version pinning scheduler started (daily 03:00 UTC)")
+
+
+def stop_version_pinning_scheduler() -> None:
+    global _scheduler
+    if _scheduler is not None:
+        _scheduler.shutdown(wait=False)
+        _scheduler = None
+        logger.info("Version pinning scheduler stopped")
diff --git a/tests/test_consolidation.py b/tests/test_consolidation.py
new file mode 100644
index 0000000..fb86782
--- /dev/null
+++ b/tests/test_consolidation.py
@@ -0,0 +1,317 @@
+"""Tests for the task-body consolidation pipeline (gate + full pass).
+
+Design: docs/superpowers/specs/2026-05-13-task-as-durable-record-design.md
+"""
+from datetime import datetime, timezone
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, 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()
+
+
+# ── Prompt builder ───────────────────────────────────────────────────────────
+
+
+def test_build_consolidation_prompt_includes_title_goal_and_logs():
+    from fabledassistant.services.consolidation import _build_consolidation_prompt
+
+    logs = [
+        SimpleNamespace(
+            content="tried REJECT, logs noisy",
+            created_at=datetime(2026, 5, 13, 10, 0, tzinfo=timezone.utc),
+        ),
+        SimpleNamespace(
+            content="switched to DROP",
+            created_at=datetime(2026, 5, 13, 11, 0, tzinfo=timezone.utc),
+        ),
+    ]
+    prompt = _build_consolidation_prompt(
+        title="firewall tuning", description="quiet the logs", logs=logs,
+    )
+    assert "firewall tuning" in prompt
+    assert "quiet the logs" in prompt
+    assert "tried REJECT" in prompt
+    assert "switched to DROP" in prompt
+
+
+def test_build_consolidation_prompt_handles_missing_description():
+    from fabledassistant.services.consolidation import _build_consolidation_prompt
+
+    logs = [
+        SimpleNamespace(
+            content="a", created_at=datetime(2026, 5, 13, tzinfo=timezone.utc),
+        ),
+    ]
+    prompt = _build_consolidation_prompt(title="t", description=None, logs=logs)
+    assert "no goal recorded" in prompt
+
+
+def test_build_consolidation_prompt_truncates_to_char_budget():
+    from fabledassistant.services.consolidation import (
+        _build_consolidation_prompt, MAX_PROMPT_INPUT_CHARS,
+    )
+
+    big = "x" * (MAX_PROMPT_INPUT_CHARS + 1000)
+    logs = [
+        SimpleNamespace(
+            content=big, created_at=datetime(2026, 5, 13, tzinfo=timezone.utc),
+        ),
+        SimpleNamespace(
+            content="should be dropped",
+            created_at=datetime(2026, 5, 13, 1, tzinfo=timezone.utc),
+        ),
+    ]
+    prompt = _build_consolidation_prompt(title="t", description=None, logs=logs)
+    # First log consumes the budget; the second is dropped.
+    assert "should be dropped" not in prompt
+
+
+# ── consolidate_task orchestration (mocked DB + LLM + embedding) ─────────────
+
+
+def _make_mock_session(task, logs):
+    """Return a mock async_session that hands out task and logs to successive
+    .execute() calls. consolidate_task calls execute three times: select task,
+    select logs (same context), then re-select task in the write-back block."""
+    mock_session = AsyncMock()
+
+    task_result = MagicMock()
+    task_result.scalars.return_value.first.return_value = task
+    logs_result = MagicMock()
+    logs_result.scalars.return_value.all.return_value = logs
+
+    mock_session.execute = AsyncMock(
+        side_effect=[task_result, logs_result, task_result],
+    )
+    mock_session.commit = AsyncMock()
+    mock_session.refresh = AsyncMock()
+    mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+    mock_session.__aexit__ = AsyncMock(return_value=False)
+    return mock_session
+
+
+async def test_consolidate_task_writes_body_and_timestamp_and_reembeds():
+    task = MagicMock()
+    task.id = 42
+    task.title = "Cert renewal"
+    task.description = "renew LE before Nov 30"
+    task.status = "in_progress"
+    task.body = "old body"
+    task.consolidated_at = None
+    logs = [
+        SimpleNamespace(
+            content="tried certbot renew",
+            created_at=datetime(2026, 5, 13, tzinfo=timezone.utc),
+        ),
+        SimpleNamespace(
+            content="switched to manual",
+            created_at=datetime(2026, 5, 13, 1, tzinfo=timezone.utc),
+        ),
+    ]
+    fake_summary = (
+        "Started with certbot renew; DNS split-horizon failed. "
+        "Switched to manual flow."
+    )
+
+    # Each consolidate_task call needs its own per-task lock state.
+    # _locks is module-level defaultdict — patch it on the module to avoid
+    # cross-test leakage of the held-lock state.
+    from collections import defaultdict
+    import asyncio as _asyncio
+
+    with patch(
+        "fabledassistant.services.consolidation._locks",
+        new=defaultdict(_asyncio.Lock),
+    ), patch(
+        "fabledassistant.services.consolidation.async_session",
+        return_value=_make_mock_session(task, logs),
+    ), patch(
+        "fabledassistant.services.llm.generate_completion",
+        new=AsyncMock(return_value=fake_summary),
+    ) as mock_llm, patch(
+        "fabledassistant.services.embeddings.upsert_note_embedding",
+        new=AsyncMock(),
+    ) as mock_embed, patch(
+        "fabledassistant.services.settings.get_setting",
+        new=AsyncMock(return_value="gemma3:4b"),
+    ):
+        from fabledassistant.services.consolidation import consolidate_task
+        await consolidate_task(1, 42)
+
+    mock_llm.assert_awaited_once()
+    mock_embed.assert_awaited_once()
+    assert task.body == fake_summary
+    assert task.consolidated_at is not None
+
+
+async def test_consolidate_task_llm_failure_leaves_body_untouched():
+    task = MagicMock()
+    task.id = 42
+    task.title = "X"
+    task.description = "y"
+    task.status = "in_progress"
+    task.body = "old body content"
+    task.consolidated_at = None
+    logs = [
+        SimpleNamespace(
+            content="a", created_at=datetime(2026, 5, 13, tzinfo=timezone.utc),
+        ),
+    ]
+
+    from collections import defaultdict
+    import asyncio as _asyncio
+
+    with patch(
+        "fabledassistant.services.consolidation._locks",
+        new=defaultdict(_asyncio.Lock),
+    ), patch(
+        "fabledassistant.services.consolidation.async_session",
+        return_value=_make_mock_session(task, logs),
+    ), patch(
+        "fabledassistant.services.llm.generate_completion",
+        new=AsyncMock(side_effect=RuntimeError("ollama down")),
+    ), patch(
+        "fabledassistant.services.embeddings.upsert_note_embedding",
+        new=AsyncMock(),
+    ) as mock_embed, patch(
+        "fabledassistant.services.settings.get_setting",
+        new=AsyncMock(return_value="gemma3:4b"),
+    ):
+        from fabledassistant.services.consolidation import consolidate_task
+        await consolidate_task(1, 42)  # must not raise
+
+    assert task.body == "old body content"
+    assert task.consolidated_at is None
+    mock_embed.assert_not_awaited()
+
+
+async def test_consolidate_task_skips_when_no_logs():
+    task = MagicMock()
+    task.id = 42
+    task.title = "X"
+    task.description = "y"
+    task.status = "in_progress"
+    task.body = "old"
+    task.consolidated_at = None
+
+    from collections import defaultdict
+    import asyncio as _asyncio
+
+    with patch(
+        "fabledassistant.services.consolidation._locks",
+        new=defaultdict(_asyncio.Lock),
+    ), patch(
+        "fabledassistant.services.consolidation.async_session",
+        return_value=_make_mock_session(task, []),
+    ), patch(
+        "fabledassistant.services.llm.generate_completion", new=AsyncMock(),
+    ) as mock_llm, patch(
+        "fabledassistant.services.settings.get_setting",
+        new=AsyncMock(return_value="gemma3:4b"),
+    ):
+        from fabledassistant.services.consolidation import consolidate_task
+        await consolidate_task(1, 42)
+
+    mock_llm.assert_not_awaited()
+    assert task.body == "old"
diff --git a/tests/test_notes_consolidation_trigger.py b/tests/test_notes_consolidation_trigger.py
new file mode 100644
index 0000000..cb8f109
--- /dev/null
+++ b/tests/test_notes_consolidation_trigger.py
@@ -0,0 +1,118 @@
+"""Tests for the status-terminal consolidation trigger in update_note."""
+from unittest.mock import AsyncMock, MagicMock, patch
+
+
+def _mock_session_for_update(mock_note):
+    mock_session = AsyncMock()
+    mock_result = MagicMock()
+    mock_result.scalars.return_value.first.return_value = mock_note
+    mock_session.execute = AsyncMock(return_value=mock_result)
+    mock_session.commit = AsyncMock()
+    mock_session.refresh = AsyncMock()
+    mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+    mock_session.__aexit__ = AsyncMock(return_value=False)
+    return mock_session
+
+
+async def test_update_note_to_done_triggers_consolidation():
+    """status: in_progress → done should fire maybe_consolidate(task_closed)."""
+    mock_note = MagicMock()
+    mock_note.id = 42
+    mock_note.status = "in_progress"   # pre-update state
+    mock_note.body = ""
+    mock_note.title = ""
+    mock_note.tags = []
+    mock_note.started_at = None
+    mock_note.completed_at = None
+    mock_note.recurrence_rule = None
+    mock_note.project_id = None
+
+    with patch(
+        "fabledassistant.services.notes.async_session",
+        return_value=_mock_session_for_update(mock_note),
+    ), patch(
+        "fabledassistant.services.consolidation.maybe_consolidate",
+        new=AsyncMock(),
+    ) as mock_mc:
+        from fabledassistant.services.notes import update_note
+        await update_note(1, 42, status="done")
+
+    mock_mc.assert_awaited_once_with(1, 42, reason="task_closed")
+
+
+async def test_update_note_to_cancelled_triggers_consolidation():
+    mock_note = MagicMock()
+    mock_note.id = 42
+    mock_note.status = "in_progress"
+    mock_note.body = ""
+    mock_note.title = ""
+    mock_note.tags = []
+    mock_note.started_at = None
+    mock_note.completed_at = None
+    mock_note.recurrence_rule = None
+    mock_note.project_id = None
+
+    with patch(
+        "fabledassistant.services.notes.async_session",
+        return_value=_mock_session_for_update(mock_note),
+    ), patch(
+        "fabledassistant.services.consolidation.maybe_consolidate",
+        new=AsyncMock(),
+    ) as mock_mc:
+        from fabledassistant.services.notes import update_note
+        await update_note(1, 42, status="cancelled")
+
+    mock_mc.assert_awaited_once_with(1, 42, reason="task_closed")
+
+
+async def test_update_note_to_in_progress_does_not_trigger_consolidation():
+    """Non-terminal status changes don't fire the trigger."""
+    mock_note = MagicMock()
+    mock_note.id = 42
+    mock_note.status = "todo"
+    mock_note.body = ""
+    mock_note.title = ""
+    mock_note.tags = []
+    mock_note.started_at = None
+    mock_note.completed_at = None
+    mock_note.recurrence_rule = None
+    mock_note.project_id = None
+
+    with patch(
+        "fabledassistant.services.notes.async_session",
+        return_value=_mock_session_for_update(mock_note),
+    ), patch(
+        "fabledassistant.services.consolidation.maybe_consolidate",
+        new=AsyncMock(),
+    ) as mock_mc:
+        from fabledassistant.services.notes import update_note
+        await update_note(1, 42, status="in_progress")
+
+    mock_mc.assert_not_awaited()
+
+
+async def test_update_note_already_done_does_not_retrigger():
+    """If the status was already 'done' and update_note(status='done') runs
+    again, no fresh trigger fires — only transitions count."""
+    mock_note = MagicMock()
+    mock_note.id = 42
+    mock_note.status = "done"
+    mock_note.body = ""
+    mock_note.title = ""
+    mock_note.tags = []
+    mock_note.started_at = None
+    mock_note.completed_at = None
+    mock_note.recurrence_rule = None
+    mock_note.project_id = None
+
+    with patch(
+        "fabledassistant.services.notes.async_session",
+        return_value=_mock_session_for_update(mock_note),
+    ), patch(
+        "fabledassistant.services.consolidation.maybe_consolidate",
+        new=AsyncMock(),
+    ) as mock_mc:
+        from fabledassistant.services.notes import update_note
+        await update_note(1, 42, status="done")
+
+    mock_mc.assert_not_awaited()
diff --git a/tests/test_notes_description_field.py b/tests/test_notes_description_field.py
new file mode 100644
index 0000000..2167d83
--- /dev/null
+++ b/tests/test_notes_description_field.py
@@ -0,0 +1,74 @@
+"""Tests for the description-field roundtrip through the notes service.
+
+The description field is the user-stated goal/context on a task; distinct
+from `body` which (post-Task-as-Durable-Record) becomes the LLM-maintained
+consolidation summary.
+"""
+from unittest.mock import AsyncMock, MagicMock, patch
+
+
+def _mock_session_for_update(mock_note):
+    mock_session = AsyncMock()
+    mock_result = MagicMock()
+    mock_result.scalars.return_value.first.return_value = mock_note
+    mock_session.execute = AsyncMock(return_value=mock_result)
+    mock_session.commit = AsyncMock()
+    mock_session.refresh = AsyncMock()
+    mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+    mock_session.__aexit__ = AsyncMock(return_value=False)
+    return mock_session
+
+
+async def test_update_note_persists_description():
+    """update_note(description=...) should assign to note.description.
+
+    update_note already accepts **fields and uses setattr, so this exercises
+    that the new model column is reachable through the existing dynamic-fields
+    path (no service-layer change required beyond the model column).
+    """
+    mock_note = MagicMock()
+    # Pre-existing state — None description, no recurrence, status untouched.
+    mock_note.description = None
+    mock_note.status = None
+    mock_note.recurrence_rule = None
+    mock_note.project_id = None
+
+    with patch(
+        "fabledassistant.services.notes.async_session",
+        return_value=_mock_session_for_update(mock_note),
+    ):
+        from fabledassistant.services.notes import update_note
+        await update_note(1, 1, description="the goal text")
+
+    assert mock_note.description == "the goal text"
+
+
+async def test_create_note_forwards_description_to_model():
+    """create_note(description=...) should pass description into the Note
+    constructor so it gets persisted alongside title/body/etc."""
+    captured: dict = {}
+
+    class FakeNote:
+        def __init__(self, **kw):
+            captured.update(kw)
+            self.id = 1
+            self.project_id = kw.get("project_id")
+            for k, v in kw.items():
+                setattr(self, k, v)
+
+    mock_session = AsyncMock()
+    mock_session.add = MagicMock()
+    mock_session.commit = AsyncMock()
+    mock_session.refresh = AsyncMock()
+    mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+    mock_session.__aexit__ = AsyncMock(return_value=False)
+
+    with patch(
+        "fabledassistant.services.notes.async_session", return_value=mock_session
+    ), patch("fabledassistant.services.notes.Note", FakeNote):
+        from fabledassistant.services.notes import create_note
+        await create_note(
+            user_id=1, title="renew cert", description="the goal text",
+        )
+
+    assert captured.get("description") == "the goal text"
diff --git a/tests/test_tools_notes.py b/tests/test_tools_notes.py
new file mode 100644
index 0000000..5298b09
--- /dev/null
+++ b/tests/test_tools_notes.py
@@ -0,0 +1,200 @@
+"""Tests for the create_note / update_note LLM tools.
+
+Verifies the new task-as-durable-record contract:
+- create_note drops `body` when a task is being created (status set).
+- create_note forwards `description` to the service.
+- update_note rejects `body` writes when the target is a task.
+- update_note accepts `description` updates on tasks.
+"""
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+
+async def test_create_note_tool_ignores_body_when_creating_task():
+    """Status present → is_task=True → body must be dropped before the
+    create_note service call; description must be forwarded."""
+    from fabledassistant.services.tools import notes as notes_tool
+
+    captured: dict = {}
+
+    async def fake_create_note(**kwargs):
+        captured.update(kwargs)
+        return SimpleNamespace(
+            id=1,
+            title=kwargs["title"],
+            status=kwargs.get("status"),
+            priority=kwargs.get("priority"),
+            due_date=None,
+            project_id=None,
+            milestone_id=None,
+            parent_id=None,
+        )
+
+    with patch(
+        "fabledassistant.services.tools.notes.create_note", new=fake_create_note,
+    ), patch(
+        "fabledassistant.services.tools.notes.check_duplicate",
+        new=AsyncMock(return_value=None),
+    ), patch(
+        "fabledassistant.services.tools.notes.suggest_tags",
+        new=AsyncMock(return_value=[]),
+    ), patch(
+        "fabledassistant.services.tools.notes.schedule_embedding",
+    ):
+        await notes_tool.create_note_tool(
+            user_id=1,
+            arguments={
+                "title": "renew cert",
+                "status": "todo",
+                "description": "the goal text",
+                "body": "this should be dropped on tasks",
+            },
+        )
+
+    assert captured.get("description") == "the goal text"
+    # Body dropped because is_task=True. consolidation owns the body field.
+    assert not captured.get("body")
+
+
+async def test_create_note_tool_preserves_body_for_knowledge_notes():
+    """No status → is_task=False → body is preserved as today."""
+    from fabledassistant.services.tools import notes as notes_tool
+
+    captured: dict = {}
+
+    async def fake_create_note(**kwargs):
+        captured.update(kwargs)
+        # Knowledge-note return path reads note.project_id; SimpleNamespace
+        # needs the attribute even when it's None.
+        return SimpleNamespace(
+            id=1, title=kwargs["title"], status=None, project_id=None,
+        )
+
+    with patch(
+        "fabledassistant.services.tools.notes.create_note", new=fake_create_note,
+    ), patch(
+        "fabledassistant.services.tools.notes.check_duplicate",
+        new=AsyncMock(return_value=None),
+    ), patch(
+        "fabledassistant.services.tools.notes.suggest_tags",
+        new=AsyncMock(return_value=[]),
+    ), patch(
+        "fabledassistant.services.tools.notes.schedule_embedding",
+    ):
+        await notes_tool.create_note_tool(
+            user_id=1,
+            arguments={
+                "title": "runbook",
+                "body": "preserved markdown content",
+            },
+        )
+
+    assert captured.get("body") == "preserved markdown content"
+
+
+async def test_update_note_tool_rejects_body_on_tasks():
+    """When the resolved note has is_task=True, providing body must error."""
+    from fabledassistant.services.tools import notes as notes_tool
+
+    # SimpleNamespace can't fake a @property; build is_task as a real attr.
+    fake_task = SimpleNamespace(
+        id=42, title="t", status="in_progress",
+        body="", tags=[], project_id=None, milestone_id=None,
+    )
+    fake_task.is_task = True
+
+    with patch(
+        "fabledassistant.services.tools.notes.get_note_by_title",
+        new=AsyncMock(return_value=fake_task),
+    ), patch(
+        "fabledassistant.services.tools.notes.update_note", new=AsyncMock(),
+    ) as mock_update:
+        result = await notes_tool.update_note_tool(
+            user_id=1,
+            arguments={"query": "t", "body": "trying to overwrite"},
+        )
+
+    assert result["success"] is False
+    err = result.get("error", "").lower()
+    assert "body" in err or "log_work" in err
+    mock_update.assert_not_awaited()
+
+
+async def test_update_note_tool_accepts_description_on_tasks():
+    """description updates flow through to the service even on tasks."""
+    from fabledassistant.services.tools import notes as notes_tool
+
+    fake_task = SimpleNamespace(
+        id=42, title="t", status="in_progress",
+        body="", tags=[], project_id=None, milestone_id=None,
+        description=None,
+    )
+    fake_task.is_task = True
+
+    captured: dict = {}
+
+    async def fake_update_note(uid, nid, **kwargs):
+        captured.update(kwargs)
+        # Apply the description so the post-update inspection makes sense.
+        for k, v in kwargs.items():
+            setattr(fake_task, k, v)
+        return fake_task
+
+    with patch(
+        "fabledassistant.services.tools.notes.get_note_by_title",
+        new=AsyncMock(return_value=fake_task),
+    ), patch(
+        "fabledassistant.services.tools.notes.update_note",
+        new=fake_update_note,
+    ), patch(
+        "fabledassistant.services.tools.notes.suggest_tags",
+        new=AsyncMock(return_value=[]),
+    ), patch(
+        "fabledassistant.services.tools.notes.schedule_embedding",
+    ):
+        result = await notes_tool.update_note_tool(
+            user_id=1,
+            arguments={"query": "t", "description": "updated goal"},
+        )
+
+    assert result["success"] is True
+    assert captured.get("description") == "updated goal"
+
+
+async def test_update_note_tool_accepts_body_on_knowledge_notes():
+    """Body writes are still allowed on non-task notes."""
+    from fabledassistant.services.tools import notes as notes_tool
+
+    fake_note = SimpleNamespace(
+        id=10, title="n", status=None,
+        body="old", tags=[], project_id=None, milestone_id=None,
+    )
+    fake_note.is_task = False
+
+    captured: dict = {}
+
+    async def fake_update_note(uid, nid, **kwargs):
+        captured.update(kwargs)
+        for k, v in kwargs.items():
+            setattr(fake_note, k, v)
+        return fake_note
+
+    with patch(
+        "fabledassistant.services.tools.notes.get_note_by_title",
+        new=AsyncMock(return_value=fake_note),
+    ), patch(
+        "fabledassistant.services.tools.notes.update_note",
+        new=fake_update_note,
+    ), patch(
+        "fabledassistant.services.tools.notes.suggest_tags",
+        new=AsyncMock(return_value=[]),
+    ), patch(
+        "fabledassistant.services.tools.notes.schedule_embedding",
+    ):
+        result = await notes_tool.update_note_tool(
+            user_id=1,
+            arguments={"query": "n", "body": "new body content"},
+        )
+
+    assert result["success"] is True
+    assert captured.get("body") == "new body content"
diff --git a/tests/test_tools_tasks.py b/tests/test_tools_tasks.py
new file mode 100644
index 0000000..8d7d4a0
--- /dev/null
+++ b/tests/test_tools_tasks.py
@@ -0,0 +1,58 @@
+"""Tests for the task tools — log_work in particular wires into the
+consolidation pipeline after every successful log."""
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+
+async def test_log_work_tool_invokes_maybe_consolidate():
+    """log_work must call maybe_consolidate(user_id, task_id, reason='log_added')
+    after a successful task_logs.create_log."""
+    from fabledassistant.services.tools import tasks as tasks_tool
+
+    fake_task = SimpleNamespace(id=42, title="X", status="in_progress")
+    fake_log = SimpleNamespace(
+        to_dict=lambda: {"id": 1, "content": "did stuff"},
+    )
+
+    # get_note_by_title is imported at the top of tools/tasks.py — patch the
+    # consumer module's bound symbol, not the source.
+    with patch(
+        "fabledassistant.services.tools.tasks.get_note_by_title",
+        new=AsyncMock(return_value=fake_task),
+    ), patch(
+        "fabledassistant.services.task_logs.create_log",
+        new=AsyncMock(return_value=fake_log),
+    ), patch(
+        "fabledassistant.services.consolidation.maybe_consolidate",
+        new=AsyncMock(),
+    ) as mock_mc:
+        result = await tasks_tool.log_work_tool(
+            user_id=1, arguments={"task": "X", "content": "did stuff"},
+        )
+
+    assert result["success"] is True
+    mock_mc.assert_awaited_once_with(1, 42, reason="log_added")
+
+
+async def test_log_work_tool_does_not_trigger_on_create_log_failure():
+    """If create_log raises ValueError, maybe_consolidate must not be called."""
+    from fabledassistant.services.tools import tasks as tasks_tool
+
+    fake_task = SimpleNamespace(id=42, title="X", status="in_progress")
+
+    with patch(
+        "fabledassistant.services.tools.tasks.get_note_by_title",
+        new=AsyncMock(return_value=fake_task),
+    ), patch(
+        "fabledassistant.services.task_logs.create_log",
+        new=AsyncMock(side_effect=ValueError("bad input")),
+    ), patch(
+        "fabledassistant.services.consolidation.maybe_consolidate",
+        new=AsyncMock(),
+    ) as mock_mc:
+        result = await tasks_tool.log_work_tool(
+            user_id=1, arguments={"task": "X", "content": "did stuff"},
+        )
+
+    assert result["success"] is False
+    mock_mc.assert_not_awaited()
diff --git a/tests/test_version_pinning_pin_unpin.py b/tests/test_version_pinning_pin_unpin.py
new file mode 100644
index 0000000..d018640
--- /dev/null
+++ b/tests/test_version_pinning_pin_unpin.py
@@ -0,0 +1,128 @@
+"""Tests for manual pin / unpin on note versions."""
+from unittest.mock import AsyncMock, MagicMock, patch
+
+
+def _mock_session_for_version(mock_version):
+    mock_session = AsyncMock()
+    result = MagicMock()
+    result.scalars.return_value.first.return_value = mock_version
+    mock_session.execute = AsyncMock(return_value=result)
+    mock_session.commit = AsyncMock()
+    mock_session.refresh = AsyncMock()
+    mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+    mock_session.__aexit__ = AsyncMock(return_value=False)
+    return mock_session
+
+
+async def test_pin_version_sets_manual_kind_and_label():
+    mock_version = MagicMock()
+    mock_version.pin_kind = None
+    mock_version.pin_label = None
+
+    with patch(
+        "fabledassistant.services.version_pinning.async_session",
+        return_value=_mock_session_for_version(mock_version),
+    ):
+        from fabledassistant.services.version_pinning import pin_version
+        result = await pin_version(
+            user_id=1, note_id=42, version_id=7, label="the runbook circa Q2",
+        )
+
+    assert result is mock_version
+    assert mock_version.pin_kind == "manual"
+    assert mock_version.pin_label == "the runbook circa Q2"
+
+
+async def test_pin_version_accepts_null_label():
+    """label is optional — pinning with no label is allowed."""
+    mock_version = MagicMock()
+    mock_version.pin_kind = None
+    mock_version.pin_label = None
+
+    with patch(
+        "fabledassistant.services.version_pinning.async_session",
+        return_value=_mock_session_for_version(mock_version),
+    ):
+        from fabledassistant.services.version_pinning import pin_version
+        await pin_version(user_id=1, note_id=42, version_id=7, label=None)
+
+    assert mock_version.pin_kind == "manual"
+    assert mock_version.pin_label is None
+
+
+async def test_pin_version_promotes_auto_to_manual_with_label_update():
+    """Pinning an already-auto-pinned version promotes it and updates label."""
+    mock_version = MagicMock()
+    mock_version.pin_kind = "auto"
+    mock_version.pin_label = "stable 2026-04-01 → 2026-04-05"
+
+    with patch(
+        "fabledassistant.services.version_pinning.async_session",
+        return_value=_mock_session_for_version(mock_version),
+    ):
+        from fabledassistant.services.version_pinning import pin_version
+        await pin_version(
+            user_id=1, note_id=42, version_id=7, label="post-network-refresh",
+        )
+
+    assert mock_version.pin_kind == "manual"
+    assert mock_version.pin_label == "post-network-refresh"
+
+
+async def test_pin_version_rejects_overlong_label():
+    """Labels are capped at 500 chars to keep the UI sane."""
+    mock_version = MagicMock()
+    mock_version.pin_kind = None
+
+    # The cap is checked before any DB access, so the session shouldn't
+    # even be entered. We still patch it to avoid accidental DB lookups
+    # if the implementation order changes.
+    with patch(
+        "fabledassistant.services.version_pinning.async_session",
+        return_value=_mock_session_for_version(mock_version),
+    ):
+        from fabledassistant.services.version_pinning import pin_version
+        try:
+            await pin_version(
+                user_id=1, note_id=42, version_id=7, label="x" * 501,
+            )
+            assert False, "expected ValueError"
+        except ValueError as e:
+            assert "500" in str(e) or "long" in str(e).lower()
+
+
+async def test_pin_version_returns_none_when_not_found():
+    mock_session = AsyncMock()
+    result = MagicMock()
+    result.scalars.return_value.first.return_value = None
+    mock_session.execute = AsyncMock(return_value=result)
+    mock_session.commit = AsyncMock()
+    mock_session.refresh = AsyncMock()
+    mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+    mock_session.__aexit__ = AsyncMock(return_value=False)
+
+    with patch(
+        "fabledassistant.services.version_pinning.async_session",
+        return_value=mock_session,
+    ):
+        from fabledassistant.services.version_pinning import pin_version
+        out = await pin_version(user_id=1, note_id=42, version_id=99, label=None)
+
+    assert out is None
+
+
+async def test_unpin_version_clears_kind_and_label():
+    mock_version = MagicMock()
+    mock_version.pin_kind = "manual"
+    mock_version.pin_label = "previous label"
+
+    with patch(
+        "fabledassistant.services.version_pinning.async_session",
+        return_value=_mock_session_for_version(mock_version),
+    ):
+        from fabledassistant.services.version_pinning import unpin_version
+        result = await unpin_version(user_id=1, note_id=42, version_id=7)
+
+    assert result is mock_version
+    assert mock_version.pin_kind is None
+    assert mock_version.pin_label is None
diff --git a/tests/test_version_pinning_prune.py b/tests/test_version_pinning_prune.py
new file mode 100644
index 0000000..c6e1ca5
--- /dev/null
+++ b/tests/test_version_pinning_prune.py
@@ -0,0 +1,83 @@
+"""Tests for the prune logic — both rolling (pin_kind IS NULL) and the
+auto-pin bucket (pin_kind='auto').
+
+Design: docs/superpowers/specs/2026-05-13-note-version-pinning-design.md
+"""
+from unittest.mock import AsyncMock, MagicMock, patch
+
+
+async def test_create_version_prune_sql_filters_to_unpinned():
+    """The DELETE statement issued by create_version's prune step must
+    include `pin_kind IS NULL` in the inner SELECT so pinned versions
+    aren't counted toward MAX_VERSIONS and can't be pruned by it."""
+    mock_session = AsyncMock()
+    select_result = MagicMock()
+    # No prior version → skips the throttle/dedupe early-return paths and
+    # proceeds straight to insert + prune.
+    select_result.scalars.return_value.first.return_value = None
+    mock_session.add = MagicMock()
+    mock_session.commit = AsyncMock()
+    mock_session.refresh = AsyncMock()
+    mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+    mock_session.__aexit__ = AsyncMock(return_value=False)
+
+    captured_sql: list[str] = []
+
+    async def execute_capture(stmt, *args, **kwargs):
+        captured_sql.append(str(stmt))
+        if len(captured_sql) == 1:
+            return select_result
+        return MagicMock()
+
+    mock_session.execute = AsyncMock(side_effect=execute_capture)
+
+    with patch(
+        "fabledassistant.services.note_versions.async_session",
+        return_value=mock_session,
+    ):
+        from fabledassistant.services.note_versions import create_version
+        await create_version(
+            user_id=1, note_id=42, body="content", title="t", tags=["a"],
+        )
+
+    assert any("pin_kind IS NULL" in s for s in captured_sql), (
+        f"Expected prune SQL to filter pin_kind IS NULL; got: {captured_sql!r}"
+    )
+
+
+async def test_prune_auto_pins_filters_to_auto_kind():
+    """prune_auto_pins must filter to pin_kind='auto' so manual pins and
+    rolling rows aren't touched, and must bind MAX_AUTO_PINS as the OFFSET."""
+    mock_session = AsyncMock()
+    mock_session.commit = AsyncMock()
+    mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+    mock_session.__aexit__ = AsyncMock(return_value=False)
+
+    captured_sql: list[str] = []
+    captured_params: list[dict] = []
+
+    async def execute_capture(stmt, *args, **kwargs):
+        captured_sql.append(str(stmt))
+        try:
+            captured_params.append(dict(stmt.compile().params))
+        except Exception:
+            captured_params.append({})
+        return MagicMock()
+
+    mock_session.execute = AsyncMock(side_effect=execute_capture)
+
+    with patch(
+        "fabledassistant.services.version_pinning.async_session",
+        return_value=mock_session,
+    ):
+        from fabledassistant.services.version_pinning import (
+            prune_auto_pins, MAX_AUTO_PINS,
+        )
+        await prune_auto_pins(user_id=1, note_id=42)
+
+    assert any("pin_kind = 'auto'" in s for s in captured_sql), (
+        f"expected auto-bucket filter in SQL; got: {captured_sql!r}"
+    )
+    assert any(
+        p.get("max_auto_pins") == MAX_AUTO_PINS for p in captured_params
+    ), f"bound params: {captured_params!r}"
diff --git a/tests/test_version_pinning_scan.py b/tests/test_version_pinning_scan.py
new file mode 100644
index 0000000..772ec87
--- /dev/null
+++ b/tests/test_version_pinning_scan.py
@@ -0,0 +1,161 @@
+"""Tests for the auto-pin scan algorithm.
+
+Tests the per-note promotion logic directly via a pure helper so the
+algorithm can be driven without standing up a real DB.
+"""
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+
+
+def _v(version_id: int, days_ago: int, pin_kind=None):
+    """Build a SimpleNamespace mimicking a NoteVersion row."""
+    return SimpleNamespace(
+        id=version_id,
+        created_at=datetime.now(timezone.utc) - timedelta(days=days_ago),
+        pin_kind=pin_kind,
+        pin_label=None,
+    )
+
+
+def test_promote_stable_versions_pins_versions_with_2_day_gap():
+    """Versions followed by another version >= 2 days later get promoted."""
+    from fabledassistant.services.version_pinning import (
+        _promote_stable_versions_for_note,
+    )
+    # 3 versions: v1 (10 days ago), v2 (5 days ago), v3 (1 day ago)
+    # Gaps: v1→v2 = 5d, v2→v3 = 4d. Both ≥ 2d → both pinned.
+    # v3 (latest) has gap to now = 1d → NOT pinned (< 2 days).
+    versions = [_v(1, 10), _v(2, 5), _v(3, 1)]
+    pinned = _promote_stable_versions_for_note(versions)
+    assert {v.id for v in pinned} == {1, 2}
+    assert versions[0].pin_kind == "auto"
+    assert versions[1].pin_kind == "auto"
+    assert versions[2].pin_kind is None
+
+
+def test_promote_stable_versions_pins_latest_if_old_enough():
+    """If the latest version is >= 2 days old with no successor, pin it."""
+    from fabledassistant.services.version_pinning import (
+        _promote_stable_versions_for_note,
+    )
+    versions = [_v(1, 10), _v(2, 3)]   # latest is 3 days old
+    pinned = _promote_stable_versions_for_note(versions)
+    assert {v.id for v in pinned} == {1, 2}
+
+
+def test_promote_stable_versions_skips_already_pinned():
+    from fabledassistant.services.version_pinning import (
+        _promote_stable_versions_for_note,
+    )
+    versions = [_v(1, 10, pin_kind="manual"), _v(2, 5)]
+    pinned = _promote_stable_versions_for_note(versions)
+    # v1 already manual — not re-pinned. v2 has 5-day gap to v1, then
+    # latest-with-no-successor gap of 5d to now → pinned auto.
+    assert {v.id for v in pinned} == {2}
+    assert versions[0].pin_kind == "manual"
+    assert versions[1].pin_kind == "auto"
+
+
+def test_promote_stable_versions_skips_active_editing():
+    """Versions a few hours apart with no >2-day gaps → no pins."""
+    base = datetime.now(timezone.utc)
+    v1 = SimpleNamespace(
+        id=1, created_at=base - timedelta(hours=10),
+        pin_kind=None, pin_label=None,
+    )
+    v2 = SimpleNamespace(
+        id=2, created_at=base - timedelta(hours=2),
+        pin_kind=None, pin_label=None,
+    )
+    from fabledassistant.services.version_pinning import (
+        _promote_stable_versions_for_note,
+    )
+    pinned = _promote_stable_versions_for_note([v1, v2])
+    assert pinned == []
+
+
+def test_promote_stable_versions_writes_descriptive_label():
+    """The auto-generated label references stability."""
+    from fabledassistant.services.version_pinning import (
+        _promote_stable_versions_for_note,
+    )
+    versions = [_v(1, 10), _v(2, 5), _v(3, 1)]
+    _promote_stable_versions_for_note(versions)
+    assert versions[0].pin_label is not None
+    assert "stable" in versions[0].pin_label.lower()
+
+
+def test_promote_stable_versions_naive_datetime_is_treated_as_utc():
+    """created_at may come in as a naive datetime from some code paths;
+    the algorithm coerces it to UTC rather than crashing on subtraction."""
+    naive_old = datetime.utcnow() - timedelta(days=10)
+    naive_recent = datetime.utcnow() - timedelta(days=1)
+    versions = [
+        SimpleNamespace(id=1, created_at=naive_old, pin_kind=None, pin_label=None),
+        SimpleNamespace(id=2, created_at=naive_recent, pin_kind=None, pin_label=None),
+    ]
+    from fabledassistant.services.version_pinning import (
+        _promote_stable_versions_for_note,
+    )
+    pinned = _promote_stable_versions_for_note(versions)
+    # v1 → v2 gap is 9 days → v1 pinned. v2 → now is 1 day → not pinned.
+    assert {v.id for v in pinned} == {1}
+
+
+async def test_scan_user_for_auto_pins_iterates_all_notes(monkeypatch):
+    """scan_user_for_auto_pins iterates every note for the user and calls
+    the per-note flow on each, summing returned counts."""
+    from fabledassistant.services import version_pinning
+
+    note_ids = [10, 20, 30]
+    seen: list[int] = []
+
+    async def fake_list_note_ids(uid):
+        return note_ids
+
+    async def fake_one_note(uid, nid):
+        seen.append(nid)
+        # Pretend each note pinned one version.
+        return 1
+
+    monkeypatch.setattr(
+        version_pinning,
+        "_list_user_note_ids_with_versions",
+        fake_list_note_ids,
+    )
+    monkeypatch.setattr(
+        version_pinning, "_scan_one_note", fake_one_note,
+    )
+
+    total = await version_pinning.scan_user_for_auto_pins(user_id=1)
+
+    assert seen == note_ids
+    assert total == 3
+
+
+async def test_scan_user_for_auto_pins_swallows_per_note_errors(monkeypatch):
+    """One bad note doesn't stop the scan; the error is logged and the
+    others continue."""
+    from fabledassistant.services import version_pinning
+
+    async def fake_list_note_ids(uid):
+        return [10, 20, 30]
+
+    async def fake_one_note(uid, nid):
+        if nid == 20:
+            raise RuntimeError("simulated DB error")
+        return 1
+
+    monkeypatch.setattr(
+        version_pinning,
+        "_list_user_note_ids_with_versions",
+        fake_list_note_ids,
+    )
+    monkeypatch.setattr(
+        version_pinning, "_scan_one_note", fake_one_note,
+    )
+
+    total = await version_pinning.scan_user_for_auto_pins(user_id=1)
+
+    # 10 and 30 each contributed 1; 20 raised and contributed 0.
+    assert total == 2