FabledScribe/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 systems as systems_svc
from scribe.services import trash as trash_svc


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 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.
    """
    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))
    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,
    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.
        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)
    data = note.to_dict()
    if system_ids:
        data["systems"] = [
            s.to_dict() for s in await systems_svc.list_record_systems(uid, note.id)
        ]
    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,
) -> 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.
    """
    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)
    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)
        ]
    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)