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