FabledScribe/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 lesson_rules as lesson_rules_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 attach_usage, record_pulled
from scribe.services import moment_delivery


# The payload shape lives in the service (`lesson_to_dict`), shared with the
# REST door — a shape spelled once per door answers the two of them
# differently the first time a field is added.
_to_dict = lessons_svc.lesson_to_dict


async def list_lessons(
    q: str = "", tag: str = "", limit: int = 50, offset: int = 0,
    project_id: int = 0, unjudged: bool = False,
) -> 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.
        unjudged: List only the lessons nobody has answered "which rule is
            this an instance of?" for — no rule named, and no "no rule fits"
            recorded. Work down this list with `update_lesson(rule_ids=…)` or
            `update_lesson(no_rule="why")`. A listing rather than a search, so
            it takes `tag` and `project_id` but not `q`.

    Returns {"lessons": [{id, title, when_to_apply, tags, preview}], "total"}.
    """
    uid = current_user_id()
    if unjudged:
        if q:
            raise ValueError(
                "unjudged lists every unjudged lesson rather than searching "
                "them — call it with tag/project_id and without q."
            )
        items, total = await lesson_rules_svc.list_unjudged(
            uid, tag=tag, project_id=project_id or None,
            limit=max(1, min(limit, 100)), offset=max(0, offset),
        )
    else:
        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)
    # One aggregate for the page, like the snippet listing — surfaced-vs-opened
    # per lesson (#4196). An agent listing lessons can see which of its own
    # triggers are firing and which are not, which is the reading that leads to
    # `update_lesson` rather than to a second lesson about the same failure.
    await attach_usage(labelled)
    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", ""),
            "usage": it["usage"],
            **({"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,
    rule_ids: list[int] | None = None,
    no_rule: str = "",
    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 POINTS AT THE RULE IT IS AN INSTANCE OF. While you write it, you
    know the situation better than anyone will again, so that is when to say
    which binding choice it falls under. Answer one of two ways, in this call
    or right after it with `update_lesson`:

    - `rule_ids=[…]` — the rule(s) or preference(s) it is an instance of. The
      rule then becomes reachable through the situation the lesson describes,
      which is closer to why it was written than its own wording often is.
    - `no_rule="why"` — no rule governs this situation, in a line. That is a
      real answer: a lesson that stands alone is what a rule nobody has
      written yet is made from, and the reason lets it be re-judged later.

    A lesson created with neither comes back with `rule_candidates` — the
    rules it most resembles, each with its trigger — and `rule_judgment:
    "unjudged"`. Read them against the lesson and answer; `list_lessons(
    unjudged=true)` gathers any left open.

    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: how the operator wants work DONE is a PREFERENCE
    (create_preference) — it guides every session rather than informing one,
    and it is kept current as they correct you. 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. At most 240
            characters and no line breaks; a longer one is refused — the
            story goes in `insight`.
        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.
        rule_ids: The rule(s) or preference(s) this lesson is an instance of —
            the binding choice its situation falls under. A lesson never
            becomes a rule; it points at the one that governs it, and the rule
            is then reachable through the situation the lesson describes
            (milestone 440). Naming a rule here confirms the link.
        no_rule: The reason no rule governs this situation, in a line — the
            other answer to "which rule?". Give one or the other, not both.
            When other lessons in the same situation also answered "no rule
            fits", the response carries `convergence`: the group, and the
            rule it may be missing.
        force: Create even if a near-duplicate exists.

    Returns the created lesson with its `rules` and `rule_judgment`, plus
    `rule_candidates` when it is unjudged (absent when the rule search could
    not run). 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."
        )

    lessons_svc.require_claim(what)
    sources = lessons_svc.normalize_sources(learned_from)
    # Validated before anything is written, so a lesson naming a rule the
    # caller cannot read fails whole rather than saving half-linked.
    linked = await lesson_rules_svc.require_rules(uid, rule_ids)
    lesson_rules_svc.require_one_answer(linked, no_rule)
    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,
            data=lessons_svc.compose_data(what, when_to_apply),
        )
        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)
    if linked:
        await lesson_rules_svc.set_lesson_rules(uid, note.id, linked)
    elif no_rule.strip():
        await lesson_rules_svc.set_no_rule(uid, note.id, no_rule)
    data = _to_dict(note)
    await systems_tools.attach_systems(uid, uid, data, note.id, project_id or None)
    await lesson_rules_svc.attach_lesson_rules(uid, [data])
    if not linked and not no_rule.strip():
        await _offer_candidates(uid, data, what, when_to_apply, project_id)
    elif no_rule.strip():
        await _name_convergence(uid, data, note.id)
    return await moment_delivery.attach_moment_rules(uid, "create_lesson", {"project_id": project_id}, data)


async def _name_convergence(uid: int, data: dict, lesson_id: int) -> None:
    """A "no rule fits" answer is the moment to notice it is not the first
    for this situation (#4634) — `convergence` names the group and the rule
    it may be missing. Absent when there is no group."""
    group = await lesson_rules_svc.convergence_for(uid, lesson_id)
    if group:
        data["convergence"] = group


async def _offer_candidates(uid: int, data: dict, what: str, trigger: str, project_id: int) -> None:
    """Put the rules an unjudged lesson resembles in front of its writer.

    `rule_judgment` is set here too, because the attach above is fail-open and
    may have left it off; a lesson just created with no answer IS unjudged.
    """
    data["rule_judgment"] = lesson_rules_svc.UNJUDGED
    candidates = await lesson_rules_svc.rule_candidates(uid, what, trigger, project_id or None)
    if candidates is not None:
        data["rule_candidates"] = candidates
    data["rule_hint"] = (
        "Which rule is this lesson an instance of? If one of rule_candidates "
        "governs its situation, update_lesson(lesson_id, rule_ids=[…]) links "
        "it; if none does, update_lesson(lesson_id, no_rule=\"why\") records "
        "that it stands alone."
    )


async def get_lesson(lesson_id: int, project_id: int = 0) -> dict:
    """Fetch one lesson by id, with its trigger and sources read back out.

    IF THIS LESSON JUST PROVED ITSELF, IT IS WORTH MORE THAN IT SAYS. You are
    reading it inside the situation it names, which makes you the one reader
    who can tell whether its trigger is keyed to what actually fired and
    whether its claim covers what you are seeing. `update_lesson` takes
    another incident into `learned_from`, a claim stated more exactly, or a
    re-keyed trigger — and the trigger is the edit that pays most, because a
    lesson keyed to a situation nobody is in looks exactly like one nobody
    needed.

    `project_id` is the project you are WORKING IN, not this record's own.
    Passing the active project is what makes "opened away from where it was
    written" answerable; 0 leaves it unreported and the pull still counts.
    """
    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).
    # Read BEFORE the pull is recorded, so the number an agent is shown is the
    # one that was true when it asked — otherwise every first read of a lesson
    # reports a pull that is its own.
    await attach_usage([out])
    await lesson_rules_svc.attach_lesson_rules(uid, [out])
    record_pulled(
        user_id=uid, note_id=int(note.id),
        source="mcp_get_lesson", project_id=project_id,
    )
    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,
    system_ids: list[int] | None = None,
    rule_ids: list[int] | None = None,
    no_rule: str = "",
) -> 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, at most 240 characters. 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.
        system_ids: Replace the Systems this lesson is filed under. None
            leaves unchanged; pass the FULL list, and `[]` to clear.

            Here because `create_lesson` took it and this did not, so a lesson
            written without one could never be filed afterwards — and the end
            of a piece of work, when a lesson is usually written, is exactly
            when that argument gets dropped (#4249). A System tag is how
            `list_system_records` gathers an area's pile, so an untagged
            lesson is reachable by search and by nothing else.
        rule_ids: Replace the rule(s) this lesson is an instance of. None
            leaves unchanged; pass the FULL list. A rule that was linked and
            is left out is recorded as REJECTED — "not an instance of this
            one" — so the pair is not proposed again; `[]` rejects them all.
            Naming a rule replaces any "no rule fits" answer.
        no_rule: Record that no rule governs this lesson's situation, with the
            reason in a line. Any rule still linked is rejected with that
            reason. Empty leaves the answer unchanged; give this or a
            non-empty `rule_ids`, not both. When other lessons in the same
            situation also answered "no rule fits", the response carries
            `convergence`: the group, and the rule it may be missing.
    """
    uid = current_user_id()
    # Only a NEW name is checked: a lesson stored with a long one must still
    # take a new trigger or a rule link — and the fix for it is this call.
    if what:
        lessons_svc.require_claim(what)
    linked = (
        await lesson_rules_svc.require_rules(uid, rule_ids)
        if rule_ids is not None else None
    )
    lesson_rules_svc.require_one_answer(linked, no_rule)
    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")
    # `is not None` rather than truthiness, so `[]` CLEARS the associations.
    # `set_record_systems` is replace-semantics; treating [] as "no change"
    # would make the one call that unfiles a lesson silently do nothing.
    if system_ids is not None:
        await systems_svc.set_record_systems(uid, lesson_id, system_ids)
        note = await lessons_svc.get_lesson(uid, lesson_id) or note
    if linked is not None:
        await lesson_rules_svc.set_lesson_rules(uid, lesson_id, linked)
    if no_rule.strip():
        await lesson_rules_svc.set_no_rule(uid, lesson_id, no_rule)
    out = _to_dict(note)
    await lesson_rules_svc.attach_lesson_rules(uid, [out])
    if no_rule.strip():
        await _name_convergence(uid, out, lesson_id)
    return out


async def judge_lesson_link(
    lesson_id: int, rule_id: int, verdict: str, note: str = "",
) -> dict:
    """Say whether a lesson is an instance of a rule: `verdict` is "confirm"
    or "reject".

    Reach for this when Scribe proposes a pair — a lesson and a rule that keep
    arriving together in different situations — or whenever you are reading a
    lesson and recognise the rule it falls under. Confirming makes the rule
    reachable through the situation the lesson describes; rejecting records
    that it is not, so the pair is not proposed again. `note` is the why, and
    the next reader judges the link by it.

    Returns the link: {lesson_id, rule_id, state, note, evidence, judged_at}.
    """
    uid = current_user_id()
    return await lesson_rules_svc.judge_link(uid, lesson_id, rule_id, verdict, 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, judge_lesson_link):
        mcp.tool(name=fn.__name__)(fn)