"""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} SUPERSEDED_HINT = ( "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 attach_relations(user_id: int, note_id: int, data: dict, *, hint: bool = False) -> None: """Add both directions of the supersession relation to a note payload. ONE seam for the REST and MCP surfaces, which must agree about what a note's payload says — or the web UI and the agent would disagree about whether a record is current. Both directions, because they answer different questions 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 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 (#2483 — a field that always says nothing trains readers to skip fields). `hint=True` (the agent surface) also attaches `superseded_note`, the one-sentence reading instruction. """ rel = await get_relations(user_id, note_id) if rel["supersedes"]: data["supersedes"] = rel["supersedes"] if rel["superseded_by"]: data["superseded_by"] = rel["superseded_by"] if hint: data["superseded_note"] = SUPERSEDED_HINT