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