FabledGrimoire/src/scribe/mcp/tools/notes.py

"""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.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 a keyword
    search against title and body. Results are ordered by last-updated descending.

    Use search for semantic/meaning-based lookup instead of exact keyword search.

    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 _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).
    """
    rel = await supersession_svc.get_relations(uid, note_id)
    if rel["supersedes"]:
        data["supersedes"] = rel["supersedes"]
    if rel["superseded_by"]:
        data["superseded_by"] = rel["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)
    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 _attach_supersession(uid, note_id, out)
    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,
    force: bool = False,
) -> dict:
    """Create a new note in Scribe.

    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.
        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.
    """
    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,
    )
    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


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,
) -> 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.
    """
    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
    note = await notes_svc.update_note(uid, note_id, **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()
    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


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,
        delete_note,
    ):
        mcp.tool(name=fn.__name__)(fn)