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

"""Lesson MCP tools: a transferable insight, retrievable by situation.

Its own module rather than `create_note(note_type="lesson")`, on the precedent
of snippets and processes — and for the reason that precedent exists. A kind
whose value depends on a field being filled needs a door that ASKS for that
field by name. `create_note` would take a lesson through a generic body
parameter, and the trigger — the whole of why a lesson is findable at all —
would be something the writer had to know to include.
"""
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 knowledge as knowledge_svc
from scribe.services import lessons as lessons_svc
from scribe.services import systems as systems_svc
from scribe.services import trash as trash_svc
from scribe.mcp.tools import systems as systems_tools
from scribe.services.note_usage import record_pulled


def _to_dict(note) -> dict:
    """A lesson as the tools return it — the composed fields read back out,
    not the raw row, so a caller sees the same vocabulary it wrote with."""
    return {
        "id": note.id,
        "title": note.title,
        "body": note.body,
        "when_to_apply": lessons_svc.lesson_trigger(note),
        "learned_from": lessons_svc.lesson_sources(note),
        "tags": list(note.tags or []),
        "project_id": note.project_id,
        "note_type": note.note_type,
        "created_at": note.created_at.isoformat() if note.created_at else None,
        "updated_at": note.updated_at.isoformat() if note.updated_at else None,
    }


async def list_lessons(
    q: str = "", tag: str = "", limit: int = 50, offset: int = 0,
    project_id: int = 0,
) -> dict:
    """List lessons — the kind enumerated, rather than only what a query
    resembles.

    Semantic search is how a lesson REACHES a session; this is how a person or
    an agent sees what exists. The two answer different questions, and without
    this one there is no way to ask "what has been learned here at all".

    Each entry carries its trigger, because a list of lessons sorted by title
    is a list of claims with the situation — the half that says when each one
    matters — left off.

    Args:
        q: Free-text search across title + body (optional).
        tag: Filter to a single tag (optional).
        limit: Max results (1-100).
        offset: Skip this many before returning — page past the cap. `total`
            is the unpaged count, so it says whether more remains.
        project_id: Narrow to where a lesson was WRITTEN. 0 = every project.
            A lesson is retrievable from anywhere regardless (step 3); this
            filters the listing, not the reach.

    Returns {"lessons": [{id, title, when_to_apply, tags, preview}], "total"}.
    """
    uid = current_user_id()
    items, total = await knowledge_svc.query_knowledge(
        user_id=uid, note_type=lessons_svc.LESSON_NOTE_TYPE,
        tags=[tag] if tag else [], sort="modified", q=q or None,
        limit=max(1, min(limit, 100)), offset=max(0, offset),
        project_id=project_id or None,
    )
    labelled = await access_svc.label_shared_items(uid, items)
    rows = [
        {
            "id": it["id"], "title": it["title"], "tags": it.get("tags", []),
            "preview": it.get("snippet", ""),
            # Projected by `_note_to_item` straight off the `data` mirror —
            # absent when the row carries none, rather than an empty string.
            "when_to_apply": it.get("when_to_apply", ""),
            **({"shared": True, "owner": it.get("owner")} if it.get("shared") else {}),
        }
        for it in labelled
    ]
    return {"lessons": rows, "total": total}


async def create_lesson(
    what: str,
    when_to_apply: str,
    insight: str = "",
    learned_from: list[int] | None = None,
    tags: list[str] | None = None,
    project_id: int = 0,
    system_ids: list[int] | None = None,
    force: bool = False,
) -> dict:
    """Record something you LEARNED, so a later session meets it at the moment
    it applies — on this project or any other.

    A LESSON OR A RULE? The difference is FORCE, not importance. A rule is
    something that must be followed; a lesson is something worth knowing. If
    ignoring it would be a mistake, it is a rule (create_rule) and needs the
    operator's yes, because a rule binds every future session. If ignoring it
    just means someone re-derives it the slow way, it is a lesson — write it
    now, and nobody is bound by it.

    That distinction is the whole reason this kind exists. Sessions holding a
    transferable insight were reaching for create_rule because it was the only
    surface that is both global and situation-keyed, and proposing rules for
    things that should never have bound anyone.

    WHAT A LESSON IS NOT: a shape to copy is a SNIPPET (create_snippet); a
    procedure followed start to finish is a PROCESS (create_process); a record
    of what happened, findable by topic, is a NOTE (create_note). A lesson is
    the claim you would want handed to you in the same situation next time.

    `when_to_apply` IS THE RECORD. Everything else is the payload.

    A lesson reaches a session by resembling the SITUATION someone is in, never
    by topic — that is what separates it from a note, and it is done by putting
    the trigger in the title and again at the head of the body, so the document
    is dominated by when it applies. A lesson written without one still saves,
    still reads correctly in every listing, and will not surface when it is
    needed. There is nothing to notice afterwards: it looks exactly like a
    lesson that works.

    So write the SYMPTOM, in the words the situation will present itself in —
    what someone would be seeing, saying or about to do. "A test fails on code
    you believe is correct" is a trigger. "Testing" is a topic, and a topic
    matches everything and surfaces for nothing.

    Args:
        what: The insight in one line — the claim itself, as you would say it.
            This becomes the title, joined with the trigger.
        when_to_apply: The situation this applies in, as a symptom. Required.
        insight: The body — what to do, and the incident that taught it.
            Write the story here for the reader; it costs the ranking nothing,
            because a long body is split into chunks that each still carry the
            trigger.
        learned_from: Ids of the issues, tasks or notes this was drawn from —
            ALL of them. A lesson that generalises three incidents into one
            claim is the good case, not the edge case, so this is a list.
        tags: Optional tags.
        project_id: Where it was learned. Kept as a fact, and it does not limit
            reach: a lesson is retrievable from every project (that is the
            point of the kind). 0 = none.
        system_ids: Systems (subsystems/areas) to file it under.
        force: Create even if a near-duplicate exists.

    Returns the created lesson. On a near-duplicate, returns the existing id
    instead of creating — two lessons about one failure class want to be one
    lesson, so update that one rather than adding a second.
    """
    uid = current_user_id()
    if not when_to_apply or not when_to_apply.strip():
        raise ValueError(
            "when_to_apply is required: it is how a lesson is found. Say the "
            "SYMPTOM — what someone would be seeing, saying or about to do "
            "when this applies — not the topic it is about. Without it this "
            "record saves, reads correctly, and never surfaces."
        )

    sources = lessons_svc.normalize_sources(learned_from)
    title, body = lessons_svc.lesson_document(
        what, when_to_apply, insight, sources,
    )
    if not force:
        dup = await dedup_svc.find_duplicate_note(
            uid, title, body, project_id=project_id or None,
            is_task=False, note_type=lessons_svc.LESSON_NOTE_TYPE,
        )
        if dup is not None:
            return dedup_svc.duplicate_response(dup, "lesson")

    note = await lessons_svc.create_lesson(
        uid, what=what, when_to_apply=when_to_apply, insight=insight,
        learned_from=sources, tags=tags, project_id=project_id or None,
    )
    if system_ids:
        await systems_svc.set_record_systems(uid, note.id, system_ids)
    data = _to_dict(note)
    await systems_tools.attach_systems(uid, uid, data, note.id, project_id or None)
    return data


