"""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. 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. See create_note for the norm-vs-constraint test that decides whether it should carry one at all; 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() batch = await trash_svc.delete(uid, "note", note_id) if batch is None: raise ValueError(f"note {note_id} not found") return {"deleted_batch_id": batch, "message": f"Note {note_id} moved to trash. Restore with restore('{batch}')."} def register(mcp) -> None: for fn in ( list_notes, get_note, create_note, update_note, find_duplicate_records, delete_note, ): mcp.tool(name=fn.__name__)(fn)