alembic/versions/0076_note_supersessions.py

@@ -0,0 +1,115 @@
+"""note_supersessions; drop the never-written notes.consolidated_at
+
+Revision ID: 0076
+Revises: 0075
+Create Date: 2026-08-07
+
+Step 1 of milestone #278. Structure only — nothing reads or writes the new
+table yet, and nothing behaves differently after this runs.
+
+## What the table is for
+
+Old records outrank newer ones on the same subject, because a similarity score
+cannot tell time. A note that accurately described how something worked in June
+is still accurate ABOUT June; it is just no longer the answer. Nothing recorded
+that, so nothing could act on it.
+
+The claim points FORWARD — the newer record names what it overtakes — because
+the older one cannot know it has been overtaken. Many-to-many and partial: a
+note may supersede parts of several others and be overtaken piecemeal by
+several later ones, which is why this is a table rather than a column. Both
+directions are queried: `superseded_id` answers "has this been overtaken?" at
+ranking time, `superseder_id` answers "what does this replace?" in a record
+view. An array column could serve one and not the other.
+
+CASCADE on both sides is safe because trashing is not a delete: `trash_svc`
+stamps `deleted_at`, so a trashed note keeps its claims and `restore` brings
+them back. The cascade fires only on `purge_trash`, where the row genuinely
+goes — and a claim about a row that no longer exists is not actionable.
+
+## What is being dropped, and why now
+
+`notes.consolidated_at` was written by NOTHING — no service, no route, no tool
+— while being serialised into every note and task payload as `null`. It cost a
+column, a line in every response, and worse: it IMPLIED a capability. A reader
+reasonably concludes notes can be consolidated and this records when.
+
+That reading was reasonable precisely because merge/unmerge exists for snippets
+and not for notes, so the column looked like the notes-side half of that
+feature, modelled and abandoned.
+
+It is dropped rather than repurposed for supersession, and the distinction is
+the point (#2483): consolidation folds several records into one survivor and
+destroys the originals. Merging two snippets is lossless — one helper, several
+call sites. Folding two dev-logs means writing a summary and losing what each
+actually said. Supersession is the opposite act: both records survive, and the
+older one is merely ranked behind. Smuggling one in under a column named for
+the other would have buried that difference in schema.
+
+## Downgrade
+
+Re-adds `consolidated_at` nullable, which is how it lived — so downgrade
+restores the shape, not the (nonexistent) data. Drops the table; any recorded
+supersession claims are lost, which costs ranking its input and nothing else,
+since no note's own content depends on them.
+"""
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0076"
+down_revision = "0075"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "note_supersessions",
+        sa.Column("id", sa.Integer, primary_key=True),
+        sa.Column(
+            "superseder_id",
+            sa.Integer,
+            sa.ForeignKey("notes.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column(
+            "superseded_id",
+            sa.Integer,
+            sa.ForeignKey("notes.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.UniqueConstraint(
+            "superseder_id", "superseded_id", name="uq_note_supersessions_pair"
+        ),
+        # Declaring that a note supersedes ITSELF is meaningless, and under flat
+        # demotion it would demote a record on its own authority. Refused in the
+        # service too, with a message — this is the backstop that holds when
+        # something writes rows directly.
+        sa.CheckConstraint(
+            "superseder_id <> superseded_id", name="ck_note_supersessions_not_self"
+        ),
+    )
+    op.create_index(
+        "ix_note_supersessions_superseder", "note_supersessions", ["superseder_id"]
+    )
+    op.create_index(
+        "ix_note_supersessions_superseded", "note_supersessions", ["superseded_id"]
+    )
+
+    op.drop_column("notes", "consolidated_at")
+
+
+def downgrade() -> None:
+    op.add_column(
+        "notes",
+        sa.Column("consolidated_at", sa.DateTime(timezone=True), nullable=True),
+    )
+    op.drop_index("ix_note_supersessions_superseded", table_name="note_supersessions")
+    op.drop_index("ix_note_supersessions_superseder", table_name="note_supersessions")
+    op.drop_table("note_supersessions")

frontend/src/types/note.ts

@@ -10,7 +10,6 @@ export interface Note {
   title: string;
   body: string;
   description: string | null;
-  consolidated_at: string | null;
   tags: string[];
   parent_id: number | null;
   parent_title?: string | null;

frontend/src/views/TaskEditorView.vue

@@ -41,7 +41,6 @@ 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");
@@ -303,7 +302,6 @@ onMounted(async () => {
       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;
@@ -1063,22 +1061,4 @@ useEditorGuards(dirty, save);
   border-color: var(--color-primary);
 }
 
-/* ── 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);
-  background: rgba(99, 102, 241, 0.06);
-  border-left: 2px solid var(--color-primary);
-  border-radius: var(--radius-sm);
-}
-.auto-summary-banner-editor .auto-summary-icon {
-  color: var(--color-primary);
-  font-style: normal;
-}
 </style>

frontend/src/views/TaskViewerView.vue

@@ -365,13 +365,6 @@ const subTaskProgress = computed(() => {
           <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"
@@ -771,17 +764,4 @@ const subTaskProgress = computed(() => {
   color: var(--color-text);
   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);
-  margin: 0 0 0.75rem;
-}
-.auto-summary-icon {
-  color: var(--color-primary);
-  font-size: 0.85rem;
-}
 </style>

plugin/.claude-plugin/plugin.json

@@ -1,7 +1,7 @@
 {
   "name": "scribe",
   "description": "Scribe system-of-record for Claude Code: MCP tools over your notes/tasks/projects/rules, a session-start push channel that surfaces your always-on rules + active-project context, process-skills (writing-plans, systematic-debugging, verification, brainstorming, reusing-code), and your saved Scribe Processes auto-surfaced as skills (/scribe:sync). Replaces superpowers + file-memory with one app-backed plugin.",
-  "version": "0.1.24",
+  "version": "0.1.25",
   "author": { "name": "Bryan Van Deusen" },
   "mcpServers": {
     "scribe": {

plugin/skills/using-scribe/SKILL.md

@@ -81,6 +81,25 @@ Two constraints on *how* that's achieved:
    (`arose_from_id`) and the subsystem it touches (`system_ids`). Don't bury a
    fix as a work-log line on whatever task happened to be open.
 
+7. **Tag records to Systems.** `enter_project` lists the project's Systems —
+   its named subsystems/areas. When you create or meaningfully update a record,
+   ask which areas it is *about* and pass `system_ids`. The test: would someone
+   investigating that subsystem want this record in the pile
+   `list_system_records` returns? If the area has no System yet, create one
+   (`create_system`: name + a one-paragraph charter) — an area that plainly
+   exists deserves naming the moment two records would share it; don't wait to
+   be asked. A record about no particular area takes none.
+
+8. **State updates in place; chronicles don't.** A dev-log records what
+   *happened* — write it once, never rewrite it. A durable finding (how a
+   subsystem works, a measured number) lives in that System's **reference
+   note** ("«System» — reference"), which you UPDATE as facts change — safe,
+   because every meaningful edit is snapshotted and the version history is the
+   changelog. The dev-log then `[[links]]` the reference note instead of
+   restating state. When a new record outright *corrects* an older one (a
+   re-measurement, a reversed decision), pass the old id in `supersedes` so the
+   stale record is demoted and labelled rather than left competing.
+
 ## Stay inside the active project's scope
 
 Once a project is in scope — you called `enter_project`, or the working repo is

src/scribe/mcp/server.py

@@ -54,10 +54,29 @@ What each part is for, and when to reach for it:
   system as a rulebook — rules are for behaviour, and tokens kept as prose
   cannot be resolved, inherited, rendered to a stylesheet, or checked against
   code.
-- System: a per-project, reusable, self-describing subsystem/area. Associate any
+- System: a per-project, reusable, self-describing subsystem/area — the
-  record (note, task, issue) with it via system_ids so research, build-work, and
+  project's vocabulary for WHERE work happens. enter_project returns the list.
-  fixes for the same area line up, and recurring problem-spots surface. Manage
+  TAG AS YOU WRITE: when you create or meaningfully update a note, task, or
-  with create_system / list_systems / get_system.
+  snippet, ask which of those areas it is about and pass system_ids. The test:
+  would someone investigating that subsystem want this record in the pile
+  list_system_records returns? Cross-cutting records take several; a record
+  about no particular area takes none — don't force it. If the area a record
+  describes has no System yet, CREATE it (create_system: name + a one-paragraph
+  charter) and tag the record — a subsystem that exists in the code deserves a
+  System the moment two records would share it, the same two-or-more test
+  snippets use; don't wait to be asked to name an area that plainly exists.
+  Read a subsystem back with list_system_records, or search(system_id=...) for
+  a ranked cut.
+- Reference note vs dev-log — STATE vs CHRONICLE. A dev-log records what
+  HAPPENED: write it once, never rewrite it. A durable finding — how a
+  subsystem works, a measured number, an architecture fact — belongs in that
+  System's REFERENCE NOTE ("«System name» — reference", tagged to the System),
+  which is UPDATED IN PLACE as the facts change. Updating loses nothing: every
+  meaningful edit is snapshotted (note versions are the changelog). Create the
+  reference note if the System lacks one; update it if it exists; have the
+  dev-log [[link]] it rather than restating state. State smeared across dated
+  logs is unreachable by search — sixteen near-identical dev-logs tie, and no
+  ranking can pick the right one, because no right one exists.
 
 Mechanics:
 - Notes and Tasks share a model; tasks are notes with is_task=True.
@@ -83,6 +102,16 @@ not something you wait to be asked for:
   record (update_note / update_task / add_task_log) rather than duplicating.
   Only pass force=true when it's genuinely a distinct record — a duplicate both
   bloats the store and surfaces as a stale competing copy in later searches.
+- When a note genuinely IS new but overtakes an older one, say so: pass the
+  older note's id in `supersedes` on create_note / update_note. Reach for it on
+  a re-measurement, a decision that reverses an earlier one, a dev-log covering
+  ground a previous one covered. The old note stays readable and keeps its
+  place; it stops competing for the same question and arrives labelled. This is
+  the third answer alongside update-instead and force: not everything that
+  resembles an existing record should be folded into it, and not everything
+  distinct should compete with it forever. If a result carries `superseded_by`,
+  a later note claims to have brought it up to date — read it as what was true
+  when written and open the newer one before acting.
 - Scope to the project in scope. When a project is active (you called
   enter_project), pass its project_id to search / list_tasks / list_notes so
   results stay inside that project. Querying with no project_id pulls in every

src/scribe/mcp/tools/notes.py

@@ -17,6 +17,7 @@ from scribe.mcp._context import current_user_id
 from scribe.services import access as access_svc
 from scribe.services import dedup as dedup_svc
 from scribe.services import notes as notes_svc
+from scribe.services import supersession as supersession_svc
 from scribe.services import systems as systems_svc
 from scribe.services import trash as trash_svc
 from scribe.services.note_usage import record_pulled
@@ -55,12 +56,42 @@ async def list_notes(
     return {"notes": [n.to_dict() for n in rows], "total": total}
 
 
+async def _attach_supersession(uid: int, note_id: int, data: dict) -> None:
+    """Add both directions of the supersession relation to a note payload.
+
+    Both, because they answer different questions and only one of them is
+    obvious. `supersedes` is what the author claimed. `superseded_by` is what a
+    READER needs and what the note itself cannot know — a stale record handed
+    over without that marker gets acted on confidently, which is worse than
+    never surfacing it.
+
+    Omitted entirely when empty, so an ordinary note's payload doesn't grow two
+    permanently-empty lists. A field that always says nothing trains readers to
+    skip fields, which is the lesson `consolidated_at` cost us (#2483).
+    """
+    rel = await supersession_svc.get_relations(uid, note_id)
+    if rel["supersedes"]:
+        data["supersedes"] = rel["supersedes"]
+    if rel["superseded_by"]:
+        data["superseded_by"] = rel["superseded_by"]
+        data["superseded_note"] = (
+            "A later note claims to bring this up to date — see superseded_by. "
+            "Read this as what was true when written, and check the newer one "
+            "before acting on it."
+        )
+
+
 async def get_note(note_id: int) -> dict:
     """Fetch the full content of a single Scribe note by its ID.
 
     Returns id, title, body (markdown), tags, project_id, created_at, updated_at.
     A note another user shared with you also carries `shared`, `owner` and
     `permission` — read it as their suggestion, not as settled practice you set.
+
+    IF THE RESULT CARRIES `superseded_by`, a later note claims to have brought
+    this one up to date. It is still here and still readable — supersession
+    demotes, it never hides — but read it as what was true when written, and
+    open the newer note before acting on it.
     """
     uid = current_user_id()
     loaded = await notes_svc.get_note_for_user(uid, note_id)
@@ -74,6 +105,7 @@ async def get_note(note_id: int) -> dict:
     # snippets would leave those permanently at zero pulls and make them look
     # like dead weight next to snippets that merely had a counter (#2085).
     record_pulled(user_id=uid, note_id=int(note.id), source="mcp_get_note")
+    await _attach_supersession(uid, note_id, out)
     return out
 
 
@@ -83,6 +115,7 @@ async def create_note(
     tags: list[str] | None = None,
     project_id: int = 0,
     system_ids: list[int] | None = None,
+    supersedes: list[int] | None = None,
     force: bool = False,
 ) -> dict:
     """Create a new note in Scribe.
@@ -94,6 +127,15 @@ async def create_note(
         project_id: Associate with a project (use 0 for no project / orphan note).
         system_ids: Ids of the project's Systems to associate this note with
             (e.g. research about a subsystem). See list_systems / create_system.
+        supersedes: Ids of EARLIER notes this one replaces or brings up to date.
+            Reach for it whenever you write something that overtakes what an
+            older note recorded — a re-measurement, a decision that reverses an
+            earlier one, a dev-log covering ground a previous one covered.
+            The older note stays readable and keeps its place in search; it
+            simply stops competing with this one for the same question, and
+            arrives labelled when it does surface. This records a CLAIM, not a
+            verdict: it never says the older note was wrong, only that it is no
+            longer the current answer.
         force: Bypass the near-duplicate gate. By default, if a title- or
             meaning-similar note already exists in the same project, creation is
             BLOCKED and the existing note's id is returned so you update it
@@ -121,11 +163,19 @@ async def create_note(
     )
     if system_ids:
         await systems_svc.set_record_systems(uid, note.id, system_ids)
+    if supersedes:
+        try:
+            await supersession_svc.set_supersedes(uid, note.id, supersedes)
+        except PermissionError as exc:
+            # The note WAS created — surface the real reason rather than a
+            # not-found, and leave the note rather than silently rolling it back.
+            raise ValueError(str(exc)) from exc
     data = note.to_dict()
     if system_ids:
         data["systems"] = [
             s.to_dict() for s in await systems_svc.list_record_systems(uid, note.id)
         ]
+    await _attach_supersession(uid, note.id, data)
     return data
 
 
@@ -136,6 +186,7 @@ async def update_note(
     tags: list[str] | None = None,
     project_id: int = 0,
     system_ids: list[int] | None = None,
+    supersedes: list[int] | None = None,
 ) -> dict:
     """Update an existing Scribe note. Only explicitly provided fields are changed.
 
@@ -147,6 +198,9 @@ async def update_note(
         project_id: New project association. Omit (or pass 0) to leave unchanged.
         system_ids: Replace this note's System associations with these ids
             (set-semantics). None = leave unchanged; [] = clear all.
+        supersedes: Replace the ids of earlier notes this one replaces
+            (set-semantics). None = leave unchanged; [] = clear all. See
+            create_note for when to reach for it.
     """
     uid = current_user_id()
     fields: dict = {}
@@ -163,11 +217,17 @@ async def update_note(
         raise ValueError(f"note {note_id} not found")
     if system_ids is not None:
         await systems_svc.set_record_systems(uid, note_id, system_ids)
+    if supersedes is not None:
+        try:
+            await supersession_svc.set_supersedes(uid, note_id, supersedes)
+        except PermissionError as exc:
+            raise ValueError(str(exc)) from exc
     data = note.to_dict()
     if system_ids is not None:
         data["systems"] = [
             s.to_dict() for s in await systems_svc.list_record_systems(uid, note_id)
         ]
+    await _attach_supersession(uid, note_id, data)
     return data
 
 

src/scribe/mcp/tools/projects.py

@@ -22,6 +22,7 @@ from scribe.services import milestones as milestones_svc
 from scribe.services import notes as notes_svc
 from scribe.services import projects as projects_svc
 from scribe.services import rulebooks as rulebooks_svc
+from scribe.services import systems as systems_svc
 from scribe.services import trash as trash_svc
 
 
@@ -54,7 +55,14 @@ async def enter_project(project_id: int) -> dict:
 
     Returns a dict with keys: project, milestone_summary, applicable_rules,
     project_rules, subscribed_rulebooks, applicable_rules_truncated,
-    open_tasks, recent_notes, design_system.
+    open_tasks, recent_notes, design_system, systems.
+
+    `systems` is the project's vocabulary of named subsystems/areas. It is
+    returned here so you can TAG as you write: when creating or meaningfully
+    updating a record, ask which of these areas it is about and pass their ids
+    as `system_ids`. If the area a record describes is missing from this list,
+    create it with create_system rather than leaving the area unmodelled. Read
+    a subsystem's accumulated records with list_system_records.
 
     `design_system` is null unless the project points at one. When present it
     carries the chain-merged guidance (the house style AND this project's
@@ -81,6 +89,11 @@ async def enter_project(project_id: int) -> dict:
         uid, is_task=False, project_id=project_id,
         sort="updated_at", limit=5,
     )
+    # The tagging vocabulary. Surfaced HERE because an instruction to "tag
+    # records to Systems" is only executable if the list is in front of the
+    # agent when it writes — which it never was, and tagging stopped within
+    # three days of the feature landing (#2546's audit).
+    systems = await systems_svc.list_systems(uid, project_id)
     # A project need not have one, and most installs won't — null is ordinary
     # here, not a missing prerequisite.
     design_system = None
@@ -91,6 +104,15 @@ async def enter_project(project_id: int) -> dict:
 
     return {
         "project": project.to_dict(),
+        # Trimmed to what tagging needs. The full charter is get_system's job —
+        # this list rides along on every session start, so it stays lean.
+        "systems": [
+            {
+                "id": s.id, "name": s.name,
+                "description": (s.description or "").split("\n")[0][:200],
+            }
+            for s in systems
+        ],
         "design_system": design_system,
         "milestone_summary": milestone_summary,
         "applicable_rules": applicable["rules"],

src/scribe/mcp/tools/search.py

@@ -20,6 +20,7 @@ async def search(
     content_type: str = "all",
     limit: int = 10,
     project_id: int = 0,
+    system_id: int = 0,
 ) -> dict:
     """Semantic search over the user's existing notes and tasks — Scribe's recall.
 
@@ -39,6 +40,11 @@ async def search(
             enter_project) — otherwise this searches across ALL projects and
             bleeds unrelated work into the result set. 0 = search everything
             (use only when you genuinely want a cross-project sweep).
+        system_id: Narrow to records tagged to one System (a named
+            subsystem/area — enter_project lists them). Use when investigating
+            a specific subsystem: it cuts the candidates to records someone
+            deliberately filed under that area. 0 = no system filter.
+            list_system_records gives the same slice unranked.
 
     Returns:
         {"results": [{"id", "title", "body", "is_task", "tags", "similarity"}],
@@ -55,6 +61,7 @@ async def search(
     raw = await semantic_search_notes(
         uid, q, limit=limit, is_task=is_task,
         project_id=project_id or None,
+        system_id=system_id or None,
         # An explicit search reaches everything the operator may read, including
         # records shared with them one-to-one.
         scope="read",

src/scribe/mcp/tools/systems.py

@@ -116,7 +116,14 @@ async def update_system(
 async def list_system_records(
     system_id: int, kind: str = "", open_only: bool = False
 ) -> dict:
-    """List records associated with a System.
+    """Everything filed under one System — the way to READ a subsystem.
+
+    Reach for this when investigating a specific area: it returns the notes,
+    tasks, issues and snippets someone deliberately tagged to it — the
+    subsystem's accumulated record, unranked. Start with its reference note if
+    one exists (titled "«System» — reference"); that is the living state, and
+    the rest is history and open work around it. For a ranked cut of the same
+    slice, search(system_id=...) filters semantic search to this association.
 
     Args:
         kind: filter by task_kind — 'issue', 'work', or 'plan'. Omit for all.

src/scribe/models/__init__.py

@@ -33,6 +33,7 @@ from scribe.models.milestone import Milestone  # noqa: E402, F401
 from scribe.models.task_log import TaskLog  # noqa: E402, F401
 from scribe.models.note_draft import NoteDraft  # noqa: E402, F401
 from scribe.models.note_version import NoteVersion  # noqa: E402, F401
+from scribe.models.note_supersession import NoteSupersession  # noqa: E402, F401
 from scribe.models.group import Group, GroupMembership  # noqa: E402, F401
 from scribe.models.share import NoteShare, ProjectShare  # noqa: E402, F401
 from scribe.models.notification import Notification  # noqa: E402, F401

src/scribe/models/note.py

@@ -33,9 +33,6 @@ class Note(Base, TimestampMixin, SoftDeleteMixin):
     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
@@ -101,9 +98,6 @@ class Note(Base, TimestampMixin, SoftDeleteMixin):
             "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,
             "arose_from_id": self.arose_from_id,

src/scribe/models/note_supersession.py

@@ -0,0 +1,70 @@
+from sqlalchemy import ForeignKey, Index, Integer, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from scribe.models import Base
+from scribe.models.base import CreatedAtMixin
+
+
+class NoteSupersession(Base, CreatedAtMixin):
+    """A newer record's claim that it has overtaken an older one.
+
+    WHY THE RELATION POINTS FORWARD
+
+    The note being WRITTEN declares what it supersedes. The old record cannot
+    know it has been overtaken — asking it to record its own obsolescence is
+    asking it to predict the future. So the claim is made by the party that has
+    the knowledge, and the demotion is derived from the far end.
+
+    WHY A TABLE RATHER THAN A COLUMN
+
+    It is genuinely many-to-many and partial: one note may supersede parts of
+    several others, and a note may be overtaken piecemeal by several later ones.
+    Both directions are queried and neither is rare —
+    `superseded_id` answers the ranking question ("has this been overtaken?"),
+    `superseder_id` answers the record view ("what does this replace?"). An
+    array column on `notes` could be indexed for one and not the other.
+
+    WHAT IT MEANS, AND WHAT IT DOES NOT
+
+    A claim, never a proof. Supersession DEMOTES a record in ranked retrieval;
+    it does not assert the older record was wrong, and it never hides it. A note
+    that accurately described how something worked in June is still accurate
+    about June — it is just no longer the answer to "how does this work".
+
+    CASCADE IS SAFE HERE BECAUSE TRASHING IS NOT A DELETE
+
+    `trash_svc` stamps `deleted_at` (an UPDATE), so a trashed note keeps its
+    claims and `restore` brings them back intact. The cascade fires only on
+    `purge_trash`, where the row genuinely goes — and a supersession claim about
+    a row that no longer exists is not a fact anyone can act on.
+    """
+
+    __tablename__ = "note_supersessions"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    # The newer record, making the claim.
+    superseder_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("notes.id", ondelete="CASCADE")
+    )
+    # The older record, demoted by it.
+    superseded_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("notes.id", ondelete="CASCADE")
+    )
+
+    __table_args__ = (
+        UniqueConstraint(
+            "superseder_id", "superseded_id", name="uq_note_supersessions_pair"
+        ),
+        # Both directions indexed — see the class docstring for why neither is
+        # the rare one.
+        Index("ix_note_supersessions_superseder", "superseder_id"),
+        Index("ix_note_supersessions_superseded", "superseded_id"),
+    )
+
+    def to_dict(self) -> dict:
+        return {
+            "id": self.id,
+            "superseder_id": self.superseder_id,
+            "superseded_id": self.superseded_id,
+            "created_at": self.created_at.isoformat() if self.created_at else None,
+        }

src/scribe/routes/notes.py

@@ -22,7 +22,25 @@ from scribe.services.notes import (
     update_note,
 )
 from scribe.services.note_drafts import upsert_draft, get_draft, delete_draft
+from scribe.services import supersession as supersession_svc
 from scribe.services.note_usage import record_pulled
+
+
+async def _attach_supersession(uid: int, note_id: int, data: dict) -> None:
+    """Both directions of the supersession relation on a note payload.
+
+    Mirrors the MCP helper of the same name — the two surfaces must agree about
+    what a note's payload says, or the web UI and the agent would disagree about
+    whether a record is current.
+
+    Omitted when empty: a field that always says nothing trains readers to skip
+    fields, which is what `consolidated_at` cost (#2483).
+    """
+    rel = await supersession_svc.get_relations(uid, note_id)
+    if rel["supersedes"]:
+        data["supersedes"] = rel["supersedes"]
+    if rel["superseded_by"]:
+        data["superseded_by"] = rel["superseded_by"]
 from scribe.services.note_versions import list_versions, get_version
 
 logger = logging.getLogger(__name__)
@@ -112,7 +130,19 @@ async def create_note_route():
         )
     except ValueError as e:
         return jsonify({"error": str(e)}), 400
-    return jsonify(note.to_dict()), 201
+
+    # Same capability as the MCP create path (#33). Without it the web UI would
+    # be the surface on which a supersession claim silently cannot be made.
+    if data.get("supersedes"):
+        try:
+            await supersession_svc.set_supersedes(uid, note.id, data["supersedes"])
+        except PermissionError as exc:
+            # 403, not 400: the request is well-formed and the caller simply
+            # may not write the target. The note itself was created.
+            return jsonify({"error": str(exc), "note": note.to_dict()}), 403
+    out = note.to_dict()
+    await _attach_supersession(uid, note.id, out)
+    return jsonify(out), 201
 
 
 @notes_bp.route("/tags", methods=["GET"])
@@ -186,6 +216,7 @@ async def get_note_route(note_id: int):
     # injected line useful?" is answered by agent pulls alone, and a human
     # clicking a link would inflate exactly the number #1038 and #2085 gate on.
     record_pulled(user_id=uid, note_id=note_id, source="rest_note")
+    await _attach_supersession(uid, note_id, data)
     return jsonify(data)
 
 
@@ -224,7 +255,18 @@ async def update_note_route(note_id: int):
         return jsonify({"error": str(e)}), 400
     if note is None:
         return not_found("Note")
-    return jsonify(note.to_dict())
+    # Set-semantics, matching MCP and the PATCH route: present-and-empty
+    # clears, absent leaves alone. Scoped by the CALLER, not owner_uid — an
+    # editor-share holder may edit this note and must not thereby inherit the
+    # owner's write access to whatever they name as superseded (#47).
+    if "supersedes" in data:
+        try:
+            await supersession_svc.set_supersedes(uid, note_id, data["supersedes"] or [])
+        except PermissionError as exc:
+            return jsonify({"error": str(exc)}), 403
+    out = note.to_dict()
+    await _attach_supersession(uid, note_id, out)
+    return jsonify(out)
 
 
 @notes_bp.route("/<int:note_id>", methods=["PATCH"])

src/scribe/routes/search.py

@@ -34,10 +34,15 @@ async def search_route():
     content_type = request.args.get("content_type", "all")
     limit = min(request.args.get("limit", 10, type=int), 50)
     is_task = _content_type_to_is_task(content_type)
+    # Same association filter the MCP tool takes (#33). The project filter this
+    # route is still missing is #2463's — it carries a default-scope UI decision
+    # this change must not preempt.
+    system_id = request.args.get("system_id", type=int)
 
     t0 = time.perf_counter()
     results = await semantic_search_notes(
         uid, q, limit=limit, is_task=is_task, threshold=_REST_SEARCH_THRESHOLD,
+        system_id=system_id,
         # The user typed this, so it reaches everything they may read.
         scope="read",
     )

src/scribe/services/backup.py

@@ -7,6 +7,7 @@ from scribe.models import async_session
 from scribe.models.milestone import Milestone
 from scribe.models.note import Note
 from scribe.models.note_draft import NoteDraft
+from scribe.models.note_supersession import NoteSupersession
 from scribe.models.note_version import NoteVersion
 from scribe.models.design_system import DesignSystem, DesignToken
 from scribe.models.note_usage import NoteUsageEvent
@@ -32,8 +33,11 @@ logger = logging.getLogger(__name__)
 # when the calendar surface was retired — old v3 events are skipped on restore.
 # v5 (2026-08) added the six tables that had accumulated outside the backup
 # entirely (#2293), and the coverage guard that stops the seventh.
+# v6 (2026-08) added note_supersessions — and the guard did stop the seventh:
+# the table shipped without a backup section and the coverage test failed the
+# build, which is the whole reason that list was written.
 # Bump when the serialized schema changes.
-BACKUP_VERSION = 5
+BACKUP_VERSION = 6
 
 # Every table this backup carries, by its REAL name. Paired with _NOT_INCLUDED
 # below, these two lists must together account for the entire schema — which is
@@ -50,7 +54,7 @@ _BACKED_UP = [
     "project_topic_suppressions",
     # v5 (2026-08): the five-year gap this list was written to stop.
     "systems", "record_systems", "design_systems", "design_tokens",
-    "note_usage_events", "repo_bindings",
+    "note_usage_events", "repo_bindings", "note_supersessions",
 ]
 
 # Tables intentionally NOT in the backup, surfaced in the payload so the gap is
@@ -110,6 +114,17 @@ def _record_system_rows(rows) -> list[dict]:
     return [{"note_id": r.note_id, "system_id": r.system_id} for r in rows]
 
 
+def _note_supersession_rows(rows) -> list[dict]:
+    """Which record has overtaken which. Carried because it is a JUDGEMENT —
+    someone decided this note replaced that one, and nothing in either note's
+    text records the decision. Lose it and the corpus silently reverts to
+    ranking stale material alongside current material."""
+    return [
+        {"superseder_id": r.superseder_id, "superseded_id": r.superseded_id}
+        for r in rows
+    ]
+
+
 def _design_system_rows(rows) -> list[dict]:
     return [
         {
@@ -171,6 +186,9 @@ async def export_full_backup() -> dict:
         settings = (await session.execute(select(Setting))).scalars().all()
         systems = (await session.execute(select(System))).scalars().all()
         record_systems = (await session.execute(select(RecordSystem))).scalars().all()
+        supersessions = (
+            await session.execute(select(NoteSupersession))
+        ).scalars().all()
         # Parent-first, so a restore can resolve parent_id as it goes rather
         # than needing a second pass — the self-FK is the only ordering
         # constraint in this payload.
@@ -354,6 +372,7 @@ async def export_full_backup() -> dict:
         "design_tokens": _design_token_rows(design_tokens),
         "note_usage_events": _usage_event_rows(usage_events),
         "repo_bindings": _repo_binding_rows(repo_bindings),
+        "note_supersessions": _note_supersession_rows(supersessions),
     }
 
 
@@ -396,6 +415,17 @@ async def export_user_backup(user_id: int) -> dict:
         record_systems = (await session.execute(
             select(RecordSystem).where(RecordSystem.system_id.in_(system_ids))
         )).scalars().all() if system_ids else []
+        # BOTH ends must be this user's notes. A claim spanning out to someone
+        # else's record cannot be restored into a single-user import — the far
+        # id would not be in the map — so carrying it would export a row that
+        # silently vanishes on the way back in. Whole-instance backups have no
+        # such problem and take every row.
+        supersessions = (await session.execute(
+            select(NoteSupersession).where(
+                NoteSupersession.superseder_id.in_(note_ids),
+                NoteSupersession.superseded_id.in_(note_ids),
+            )
+        )).scalars().all() if note_ids else []
         design_systems = (await session.execute(
             select(DesignSystem).where(DesignSystem.owner_user_id == user_id)
             .order_by(DesignSystem.parent_id.nullsfirst(), DesignSystem.id)
@@ -597,6 +627,7 @@ async def export_user_backup(user_id: int) -> dict:
         "design_tokens": _design_token_rows(design_tokens),
         "note_usage_events": _usage_event_rows(usage_events),
         "repo_bindings": _repo_binding_rows(repo_bindings),
+        "note_supersessions": _note_supersession_rows(supersessions),
     }
 
 
@@ -700,6 +731,7 @@ async def _restore_v2(data: dict) -> dict:
         "topic_suppressions": 0,
         "systems": 0, "record_systems": 0, "design_systems": 0,
         "design_tokens": 0, "note_usage_events": 0, "repo_bindings": 0,
+        "note_supersessions": 0,
     }
 
     async with async_session() as session:
@@ -990,6 +1022,24 @@ async def _restore_v2(data: dict) -> dict:
             session.add(RecordSystem(note_id=mapped_nid, system_id=mapped_sid))
             stats["record_systems"] += 1
 
+        # 16b. Supersession claims. Guarded by `data.get` like every other
+        # post-v2 section, so a v5 or older payload restores cleanly without it.
+        #
+        # Both ends must map. A claim is about a PAIR — half of one is not a
+        # weaker claim, it is a dangling row pointing at whatever note happens
+        # to hold that id next.
+        for sup in data.get("note_supersessions", []):
+            mapped_new = note_id_map.get(sup.get("superseder_id", 0))
+            mapped_old = note_id_map.get(sup.get("superseded_id", 0))
+            if mapped_new is None or mapped_old is None or mapped_new == mapped_old:
+                continue
+            session.add(
+                NoteSupersession(
+                    superseder_id=mapped_new, superseded_id=mapped_old
+                )
+            )
+            stats["note_supersessions"] += 1
+
         # 17. Design systems. The export orders these parent-first, so a
         # parent's new id is always in the map by the time a child needs it —
         # no second pass, and a child whose parent is missing lands as a root

src/scribe/services/dedup.py

@@ -258,7 +258,13 @@ async def find_duplicate_note(
 
     # --- Signal 3: semantic similarity (only with a substantial body) ---
     if body and len(body.strip()) >= _MIN_BODY_FOR_SEMANTIC:
-        query = f"{title}\n{body}".strip()
+        # Built by the SAME function the corpus was embedded with. This one is
+        # the copy that mattered most and was easiest to miss: it is a QUERY
+        # document, compared against embedded ones. Shaped differently from the
+        # corpus it searches, the gate degrades silently — it still returns
+        # neighbours, just less apt ones, and no signal says the query and the
+        # index stopped agreeing (found by the guard in test_embedding_text).
+        query = embeddings_svc.embedding_text(title, body)
         # Scope the semantic check the same way as the title check: a record in
         # project P compares only to P; a project-less (orphan) record compares
         # only to other orphans (orphan_only), NOT across every project — without
@@ -275,6 +281,12 @@ async def find_duplicate_note(
             # would refuse their write and point them at something they may not
             # be able to edit.
             scope="own",
+            # NOT demoted by supersession (#278). A superseded record is still a
+            # duplicate of what you are about to write — the claim is that it is
+            # no longer CURRENT, not that it is gone. Demoting it here would let
+            # the same note be recorded a second time, and the second copy would
+            # be the one nothing warns about.
+            demote_superseded=False,
         )
         for score, note in hits:
             # semantic_search_notes doesn't filter note_type — enforce it here so

src/scribe/services/embeddings.py

@@ -86,6 +86,94 @@ def _cosine_similarity(a: list[float], b: list[float]) -> float:
     return dot / (mag_a * mag_b)
 
 
+# How much a superseded record is pushed down the ranking (#278).
+#
+# Chosen against a measurement, not by feel. On 2026-08-07 dev-log #2420 sat at
+# 0.6120 on a query made of its own title phrase, 8th, behind #1759 at 0.6506 —
+# a deficit of 0.039 to the top and ~0.014 to its nearest neighbours. A penalty
+# of 0.05 clears that whole band, so demoting a cluster's stale members actually
+# reorders it rather than shuffling within a tie.
+#
+# It is deliberately NOT large. Supersession is a claim about SOME of a record's
+# content, so a superseded note that strongly answers a question nothing else
+# answers should still surface — just behind anything comparable that is
+# current. A penalty big enough to bury it outright would be hiding by another
+# name, which is the thing the operator ruled out.
+_SUPERSESSION_PENALTY = 0.05
+
+# Candidates fetched per requested result when a re-rank follows. Three ranks of
+# headroom is far more than a 0.05 penalty can move anything through in a corpus
+# whose neighbours sit ~0.01-0.02 apart.
+_SUPERSESSION_OVERFETCH = 3
+
+
+async def _apply_supersession_penalty(
+    scored: list[tuple[float, "Note"]], limit: int
+) -> list[tuple[float, "Note"]]:
+    """Push superseded records below their equals, then take the top `limit`.
+
+    The penalty is applied to the RANKING score and the returned score, so
+    downstream gates see the adjusted value — the auto-inject margin band in
+    particular, which exists to stop near-ties dragging in neighbours and would
+    otherwise re-tie exactly what this just separated.
+
+    It is NOT applied to the relevance threshold: the floor decides whether a
+    record is relevant at all, the penalty decides which relevant record comes
+    first. Applying it to the floor would drop a superseded record out of the
+    results entirely — hiding, which is the one thing this must not do.
+
+    Stable within a tie: Python's sort preserves the distance order the database
+    already established, so equal-scoring records keep their original sequence
+    rather than reshuffling per call.
+    """
+    if not scored:
+        return []
+    from scribe.services.supersession import superseded_ids
+
+    try:
+        stale = await superseded_ids([int(note.id) for _score, note in scored])
+    except Exception:
+        # Fail OPEN, and the direction matters: ranking without the penalty is
+        # the behaviour that shipped for months. Returning nothing, or raising,
+        # would turn a supersession-lookup hiccup into a broken search.
+        logger.warning("Supersession lookup failed — ranking unpenalised", exc_info=True)
+        return scored[:limit]
+
+    if not stale:
+        return scored[:limit]
+    adjusted = [
+        (score - _SUPERSESSION_PENALTY if int(note.id) in stale else score, note)
+        for score, note in scored
+    ]
+    adjusted.sort(key=lambda pair: pair[0], reverse=True)
+    return adjusted[:limit]
+
+
+def embedding_text(title: str | None, body: str | None) -> str:
+    """The document a record is embedded AS.
+
+    One definition, deliberately. This was written out three times — the write
+    path (`notes.embed_note`), the recurring-task spawn, and the startup
+    backfill — and identical copies of a formatting rule are three chances to
+    change one and not the others. The spawn path is the dangerous one: a
+    recurring task embedded to a different shape than everything else would be
+    ranked against a corpus it doesn't match, and nothing would report it.
+
+    It is also a PRECONDITION for changing the shape at all (#2486). Measured,
+    a dev-log's vector separates from five unrelated dev-logs by 0.023 while a
+    snippet's separates by 0.153 — the difference being that a snippet states
+    its purpose twice in a short document, so the purpose dominates. Testing an
+    alternative shape against three copies would mean testing a shape that is
+    not the one in production.
+
+    Whether `title\\n{body}` is the RIGHT shape is the open question. That it is
+    one shape is what makes the question answerable.
+    """
+    title = title or ""
+    body = body or ""
+    return f"{title}\n{body}".strip() if body else title
+
+
 async def upsert_note_embedding(note_id: int, user_id: int, text: str) -> None:
     """Generate and persist an embedding for a note. Safe to fire-and-forget."""
     if not text or not text.strip():
@@ -120,6 +208,8 @@ async def semantic_search_notes(
     task_kind: str | Sequence[str] | None = None,
     orphan_only: bool = False,
     scope: str = "own",
+    demote_superseded: bool = True,
+    system_id: int | None = None,
 ) -> list[tuple[float, Note]]:
     """Return up to *limit* (score, note) pairs most relevant to *query*.
 
@@ -151,6 +241,13 @@ async def semantic_search_notes(
     so a similarity floor of *threshold* is a distance ceiling of
     ``1 - threshold`` and similarity is recovered as ``1 - distance``.
 
+    `demote_superseded` applies the supersession penalty (#278): a record a
+    later note claims to have overtaken ranks below its equals. Callers asking
+    "what is the current answer" want it; the near-duplicate gate does NOT, and
+    passes False — a superseded record is still a duplicate of what you are
+    about to write, and demoting it there would let the same note be recorded
+    twice, the second time invisibly.
+
     Returns an empty list if the embedder is unavailable or on any error.
     """
     if not query or not query.strip():
@@ -185,6 +282,19 @@ async def semantic_search_notes(
                 stmt = stmt.where(Note.project_id.is_(None))
             elif project_id is not None:
                 stmt = stmt.where(Note.project_id == project_id)
+            # Narrow to records tagged to one System (subsystem/area). An
+            # association filter, not a ranking signal — membership in the
+            # candidate set, decided before scoring, like project_id above.
+            if system_id is not None:
+                from scribe.models.system import RecordSystem
+                stmt = stmt.where(
+                    select(RecordSystem.id)
+                    .where(
+                        RecordSystem.note_id == Note.id,
+                        RecordSystem.system_id == system_id,
+                    )
+                    .exists()
+                )
             if is_task is True:
                 stmt = stmt.where(Note.status.isnot(None))
             elif is_task is False:
@@ -207,14 +317,38 @@ async def semantic_search_notes(
                 )
             if exclude_ids:
                 stmt = stmt.where(NoteEmbedding.note_id.notin_(exclude_ids))
-            stmt = stmt.where(distance <= max_distance).order_by(distance.asc()).limit(limit)
+            # OVER-FETCH when a re-rank follows, so the demotion can actually
+            # move something. Demoting after a LIMIT k would be theatre: the cut
+            # already happened, so a superseded record pushed down still sits in
+            # the results and the live record that should have replaced it was
+            # never fetched.
+            #
+            # Ordering stays on RAW distance so pgvector's HNSW index still
+            # serves it (migration 0067). Ordering by `distance + penalty`
+            # instead would be exact, and would turn an indexed top-k into a
+            # scan-and-sort of every embedded note.
+            #
+            # The cost of that trade, stated plainly: a live record outside the
+            # over-fetch window cannot be promoted into the results. With a
+            # penalty far smaller than the window's score spread, that case
+            # needs the true answer to be more than _SUPERSESSION_OVERFETCH
+            # ranks down, which no observed query comes close to.
+            fetch = limit * _SUPERSESSION_OVERFETCH if demote_superseded else limit
+            stmt = (
+                stmt.where(distance <= max_distance)
+                .order_by(distance.asc())
+                .limit(fetch)
+            )
             rows = list((await session.execute(stmt)).all())
     except Exception:
         logger.warning("Failed to query note embeddings", exc_info=True)
         return []
 
     # Recover similarity (1 - distance) and preserve the highest-first contract.
-    return [(1.0 - float(dist), note) for note, dist in rows]
+    scored = [(1.0 - float(dist), note) for note, dist in rows]
+    if not demote_superseded:
+        return scored[:limit]
+    return await _apply_supersession_penalty(scored, limit)
 
 
 async def backfill_note_embeddings() -> None:
@@ -248,7 +382,7 @@ async def backfill_note_embeddings() -> None:
     logger.info("Embedding backfill: generating embeddings for %d notes", len(notes_to_embed))
     success = 0
     for note_id, user_id, title, body in notes_to_embed:
-        text = f"{title}\n{body}".strip() if body else (title or "")
+        text = embedding_text(title, body)
         if not text:
             continue
         await upsert_note_embedding(note_id, user_id, text)

src/scribe/services/notes.py

@@ -30,13 +30,13 @@ def embed_note(note) -> None:
     index refresh. No running loop (unit tests, scripts) is an ordinary case,
     not an error.
     """
-    text = f"{note.title}\n{note.body}".strip() if note.body else (note.title or "")
-    if not text:
-        return
     try:
         import asyncio
 
-        from scribe.services.embeddings import upsert_note_embedding
+        from scribe.services.embeddings import embedding_text, upsert_note_embedding
+        text = embedding_text(note.title, note.body)
+        if not text:
+            return
         asyncio.create_task(upsert_note_embedding(note.id, note.user_id, text))
     except RuntimeError:
         pass          # no running loop — a sync caller, not a failure
@@ -373,17 +373,16 @@ async def update_note(user_id: int, note_id: int, **fields: object) -> Note | No
     return note
 
 
-async def delete_note(user_id: int, note_id: int) -> bool:
+# A hard `delete_note(user_id, note_id)` lived here with ZERO callers, and was
-    async with async_session() as session:
+# removed with #278 step 1. It is recorded rather than silently dropped because
-        result = await session.execute(
+# the danger was never that it ran — it is that it was findable by name. Someone
-            select(Note).where(Note.id == note_id, Note.user_id == user_id)
+# wanting to delete a note greps `delete_note`, finds a function in the notes
-        )
+# service with exactly the right signature, and permanently destroys a record
-        note = result.scalars().first()
+# every path downstream expects to be recoverable.
-        if note is None:
+#
-            return False
+# The delete path is `trash_svc.delete`, which soft-deletes an entity AND its
-        await session.delete(note)
+# descendants under one batch_id so `restore(batch)` works. `purge_trash` owns
-        await session.commit()
+# permanent deletion. Both are reachable; neither is spelled `delete_note`.
-        return True
 
 
 async def get_all_tags(user_id: int, q: str | None = None) -> list[str]:

src/scribe/services/plugin_context.py

@@ -31,6 +31,7 @@ from scribe.services import snippets as snippets_svc
 from scribe.services.access import label_shared_items, owner_names_for
 from scribe.services.embeddings import semantic_search_notes
 from scribe.services.note_usage import record_surfaced
+from scribe.services.supersession import superseded_ids
 from scribe.services.retrieval_telemetry import record_retrieval
 from scribe.services.settings import get_setting
 
@@ -425,11 +426,19 @@ async def build_autoinject_hint(
         "`get_note(id)`, or `get_snippet` / `get_process` for those kinds "
         "(titles only; injected once per session):",
     ]
+    # A superseded record is DEMOTED, not removed (#278) — so one can still reach
+    # this menu, and when it does the reader has to be told. An agent handed
+    # stale material with nothing marking it acts on it with full confidence,
+    # which is worse than never having surfaced it. One query for the whole menu.
+    stale = await superseded_ids([int(n.id) for _s, n in kept])
+
     note_ids: list[int] = []
     for score, note in kept:
         note_ids.append(int(note.id))
         title = (note.title or "(untitled)").replace("\n", " ").strip()
         line = f"> - #{note.id} [{_record_kind(note)}] \"{title}\" ({score:.2f})"
+        if int(note.id) in stale:
+            line += " — SUPERSEDED, a later record covers this; check that first"
         if note.user_id != user_id:
             who = owners.get(int(note.user_id)) or "another user"
             line += f" — shared by {who}, treat as a suggestion"

src/scribe/services/recurrence.py

@@ -103,7 +103,7 @@ async def spawn_recurring_tasks() -> int:
 
     Returns the number of tasks spawned.
     """
-    from scribe.services.embeddings import upsert_note_embedding
+    from scribe.services.embeddings import embedding_text, upsert_note_embedding
     from scribe.services.notes import create_note
 
     now = datetime.now(timezone.utc)
@@ -139,7 +139,7 @@ async def spawn_recurring_tasks() -> int:
                 milestone_id=task.milestone_id,
                 recurrence_rule=task.recurrence_rule,
             )
-            text = f"{child.title}\n{child.body}".strip() if child.body else (child.title or "")
+            text = embedding_text(child.title, child.body)
             if text:
                 asyncio.create_task(upsert_note_embedding(child.id, task.user_id, text))
         except Exception:

src/scribe/services/supersession.py

@@ -0,0 +1,183 @@
+"""Which records have been overtaken by which — the claim, not the ranking.
+
+Step 2 of milestone #278. This module only records and reads the relation; the
+demotion that makes it matter lives in the retrieval layer.
+
+WHY THE CLAIM POINTS FORWARD
+
+The note being written declares what it supersedes. The older record cannot
+know it has been overtaken — asking it to record its own obsolescence is asking
+it to predict the future. So the party with the knowledge makes the claim, and
+"has this been superseded?" is derived by looking at the far end.
+
+WHAT IT MEANS
+
+A claim, never a proof. It demotes a record in ranked retrieval; it does not
+assert the older record was wrong and it never hides it. A note that accurately
+described how something worked in June is still accurate about June.
+
+Partial and many-to-many by nature: one note may supersede parts of several
+others, and be overtaken piecemeal by several later ones.
+"""
+from __future__ import annotations
+
+import logging
+
+from sqlalchemy import delete, select
+
+from scribe.models import async_session
+from scribe.models.note import Note
+from scribe.models.note_supersession import NoteSupersession
+from scribe.services import access
+
+logger = logging.getLogger(__name__)
+
+
+async def _closes_a_cycle(session, superseder_id: int, superseded_id: int) -> bool:
+    """True if `superseder -> superseded` would complete a loop.
+
+    Walks the existing graph from `superseded_id` following superseder→superseded
+    edges. If the walk reaches `superseder_id`, the new edge closes a cycle.
+
+    Why refuse rather than tolerate: a cycle claims every member is obsolete, and
+    under FLAT demotion (see the milestone) that demotes all of them equally —
+    so a set of records that supersede each other in a ring would vanish from
+    ranked retrieval together, which is the opposite of the intent. Nothing about
+    the data would say why.
+
+    Iterative with a visited set, not recursion: the graph is user-supplied and
+    a deep chain must not become a stack overflow in a write path.
+    """
+    seen: set[int] = set()
+    frontier = [superseded_id]
+    while frontier:
+        current = frontier.pop()
+        if current == superseder_id:
+            return True
+        if current in seen:
+            continue
+        seen.add(current)
+        rows = (await session.execute(
+            select(NoteSupersession.superseded_id)
+            .where(NoteSupersession.superseder_id == current)
+        )).scalars().all()
+        frontier.extend(int(r) for r in rows)
+    return False
+
+
+async def set_supersedes(
+    user_id: int, note_id: int, superseded_ids: list[int]
+) -> list[int] | None:
+    """Replace what `note_id` claims to supersede (set semantics).
+
+    Returns the resulting list, or None if the caller cannot write the note
+    making the claim.
+
+    WHAT IS SILENTLY DROPPED, and why each is a drop rather than an error:
+      - ids that don't exist or are trashed — the claim has no subject
+      - the note's own id — meaningless, and the DB CHECK would refuse it anyway
+      - an id that would close a cycle — see _closes_a_cycle
+
+    WHAT IS REFUSED OUTRIGHT: a target the caller cannot WRITE. That is not a
+    silent drop, because it is the one case where the caller might reasonably
+    believe they succeeded and be wrong in a way that matters — demoting someone
+    else's record out of their retrieval is damage you cannot see from the
+    outside. Rule #47.
+    """
+    if not await access.can_write_note(user_id, note_id):
+        return None
+
+    async with async_session() as session:
+        wanted: list[int] = []
+        for target in dict.fromkeys(superseded_ids):   # de-dup, keep order
+            target = int(target)
+            if target == note_id:
+                continue
+            note = await session.get(Note, target)
+            if note is None or note.deleted_at is not None:
+                continue
+            if not await access.can_write_note(user_id, target):
+                raise PermissionError(
+                    f"note {target} is not yours to supersede — you need write "
+                    f"access to it, not just read. Superseding demotes a record "
+                    f"in its owner's retrieval too."
+                )
+            if await _closes_a_cycle(session, note_id, target):
+                continue
+            wanted.append(target)
+
+        existing = set((await session.execute(
+            select(NoteSupersession.superseded_id)
+            .where(NoteSupersession.superseder_id == note_id)
+        )).scalars().all())
+        wanted_set = set(wanted)
+
+        to_remove = existing - wanted_set
+        if to_remove:
+            await session.execute(
+                delete(NoteSupersession).where(
+                    NoteSupersession.superseder_id == note_id,
+                    NoteSupersession.superseded_id.in_(to_remove),
+                )
+            )
+        for target in wanted:
+            if target not in existing:
+                session.add(
+                    NoteSupersession(superseder_id=note_id, superseded_id=target)
+                )
+        await session.commit()
+        return wanted
+
+
+async def get_relations(user_id: int, note_id: int) -> dict[str, list[int]]:
+    """Both directions for one note: what it supersedes, and what supersedes it.
+
+    ONE query and ONE ACL check, because this runs on every note read. Asking
+    the two questions separately doubled the round trips on the hottest path in
+    the product to save a two-line partition — the wrong trade, and one I made
+    on the first attempt.
+
+    Returns {"supersedes": [...], "superseded_by": [...]}, both sorted. Empty
+    lists when the caller cannot read the note.
+
+    `superseded_by` is the direction that matters to a READER and the one the
+    note itself cannot know. An agent handed a stale record with no marker acts
+    on it confidently, which is worse than never surfacing it at all.
+    """
+    empty: dict[str, list[int]] = {"supersedes": [], "superseded_by": []}
+    if not await access.can_read_note(user_id, note_id):
+        return empty
+    async with async_session() as session:
+        rows = (await session.execute(
+            select(
+                NoteSupersession.superseder_id, NoteSupersession.superseded_id
+            ).where(
+                (NoteSupersession.superseder_id == note_id)
+                | (NoteSupersession.superseded_id == note_id)
+            )
+        )).all()
+    supersedes = sorted(
+        int(old) for new, old in rows if int(new) == note_id
+    )
+    superseded_by = sorted(
+        int(new) for new, old in rows if int(old) == note_id
+    )
+    return {"supersedes": supersedes, "superseded_by": superseded_by}
+
+
+async def superseded_ids(note_ids: list[int]) -> set[int]:
+    """Of `note_ids`, which have been superseded by anything. One query.
+
+    Deliberately NOT ACL-scoped: this feeds ranking over a candidate set the
+    caller has already been authorised to see, and re-checking per candidate
+    would be a per-result round trip on a hot path. Callers must pass an
+    already-scoped set — which is why this takes ids rather than a user.
+    """
+    if not note_ids:
+        return set()
+    async with async_session() as session:
+        rows = (await session.execute(
+            select(NoteSupersession.superseded_id)
+            .where(NoteSupersession.superseded_id.in_(note_ids))
+        )).scalars().all()
+    return {int(r) for r in rows}

tests/test_embedding_text.py

@@ -0,0 +1,81 @@
+"""The document shape a record is embedded as, and the guard that keeps it one.
+
+WHY THIS EXISTS
+
+`f"{title}\\n{body}"` was written out three times — the write path, the
+recurring-task spawn, and the startup backfill. Identical copies of a formatting
+rule are three chances to change one and not the others, and the spawn path is
+the dangerous one: a recurring task embedded to a different shape than the rest
+of the corpus is ranked against documents it doesn't match, and nothing reports
+it. A wrong vector returns results; it just returns the wrong ones.
+
+It is also the precondition for #2486. A dev-log's vector separates from five
+unrelated dev-logs by 0.023 where a snippet separates by 0.153, and the leading
+explanation is shape — a snippet states its purpose twice in a short document.
+Testing an alternative against three copies would mean testing a shape that
+isn't the one in production.
+"""
+from __future__ import annotations
+
+import ast
+import pathlib
+
+from scribe.services.embeddings import embedding_text
+
+SERVICES = pathlib.Path(__file__).resolve().parents[1] / "src" / "scribe"
+
+
+def test_title_and_body_are_joined_by_a_newline():
+    assert embedding_text("A title", "A body") == "A title\nA body"
+
+
+def test_a_bodyless_record_embeds_as_its_title_alone():
+    """Not "title\\n" — the trailing separator would be a token's worth of noise
+    on the shortest documents, which are the ones least able to spare it."""
+    assert embedding_text("Just a title", "") == "Just a title"
+    assert embedding_text("Just a title", None) == "Just a title"
+
+
+def test_an_empty_record_yields_an_empty_string():
+    """Callers gate on falsiness to skip embedding entirely, so this must be
+    empty rather than a stray newline."""
+    assert embedding_text("", "") == ""
+    assert embedding_text(None, None) == ""
+
+
+def test_surrounding_whitespace_is_stripped():
+    assert embedding_text("  A title  ", "  A body  ") == "A title  \n  A body"
+
+
+def test_nothing_else_builds_the_embedding_document_itself():
+    """The guard. A fourth copy is how the first three happened.
+
+    Source inspection, because this is the shape no behavioural test catches:
+    an inlined copy produces the same string today and diverges silently the day
+    the shape changes. Matches the f-string pattern itself rather than a
+    variable name, so a copy that renames its locals is still caught.
+    """
+    offenders = []
+    for path in SERVICES.rglob("*.py"):
+        source = path.read_text()
+        for node in ast.walk(ast.parse(source)):
+            if not isinstance(node, ast.JoinedStr):
+                continue
+            # An f-string whose literal parts are exactly a newline, with a
+            # substitution either side: `f"{x}\n{y}"`.
+            literals = [
+                v.value for v in node.values
+                if isinstance(v, ast.Constant) and isinstance(v.value, str)
+            ]
+            subs = [v for v in node.values if isinstance(v, ast.FormattedValue)]
+            if literals == ["\n"] and len(subs) == 2:
+                offenders.append(f"{path.relative_to(SERVICES)}:{node.lineno}")
+
+    # embeddings.py holds the one definition.
+    offenders = [o for o in offenders if not o.startswith("services/embeddings.py")]
+    assert not offenders, (
+        f"these build the embedding document inline instead of calling "
+        f"embedding_text(): {offenders}. One definition — an inlined copy is "
+        f"ranked against a corpus it no longer matches the moment the shape "
+        f"changes, and nothing reports it (#2486)."
+    )

tests/test_mcp_tool_notes.py

@@ -17,6 +17,21 @@ def _bind_user():
     _user_id_ctx.reset(token)
 
 
+@pytest.fixture(autouse=True)
+def _no_supersession():
+    """Every note read/write now asks for its supersession relations (#278).
+
+    These are unit tests of the TOOL layer and this job has no database — the
+    same hazard the `_fake_note` comment below records for note 2109. Stubbed
+    to "no relations", which is the state of essentially every note; the
+    relation's own behaviour is covered in test_services_supersession.py, and
+    the attachment is covered explicitly below.
+    """
+    with patch("scribe.mcp.tools.notes.supersession_svc.get_relations",
+               AsyncMock(return_value={"supersedes": [], "superseded_by": []})):
+        yield
+
+
 def _fake_note(*, user_id: int = 7, **overrides) -> MagicMock:
     note = MagicMock()
     base = {"id": 1, "title": "t", "body": "b", "tags": [], "is_task": False}
@@ -123,6 +138,30 @@ async def test_get_note_returns_dict():
     assert out["title"] == "found"
     # Own record: no provenance noise.
     assert "shared" not in out
+    # No supersession relations: both keys ABSENT, not present-and-empty. A
+    # field that always says nothing trains readers to skip fields (#2483).
+    assert "supersedes" not in out
+    assert "superseded_by" not in out
+
+
+@pytest.mark.asyncio
+async def test_get_note_warns_in_words_when_a_later_note_overtook_it():
+    """The label is the point, not the ids.
+
+    A superseded record still surfaces — supersession demotes, it never hides —
+    so an agent WILL read stale material. Handing it over with only a numeric
+    field to notice would be worse than not surfacing it, because the reader
+    acts on it confidently either way.
+    """
+    fake = _fake_note(id=5, title="June's answer")
+    with patch("scribe.mcp.tools.notes.notes_svc.get_note_for_user",
+               AsyncMock(return_value=(fake, "owner"))), \
+         patch("scribe.mcp.tools.notes.supersession_svc.get_relations",
+               AsyncMock(return_value={"supersedes": [], "superseded_by": [9]})):
+        out = await get_note(note_id=5)
+    assert out["superseded_by"] == [9]
+    assert "superseded_note" in out
+    assert "before acting" in out["superseded_note"]
 
 
 @pytest.mark.asyncio

tests/test_mcp_tool_projects.py

@@ -17,6 +17,18 @@ def _bind_user():
     _user_id_ctx.reset(token)
 
 
+@pytest.fixture(autouse=True)
+def _no_systems():
+    """enter_project now surfaces the project's Systems as the tagging
+    vocabulary (#2546). These are tool-layer unit tests with no database, so
+    the lookup is stubbed to the common case — a project with none. The
+    populated shape is asserted in its own test below.
+    """
+    with patch("scribe.mcp.tools.projects.systems_svc.list_systems",
+               AsyncMock(return_value=[])):
+        yield
+
+
 def _fake_project(design_system_id=None, **overrides) -> MagicMock:
     p = MagicMock()
     base = {"id": 1, "title": "P", "description": "", "goal": "",
@@ -184,6 +196,48 @@ async def test_enter_project_composes_full_context():
     # absent. A caller that has to distinguish "no key" from "no system" will
     # eventually get it wrong.
     assert out["design_system"] is None
+    # No Systems -> present-and-empty, NOT absent: this key is the tagging
+    # vocabulary, and "this project has no named areas yet" is information the
+    # create-the-System instruction acts on.
+    assert out["systems"] == []
+
+
+@pytest.mark.asyncio
+async def test_enter_project_surfaces_the_systems_vocabulary():
+    """The tagging instruction is only executable if the vocabulary is in
+    front of the agent when it writes. It never was, and tagging stopped three
+    days after the feature landed — one System, nothing tagged since July 28
+    (#2546's audit). Trimmed to id/name/first-line: it rides on every session
+    start, and the full charter is get_system's job."""
+    p = _fake_project(id=5)
+    sys1 = MagicMock()
+    sys1.id = 3
+    sys1.name = "retrieval"
+    sys1.description = "Embeddings, ranking, auto-inject.\nLong detail below."
+
+    with patch(
+        "scribe.mcp.tools.projects.projects_svc.get_project",
+        AsyncMock(return_value=p),
+    ), patch(
+        "scribe.mcp.tools.projects.rulebooks_svc.get_applicable_rules",
+        AsyncMock(return_value={"rules": [], "truncated": False,
+                                "subscribed_rulebooks": []}),
+    ), patch(
+        "scribe.mcp.tools.projects.milestones_svc.get_project_milestone_summary",
+        AsyncMock(return_value=[]),
+    ), patch(
+        "scribe.mcp.tools.projects.notes_svc.list_notes",
+        AsyncMock(side_effect=[([], 0), ([], 0)]),
+    ), patch(
+        "scribe.mcp.tools.projects.systems_svc.list_systems",
+        AsyncMock(return_value=[sys1]),
+    ):
+        out = await enter_project(project_id=5)
+
+    assert out["systems"] == [
+        {"id": 3, "name": "retrieval",
+         "description": "Embeddings, ranking, auto-inject."}
+    ]
 
 
 @pytest.mark.asyncio

tests/test_note_usage.py

@@ -9,6 +9,22 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
+
+@pytest.fixture(autouse=True)
+def _no_supersession():
+    """The auto-inject menu now asks which of its lines are superseded (#278).
+
+    That is a real database call on a path these tests exercise without one.
+    Stubbed to "nothing superseded" — the ordinary state — rather than hidden
+    behind a try/except in the product, which would make the code lie about
+    what it does. The label's own behaviour is covered in
+    tests/test_supersession_ranking.py.
+    """
+    with patch("scribe.services.plugin_context.superseded_ids",
+               AsyncMock(return_value=set())):
+        yield
+
+
 from scribe.services import note_usage
 from scribe.services.note_usage import (
     empty_usage,

tests/test_retrieval_scopes.py

@@ -17,6 +17,22 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 
 
+@pytest.fixture(autouse=True)
+def _no_supersession():
+    """The auto-inject menu now asks which of its lines are superseded (#278).
+
+    That is a real database call on a path these tests exercise without one.
+    Stubbed to "nothing superseded" — the ordinary state — rather than hidden
+    behind a try/except in the product, which would make the code lie about
+    what it does. The label's own behaviour is covered in
+    tests/test_supersession_ranking.py.
+    """
+    with patch("scribe.services.plugin_context.superseded_ids",
+               AsyncMock(return_value=set())):
+        yield
+
+
+
 def _note(id=1, user_id=7, title="A note", note_type="note",
           is_task=False, task_kind="work"):
     n = MagicMock()

tests/test_services_backup.py

@@ -13,8 +13,11 @@ import pytest
 from scribe.services import backup
 
 
-def test_backup_version_is_v5():
+def test_backup_version_is_v6():
-    assert backup.BACKUP_VERSION == 5
+    """v6 added note_supersessions (#278). The bump is the point of the test —
+    a payload section added without moving the version produces backups that
+    are structurally different and indistinguishable by inspection."""
+    assert backup.BACKUP_VERSION == 6
 
 
 def test_not_included_lists_the_known_gaps():
@@ -102,11 +105,26 @@ async def test_export_full_backup_contains_every_declared_section():
     assert out["version"] == backup.BACKUP_VERSION
     assert out["scope"] == "full"
     assert "api_keys" in out["_not_included"]
-    # The sections v2 silently dropped, plus the six v5 added (empty here).
+    # The sections v2 silently dropped, the six v5 added, and v6's
+    # note_supersessions (all empty here).
     for key in ("rulebooks", "rulebook_topics", "rules",
                 "rulebook_subscriptions", "rule_suppressions",
                 "topic_suppressions",
                 "systems", "record_systems", "design_systems",
-                "design_tokens", "note_usage_events", "repo_bindings"):
+                "design_tokens", "note_usage_events", "repo_bindings",
+                "note_supersessions"):
         assert key in out, f"missing export section: {key}"
         assert out[key] == []
+
+
+def test_supersession_rows_serialise_the_pair():
+    """The row builder is a plain function precisely so it can be tested with
+    no database — same reason as the other v5/v6 builders."""
+    class _Row:
+        def __init__(self, a, b):
+            self.superseder_id, self.superseded_id = a, b
+
+    assert backup._note_supersession_rows([_Row(9, 4), _Row(9, 5)]) == [
+        {"superseder_id": 9, "superseded_id": 4},
+        {"superseder_id": 9, "superseded_id": 5},
+    ]

tests/test_services_plugin_context.py

@@ -3,6 +3,22 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 
 
+@pytest.fixture(autouse=True)
+def _no_supersession():
+    """The auto-inject menu now asks which of its lines are superseded (#278).
+
+    That is a real database call on a path these tests exercise without one.
+    Stubbed to "nothing superseded" — the ordinary state — rather than hidden
+    behind a try/except in the product, which would make the code lie about
+    what it does. The label's own behaviour is covered in
+    tests/test_supersession_ranking.py.
+    """
+    with patch("scribe.services.plugin_context.superseded_ids",
+               AsyncMock(return_value=set())):
+        yield
+
+
+
 def _rule(rid, title, topic_id):
     r = MagicMock()
     r.id, r.title, r.topic_id = rid, title, topic_id

tests/test_services_supersession.py

@@ -0,0 +1,175 @@
+"""The supersession claim — who may make it, and what it refuses.
+
+Step 2 of #278. Ranking behaviour is step 3; this covers only recording and
+reading the relation.
+
+The cycle tests are the ones worth reading. Under FLAT demotion a ring of
+records that supersede each other claims every member is obsolete, so all of
+them get demoted equally and the whole set drops out of ranked retrieval
+together — with nothing in the data saying why.
+"""
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from scribe.services import supersession
+
+
+def _session(scalars_sequence=None, get_returns=None):
+    """A mocked async_session whose execute() yields successive scalar lists."""
+    s = AsyncMock()
+    s.__aenter__ = AsyncMock(return_value=s)
+    s.__aexit__ = AsyncMock(return_value=False)
+
+    results = []
+    for scalars in scalars_sequence or []:
+        r = MagicMock()
+        r.scalars.return_value.all.return_value = scalars
+        results.append(r)
+    s.execute = AsyncMock(side_effect=results or None)
+    s.get = AsyncMock(side_effect=get_returns) if get_returns else AsyncMock()
+    s.commit = AsyncMock()
+    s.add = MagicMock()
+    return s
+
+
+def _live_note(note_id=1):
+    n = MagicMock()
+    n.id, n.deleted_at = note_id, None
+    return n
+
+
+@pytest.mark.asyncio
+async def test_a_caller_who_cannot_write_the_note_gets_none():
+    with patch("scribe.services.supersession.access.can_write_note",
+               AsyncMock(return_value=False)):
+        assert await supersession.set_supersedes(7, 1, [2]) is None
+
+
+@pytest.mark.asyncio
+async def test_superseding_a_note_you_can_only_READ_is_refused_not_dropped():
+    """The one case that raises rather than silently skipping.
+
+    Demoting someone else's record out of their retrieval is damage that is
+    invisible from the outside — the caller would believe it worked, and the
+    owner would have no symptom to trace. Rule #47.
+    """
+    # writable for the superseder (id 1), not for the target (id 2)
+    writable = AsyncMock(side_effect=lambda uid, nid: nid == 1)
+    session = _session(get_returns=[_live_note(2)])
+    with patch("scribe.services.supersession.access.can_write_note", writable), \
+         patch("scribe.services.supersession.async_session", return_value=session):
+        with pytest.raises(PermissionError, match="not yours to supersede"):
+            await supersession.set_supersedes(7, 1, [2])
+
+
+@pytest.mark.asyncio
+async def test_self_supersession_is_dropped_silently():
+    """Meaningless rather than dangerous, and the DB CHECK refuses it anyway —
+    so it is a drop, not an error the caller has to handle."""
+    session = _session(scalars_sequence=[[]])
+    with patch("scribe.services.supersession.access.can_write_note",
+               AsyncMock(return_value=True)), \
+         patch("scribe.services.supersession.async_session", return_value=session):
+        assert await supersession.set_supersedes(7, 5, [5]) == []
+    session.add.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_a_trashed_target_is_dropped():
+    """A claim needs a subject. `get` returns the row, `deleted_at` says it is
+    in the trash, so there is nothing to demote."""
+    trashed = MagicMock()
+    trashed.deleted_at = "2026-08-08"
+    session = _session(scalars_sequence=[[]], get_returns=[trashed])
+    with patch("scribe.services.supersession.access.can_write_note",
+               AsyncMock(return_value=True)), \
+         patch("scribe.services.supersession.async_session", return_value=session):
+        assert await supersession.set_supersedes(7, 1, [2]) == []
+
+
+@pytest.mark.asyncio
+async def test_a_direct_cycle_is_refused():
+    """B already supersedes A; A may not now supersede B.
+
+    Walk from the proposed target (B) and see whether it reaches the proposer
+    (A). It does — B -> A — so the edge would close a ring.
+    """
+    session = _session(scalars_sequence=[[1]])   # B supersedes A(=1)
+    with patch("scribe.services.supersession.async_session", return_value=session):
+        assert await supersession._closes_a_cycle(session, 1, 2) is True
+
+
+@pytest.mark.asyncio
+async def test_an_indirect_cycle_is_refused():
+    """A -> B -> C exists; C may not supersede A.
+
+    Walking from A follows A -> B, then B -> C... and the walk must reach the
+    proposer. Here the proposer is C and the target is A, so: A -> B -> C.
+    """
+    session = _session(scalars_sequence=[[2], [3]])   # A->B, B->C
+    with patch("scribe.services.supersession.async_session", return_value=session):
+        assert await supersession._closes_a_cycle(session, 3, 1) is True
+
+
+@pytest.mark.asyncio
+async def test_a_chain_that_does_not_loop_is_allowed():
+    """A -> B exists; C may supersede A. Walking from A reaches only B."""
+    session = _session(scalars_sequence=[[2], []])
+    with patch("scribe.services.supersession.async_session", return_value=session):
+        assert await supersession._closes_a_cycle(session, 3, 1) is False
+
+
+@pytest.mark.asyncio
+async def test_the_cycle_walk_terminates_on_an_existing_ring():
+    """Defensive: if a ring somehow exists (written directly to the DB), the
+    walk must not spin. The visited set is what guarantees it, and this pins
+    that guarantee rather than trusting it."""
+    # 1 -> 2, 2 -> 1: a ring that does not contain the proposer (99).
+    session = _session(scalars_sequence=[[2], [1], []])
+    with patch("scribe.services.supersession.async_session", return_value=session):
+        assert await supersession._closes_a_cycle(session, 99, 1) is False
+
+
+@pytest.mark.asyncio
+async def test_superseded_ids_is_empty_for_an_empty_candidate_set():
+    """Ranking calls this per query. An empty candidate set must not become a
+    `WHERE id IN ()`, which Postgres accepts and every reader misreads."""
+    assert await supersession.superseded_ids([]) == set()
+
+
+@pytest.mark.asyncio
+async def test_reads_are_empty_when_the_caller_cannot_read_the_note():
+    with patch("scribe.services.supersession.access.can_read_note",
+               AsyncMock(return_value=False)):
+        assert await supersession.get_relations(7, 1) == {
+            "supersedes": [], "superseded_by": []
+        }
+
+
+@pytest.mark.asyncio
+async def test_get_relations_partitions_both_directions_from_one_query():
+    """ONE round trip for both directions, because this runs on every note read.
+
+    Note 5 supersedes 2 and 3, and is itself superseded by 9. All four rows come
+    back from a single OR query and are partitioned by which column holds 5.
+    """
+    session = AsyncMock()
+    session.__aenter__ = AsyncMock(return_value=session)
+    session.__aexit__ = AsyncMock(return_value=False)
+    result = MagicMock()
+    result.all.return_value = [(5, 3), (5, 2), (9, 5)]   # (superseder, superseded)
+    session.execute = AsyncMock(return_value=result)
+
+    with patch("scribe.services.supersession.access.can_read_note",
+               AsyncMock(return_value=True)), \
+         patch("scribe.services.supersession.async_session", return_value=session):
+        rel = await supersession.get_relations(7, 5)
+
+    assert rel == {"supersedes": [2, 3], "superseded_by": [9]}
+    assert session.execute.await_count == 1, (
+        "both directions must come from one query — asking separately doubles "
+        "the round trips on the hottest path in the product"
+    )

tests/test_supersession_ranking.py

@@ -0,0 +1,154 @@
+"""Demotion: superseded records rank behind their equals, and are never hidden.
+
+Step 3 of #278. The distinction these tests exist to protect is DEMOTE vs
+FILTER. The operator was explicit:
+
+    "the failure is pollution, not existence"
+
+Hiding a superseded record would turn every one of them into something you must
+already know exists in order to find, and would destroy "what did we think
+then" — half the reason a log is kept. So every test here that could be
+satisfied by dropping a record instead checks that it is still present.
+"""
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from scribe.services.embeddings import (
+    _SUPERSESSION_PENALTY,
+    _apply_supersession_penalty,
+)
+
+
+def _note(note_id: int):
+    n = MagicMock()
+    n.id = note_id
+    return n
+
+
+def _stale(*ids):
+    return patch(
+        "scribe.services.supersession.superseded_ids",
+        AsyncMock(return_value=set(ids)),
+    )
+
+
+@pytest.mark.asyncio
+async def test_a_superseded_record_falls_behind_an_equal_live_one():
+    scored = [(0.70, _note(1)), (0.69, _note(2))]      # 1 leads on raw score
+    with _stale(1):
+        out = await _apply_supersession_penalty(scored, limit=5)
+    assert [int(n.id) for _s, n in out] == [2, 1]
+
+
+@pytest.mark.asyncio
+async def test_a_strong_superseded_record_still_beats_a_weak_live_one():
+    """Demote, not filter — and this is why the penalty is small.
+
+    Supersession is a claim about SOME of a record's content. One that strongly
+    answers a question nothing else answers should still surface, just behind
+    anything comparable that is current.
+    """
+    scored = [(0.90, _note(1)), (0.50, _note(2))]
+    with _stale(1):
+        out = await _apply_supersession_penalty(scored, limit=5)
+    assert [int(n.id) for _s, n in out] == [1, 2]
+    assert out[0][0] == pytest.approx(0.90 - _SUPERSESSION_PENALTY)
+
+
+@pytest.mark.asyncio
+async def test_the_superseded_record_is_still_returned():
+    """The whole point. A test that only checked ordering would pass just as
+    happily against an implementation that dropped it."""
+    scored = [(0.70, _note(1))]
+    with _stale(1):
+        out = await _apply_supersession_penalty(scored, limit=5)
+    assert [int(n.id) for _s, n in out] == [1]
+
+
+@pytest.mark.asyncio
+async def test_the_returned_score_is_the_adjusted_one():
+    """Downstream gates must see the adjusted value — the auto-inject margin
+    band in particular, which exists to stop near-ties dragging in neighbours
+    and would otherwise re-tie exactly what this just separated."""
+    scored = [(0.70, _note(1)), (0.68, _note(2))]
+    with _stale(1):
+        out = await _apply_supersession_penalty(scored, limit=5)
+    by_id = {int(n.id): s for s, n in out}
+    assert by_id[1] == pytest.approx(0.65)
+    assert by_id[2] == pytest.approx(0.68)
+
+
+@pytest.mark.asyncio
+async def test_nothing_superseded_leaves_the_order_untouched():
+    scored = [(0.70, _note(1)), (0.69, _note(2)), (0.60, _note(3))]
+    with _stale():
+        out = await _apply_supersession_penalty(scored, limit=5)
+    assert [int(n.id) for _s, n in out] == [1, 2, 3]
+
+
+@pytest.mark.asyncio
+async def test_ties_keep_their_database_order():
+    """Stable sort. Equal scores must not reshuffle per call — a menu that
+    reorders between identical queries reads as nondeterminism and sends
+    someone hunting for a bug that isn't there."""
+    scored = [(0.70, _note(1)), (0.70, _note(2)), (0.70, _note(3))]
+    with _stale():
+        out = await _apply_supersession_penalty(scored, limit=5)
+    assert [int(n.id) for _s, n in out] == [1, 2, 3]
+
+
+@pytest.mark.asyncio
+async def test_the_limit_is_applied_after_reordering():
+    """Over-fetching is pointless if the cut happens first. Three candidates,
+    limit 2, and the demoted leader must be the one that falls out."""
+    scored = [(0.70, _note(1)), (0.69, _note(2)), (0.68, _note(3))]
+    with _stale(1):
+        out = await _apply_supersession_penalty(scored, limit=2)
+    assert [int(n.id) for _s, n in out] == [2, 3]
+
+
+@pytest.mark.asyncio
+async def test_a_failed_lookup_returns_unpenalised_results_not_none():
+    """Fail OPEN, and the direction matters. Ranking without the penalty is the
+    behaviour that shipped for months; returning nothing would turn a
+    supersession hiccup into a broken search."""
+    scored = [(0.70, _note(1)), (0.69, _note(2))]
+    with patch("scribe.services.supersession.superseded_ids",
+               AsyncMock(side_effect=RuntimeError("db gone"))):
+        out = await _apply_supersession_penalty(scored, limit=5)
+    assert [int(n.id) for _s, n in out] == [1, 2]
+    assert out[0][0] == pytest.approx(0.70)
+
+
+@pytest.mark.asyncio
+async def test_an_empty_candidate_set_short_circuits():
+    """No candidates means no lookup — this runs on every ranked query, and a
+    round trip to learn nothing is a round trip too many."""
+    called = AsyncMock(return_value=set())
+    with patch("scribe.services.supersession.superseded_ids", called):
+        assert await _apply_supersession_penalty([], limit=5) == []
+    called.assert_not_awaited()
+
+
+def test_the_penalty_is_sized_to_reorder_a_cluster_not_shuffle_within_it():
+    """Sized against a measurement, not by feel.
+
+    Measured 2026-08-07 (#2486): the notes competing on a dev-log's own title
+    phrase sat within ~0.014 of each other, spanning 0.6506 down to 0.6120. A
+    penalty smaller than that spread would move a record within a tie without
+    changing which one wins — the failure #2486 already proved cannot be tuned
+    away, because the neighbours are not barely passing, they are tied.
+
+    The upper bound is the operator's constraint, not an optimisation: a penalty
+    large enough to bury a superseded record outright is hiding by another name.
+    """
+    assert _SUPERSESSION_PENALTY > 0.014, (
+        "must exceed the measured neighbour spread, or it reorders nothing"
+    )
+    assert _SUPERSESSION_PENALTY < 0.15, (
+        "must not bury a superseded record outright — that is hiding, which "
+        "the operator ruled out"
+    )