async def get_lesson(lesson_id: int) -> dict:
    """Fetch one lesson by id, with its trigger and sources read back out."""
    uid = current_user_id()
    note = await lessons_svc.get_lesson(uid, lesson_id)
    if note is None:
        raise ValueError(f"lesson {lesson_id} not found")
    out = _to_dict(note)
    out.update(await access_svc.describe_provenance(uid, note))
    # Every explicit open records the pull. A lesson is surfaced by the same
    # retrieval as any other note, so a getter that records nothing would leave
    # the kind permanently at zero pulls — reading as dead weight beside kinds
    # that merely had a counter (#2476, the repeat of #2245).
    record_pulled(user_id=uid, note_id=int(note.id), source="mcp_get_lesson")
    return out


async def update_lesson(
    lesson_id: int,
    what: str = "",
    when_to_apply: str = "",
    insight: str = "",
    learned_from: list[int] | None = None,
    tags: list[str] | None = None,
) -> dict:
    """Update a lesson. Empty fields are left unchanged.

    REWORDING A LESSON IS ORDINARY WORK. Understanding improves, and a trigger
    that turned out to fire on the wrong situation is the single most valuable
    thing to fix here — a lesson nobody is reaching is usually not wrong, it is
    keyed to a situation nobody is in.

    Title, body and the indexed mirror are re-composed together from the merged
    fields, so a partial update cannot leave the trigger saying one thing in
    the title and another in the body.

    Args:
        lesson_id: Lesson to update.
        what: New one-line claim. Empty leaves unchanged.
        when_to_apply: New trigger, as a symptom. Empty leaves unchanged.
        insight: New body. Empty leaves unchanged.
        learned_from: Replace the source ids. None leaves unchanged; pass the
            FULL list, including the ones already there.
        tags: Replace tags. None leaves unchanged.
    """
    uid = current_user_id()
    note = await lessons_svc.update_lesson(
        uid, lesson_id,
        what=what or None,
        when_to_apply=when_to_apply or None,
        insight=insight or None,
        learned_from=learned_from,
        tags=tags,
    )
    if note is None:
        raise ValueError(f"lesson {lesson_id} not found")
    return _to_dict(note)


async def delete_lesson(lesson_id: int) -> dict:
    """Retire a lesson — it moves to the trash and is recoverable.

    Reach for this when a lesson turned out to be wrong, or was superseded by
    a better one. A lesson that is merely NOT REACHING anyone is usually not a
    deletion: its trigger is keyed to a situation nobody is in, and rewording
    that with update_lesson keeps what was learned.

    Deletion was always possible through `delete_note` — a lesson is a note and
    the trash is kind-agnostic — but nothing said so, and a kind whose own
    tools offer create/read/update reads as one you cannot retire (#2250, which
    recorded exactly this for processes).
    """
    uid = current_user_id()
    note = await lessons_svc.get_lesson(uid, lesson_id)
    # The KIND is checked before deleting: this tool is reached for by name, so
    # letting it trash an ordinary note because the id happened to resolve
    # would be a destructive action taken on a mistyped argument.
    if note is None:
        raise ValueError(f"lesson {lesson_id} not found")
    batch = await trash_svc.delete(uid, "note", lesson_id)
    if batch is None:
        raise ValueError(f"lesson {lesson_id} not found")
    return {
        "deleted_batch_id": batch,
        "message": (
            f"Lesson {lesson_id} moved to trash. Restore with restore('{batch}')."
        ),
    }


def register(mcp) -> None:
    for fn in (list_lessons, create_lesson, get_lesson, update_lesson,
               delete_lesson):
        mcp.tool(name=fn.__name__)(fn)