src/scribe/mcp/server.py

@@ -83,6 +83,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,43 @@ 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).
+    """
+    supersedes = await supersession_svc.get_supersedes(uid, note_id)
+    superseded_by = await supersession_svc.get_superseded_by(uid, note_id)
+    if supersedes:
+        data["supersedes"] = supersedes
+    if superseded_by:
+        data["superseded_by"] = 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 +106,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 +116,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 +128,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 +164,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 +187,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 +199,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 +218,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/routes/notes.py

@@ -22,7 +22,26 @@ 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).
+    """
+    supersedes = await supersession_svc.get_supersedes(uid, note_id)
+    superseded_by = await supersession_svc.get_superseded_by(uid, note_id)
+    if supersedes:
+        data["supersedes"] = supersedes
+    if superseded_by:
+        data["superseded_by"] = superseded_by
 from scribe.services.note_versions import list_versions, get_version
 
 logger = logging.getLogger(__name__)
@@ -112,7 +131,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 +217,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 +256,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/services/supersession.py

@@ -0,0 +1,176 @@
+"""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_supersedes(user_id: int, note_id: int) -> list[int]:
+    """Ids this note claims to supersede. Empty if the caller can't read it."""
+    if not await access.can_read_note(user_id, note_id):
+        return []
+    async with async_session() as session:
+        return [int(i) for i in (await session.execute(
+            select(NoteSupersession.superseded_id)
+            .where(NoteSupersession.superseder_id == note_id)
+            .order_by(NoteSupersession.superseded_id)
+        )).scalars().all()]
+
+
+async def get_superseded_by(user_id: int, note_id: int) -> list[int]:
+    """Ids claiming to supersede this note.
+
+    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;
+    that is worse than never surfacing it at all.
+    """
+    if not await access.can_read_note(user_id, note_id):
+        return []
+    async with async_session() as session:
+        return [int(i) for i in (await session.execute(
+            select(NoteSupersession.superseder_id)
+            .where(NoteSupersession.superseded_id == note_id)
+            .order_by(NoteSupersession.superseder_id)
+        )).scalars().all()]
+
+
+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_services_supersession.py

@@ -0,0 +1,148 @@
+"""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_supersedes(7, 1) == []
+        assert await supersession.get_superseded_by(7, 1) == []