src/scribe/mcp/server.py

@@ -106,6 +106,7 @@ _READ_ONLY_TOOLS = frozenset({
     # free-text records and withholds the structured ones (#2496).
     "get_snippet", "list_snippets",
     "get_process", "list_processes",
+    "get_lesson",
     # Design systems: read, resolve (inheritance + mode), render, and compare
     # against recorded snippets. All four compute from stored records and write
     # nothing — the drift report is a report, and applying it is a separate
@@ -158,6 +159,7 @@ _READ_ONLY_TOOLS = frozenset({
 _WRITE_TOOLS = frozenset({
     # notes, tasks, planning
     "create_note", "update_note", "delete_note",
+    "create_lesson", "update_lesson",
     "create_task", "update_task", "delete_task", "add_task_log",
     "create_records", "start_planning",
     "create_milestone", "update_milestone", "delete_milestone",

src/scribe/mcp/tools/__init__.py

@@ -5,7 +5,8 @@ to a FastMCP instance. `register_all(mcp)` is the single entry point called
 from `mcp.server.build_mcp_server`.
 """
 from scribe.mcp.tools import (
-    design_systems, milestones, notes, processes, projects, recent, repos, retrieval_tuning,
+    design_systems, lessons, milestones, notes, processes, projects, recent, repos,
+    retrieval_tuning,
     wide_net,
     rulebooks, search, shapes, snippets, systems, tags, tasks, trash,
 )
@@ -27,6 +28,7 @@ def register_all(mcp) -> None:
     repos.register(mcp)
     processes.register(mcp)
     snippets.register(mcp)
+    lessons.register(mcp)
     shapes.register(mcp)
     rulebooks.register(mcp)
     trash.register(mcp)

src/scribe/mcp/tools/lessons.py

@@ -0,0 +1,197 @@
+"""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 lessons as lessons_svc
+from scribe.services import systems as systems_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 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)
+
+
+def register(mcp) -> None:
+    for fn in (create_lesson, get_lesson, update_lesson):
+        mcp.tool(name=fn.__name__)(fn)

src/scribe/services/dedup.py

@@ -38,6 +38,7 @@ from scribe.services import embeddings as embeddings_svc
 # Imported rather than redeclared: no service imports this module (the create
 # gate is called from the routes/tools layer), so there is no cycle to dodge,
 # and a second copy of the constant is a thing to drift.
+from scribe.services.lessons import LESSON_NOTE_TYPE
 from scribe.services.snippets import SNIPPET_NOTE_TYPE
 
 logger = logging.getLogger(__name__)
@@ -71,6 +72,28 @@ _SEMANTIC_THRESHOLD = 0.90
 # structural signals cannot see.
 _SNIPPET_SEMANTIC_THRESHOLD = 0.96
 
+# A LESSON is measured the same way, for the first of those reasons and not the
+# second. Its document is `{what} — {trigger}` over a body that opens by
+# restating the trigger (milestone 385 step 3) — the same prose-about-the-thing
+# shape, so two GENUINELY DIFFERENT lessons about one area ("CI cannot see this
+# class of failure") land in the same sibling band that refused .btn-danger
+# against .btn-danger-outline at 0.92.
+#
+# Its own constant rather than reusing the snippet's, because the two are
+# separate facts that happen to coincide: this number is INHERITED from #2518's
+# measurement of a structurally analogous corpus, not measured on lessons —
+# there are none yet to measure. When there are, this moves without dragging
+# snippets with it.
+#
+# The trade-off differs and is worth naming. A snippet has structural signals
+# (code, repo·path·symbol) to catch the literal copy a high bar lets through; a
+# lesson has none, and is not in `_REPORT_KINDS` either, so the duplicate report
+# is not a backstop for it yet. What remains is the exact-title check, which
+# still fires. That is the right way round for a gate that BLOCKS: a false
+# positive refuses a real lesson outright, while a false negative leaves two
+# records that can still be merged by hand.
+_LESSON_SEMANTIC_THRESHOLD = 0.96
+
 # The gate queries per CHUNK of the candidate (#280) — this caps how many
 # searches one save may cost. Eight chunks ≈ five thousand words of candidate;
 # a duplicate hiding past that is the duplicate report's job to find, not a
@@ -205,6 +228,17 @@ async def _find_snippet_by_structure(
     return None
 
 
+def _semantic_threshold(note_type: str) -> float:
+    """The semantic bar for this kind — a lookup, so the kinds that need a
+    different one are named in a single place rather than in a conditional
+    that grows a branch per kind."""
+    if note_type == SNIPPET_NOTE_TYPE:
+        return _SNIPPET_SEMANTIC_THRESHOLD
+    if note_type == LESSON_NOTE_TYPE:
+        return _LESSON_SEMANTIC_THRESHOLD
+    return _SEMANTIC_THRESHOLD
+
+
 async def find_duplicate_note(
     user_id: int,
     title: str,
@@ -286,8 +320,7 @@ async def find_duplicate_note(
                 user_id, query, project_id=project_id, is_task=is_task,
                 orphan_only=(project_id is None),
                 limit=3,
-                threshold=(_SNIPPET_SEMANTIC_THRESHOLD
+                threshold=_semantic_threshold(note_type),
-                           if note_type == SNIPPET_NOTE_TYPE else _SEMANTIC_THRESHOLD),
                 # Owner-only, deliberately: this gate BLOCKS a create and tells
                 # the caller to update the match instead. Matching someone
                 # else's record would refuse their write and point them at

src/scribe/services/lessons.py

@@ -95,6 +95,10 @@ LESSON_NOTE_TYPE = "lesson"
 # a second word for it.
 TRIGGER_KEY = "when_to_apply"
 
+# What taught this lesson: the ids of the issues, tasks or notes it was drawn
+# from. A LIST, and that is the whole decision — see `normalize_sources`.
+SOURCES_KEY = "taught_by"
+
 # The body's trigger line, and the pattern that reads it back. The body is the
 # readable form and the thing that gets embedded; `data` is the queryable
 # mirror. Reads prefer the mirror and fall back to this, which is the discipline
@@ -102,6 +106,12 @@ TRIGGER_KEY = "when_to_apply"
 # existed is still readable.
 _BODY_TRIGGER_RE = re.compile(r"^\*\*When to apply:\*\*\s*(.+?)\s*$", re.M)
 
+# The readable mirror of `data[SOURCES_KEY]`, and the pattern that reads it
+# back — the shape snippets use for `**Merged from:** #ids`, for the same
+# reason: the body is what a human sees and what survives a row with no `data`.
+_BODY_SOURCES_RE = re.compile(r"^\*\*Learned from:\*\*\s*(.+?)\s*$", re.M)
+_ID_RE = re.compile(r"#(\d+)")
+
 
 def lesson_trigger(note) -> str:
     """When this lesson applies, or "" — the mirror first, then the body.
@@ -120,6 +130,80 @@ def lesson_trigger(note) -> str:
     return match.group(1).strip() if match else ""
 
 
+def normalize_sources(entries: list | None) -> list[int]:
+    """The ids that taught this lesson — ints, de-duplicated, in the order given.
+
+    THE CARDINALITY DECISION, and why it is a list.
+
+    `arose_from_id` already exists and holds ONE id, which is the obvious first
+    answer and the wrong one. The lesson that started this milestone generalised
+    THREE incidents — a badge collision, an un-backfilled column, a duplicated
+    const — into one claim about failure classes no CI lane can see. Generalising
+    across incidents is the shape a good lesson HAS, not an edge case. A single
+    id would keep the first and silently drop the rest, and a record that drops
+    two of its three sources is worse than one that names none, because it reads
+    as complete.
+
+    It lives in `notes.data` rather than a join table for exactly the reason
+    decision #4157 put the trigger there: a join table would settle, for every
+    note kind at once, whether provenance is multi-valued — a question nothing
+    has measured. `data` is JSONB with a GIN index (0070), so the list is
+    queryable today and a table can be migrated to later if the need is shown.
+
+    Order is history, not sorting: the incidents stay in the sequence the writer
+    named them, which is the order they were learned in.
+    """
+    out: list[int] = []
+    seen: set[int] = set()
+    for raw in entries or []:
+        ident = raw.get("id") if isinstance(raw, dict) else raw
+        try:
+            i = int(ident)
+        except (TypeError, ValueError):
+            continue
+        if i > 0 and i not in seen:
+            seen.add(i)
+            out.append(i)
+    return out
+
+
+def lesson_sources(note) -> list[int]:
+    """What taught this lesson — the mirror first, then the body, then
+    `arose_from_id`.
+
+    Three fallbacks rather than two, because the third is what keeps this
+    honest on a record written before the kind existed: a note carrying only
+    `arose_from_id` has exactly one source and this returns it, so a caller
+    never has to ask which field to read.
+    """
+    data = getattr(note, "data", None) or {}
+    if isinstance(data, dict):
+        from_mirror = normalize_sources(data.get(SOURCES_KEY))
+        if from_mirror:
+            return from_mirror
+    match = _BODY_SOURCES_RE.search(getattr(note, "body", None) or "")
+    if match:
+        found = normalize_sources(_ID_RE.findall(match.group(1)))
+        if found:
+            return found
+    single = getattr(note, "arose_from_id", None)
+    return normalize_sources([single]) if single else []
+
+
+def sole_source(sources: list[int] | None) -> int | None:
+    """`arose_from_id` for this lesson: the id when there is exactly ONE.
+
+    Left NULL for a lesson drawn from several, deliberately. Every existing
+    surface that renders provenance reads `arose_from_id` and renders it as
+    THE origin; handing it one of three would make those surfaces state
+    something false. Showing nothing there is accurate — there is no single
+    origin — and `data[SOURCES_KEY]` carries all of them for the surfaces that
+    know to ask.
+    """
+    ids = normalize_sources(sources)
+    return ids[0] if len(ids) == 1 else None
+
+
 def compose_title(what: str, when_to_apply: str = "") -> str:
     """`{what} — {when it applies}`, the half of the document that ranks.
 
@@ -138,7 +222,9 @@ def compose_title(what: str, when_to_apply: str = "") -> str:
     return trigger_title(what, when_to_apply)
 
 
-def compose_body(insight: str, when_to_apply: str = "") -> str:
+def compose_body(
+    insight: str, when_to_apply: str = "", learned_from: list[int] | None = None,
+) -> str:
     """The lesson body — the trigger line first, the insight after.
 
     The mirror of `compose_title` on the other half of the document, and the
@@ -172,11 +258,19 @@ def compose_body(insight: str, when_to_apply: str = "") -> str:
     insight = (insight or "").strip()
     if insight:
         lines.append(insight)
+    sources = normalize_sources(learned_from)
+    if sources:
+        # LAST, not beside the trigger. The first line has to be what this
+        # lesson is FOR; a provenance line above the insight would push the
+        # thing the reader came for below a list of ids, and would put
+        # numbers where the trigger's second appearance does its work.
+        lines.append("**Learned from:** " + ", ".join(f"#{i}" for i in sources))
     return "\n\n".join(lines)
 
 
 def lesson_document(
     what: str, when_to_apply: str = "", insight: str = "",
+    learned_from: list[int] | None = None,
 ) -> tuple[str, str]:
     """The (title, body) a lesson is STORED — and therefore embedded — as.
 
@@ -193,4 +287,156 @@ def lesson_document(
     instead — the stored record IS the sharp document — which is why nothing
     re-embeds and `CHUNKER_VERSION` does not move.
     """
-    return compose_title(what, when_to_apply), compose_body(insight, when_to_apply)
+    return (
+        compose_title(what, when_to_apply),
+        compose_body(insight, when_to_apply, learned_from),
+    )
+
+
+def compose_data(
+    what: str, when_to_apply: str = "", learned_from: list[int] | None = None,
+) -> dict:
+    """The indexed mirror of the same fields the body renders (0070).
+
+    Written together with the body by the one caller that composes both, so the
+    two can never describe different things — the discipline `compose_data`
+    follows for snippets, and the reason `lesson_trigger` can prefer `data`
+    without checking whether it agrees with the prose.
+
+    Empty values are omitted so the column stays sparse: a lesson with no
+    sources has no `taught_by` key rather than an empty list, which keeps a
+    `?` containment query honest.
+    """
+    data: dict = {"what": what.strip()} if what and what.strip() else {}
+    trigger = (when_to_apply or "").strip()
+    if trigger:
+        data[TRIGGER_KEY] = trigger
+    sources = normalize_sources(learned_from)
+    if sources:
+        data[SOURCES_KEY] = sources
+    return data
+
+
+async def create_lesson(
+    user_id: int,
+    *,
+    what: str,
+    when_to_apply: str = "",
+    insight: str = "",
+    learned_from: list[int] | None = None,
+    tags: list[str] | None = None,
+    project_id: int | None = None,
+):
+    """Create a lesson note. Returns the created Note.
+
+    The title, the body and the `data` mirror are composed HERE from named
+    parameters rather than asked of the caller. That is the whole evidence base
+    for this design and not a convenience: the snippet corpus carries a trigger
+    on 164 of 164 records with no guard anywhere, because a service builds the
+    title from a parameter — what is at 100% is a named structured field, not an
+    agent typing a convention correctly.
+
+    `project_id` is accepted and kept, even though a lesson is reachable from
+    every project (step 3). Where it was learned is a fact worth keeping; it
+    simply stops being the limit of where it can be found.
+    """
+    from scribe.services import notes as notes_svc
+
+    sources = normalize_sources(learned_from)
+    title, body = lesson_document(what, when_to_apply, insight, sources)
+    return await notes_svc.create_note(
+        user_id,
+        title=title,
+        body=body,
+        note_type=LESSON_NOTE_TYPE,
+        tags=tags,
+        project_id=project_id,
+        # NULL unless there is exactly one source — see `sole_source`.
+        arose_from_id=sole_source(sources),
+        data=compose_data(what, when_to_apply, sources),
+    )
+
+
+async def get_lesson(user_id: int, lesson_id: int):
+    """Fetch a lesson by id, or None if it isn't one / isn't readable.
+
+    Share-aware (rule 78): a fetch by id is an explicit act, so it resolves the
+    caller's full read scope rather than ownership alone — without this, a
+    lesson a search legitimately surfaced could not then be opened (#2093).
+    """
+    from scribe.services import notes as notes_svc
+
+    result = await notes_svc.get_note_for_user(user_id, lesson_id)
+    if result is None:
+        return None
+    note, _permission = result
+    if note.note_type != LESSON_NOTE_TYPE or note.deleted_at is not None:
+        return None
+    return note
+
+
+async def update_lesson(
+    user_id: int,
+    lesson_id: int,
+    *,
+    what: str | None = None,
+    when_to_apply: str | None = None,
+    insight: str | None = None,
+    learned_from: list[int] | None = None,
+    tags: list[str] | None = None,
+):
+    """Update a lesson, re-composing title, body and mirror from the merged
+    fields. Returns the updated Note, or None if it isn't a readable lesson.
+
+    READ-MODIFY-WRITE over the whole record rather than patching one half.
+    The three fields are not independent: the trigger appears in the title AND
+    at the head of the body, so editing it in place would need two edits that
+    a caller could do one of. Re-composing from the merged values means a
+    partial update cannot leave the halves disagreeing — which, because the
+    document is what ranks, would be a lesson that still reads correctly and
+    quietly stops being retrievable.
+    """
+    from scribe.services import notes as notes_svc
+
+    note = await get_lesson(user_id, lesson_id)
+    if note is None:
+        return None
+
+    current = note.data if isinstance(note.data, dict) else {}
+    merged_what = current.get("what") or "" if what is None else what
+    merged_trigger = lesson_trigger(note) if when_to_apply is None else when_to_apply
+    merged_sources = (
+        lesson_sources(note) if learned_from is None
+        else normalize_sources(learned_from)
+    )
+    if insight is None:
+        insight = _strip_composed_lines(note.body)
+
+    title, body = lesson_document(
+        merged_what, merged_trigger, insight, merged_sources,
+    )
+    fields: dict = {
+        "title": title,
+        "body": body,
+        "arose_from_id": sole_source(merged_sources),
+        "data": compose_data(merged_what, merged_trigger, merged_sources),
+    }
+    if tags is not None:
+        fields["tags"] = tags
+    return await notes_svc.update_note(user_id, lesson_id, **fields)
+
+
+def _strip_composed_lines(body: str | None) -> str:
+    """The insight alone — the body with the lines `compose_body` wrote removed.
+
+    An update that keeps the insight has to hand it back to `compose_body`,
+    which will re-add the trigger and provenance lines. Without this the two
+    composed lines accumulate a copy per edit, and since the trigger line is
+    half of what makes the document rank, the duplicates would look like the
+    shape working rather than a bug.
+    """
+    kept = [
+        line for line in (body or "").splitlines()
+        if not _BODY_TRIGGER_RE.match(line) and not _BODY_SOURCES_RE.match(line)
+    ]
+    return "\n".join(kept).strip()

tests/test_lesson_write_path.py

@@ -0,0 +1,200 @@
+"""Creating a lesson, and tying it to what taught it (milestone 385 step 4).
+
+THE TRIGGER IS THE RECORD, so the door refuses a lesson without one.
+
+Step 1 could have chosen "flag it visibly" instead. Refusing is the stronger
+answer for the same reason `create_rule` makes enforcement the deciding
+question: a lesson with no trigger is not a weaker lesson, it is a note that
+will never surface, and nothing downstream can tell the difference. It saves,
+it reads correctly in every listing, and it is silently absent from the one
+moment it was written for. A flag would be a warning nobody is present to read
+— the write path is where the writer still is.
+
+THE SOURCE IDS ARE A LIST, and `test_a_lesson_keeps_every_incident_that_taught
+_it` is why. The founding example generalised three incidents into one claim;
+`arose_from_id` holds one, and a record that keeps the first and drops two
+reads as complete.
+"""
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from scribe.mcp._context import _user_id_ctx
+from scribe.mcp.tools.lessons import create_lesson, update_lesson
+from scribe.services import lessons as lessons_svc
+
+TRIGGER = "a test fails on code you believe is correct"
+SUBJECT = "Suspect the guard before the code"
+
+
+def _stub_note(**kw):
+    """A stand-in lesson row — every attribute the tool's _to_dict reads."""
+    base = dict(
+        id=1, title="t", body="b", tags=[], project_id=None,
+        note_type="lesson", data={}, arose_from_id=None,
+        created_at=None, updated_at=None,
+    )
+    base.update(kw)
+    return SimpleNamespace(**base)
+
+
+@pytest.mark.asyncio
+async def test_a_lesson_without_a_trigger_is_refused():
+    """THE guard. Falsifiable: it names the parameter, so a door that stopped
+    asking for it fails here rather than quietly writing an unfindable record."""
+    _user_id_ctx.set(7)
+    with pytest.raises(ValueError) as err:
+        await create_lesson(what=SUBJECT, when_to_apply="")
+
+    message = str(err.value)
+    assert "when_to_apply" in message
+    # It says what to write, not just that something is missing — the writer is
+    # here now, and "required" alone produces a topic where a symptom was wanted.
+    assert "symptom" in message.lower()
+
+
+@pytest.mark.asyncio
+async def test_whitespace_is_not_a_trigger():
+    """The refusal reads the stripped value. A door checking only falsiness
+    accepts a space and produces exactly the record it meant to prevent."""
+    _user_id_ctx.set(7)
+    with pytest.raises(ValueError):
+        await create_lesson(what=SUBJECT, when_to_apply="   ")
+
+
+@pytest.mark.asyncio
+async def test_a_lesson_keeps_every_incident_that_taught_it():
+    """Three sources in, three sources stored — and `arose_from_id` left NULL,
+    because one of three on a surface that renders it as THE origin would make
+    that surface state something false."""
+    _user_id_ctx.set(7)
+    created = AsyncMock(return_value=_stub_note())
+    with patch.object(lessons_svc, "create_lesson", created), \
+         patch("scribe.mcp.tools.lessons.dedup_svc.find_duplicate_note",
+               AsyncMock(return_value=None)), \
+         patch("scribe.mcp.tools.lessons.systems_tools.attach_systems", AsyncMock()):
+        await create_lesson(
+            what=SUBJECT, when_to_apply=TRIGGER, learned_from=[11, 22, 33],
+        )
+
+    assert created.await_args.kwargs["learned_from"] == [11, 22, 33]
+    assert lessons_svc.sole_source([11, 22, 33]) is None
+    assert lessons_svc.compose_data(SUBJECT, TRIGGER, [11, 22, 33])["taught_by"] == [
+        11, 22, 33,
+    ]
+
+
+def test_a_single_source_still_reaches_arose_from_id():
+    """The existing provenance field keeps working for the ordinary case — a
+    lesson drawn from one issue is not made less connected by the list."""
+    assert lessons_svc.sole_source([11]) == 11
+
+
+@pytest.mark.asyncio
+async def test_the_duplicate_gate_runs_before_anything_is_created():
+    """A near-match returns the existing id and writes nothing: two lessons
+    about one failure class want to be one lesson."""
+    _user_id_ctx.set(7)
+    hit = SimpleNamespace(id=99, title="already recorded", similarity=0.97,
+                          reason="semantic")
+    created = AsyncMock()
+    with patch("scribe.mcp.tools.lessons.dedup_svc.find_duplicate_note",
+               AsyncMock(return_value=hit)), \
+         patch.object(lessons_svc, "create_lesson", created):
+        out = await create_lesson(what=SUBJECT, when_to_apply=TRIGGER)
+
+    assert created.await_count == 0
+    assert out["duplicate"] is True
+    assert out["existing_id"] == 99
+    # The hint names the lesson's own updater, so the caller is pointed at a
+    # tool that exists rather than at update_note.
+    assert "update_lesson" in out["message"]
+
+
+@pytest.mark.asyncio
+async def test_the_gate_compares_the_composed_document_not_the_raw_fields():
+    """What reaches the gate is the title and body a lesson will actually be
+    stored as. Comparing `what` alone would miss that the trigger is half the
+    document, and would judge two lessons alike that rank nothing alike."""
+    _user_id_ctx.set(7)
+    gate = AsyncMock(return_value=None)
+    with patch("scribe.mcp.tools.lessons.dedup_svc.find_duplicate_note", gate), \
+         patch.object(lessons_svc, "create_lesson",
+                      AsyncMock(return_value=_stub_note())), \
+         patch("scribe.mcp.tools.lessons.systems_tools.attach_systems", AsyncMock()):
+        await create_lesson(what=SUBJECT, when_to_apply=TRIGGER, insight="Look.")
+
+    title, body = gate.await_args.args[1], gate.await_args.args[2]
+    assert title == f"{SUBJECT} — {TRIGGER}"
+    assert body.startswith(f"**When to apply:** {TRIGGER}")
+    assert gate.await_args.kwargs["note_type"] == "lesson"
+
+
+def test_a_lesson_is_judged_at_the_trigger_dominated_bar():
+    """#2518 measured deliberately-parallel siblings at 0.92 on a document that
+    is mostly prose ABOUT the thing. A lesson's document is that shape, so the
+    bar sits above that band — otherwise two different lessons about one area
+    block each other, which is the failure step 4 asked to check for."""
+    from scribe.services.dedup import (
+        _LESSON_SEMANTIC_THRESHOLD,
+        _SEMANTIC_THRESHOLD,
+        _semantic_threshold,
+    )
+
+    assert _semantic_threshold("lesson") == _LESSON_SEMANTIC_THRESHOLD
+    assert _LESSON_SEMANTIC_THRESHOLD > 0.92, (
+        "below the observed sibling band, so two genuinely different lessons "
+        "about one area would refuse each other"
+    )
+    assert _LESSON_SEMANTIC_THRESHOLD > _SEMANTIC_THRESHOLD
+    # An ordinary note is untouched — the carve-out is per kind, not a
+    # loosening of the gate.
+    assert _semantic_threshold("note") == _SEMANTIC_THRESHOLD
+
+
+@pytest.mark.asyncio
+async def test_an_update_recomposes_both_halves_of_the_document():
+    """A new trigger has to reach the title AND the head of the body. Patching
+    one would leave a lesson that reads correctly and ranks on the old
+    situation — the failure mode with no symptom."""
+    _user_id_ctx.set(7)
+    stored = _stub_note(
+        title=f"{SUBJECT} — {TRIGGER}",
+        body=f"**When to apply:** {TRIGGER}\n\nLook at the guard.",
+        data={"what": SUBJECT, "when_to_apply": TRIGGER},
+    )
+    updated = AsyncMock(return_value=_stub_note())
+    with patch.object(lessons_svc, "get_lesson", AsyncMock(return_value=stored)), \
+         patch("scribe.services.notes.update_note", updated):
+        await lessons_svc.update_lesson(7, 1, when_to_apply="a guard goes red")
+
+    fields = updated.await_args.kwargs
+    assert fields["title"] == f"{SUBJECT} — a guard goes red"
+    assert fields["body"].startswith("**When to apply:** a guard goes red")
+    assert fields["data"]["when_to_apply"] == "a guard goes red"
+
+
+@pytest.mark.asyncio
+async def test_an_update_does_not_stack_the_composed_lines():
+    """The insight is handed back to `compose_body`, which re-adds the trigger
+    and provenance lines. Without stripping them first they accumulate a copy
+    per edit — and because the trigger line is half of what makes the document
+    rank, the duplicates would look like the shape working."""
+    _user_id_ctx.set(7)
+    stored = _stub_note(
+        body=f"**When to apply:** {TRIGGER}\n\nLook at the guard."
+             "\n\n**Learned from:** #5",
+        data={"what": SUBJECT, "when_to_apply": TRIGGER, "taught_by": [5]},
+    )
+    updated = AsyncMock(return_value=_stub_note())
+    with patch.object(lessons_svc, "get_lesson", AsyncMock(return_value=stored)), \
+         patch("scribe.services.notes.update_note", updated):
+        await lessons_svc.update_lesson(7, 1, what="Suspect the guard")
+
+    body = updated.await_args.kwargs["body"]
+    assert body.count("**When to apply:**") == 1
+    assert body.count("**Learned from:**") == 1
+    assert "Look at the guard." in body