"""Note CRUD MCP tools. Thin wrappers over services/notes.py with is_task=False. Signatures mirror the existing fable-mcp contracts exactly so client behavior is preserved. Sentinel conventions (inherited from existing fable-mcp tools): - `tag: str = ""` / `search_text: str = ""` — empty means "no filter" - `project_id: int = 0` — on create: orphan note (no project); on update: "leave unchanged" (there is no "remove from project" through this tool, which is a pre-existing limitation) - `title: str = ""` / `body: str = ""` on update — empty means "leave unchanged" - `tags: list[str] | None = None` — None means "leave unchanged"; [] clears """ from __future__ import annotations from scribe.mcp._context import current_user_id from scribe.mcp.tools import systems as systems_tools 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 async def list_notes( limit: int = 20, offset: int = 0, tag: str = "", search_text: str = "", project_id: int = 0, ) -> dict: """List notes (non-task documents) stored in Scribe. Optionally filter by a single tag (plain string, no # prefix) or by `search_text`, which matches on MEANING — the same semantic match every search surface uses (#2462; keyword ILIKE is only the fallback when the embedder is down). With `search_text`, results order by relevance; without it, by last-updated descending. Reaches your records plus shared-project records, like every other list. Prefer `search` for a pure lookup — it returns scores and reaches everything readable; this adds the lifecycle filters on top. Args: project_id: Scope to one project. PASS THE ACTIVE PROJECT'S ID whenever a project is in scope so you list that project's notes, not every project's. 0 = no filter (all projects — use only for a deliberate cross-project view). """ uid = current_user_id() rows, total = await notes_svc.list_notes( uid, q=search_text or None, tags=[tag] if tag else None, is_task=False, project_id=project_id or None, limit=max(1, min(limit, 100)), offset=max(0, offset), ) return {"notes": [n.to_dict() for n in rows], "total": total} 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 — plus `systems` (the areas this note is filed under) or, for an untagged project note, the `systems_hint` question. 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) note = loaded[0] if loaded else None if note is None or note.deleted_at is not None: raise ValueError(f"note {note_id} not found") out = note.to_dict() out.update(await access_svc.describe_provenance(uid, note)) # Records the pull for ANY note kind, not just snippets: the auto-inject # menu surfaces notes, tasks and processes too, so restricting this to # 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 supersession_svc.attach_relations(uid, note_id, out, hint=True) await systems_tools.attach_systems( uid, getattr(note, "user_id", uid) or uid, out, note.id, note.project_id ) return out async def create_note( title: str, body: str = "", tags: list[str] | None = None, project_id: int = 0, system_ids: list[int] | None = None, supersedes: list[int] | None = None, verify_with: str = "", expires_when: str = "", force: bool = False, ) -> dict: """Create a new note in Scribe. WHAT ELSE COULD HOLD THIS? A note is the right home when the answer is "nothing": it records what you know, nobody owes anything on it, and nothing enforces it. Otherwise — - someone has to DO something -> create_task. A note titled "we should…" is a task nobody will ever see again. - future sessions must OBEY it -> create_rule. The test is whether ignoring it would be a mistake, not merely uninformed. - reusable code with a place in a repo -> create_snippet. The location is what lets it be found from the file someone is about to edit. - a procedure followed start to finish -> create_process. Args: title: Note title (required). body: Markdown content. Supports [[wikilinks]] to other notes by title. tags: List of plain-string tags without # prefix, e.g. ["python", "ideas"]. 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. verify_with: HOW TO CHECK this note is still true. Leave empty for almost every note — that is the normal case, not an unfinished one. A note is a NORM or a CONSTRAINT. A norm is a decision ("we derive versions from commit time"): it has no truth value and changes only when its author changes it, which they know they did. A CONSTRAINT asserts a fact about someone else's software ("AMO refuses to re-sign a version", "this forge numbers CI runs per repository"), and it goes false with nobody watching. Only constraints get a check. The test, in one question: COULD THIS NOTE BECOME FALSE WITHOUT ANYONE EDITING IT? If no, leave this empty. Sharper still: is the thing this note describes YOURS TO CHANGE? If yes it is a decision. Measured against a real corpus, every note that earned a check was about somebody else's software. Three that look like candidates and are not: a resume pointer or "current state" note (goes stale fastest, but the cure is to update it, not to check it); a measurement of your own system (it goes false because you changed something, and you knew); and a decision that RESTS on someone else's behaviour (check the note asserting the fact, not the decision). A command, a path, a URL, a query. Prose is allowed; something runnable is better. expires_when: The STATE that ends it — deliberately not a date. "When Forgejo issues run numbers per workflow rather than per repository", not "in six months". Constraints expire when the ground moves, not on a schedule. 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 instead (no duplicate bloat / no stale RAG copies). Set true only when you're sure this is a genuinely distinct note. Returns the created note object including its assigned id, OR — when a near-duplicate is found and force is false — {"duplicate": true, "existing_id": ..., "message": ...} and nothing is created. A tagged record shows its `systems`; created untagged in a project, the response carries the `systems_hint` question instead — answer it: tag the record, create the missing System, or deliberately leave it untagged. """ uid = current_user_id() if not force: dup = await dedup_svc.find_duplicate_note( uid, title, body, project_id=project_id or None, is_task=False, note_type="note", ) if dup is not None: return dedup_svc.duplicate_response(dup, "note") note = await notes_svc.create_note( uid, title=title, body=body, tags=tags, project_id=project_id or None, verify_with=verify_with, expires_when=expires_when, ) 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() await systems_tools.attach_systems(uid, uid, data, note.id, project_id or None) await supersession_svc.attach_relations(uid, note.id, data, hint=True) return data async def update_note( note_id: int, title: str = "", body: str = "", tags: list[str] | None = None, project_id: int = 0, system_ids: list[int] | None = None, supersedes: list[int] | None = None, verify_with: str = "", expires_when: str = "", clear: list[str] | None = None, ) -> dict: """Update an existing Scribe note. Only explicitly provided fields are changed. Args: note_id: ID of the note to update. title: New title, or omit to leave unchanged. body: New markdown body, or omit to leave unchanged. tags: Replaces the full tag list. Pass [] to clear all tags. Omit to leave unchanged. 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. verify_with: How to check this note is still true. Almost every note should leave this empty — that is the normal, finished state, not a gap: an empty check is the marker for "this is a decision, there is nothing to go and check". See create_note for the full norm-vs-constraint test; the short form is "could this become false without anyone editing it?". expires_when: The STATE that ends it, not a date. clear: Names of fields to UNSET — "verify_with", "expires_when". Needed because "" means "leave this alone" here, so there is no value that removes a field: an agent updating a body must not silently wipe a check it was not asked about. A note that stops being a constraint is cleared by naming the field, which cannot happen by accident. Rewriting `verify_with` drops the note's verification stamp: a stamp certifies a particular check, and carrying it across a rewrite would vouch for something nobody has looked at. A task and a snippet are both REFUSED a check, with a message saying where to go instead — a task's decay is its status, and a snippet has verify_snippet. """ uid = current_user_id() fields: dict = {} if title: fields["title"] = title if body: fields["body"] = body if tags is not None: fields["tags"] = tags if project_id: fields["project_id"] = project_id if verify_with: fields["verify_with"] = verify_with if expires_when: fields["expires_when"] = expires_when note = await notes_svc.update_note( uid, note_id, clear=clear or (), **fields ) if note is None: 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() await systems_tools.attach_systems( uid, getattr(note, "user_id", uid) or uid, data, note_id, note.project_id ) await supersession_svc.attach_relations(uid, note_id, data, hint=True) return data async def find_duplicate_records(kind: str = "note", threshold: float = 0.0) -> dict: """Notes or tasks already recorded that closely resemble each other. The create gate PREVENTS a duplicate arriving through an agent; the web UI deliberately has no gate (a human mid-thought must not be blocked by a 409), which makes YOU the corrections system — and a corrector has to be able to SEE what needs correcting. This is the finder. Run it when tidying a project's records, or when you suspect the same ground was covered twice. Args: kind: "note" (documents) or "task". Snippets have their own report, find_duplicate_snippets, whose groups propose a lossless merge. threshold: Similarity floor, 0-1. 0 uses the configured setting. READ THE `suggestion` FIELD BEFORE ACTING, because the right fix differs by what the records ARE, not how alike they score. For notes: a correction pair → declare `supersedes` on the newer; state smeared across dated records → extract it into the System's reference note and leave these as history; genuinely parallel records → leave them alone. NEVER merge notes. Each group carries `members` with dates and any `existing_supersessions` already declared inside it — a pair someone has ruled on is not an open question. """ uid = current_user_id() if kind not in ("note", "task"): raise ValueError('kind must be "note" or "task" — snippets have their ' "own report, find_duplicate_snippets") return await dedup_svc.find_duplicate_records( uid, kind=kind, threshold=threshold if threshold > 0 else None, ) async def delete_note(note_id: int) -> dict: """Move a Scribe note to the trash (recoverable). Restore via restore(batch_id).""" uid = current_user_id() # Read the title BEFORE the delete: afterwards the row is trashed and the # confirmation could only echo the number back. A deletion the operator # cannot recognise is one they cannot tell was the wrong one. # Fail-open: the title is a COURTESY on top of the delete, so a lookup # that errors must not stop the delete happening. Same posture the # staleness marker takes — a decoration may never break its payload. try: loaded = await notes_svc.get_note_for_user(uid, note_id) title = getattr(loaded[0], "title", "") if loaded else "" except Exception: title = "" batch = await trash_svc.delete(uid, "note", note_id) if batch is None: raise ValueError(f"note {note_id} not found") return {"deleted": note_id, "title": title, "deleted_batch_id": batch, "message": f'Note {note_id} ("{title}") moved to trash. ' f"Restore with restore('{batch}')."} async def notes_due_for_verification( older_than_days: int = 0, project_id: int = 0, never_only: bool = False, ) -> dict: """Which notes assert a FACT that nobody has confirmed lately. A corpus of notes holds two kinds of thing. Most are DECISIONS or records of what happened — they have no truth value and cannot rot. A few assert a fact about someone else's software: what a signing service does on a duplicate upload, how a forge numbers its CI runs, what an updater compares. Those go false silently, with nobody present, and a cross-project reference note keeps being read as current by every project that cites it. This lists the second kind, oldest verification first, NEVER-CHECKED AT THE TOP — a note nobody has ever confirmed is a claim with no evidence behind it at all. Each row carries `verify_with` in full, because you are about to go and run it, plus `expires_when` and `days_since_verified`. Reach for it when curating, when a note's claim just contradicted what you observed, or periodically. Then, per row: run the check, and call mark_note_verified with what you found. Notes with no `verify_with` never appear, and that is correct — they are decisions, and there is nothing to go and check. Do not "fix" their absence by giving them checks: this list is only worth reading while everything on it genuinely can go false. Args: older_than_days: only notes last verified longer ago than this. Never-checked notes always qualify — they are the most overdue thing there is. 0 = no age filter. project_id: narrow to one project. 0 = every project. Unlike the rules sweep, this filter is safe: a note belongs to at most one project outright, with none of the subscription and always-on paths that would make a project filter UNDER-report a rule. never_only: only notes nobody has ever verified. """ uid = current_user_id() notes = await notes_svc.notes_due_for_verification( uid, older_than_days=older_than_days, project_id=project_id or None, never_only=never_only, ) return { "notes": [notes_svc.verification_row(n) for n in notes], "total": len(notes), } async def mark_note_verified(note_id: int, still_true: bool = True) -> dict: """Record that you ran a note's check — and what it said. Call this AFTER actually running the note's `verify_with`, never on the strength of the claim sounding plausible. A stamp nobody earned is worse than no stamp: it moves the note to the bottom of the sweep and buys the claim another long silence. `still_true=False` writes NOTHING. A note whose check failed is not in a special state to be recorded — it is WRONG, and the honest next moves are to correct it, supersede it, or find out why. So it stays at the top of the sweep until someone deals with it, and the response tells you what the note said would end it. Args: note_id: the note whose check you ran. still_true: True if the check passed. False if the fact it asserts is no longer true — say so, that is the outcome worth having. """ uid = current_user_id() note = await notes_svc.mark_note_verified(note_id, uid, still_true) if note is None: raise ValueError( f"note {note_id} not found, not writable by you, or carries no " f"verify_with (nothing to verify is not the same as verified)" ) data = notes_svc.verification_row(note) data["verified"] = bool(still_true) if not still_true: data["next"] = ( "This note is no longer true and is still being read as current " "by anything that cites it. Correct it with update_note, write " "the replacement with create_note(supersedes=[...]), or clear its " "check with update_note(clear=[\"verify_with\"]) if it has stopped " "asserting a fact at all. It stays at the top of " "notes_due_for_verification until one of those happens." ) return data def register(mcp) -> None: for fn in ( list_notes, get_note, create_note, update_note, find_duplicate_records, delete_note, notes_due_for_verification, mark_note_verified, ): mcp.tool(name=fn.__name__)(fn)