feat(supersession): declare it — supersedes on both write paths

Step 2 of #278. Records and reads the claim; the demotion that makes it matter
is step 3.

`services/supersession.py` with set/get on both directions, following the
set_record_systems shape since this is the same kind of mutable M2M at the
tool/route layer rather than inside notes_svc.

## Both directions are exposed, and only one 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 with no
marker gets acted on confidently, which is worse than never surfacing it. So
get_note carries it, says so in its docstring, and adds a plain-language line
telling the reader to open the newer note first.

Both are OMITTED when empty rather than serialised as empty lists. A field that
always says nothing trains readers to skip fields — the lesson consolidated_at
cost, removed in the previous commit.

## Refuse vs drop, which is the one real judgement here

Dropped silently: a target that doesn't exist, is trashed, is the note itself,
or would close a cycle. Each is a claim with no subject or no meaning; none is
something the caller can act on.

REFUSED with PermissionError: a target the caller can read but not write.
That is the single case where the caller could believe they succeeded and be
wrong in a way that matters — demoting someone else's record out of their
retrieval is damage invisible from the outside, with no symptom for the owner
to trace. Rule #47, and PermissionError because services/snippets.py already
uses it for read-but-not-write with both surfaces catching it.

The PATCH/PUT routes scope by the CALLER, not owner_uid: an editor-share holder
may edit the note and must not thereby inherit the owner's write access to
whatever they name as superseded.

## Cycles

A ring claims every member is obsolete. Under flat demotion that demotes them
all equally, so the set drops out of ranked retrieval together with nothing in
the data saying why. Refused by walking the existing graph from the proposed
target — iteratively with a visited set, because the graph is user-supplied and
a deep chain must not become a stack overflow on a write path. The visited set
also makes the walk terminate on a ring that already exists, which is pinned by
its own test rather than trusted.

Both surfaces (#33), the instruction surface per #119 — framed as the third
answer beside update-instead and force=true: not everything resembling an
existing record should be folded into it, and not everything distinct should
compete with it forever.

Refs #278

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) == []