src/scribe/app.py

@@ -32,6 +32,7 @@ from scribe.routes.trash import trash_bp
 from scribe.routes.dashboard import dashboard_bp
 from scribe.routes.systems import systems_bp
 from scribe.routes.canonical_systems import canonical_systems_bp
+from scribe.routes.lessons import lessons_bp
 from scribe.routes.snippets import snippets_bp
 from scribe.routes.webhooks import webhooks_bp
 from scribe.mcp import mount_mcp
@@ -92,6 +93,7 @@ def create_app() -> Quart:
     app.register_blueprint(search_bp)
     app.register_blueprint(profile_bp)
     app.register_blueprint(knowledge_bp)
+    app.register_blueprint(lessons_bp)
     app.register_blueprint(rulebooks_bp)
     app.register_blueprint(plugin_bp)
     app.register_blueprint(design_systems_bp)

src/scribe/mcp/tools/lessons.py

@@ -20,21 +20,10 @@ 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,
-    }
+# 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(

src/scribe/routes/lessons.py

@@ -0,0 +1,251 @@
+"""REST routes for lessons — a transferable insight, retrievable by situation.
+
+A lesson is a note with note_type='lesson' (see services/lessons.py). These
+routes feed the web UI; the MCP tools (mcp/tools/lessons.py) are the
+agent-facing surface. Both go through services/lessons.py, so the compose/parse
+contract and the `data` mirror live in one place — the same division snippets
+use, and for the same reason: two doors that each compose a lesson would
+compose it two ways, and the document IS what ranks.
+
+ACL (rule #78): reads and writes of a single lesson resolve through the
+share-aware `get_lesson` / `can_write_note`, and writes are performed as the
+OWNER so a shared editor isn't rejected by the owner-scoped service —
+mirroring routes/snippets.py and routes/notes.py.
+
+WHY THE TRIGGER IS A NAMED FIELD HERE TOO. The web editor could have posted a
+body and let the service parse it. It doesn't, because the evidence behind
+this kind (step 1) is that a trigger gets filled when a door ASKS for it by
+name — the snippet corpus is at 100% on its trigger with no guard anywhere,
+because a service composes the title from a parameter. A form that offered one
+markdown box would be the option milestone 385 rejected, wearing a different
+hat.
+"""
+import logging
+
+from quart import Blueprint, jsonify, request
+
+from scribe.auth import get_current_user_id, login_required
+from scribe.routes.utils import not_found, parse_pagination
+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.services.access import (
+    can_write_note,
+    describe_provenance,
+    label_shared_items,
+)
+from scribe.services.note_usage import empty_usage, record_pulled, usage_for_notes
+
+logger = logging.getLogger(__name__)
+
+lessons_bp = Blueprint("lessons", __name__, url_prefix="/api/lessons")
+
+
+@lessons_bp.route("", methods=["GET"])
+@login_required
+async def list_lessons_route():
+    """The kind enumerated, rather than only what a query resembles.
+
+    Semantic search is how a lesson REACHES a session; this is how a person
+    sees what exists at all. Each row carries its trigger, because a list of
+    lessons without them is a list of claims with the half that says when each
+    one matters left off.
+    """
+    uid = get_current_user_id()
+    q = request.args.get("q") or None
+    tag = request.args.get("tag", "")
+    try:
+        project_id = int(request.args.get("project_id", 0) or 0) or None
+    except (TypeError, ValueError):
+        project_id = None
+    limit, offset = parse_pagination(default_limit=24, max_limit=100)
+
+    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,
+        limit=limit,
+        offset=offset,
+        project_id=project_id,
+    )
+    return jsonify({
+        "lessons": await label_shared_items(uid, items),
+        "total": total,
+    })
+
+
+@lessons_bp.route("/taught-by/<int:record_id>", methods=["GET"])
+@login_required
+async def lessons_taught_by_route(record_id: int):
+    """The lessons drawn FROM one record — the reverse of `learned_from`.
+
+    Registered ABOVE the `/<int:lesson_id>` routes on purpose: Quart matches
+    in registration order, and `taught-by` would otherwise never be reached
+    if the converter ever widened. The same ordering snippets' `/duplicates`
+    route documents.
+
+    This is the direction that gets forgotten. A reader opening an old issue
+    wants to know what was learned from it, and without this the relation is
+    only navigable from the lesson's side.
+    """
+    uid = get_current_user_id()
+    notes = await lessons_svc.lessons_taught_by(uid, record_id)
+    return jsonify({
+        "lessons": [lessons_svc.lesson_to_dict(n) for n in notes],
+        "taught_by": record_id,
+    })
+
+
+@lessons_bp.route("", methods=["POST"])
+@login_required
+async def create_lesson_route():
+    uid = get_current_user_id()
+    data = await request.get_json() or {}
+    what = (data.get("what") or "").strip()
+    when_to_apply = (data.get("when_to_apply") or "").strip()
+    if not what:
+        return jsonify({"error": "what is required"}), 400
+    # The trigger is not optional at this door even though the service will
+    # store a lesson without one. A lesson with no trigger saves, reads
+    # correctly in every listing, and never surfaces — there is nothing to
+    # notice afterwards, which is exactly why the form has to refuse it here
+    # rather than leave the writer a record that looks finished.
+    if not when_to_apply:
+        return jsonify({
+            "error": "when_to_apply is required",
+            "detail": (
+                "A lesson is found by the SITUATION it applies to. Without a "
+                "trigger it still saves and still reads correctly, and it "
+                "never reaches anyone — so it is refused here rather than "
+                "stored as a record that looks finished."
+            ),
+        }), 400
+
+    project_id = data.get("project_id") or None
+    learned_from = data.get("learned_from") or []
+
+    # The same near-duplicate gate the MCP create path applies. Two lessons
+    # under one trigger compete in a single ranked list for one reserved slot,
+    # so the duplicate does not merely clutter — it displaces.
+    if not data.get("force"):
+        title, body = lessons_svc.lesson_document(
+            what, when_to_apply, data.get("insight", ""), learned_from,
+        )
+        dup = await dedup_svc.find_duplicate_note(
+            uid, title, body,
+            project_id=project_id,
+            is_task=False,
+            note_type=lessons_svc.LESSON_NOTE_TYPE,
+        )
+        if dup is not None:
+            return jsonify(dedup_svc.duplicate_response(dup, "lesson")), 409
+
+    note = await lessons_svc.create_lesson(
+        uid,
+        what=what,
+        when_to_apply=when_to_apply,
+        insight=data.get("insight", ""),
+        learned_from=learned_from,
+        tags=data.get("tags"),
+        project_id=project_id,
+    )
+    if data.get("system_ids") is not None:
+        await systems_svc.set_record_systems(uid, note.id, data["system_ids"])
+    out = lessons_svc.lesson_to_dict(note)
+    out["systems"] = [
+        s.to_dict() for s in await systems_svc.list_record_systems(uid, note.id)
+    ]
+    return jsonify(out), 201
+
+
+@lessons_bp.route("/<int:lesson_id>", methods=["GET"])
+@login_required
+async def get_lesson_route(lesson_id: int):
+    uid = get_current_user_id()
+    note = await lessons_svc.get_lesson(uid, lesson_id)
+    if note is None:
+        return not_found("Lesson")
+    out = lessons_svc.lesson_to_dict(note)
+    # As the OWNER: a shared reader isn't scoped to the owner's project, so
+    # their own id would come back empty (the write-as-owner pattern this
+    # module already uses, read side).
+    out["systems"] = [
+        s.to_dict()
+        for s in await systems_svc.list_record_systems(note.user_id, lesson_id)
+    ]
+    out.update(await describe_provenance(uid, note))
+    out["usage"] = (await usage_for_notes([lesson_id])).get(
+        lesson_id, empty_usage()
+    )
+    # Opening the detail view IS a pull — the operator chose to look. Tagged
+    # apart from the MCP sources so "an agent was handed it" and "a human read
+    # it" stay distinguishable; they mean different things for pruning (#2085).
+    record_pulled(user_id=uid, note_id=lesson_id, source="rest_lesson")
+    return jsonify(out)
+
+
+@lessons_bp.route("/<int:lesson_id>", methods=["PATCH"])
+@login_required
+async def update_lesson_route(lesson_id: int):
+    uid = get_current_user_id()
+    note = await lessons_svc.get_lesson(uid, lesson_id)
+    if note is None:
+        return not_found("Lesson")
+    if not await can_write_note(uid, lesson_id):
+        return jsonify({"error": "Permission denied"}), 403
+    owner_uid = note.user_id
+    data = await request.get_json() or {}
+
+    # Partial update: only keys present in the payload change, and the service
+    # re-composes title, body and mirror from the merged set — so a form that
+    # sends one field cannot leave the halves of the document disagreeing.
+    kwargs = {
+        k: data[k]
+        for k in ("what", "when_to_apply", "insight", "learned_from", "tags")
+        if k in data
+    }
+    # An empty trigger would save and silently stop the lesson surfacing, so
+    # clearing it is refused for the same reason creating without one is.
+    if "when_to_apply" in kwargs and not (kwargs["when_to_apply"] or "").strip():
+        return jsonify({
+            "error": "when_to_apply cannot be cleared",
+            "detail": (
+                "A lesson with no trigger never surfaces, and nothing about "
+                "the stored record would show it. Rewrite the trigger rather "
+                "than emptying it."
+            ),
+        }), 400
+
+    updated = await lessons_svc.update_lesson(owner_uid, lesson_id, **kwargs)
+    if updated is None:
+        return not_found("Lesson")
+    if data.get("system_ids") is not None:
+        await systems_svc.set_record_systems(
+            owner_uid, lesson_id, data["system_ids"]
+        )
+    out = lessons_svc.lesson_to_dict(updated)
+    out["systems"] = [
+        s.to_dict()
+        for s in await systems_svc.list_record_systems(owner_uid, lesson_id)
+    ]
+    return jsonify(out)
+
+
+@lessons_bp.route("/<int:lesson_id>", methods=["DELETE"])
+@login_required
+async def delete_lesson_route(lesson_id: int):
+    """Trash, not erase — recoverable from the trash like every other kind."""
+    uid = get_current_user_id()
+    note = await lessons_svc.get_lesson(uid, lesson_id)
+    if note is None:
+        return not_found("Lesson")
+    if not await can_write_note(uid, lesson_id):
+        return jsonify({"error": "Permission denied"}), 403
+    batch_id = await trash_svc.delete(note.user_id, "note", lesson_id)
+    if batch_id is None:
+        return not_found("Lesson")
+    return jsonify({"deleted": lesson_id, "deleted_batch_id": batch_id})

src/scribe/services/lessons.py

@@ -88,6 +88,9 @@ from __future__ import annotations
 
 import re
 
+from scribe.models import async_session
+from scribe.models.note import Note
+
 LESSON_NOTE_TYPE = "lesson"
 
 # The key in `notes.data`. Named for the field it mirrors on `rules`, because it
@@ -415,6 +418,78 @@ async def get_lesson(user_id: int, lesson_id: int):
     return note
 
 
+def lesson_to_dict(note) -> dict:
+    """A lesson as either door returns it — the composed fields read back out,
+    not the raw row, so a caller sees the same vocabulary it wrote with.
+
+    In the SERVICE rather than in each door, on the `snippet_to_dict`
+    precedent: the REST route feeds the web UI and the MCP tools feed an
+    agent, and a shape spelled once per door is a shape that answers the two
+    of them differently the first time a field is added.
+    """
+    return {
+        "id": note.id,
+        "title": note.title,
+        "body": note.body,
+        # The composed vocabulary, not the storage: a caller that wrote
+        # `when_to_apply` reads `when_to_apply` back.
+        "what": (note.data or {}).get("what", "") if isinstance(note.data, dict) else "",
+        "when_to_apply": lesson_trigger(note),
+        "learned_from": lesson_sources(note),
+        "insight": _strip_composed_lines(note.body),
+        "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 lessons_taught_by(user_id: int, record_id: int, limit: int = 20):
+    """The lessons drawn FROM one record — the reverse of `learned_from`.
+
+    THE DIRECTION THAT GETS FORGOTTEN, and arguably the more useful one: a
+    reader opening an old issue wants to know what was learned from it, and
+    without this the relation is only navigable from the lesson's side. A
+    record that taught something should say so on its own page.
+
+    Queried through `data[SOURCES_KEY]` rather than by scanning bodies: the
+    mirror is JSONB with a GIN index (0070), which is the whole reason step 4
+    put the list there. `path_exists` is the same dialect the snippet location
+    lookup uses, so both reverse lookups read the index the same way.
+
+    Share-aware (rule 78) via `readable_notes_clause`: this renders beside a
+    record the caller can already see, and a lesson someone shared with them
+    belongs in that list exactly as their own does.
+    """
+    from sqlalchemy import select
+
+    from scribe.services.access import readable_notes_clause
+
+    try:
+        wanted = int(record_id)
+    except (TypeError, ValueError):
+        return []
+    if wanted <= 0:
+        return []
+
+    # The id is an int we just validated, never caller text, so it cannot
+    # break out of the expression — the same guarantee `location_jsonpath`
+    # gets from JSON-quoting its values.
+    jsonpath = f"$.{SOURCES_KEY}[*] ? (@ == {wanted})"
+    async with async_session() as session:
+        result = await session.execute(
+            select(Note)
+            .where(Note.note_type == LESSON_NOTE_TYPE)
+            .where(Note.deleted_at.is_(None))
+            .where(Note.data.path_exists(jsonpath))
+            .where(readable_notes_clause(user_id))
+            .order_by(Note.updated_at.desc())
+            .limit(max(1, min(limit, 100)))
+        )
+        return list(result.scalars().all())
+
+
 async def update_lesson(
     user_id: int,
     lesson_id: int,

tests/test_lesson_rest_door.py

@@ -0,0 +1,246 @@
+"""The REST door for lessons — the half the web UI can actually reach (#3734).
+
+WHY THIS EXISTS AT ALL
+
+Milestone 385 built the lesson kind through the MCP tools, which is the surface
+an agent uses. The Vue app speaks REST, so until this blueprint existed a lesson
+was a record a person could not create, read, edit or retire from the UI — rule
+27's "no UI, no ship" failing at the door rather than in the view.
+
+WHAT THESE PIN
+
+Three things a second door tends to get wrong, and one that is specific to this
+kind:
+
+  - PARITY. Both doors go through services/lessons.py, so the composed
+    document is identical whichever one wrote it. A REST door that composed its
+    own title would produce lessons that rank differently from the agent's, and
+    the document IS what ranks.
+
+  - ACL (rule 78). Share-aware resolve, write as the owner — the pattern
+    routes/snippets.py sets — so a shared editor isn't rejected by the
+    owner-scoped service.
+
+  - THE TRIGGER IS REFUSED WHEN EMPTY. This is the kind-specific one and the
+    reason the door is not a thin wrapper. The service will happily store a
+    lesson with no trigger: it saves, it reads correctly in every listing, and
+    it never surfaces. There is nothing to notice afterwards — it looks exactly
+    like a lesson that works. So the door refuses it at both create and update
+    rather than handing back a record that looks finished.
+
+  - THE REVERSE DIRECTION. `taught-by/<id>` answers "what was learned from this
+    record", which the task body calls the direction that gets forgotten and
+    arguably the more useful one.
+"""
+import inspect
+
+import pytest
+
+
+# ── parity: one composer, two doors ──────────────────────────────────────────
+
+
+def test_both_doors_share_one_serializer():
+    """A payload shape spelled once per door answers the two of them
+    differently the first time a field is added."""
+    from scribe.mcp.tools import lessons as mcp_lessons
+    from scribe.services import lessons as lessons_svc
+
+    assert mcp_lessons._to_dict is lessons_svc.lesson_to_dict
+
+
+def test_the_rest_door_composes_nothing_itself():
+    """Asserted on structure (rule 167). The document is what ranks, so a door
+    that built its own title would produce lessons that rank differently from
+    the ones the agent writes."""
+    from scribe.routes import lessons as routes
+
+    src = inspect.getsource(routes)
+    # It may CALL the service's composer (the dedup gate needs the document),
+    # but it must not assemble a title or a trigger line itself.
+    assert "trigger_title" not in src
+    assert "**When to apply:**" not in src
+    assert "lessons_svc.lesson_document" in src, (
+        "the dedup gate must hash the same document the service will store"
+    )
+
+
+def test_the_door_is_registered():
+    """A blueprint nobody registers is a file, not a door."""
+    from scribe import app as app_module
+
+    src = inspect.getsource(app_module)
+    assert "from scribe.routes.lessons import lessons_bp" in src
+    assert "app.register_blueprint(lessons_bp)" in src
+
+
+def test_the_blueprint_is_mounted_where_the_client_looks():
+    from scribe.routes.lessons import lessons_bp
+
+    assert lessons_bp.url_prefix == "/api/lessons"
+
+
+def test_the_reverse_lookup_is_registered_before_the_id_route():
+    """Quart matches in registration order. The int converter protects
+    `taught-by` today, but the ordering is what keeps that true if the
+    converter is ever widened — the same care snippets' `/duplicates` takes."""
+    from scribe.routes import lessons as routes
+
+    src = inspect.getsource(routes)
+    assert src.index('"/taught-by/<int:record_id>"') < src.index(
+        '"/<int:lesson_id>"'
+    )
+
+
+# ── the trigger is not optional at this door ─────────────────────────────────
+
+
+def test_create_refuses_a_lesson_with_no_trigger():
+    """The kind-specific guard. A triggerless lesson saves and never surfaces,
+    and nothing about the stored record shows it."""
+    from scribe.routes import lessons as routes
+
+    src = inspect.getsource(routes.create_lesson_route)
+    assert "when_to_apply is required" in src
+    assert "never reaches anyone" in src, (
+        "the refusal must say WHY, or the next person reads it as a nag and "
+        "removes it"
+    )
+
+
+def test_update_refuses_to_clear_the_trigger():
+    """The other half. Creating without one is refused; emptying one later
+    would reach the same broken state by a different path."""
+    from scribe.routes import lessons as routes
+
+    src = inspect.getsource(routes.update_lesson_route)
+    assert "cannot be cleared" in src
+
+
+def test_the_subject_is_required_too():
+    from scribe.routes import lessons as routes
+
+    src = inspect.getsource(routes.create_lesson_route)
+    assert "what is required" in src
+
+
+# ── ACL: rule 78's pattern, not a bare owner filter ──────────────────────────
+
+
+@pytest.mark.parametrize(
+    "handler", ["update_lesson_route", "delete_lesson_route"]
+)
+def test_writes_check_permission_and_act_as_the_owner(handler):
+    """Resolve share-aware, then write as the owner — otherwise a shared
+    editor is rejected by the owner-scoped service."""
+    from scribe.routes import lessons as routes
+
+    src = inspect.getsource(getattr(routes, handler))
+    assert "can_write_note" in src, f"{handler} does not check write permission"
+    assert "note.user_id" in src, (
+        f"{handler} writes as the caller rather than as the owner, which "
+        f"rejects a legitimately shared editor (rule 78)"
+    )
+
+
+def test_no_handler_builds_its_own_owner_filter():
+    """Rule 78's actual failure mode: a route assembling its own
+    `Note.user_id == uid` clause instead of going through the service and the
+    access helpers. Passing `user_id=uid` INTO a service is the correct call
+    and is not what this looks for."""
+    from scribe.routes import lessons as routes
+
+    src = inspect.getsource(routes)
+    assert "Note.user_id" not in src
+    assert "select(" not in src, (
+        "a route composing its own query has bypassed the access helpers"
+    )
+
+
+def test_delete_trashes_recoverably():
+    """Every kind's delete is a trash, and the batch id is what restores it."""
+    from scribe.routes import lessons as routes
+
+    src = inspect.getsource(routes.delete_lesson_route)
+    assert "trash_svc.delete" in src
+    assert "deleted_batch_id" in src
+
+
+# ── the reverse direction ────────────────────────────────────────────────────
+
+
+def test_the_reverse_lookup_reads_the_indexed_mirror():
+    """Not by scanning bodies: `data[taught_by]` is JSONB with a GIN index
+    (0070), which is the whole reason step 4 put the list there."""
+    from scribe.services import lessons as lessons_svc
+
+    src = inspect.getsource(lessons_svc.lessons_taught_by)
+    assert "path_exists" in src
+    assert "SOURCES_KEY" in src
+
+
+def test_the_reverse_lookup_is_share_aware():
+    """It renders beside a record the caller can already see, so a lesson
+    someone shared with them belongs in the list exactly as their own does."""
+    from scribe.services import lessons as lessons_svc
+
+    src = inspect.getsource(lessons_svc.lessons_taught_by)
+    assert "readable_notes_clause" in src
+    assert "deleted_at" in src
+
+
+def test_the_reverse_lookup_refuses_a_nonsense_id_rather_than_interpolating():
+    """The jsonpath is built by formatting, so the id has to be an int before
+    it gets near the expression."""
+    from scribe.services import lessons as lessons_svc
+
+    src = inspect.getsource(lessons_svc.lessons_taught_by)
+    assert "int(record_id)" in src
+
+
+@pytest.mark.asyncio
+async def test_a_bad_id_returns_nothing_and_raises_nothing():
+    from scribe.services.lessons import lessons_taught_by
+
+    assert await lessons_taught_by(1, 0) == []
+    assert await lessons_taught_by(1, -3) == []
+    assert await lessons_taught_by(1, "not a number") == []
+
+
+# ── the serializer speaks the vocabulary the caller wrote with ───────────────
+
+
+def test_the_payload_reads_back_the_composed_fields():
+    """A caller that wrote `when_to_apply` reads `when_to_apply` back, not a
+    body it has to parse."""
+    from tests.helpers import fake_lesson
+    from scribe.services.lessons import compose_body, compose_title, lesson_to_dict
+
+    what = "Read the job log before waiting longer"
+    trigger = "a CI run has sat in_progress longer than its suite takes"
+    note = fake_lesson(
+        id=7,
+        title=compose_title(what, trigger),
+        body=compose_body("The work is usually done.", trigger, [4181]),
+        data={"what": what, "when_to_apply": trigger, "taught_by": [4181]},
+        project_id=None,
+        created_at=None,
+        updated_at=None,
+    )
+    out = lesson_to_dict(note)
+    assert out["what"] == what
+    assert out["when_to_apply"] == trigger
+    assert out["learned_from"] == [4181]
+    # The insight comes back WITHOUT the lines compose_body added, so an edit
+    # form round-trips instead of accumulating a copy of them per save.
+    assert out["insight"] == "The work is usually done."
+    assert "**When to apply:**" not in out["insight"]
+
+
+def test_the_rest_guards_can_fail():
+    """Rule 167: shown turning red once."""
+    from scribe.routes import lessons as routes
+
+    src = inspect.getsource(routes)
+    assert "a phrase that is definitely not in this module" not in src
+    assert "lessons_bp" in src