src/scribe/app.py

@@ -162,6 +162,13 @@ def create_app() -> Quart:
     async def startup():
         import asyncio
 
+        # Set on the last line of this hook; the deferred backfill below waits
+        # on it. An Event rather than a bare bool so the waiter is woken
+        # instead of polling, and declared here — inside the hook, where
+        # `asyncio` is in scope and a running loop exists — rather than in
+        # `create_app`, which runs before either is true.
+        _startup_finished = asyncio.Event()
+
         from scribe.services.auth import start_auth_token_retention_loop
         from scribe.services.embeddings import (
             backfill_milestone_embeddings, backfill_note_embeddings, backfill_rule_embeddings,
@@ -174,8 +181,23 @@ def create_app() -> Quart:
         start_auth_token_retention_loop()
 
         # Backfill embeddings for any notes that don't have one. Runs in the
-        # background so it never blocks the server from accepting requests.
+        # background so it never blocks the server from accepting requests —
+        # and, since #4181, not until the rest of this hook has finished.
+        #
+        # "Background" was true of REQUESTS and false of STARTUP. A task
+        # created here begins immediately, while `before_serving` is still
+        # running, so its four passes competed with the startup hook's own
+        # database reads for the same connection pool. That is survivable on a
+        # healthy disk and fatal on a sick one: on 2026-09-19 both this
+        # backfill's first query and the hook's maintenance-hour read were
+        # cancelled together when Hypercorn killed the worker at its lifespan
+        # timeout, and the instance stayed down for three hours.
+        #
+        # The wait is on the SERVING FLAG rather than a sleep, because a sleep
+        # would be a guess about how long the rest of the hook takes and would
+        # be wrong in exactly the conditions that matter.
         async def _delayed_backfill() -> None:
+            await _startup_finished.wait()
             try:
                 await backfill_note_embeddings()
             except Exception:
@@ -202,36 +224,56 @@ def create_app() -> Quart:
             except Exception:
                 logger.warning("Snippet data backfill failed", exc_info=True)
 
+        # Created here but gated on the flag released at the END of this hook,
+        # so the task exists (nothing can forget to start it) while none of its
+        # work overlaps startup's own.
         asyncio.create_task(_delayed_backfill())
 
-        # Recurrence scheduler (recurring-task spawn every 15m)
-        from scribe.services.recurrence_scheduler import start_recurrence_scheduler
-        start_recurrence_scheduler(asyncio.get_running_loop())
+        # RELEASED IN A `finally`, never after the work (rules 156 and 157).
+        # The waiter above has no deadline of its own, and the way to make an
+        # undeadlined wait safe is to make the release unmissable: if anything
+        # below raises, the flag is still set and the task ends instead of
+        # living on as a coroutine nobody will ever wake.
+        try:
+            # Recurrence scheduler (recurring-task spawn every 15m)
+            from scribe.services.recurrence_scheduler import start_recurrence_scheduler
+            start_recurrence_scheduler(asyncio.get_running_loop())
 
-        # Version-pinning scheduler (daily auto-pin scan at 03:00 UTC)
-        from scribe.services.version_pinning_scheduler import (
-            start_version_pinning_scheduler,
-        )
-        start_version_pinning_scheduler(asyncio.get_running_loop())
+            # Version-pinning scheduler (daily auto-pin scan at 03:00 UTC)
+            from scribe.services.version_pinning_scheduler import (
+                start_version_pinning_scheduler,
+            )
+            start_version_pinning_scheduler(asyncio.get_running_loop())
 
-        # Trash retention scheduler (daily expired-trash purge at 03:30 UTC)
-        from scribe.services.trash_scheduler import start_trash_scheduler
-        start_trash_scheduler(asyncio.get_running_loop())
+            # Trash retention scheduler (daily expired-trash purge at 03:30 UTC)
+            from scribe.services.trash_scheduler import start_trash_scheduler
+            start_trash_scheduler(asyncio.get_running_loop())
 
-        # DB maintenance scheduler (daily targeted VACUUM ANALYZE, default 04:00 UTC)
-        from scribe.services.db_maintenance_scheduler import (
-            get_maintenance_hour,
-            start_db_maintenance_scheduler,
-        )
-        start_db_maintenance_scheduler(
-            asyncio.get_running_loop(), await get_maintenance_hour()
-        )
+            # DB maintenance scheduler (daily targeted VACUUM ANALYZE, default
+            # 04:00 UTC). `get_maintenance_hour` is the FIRST database read in
+            # this hook and is bounded for that reason (#4181) — see the long
+            # comment above `_STARTUP_READ_TIMEOUT`. Anything added here that
+            # touches the database needs the same treatment: a lifespan hook
+            # that does not return is a worker that never serves.
+            from scribe.services.db_maintenance_scheduler import (
+                get_maintenance_hour,
+                start_db_maintenance_scheduler,
+            )
+            start_db_maintenance_scheduler(
+                asyncio.get_running_loop(), await get_maintenance_hour()
+            )
 
-        # Diagnostic instrumentation — heartbeat, signal handlers, asyncio
-        # exception hook. Cheap (~1 log line/min), high diagnostic value when
-        # the app crashes mysteriously. See services/diagnostics.py.
-        from scribe.services.diagnostics import start_diagnostics
-        start_diagnostics(asyncio.get_running_loop())
+            # Diagnostic instrumentation — heartbeat, signal handlers, asyncio
+            # exception hook. Cheap (~1 log line/min), high diagnostic value when
+            # the app crashes mysteriously. See services/diagnostics.py.
+            from scribe.services.diagnostics import start_diagnostics
+            start_diagnostics(asyncio.get_running_loop())
+        finally:
+            # STARTUP IS OVER — release the backfill (#4181). Everything above
+            # is work the server needs done before it serves; everything the
+            # backfill does is work that can wait for a server already up.
+            logger.info("Startup complete; releasing deferred backfill")
+            _startup_finished.set()
 
     @app.after_serving
     async def shutdown():

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", "list_lessons",
     # 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", "delete_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,280 @@
+"""Lesson MCP tools: a transferable insight, retrievable by situation.
+
+Its own module rather than `create_note(note_type="lesson")`, on the precedent
+of snippets and processes — and for the reason that precedent exists. A kind
+whose value depends on a field being filled needs a door that ASKS for that
+field by name. `create_note` would take a lesson through a generic body
+parameter, and the trigger — the whole of why a lesson is findable at all —
+would be something the writer had to know to include.
+"""
+from __future__ import annotations
+
+from scribe.mcp._context import current_user_id
+from scribe.services import access as access_svc
+from scribe.services import dedup as dedup_svc
+from scribe.services import knowledge as knowledge_svc
+from scribe.services import lessons as lessons_svc
+from scribe.services import systems as systems_svc
+from scribe.services import trash as trash_svc
+from scribe.mcp.tools import systems as systems_tools
+from scribe.services.note_usage import record_pulled
+
+
+def _to_dict(note) -> dict:
+    """A lesson as the tools return it — the composed fields read back out,
+    not the raw row, so a caller sees the same vocabulary it wrote with."""
+    return {
+        "id": note.id,
+        "title": note.title,
+        "body": note.body,
+        "when_to_apply": lessons_svc.lesson_trigger(note),
+        "learned_from": lessons_svc.lesson_sources(note),
+        "tags": list(note.tags or []),
+        "project_id": note.project_id,
+        "note_type": note.note_type,
+        "created_at": note.created_at.isoformat() if note.created_at else None,
+        "updated_at": note.updated_at.isoformat() if note.updated_at else None,
+    }
+
+
+async def list_lessons(
+    q: str = "", tag: str = "", limit: int = 50, offset: int = 0,
+    project_id: int = 0,
+) -> dict:
+    """List lessons — the kind enumerated, rather than only what a query
+    resembles.
+
+    Semantic search is how a lesson REACHES a session; this is how a person or
+    an agent sees what exists. The two answer different questions, and without
+    this one there is no way to ask "what has been learned here at all".
+
+    Each entry carries its trigger, because a list of lessons sorted by title
+    is a list of claims with the situation — the half that says when each one
+    matters — left off.
+
+    Args:
+        q: Free-text search across title + body (optional).
+        tag: Filter to a single tag (optional).
+        limit: Max results (1-100).
+        offset: Skip this many before returning — page past the cap. `total`
+            is the unpaged count, so it says whether more remains.
+        project_id: Narrow to where a lesson was WRITTEN. 0 = every project.
+            A lesson is retrievable from anywhere regardless (step 3); this
+            filters the listing, not the reach.
+
+    Returns {"lessons": [{id, title, when_to_apply, tags, preview}], "total"}.
+    """
+    uid = current_user_id()
+    items, total = await knowledge_svc.query_knowledge(
+        user_id=uid, note_type=lessons_svc.LESSON_NOTE_TYPE,
+        tags=[tag] if tag else [], sort="modified", q=q or None,
+        limit=max(1, min(limit, 100)), offset=max(0, offset),
+        project_id=project_id or None,
+    )
+    labelled = await access_svc.label_shared_items(uid, items)
+    rows = [
+        {
+            "id": it["id"], "title": it["title"], "tags": it.get("tags", []),
+            "preview": it.get("snippet", ""),
+            # Projected by `_note_to_item` straight off the `data` mirror —
+            # absent when the row carries none, rather than an empty string.
+            "when_to_apply": it.get("when_to_apply", ""),
+            **({"shared": True, "owner": it.get("owner")} if it.get("shared") else {}),
+        }
+        for it in labelled
+    ]
+    return {"lessons": rows, "total": total}
+
+
+async def create_lesson(
+    what: str,
+    when_to_apply: str,
+    insight: str = "",
+    learned_from: list[int] | None = None,
+    tags: list[str] | None = None,
+    project_id: int = 0,
+    system_ids: list[int] | None = None,
+    force: bool = False,
+) -> dict:
+    """Record something you LEARNED, so a later session meets it at the moment
+    it applies — on this project or any other.
+
+    A LESSON OR A RULE? The difference is FORCE, not importance. A rule is
+    something that must be followed; a lesson is something worth knowing. If
+    ignoring it would be a mistake, it is a rule (create_rule) and needs the
+    operator's yes, because a rule binds every future session. If ignoring it
+    just means someone re-derives it the slow way, it is a lesson — write it
+    now, and nobody is bound by it.
+
+    That distinction is the whole reason this kind exists. Sessions holding a
+    transferable insight were reaching for create_rule because it was the only
+    surface that is both global and situation-keyed, and proposing rules for
+    things that should never have bound anyone.
+
+    WHAT A LESSON IS NOT: a shape to copy is a SNIPPET (create_snippet); a
+    procedure followed start to finish is a PROCESS (create_process); a record
+    of what happened, findable by topic, is a NOTE (create_note). A lesson is
+    the claim you would want handed to you in the same situation next time.
+
+    `when_to_apply` IS THE RECORD. Everything else is the payload.
+
+    A lesson reaches a session by resembling the SITUATION someone is in, never
+    by topic — that is what separates it from a note, and it is done by putting
+    the trigger in the title and again at the head of the body, so the document
+    is dominated by when it applies. A lesson written without one still saves,
+    still reads correctly in every listing, and will not surface when it is
+    needed. There is nothing to notice afterwards: it looks exactly like a
+    lesson that works.
+
+    So write the SYMPTOM, in the words the situation will present itself in —
+    what someone would be seeing, saying or about to do. "A test fails on code
+    you believe is correct" is a trigger. "Testing" is a topic, and a topic
+    matches everything and surfaces for nothing.
+
+    Args:
+        what: The insight in one line — the claim itself, as you would say it.
+            This becomes the title, joined with the trigger.
+        when_to_apply: The situation this applies in, as a symptom. Required.
+        insight: The body — what to do, and the incident that taught it.
+            Write the story here for the reader; it costs the ranking nothing,
+            because a long body is split into chunks that each still carry the
+            trigger.
+        learned_from: Ids of the issues, tasks or notes this was drawn from —
+            ALL of them. A lesson that generalises three incidents into one
+            claim is the good case, not the edge case, so this is a list.
+        tags: Optional tags.
+        project_id: Where it was learned. Kept as a fact, and it does not limit
+            reach: a lesson is retrievable from every project (that is the
+            point of the kind). 0 = none.
+        system_ids: Systems (subsystems/areas) to file it under.
+        force: Create even if a near-duplicate exists.
+
+    Returns the created lesson. On a near-duplicate, returns the existing id
+    instead of creating — two lessons about one failure class want to be one
+    lesson, so update that one rather than adding a second.
+    """
+    uid = current_user_id()
+    if not when_to_apply or not when_to_apply.strip():
+        raise ValueError(
+            "when_to_apply is required: it is how a lesson is found. Say the "
+            "SYMPTOM — what someone would be seeing, saying or about to do "
+            "when this applies — not the topic it is about. Without it this "
+            "record saves, reads correctly, and never surfaces."
+        )
+
+    sources = lessons_svc.normalize_sources(learned_from)
+    title, body = lessons_svc.lesson_document(
+        what, when_to_apply, insight, sources,
+    )
+    if not force:
+        dup = await dedup_svc.find_duplicate_note(
+            uid, title, body, project_id=project_id or None,
+            is_task=False, note_type=lessons_svc.LESSON_NOTE_TYPE,
+        )
+        if dup is not None:
+            return dedup_svc.duplicate_response(dup, "lesson")
+
+    note = await lessons_svc.create_lesson(
+        uid, what=what, when_to_apply=when_to_apply, insight=insight,
+        learned_from=sources, tags=tags, project_id=project_id or None,
+    )
+    if system_ids:
+        await systems_svc.set_record_systems(uid, note.id, system_ids)
+    data = _to_dict(note)
+    await systems_tools.attach_systems(uid, uid, data, note.id, project_id or None)
+    return data
+
+
+async def get_lesson(lesson_id: int) -> dict:
+    """Fetch one lesson by id, with its trigger and sources read back out."""
+    uid = current_user_id()
+    note = await lessons_svc.get_lesson(uid, lesson_id)
+    if note is None:
+        raise ValueError(f"lesson {lesson_id} not found")
+    out = _to_dict(note)
+    out.update(await access_svc.describe_provenance(uid, note))
+    # Every explicit open records the pull. A lesson is surfaced by the same
+    # retrieval as any other note, so a getter that records nothing would leave
+    # the kind permanently at zero pulls — reading as dead weight beside kinds
+    # that merely had a counter (#2476, the repeat of #2245).
+    record_pulled(user_id=uid, note_id=int(note.id), source="mcp_get_lesson")
+    return out
+
+
+async def update_lesson(
+    lesson_id: int,
+    what: str = "",
+    when_to_apply: str = "",
+    insight: str = "",
+    learned_from: list[int] | None = None,
+    tags: list[str] | None = None,
+) -> dict:
+    """Update a lesson. Empty fields are left unchanged.
+
+    REWORDING A LESSON IS ORDINARY WORK. Understanding improves, and a trigger
+    that turned out to fire on the wrong situation is the single most valuable
+    thing to fix here — a lesson nobody is reaching is usually not wrong, it is
+    keyed to a situation nobody is in.
+
+    Title, body and the indexed mirror are re-composed together from the merged
+    fields, so a partial update cannot leave the trigger saying one thing in
+    the title and another in the body.
+
+    Args:
+        lesson_id: Lesson to update.
+        what: New one-line claim. Empty leaves unchanged.
+        when_to_apply: New trigger, as a symptom. Empty leaves unchanged.
+        insight: New body. Empty leaves unchanged.
+        learned_from: Replace the source ids. None leaves unchanged; pass the
+            FULL list, including the ones already there.
+        tags: Replace tags. None leaves unchanged.
+    """
+    uid = current_user_id()
+    note = await lessons_svc.update_lesson(
+        uid, lesson_id,
+        what=what or None,
+        when_to_apply=when_to_apply or None,
+        insight=insight or None,
+        learned_from=learned_from,
+        tags=tags,
+    )
+    if note is None:
+        raise ValueError(f"lesson {lesson_id} not found")
+    return _to_dict(note)
+
+
+async def delete_lesson(lesson_id: int) -> dict:
+    """Retire a lesson — it moves to the trash and is recoverable.
+
+    Reach for this when a lesson turned out to be wrong, or was superseded by
+    a better one. A lesson that is merely NOT REACHING anyone is usually not a
+    deletion: its trigger is keyed to a situation nobody is in, and rewording
+    that with update_lesson keeps what was learned.
+
+    Deletion was always possible through `delete_note` — a lesson is a note and
+    the trash is kind-agnostic — but nothing said so, and a kind whose own
+    tools offer create/read/update reads as one you cannot retire (#2250, which
+    recorded exactly this for processes).
+    """
+    uid = current_user_id()
+    note = await lessons_svc.get_lesson(uid, lesson_id)
+    # The KIND is checked before deleting: this tool is reached for by name, so
+    # letting it trash an ordinary note because the id happened to resolve
+    # would be a destructive action taken on a mistyped argument.
+    if note is None:
+        raise ValueError(f"lesson {lesson_id} not found")
+    batch = await trash_svc.delete(uid, "note", lesson_id)
+    if batch is None:
+        raise ValueError(f"lesson {lesson_id} not found")
+    return {
+        "deleted_batch_id": batch,
+        "message": (
+            f"Lesson {lesson_id} moved to trash. Restore with restore('{batch}')."
+        ),
+    }
+
+
+def register(mcp) -> None:
+    for fn in (list_lessons, create_lesson, get_lesson, update_lesson,
+               delete_lesson):
+        mcp.tool(name=fn.__name__)(fn)

src/scribe/mcp/tools/notes.py

@@ -310,8 +310,14 @@ async def find_duplicate_records(kind: str = "note", threshold: float = 0.0) ->
     project's records, or when you suspect the same ground was covered twice.
 
     Args:
-        kind: "note" (documents) or "task". Snippets have their own report,
+        kind: "note" (documents), "task", or "lesson". Each kind is compared
+            only against its own — a to-do and a write-up are not alternatives
+            to each other. Snippets have their own report,
             find_duplicate_snippets, whose groups propose a lossless merge.
+            LESSONS are the kind most worth running this on: their create gate
+            is deliberately permissive (it sits above the band where two
+            genuinely different lessons about one area would block each other),
+            so this report is where that tolerance gets reviewed.
         threshold: Similarity floor, 0-1. 0 uses the configured setting.
 
     READ THE `suggestion` FIELD BEFORE ACTING, because the right fix differs by
@@ -324,9 +330,15 @@ async def find_duplicate_records(kind: str = "note", threshold: float = 0.0) ->
     question.
     """
     uid = current_user_id()
-    if kind not in ("note", "task"):
-        raise ValueError('kind must be "note" or "task" — snippets have their '
-                         "own report, find_duplicate_snippets")
+    # Validated against the service's own list rather than a second copy here,
+    # so a kind added there reaches this door instead of being refused by a
+    # literal nobody remembered to widen.
+    allowed = tuple(k for k in dedup_svc._REPORT_KINDS if k != "snippet")
+    if kind not in allowed:
+        raise ValueError(
+            f"kind must be one of {', '.join(allowed)} — snippets have their "
+            "own report, find_duplicate_snippets"
+        )
     return await dedup_svc.find_duplicate_records(
         uid, kind=kind, threshold=threshold if threshold > 0 else None,
     )

src/scribe/models/__init__.py

@@ -3,11 +3,39 @@ from sqlalchemy.orm import DeclarativeBase
 
 from scribe.config import Config
 
+# Named rather than inlined so the deadline below is READABLE. SQLAlchemy
+# captures `connect_args` in a closure and merges it at connect time, so an
+# inline dict cannot be recovered from the engine — and a guard that cannot
+# read the value it guards is a guard that passes forever.
+_CONNECT_ARGS: dict = {
+    # A DEADLINE ON ESTABLISHING A CONNECTION (#4181, rule 156).
+    #
+    # asyncpg's `timeout` bounds the CONNECT — the TCP handshake plus session
+    # setup — and nothing else. Against a host whose storage has wedged, that
+    # handshake does not fail, it waits, and without this the wait is
+    # asyncpg's own 60-second default: the entire lifespan budget spent before
+    # a single query is even sent. Ten seconds is far longer than a healthy
+    # local connect (single-digit milliseconds) and short enough to leave room
+    # to fail usefully rather than be killed.
+    #
+    # WHAT THIS DOES NOT COVER, said plainly so the next reader doesn't assume
+    # it does: a query on an already-open connection, which includes
+    # `pool_pre_ping`'s liveness check. Bounding those is `command_timeout`,
+    # and that is deliberately NOT set here — it would apply to every
+    # statement, and this app legitimately runs long ones (the embedding
+    # backfills, VACUUM ANALYZE). A blanket statement deadline would trade
+    # this failure mode for a worse one. Callers that must not hang — the
+    # lifespan hook above all — bound their own await instead; see
+    # `_STARTUP_READ_TIMEOUT` in services/db_maintenance_scheduler.py.
+    "timeout": 10,
+}
+
 engine = create_async_engine(
     Config.DATABASE_URL,
     echo=False,
     pool_pre_ping=True,
     pool_recycle=1800,
+    connect_args=_CONNECT_ARGS,
 )
 async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
 

src/scribe/services/db_maintenance_scheduler.py

@@ -25,10 +25,62 @@ logger = logging.getLogger(__name__)
 _JOB_ID = "db_maintenance_vacuum"
 _DEFAULT_HOUR = 4
 
+# HOW LONG A COLD START WILL WAIT FOR THIS ONE SETTING (#4181).
+#
+# This read is the FIRST database call in the app's `before_serving` hook, and
+# a lifespan hook that does not return is a worker that never serves. On
+# 2026-09-19 a host storage stall made one Postgres checkpoint of 14 buffers
+# take 281 seconds against a 1.3-second baseline; the app restarted into the
+# tail of it, this query hung with no deadline, Hypercorn killed the worker at
+# its 60-second lifespan timeout, and nothing retries a failed lifespan. A
+# five-minute disk hiccup became a three-hour outage that only a human restart
+# could clear.
+#
+# Three seconds because the honest requirement is "don't hold up the boot",
+# not "get the right hour". The value is one small indexed row on a local
+# database: under any healthy condition this returns in single-digit
+# milliseconds, so the timeout can only ever fire when something is already
+# badly wrong — which is exactly the moment the app must come up anyway.
+_STARTUP_READ_TIMEOUT = 3.0
+
 
 async def get_maintenance_hour() -> int:
-    """The configured run-hour (UTC, 0–23), clamped; default 04:00."""
-    raw = await get_admin_setting("db_maintenance_hour", str(_DEFAULT_HOUR))
+    """The configured run-hour (UTC, 0–23), clamped; default 04:00.
+
+    BOUNDED, because the caller is a lifespan hook (rule 156). The fallback is
+    not new behaviour invented for the timeout — this function already answers
+    `_DEFAULT_HOUR` for a value it cannot parse, and a database that will not
+    answer in three seconds is the same class of "no usable value here". What
+    changes is that the failure is now a logged line and a default hour rather
+    than the application failing to start.
+
+    Degrading to the default is the right trade in both directions: the cost of
+    being wrong is that a VACUUM runs at 04:00 instead of the configured hour,
+    for one boot, on an instance whose disk is in trouble. The cost of waiting
+    is the whole instance.
+    """
+    try:
+        raw = await asyncio.wait_for(
+            get_admin_setting("db_maintenance_hour", str(_DEFAULT_HOUR)),
+            timeout=_STARTUP_READ_TIMEOUT,
+        )
+    except (TimeoutError, asyncio.TimeoutError):
+        # WARNING, not debug: this is never normal, and it is the breadcrumb
+        # that would have named #4181 in seconds instead of requiring
+        # Postgres's own log to be read.
+        logger.warning(
+            "db maintenance: reading db_maintenance_hour exceeded %.1fs; "
+            "starting with the default %02d:00 UTC. The database is slow or "
+            "unreachable — this is a symptom, not the disease.",
+            _STARTUP_READ_TIMEOUT, _DEFAULT_HOUR,
+        )
+        return _DEFAULT_HOUR
+    except Exception:
+        logger.warning(
+            "db maintenance: could not read db_maintenance_hour; starting "
+            "with the default %02d:00 UTC", _DEFAULT_HOUR, exc_info=True,
+        )
+        return _DEFAULT_HOUR
     try:
         hour = int(raw)
     except (TypeError, ValueError):

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
-                           if note_type == SNIPPET_NOTE_TYPE else _SEMANTIC_THRESHOLD),
+                threshold=_semantic_threshold(note_type),
                 # 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
@@ -341,8 +374,24 @@ DUPLICATE_THRESHOLD_KEYS = {
     "snippet": "kb_duplicate_threshold_snippet",
     "note": "kb_duplicate_threshold_note",
     "task": "kb_duplicate_threshold_task",
+    "lesson": "kb_duplicate_threshold_lesson",
+    "process": "kb_duplicate_threshold_process",
+}
+# The lesson default is the GENERAL semantic floor, and deliberately below its
+# own write-path bar. The gate sits at 0.96 so it does not refuse two genuinely
+# different lessons whose triggers read alike — and that tolerance is precisely
+# what wants reviewing later. So the report looks at the band the gate was told
+# to let through. Safe here and not at the gate, because a report proposes and
+# the operator picks, where the gate blocks a write outright.
+DUPLICATE_DEFAULT_THRESHOLDS = {
+    "snippet": 0.82, "note": 0.93, "task": 0.93, "lesson": 0.90,
+    # A process is prose like a note, and its gate is the general one, so it
+    # takes the note's floor. It is here because every typed kind earns a
+    # report — a kind with create/read/update/delete and no way to ask "did we
+    # record this twice" is one whose duplicates are only ever found by
+    # accident.
+    "process": 0.93,
 }
-DUPLICATE_DEFAULT_THRESHOLDS = {"snippet": 0.82, "note": 0.93, "task": 0.93}
 # Hard cap on returned pairs. A pathologically uniform corpus is O(n²) pairs, and
 # a report nobody can read is not a report.
 _MAX_DUPLICATE_PAIRS = 200
@@ -474,22 +523,47 @@ _KIND_SUGGESTION = {
         "a work-log, and cancel the other with a pointer. If one CORRECTS the "
         "other's conclusions, supersession also works for tasks."
     ),
+    "lesson": (
+        "Read both TRIGGERS before anything else — a lesson is retrieved by "
+        "the situation it names, so two alike insights under different "
+        "triggers are two lessons and belong apart. Same trigger, same "
+        "teaching means one lesson learned twice: keep the clearer one, "
+        "update_lesson it with the union of both `taught_by` lists and "
+        "anything the other said that it doesn't, then delete_lesson the "
+        "other. The sources are the point — a lesson that loses an incident "
+        "loses its evidence."
+    ),
+    "process": (
+        "A process arrives at the agent as a skill, so two alike processes "
+        "compete for the same moment and whichever wins retrieval is the one "
+        "that runs — a duplicate here executes the wrong procedure rather "
+        "than merely cluttering a list. Keep the one actually in use, "
+        "update_process it with any step the other has, then delete_process "
+        "the loser. If they are genuinely different procedures that share "
+        "vocabulary, sharpen the titles instead so the right one wins."
+    ),
 }
 
 # kind → the Note-model predicate for BOTH sides of the self-join. Tasks are
 # notes with a status, not a note_type of their own — the same split every
 # list surface makes.
-_REPORT_KINDS = ("snippet", "note", "task")
+_REPORT_KINDS = ("snippet", "note", "task", "lesson", "process")
 
 
 def _kind_clauses(kind: str, note_alias):
     """The WHERE terms that make an aliased Note row one `kind` of record."""
     if kind == "snippet":
         return (note_alias.note_type == SNIPPET_NOTE_TYPE,)
+    if kind == "lesson":
+        return (note_alias.note_type == LESSON_NOTE_TYPE,)
+    if kind == "process":
+        return (note_alias.note_type == "process",)
     if kind == "task":
         return (note_alias.note_type == "note", note_alias.status.isnot(None))
-    # kind == "note": documents only — a task is a note with a status, and
-    # mixing them would propose folding a to-do into a write-up.
+    # kind == "note": documents only. A task is a note with a status, and
+    # mixing them would propose folding a to-do into a write-up; the typed
+    # kinds are excluded by the note_type equality for the same reason, so
+    # each kind is only ever compared against its own.
     return (note_alias.note_type == "note", note_alias.status.is_(None))
 
 

src/scribe/services/knowledge.py

@@ -240,6 +240,15 @@ def _note_to_item(note: Note) -> dict:
     if language:
         item["language"] = language
 
+    # WHEN a lesson applies, same reasoning as the language above: a plain
+    # projection of the `data` mirror, no body parsing. A listing of lessons
+    # without it is a list of claims with the situation — the half that says
+    # when each one matters — left off, and the title's own copy cannot be
+    # recovered by splitting on the em dash, because a claim may contain one.
+    trigger = (note.data or {}).get("when_to_apply") if note.data else None
+    if trigger:
+        item["when_to_apply"] = trigger
+
     verdict = (note.data or {}).get("verification") if note.data else None
     if verdict and verdict.get("status"):
         item["verification"] = {

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()

src/scribe/services/plugin_context.py

@@ -28,6 +28,7 @@ from scribe.services import shape_ledger as shape_ledger_svc
 from scribe.services import snippets as snippets_svc
 from scribe.services.access import label_shared_items, owner_names_for
 from scribe.services.embeddings import semantic_search_notes, semantic_search_rules
+from scribe.services.lessons import LESSON_NOTE_TYPE
 from scribe.services.note_usage import record_surfaced
 from scribe.services.rule_usage import record_rule_surfaced
 from scribe.services.supersession import superseded_ids
@@ -759,6 +760,122 @@ async def _reserve_slot_for_reuse(
     return (kept + fresh)[:top_k]
 
 
+async def _reserve_slot_for_lesson(
+    user_id: int,
+    query: str,
+    kept: list,
+    cfg: dict,
+    *,
+    project_id: int | None,
+    already: set[int],
+) -> tuple[list, int | None]:
+    """Guarantee a lesson one slot, if one clears the bar (milestone 385 step 5).
+
+    WHY A SLOT, AND WHAT IT ACTUALLY DISPLACES
+
+    The step's own framing was that "a slot spent on a lesson is a slot not
+    spent on a rule that binds". That is not what happens here, and the
+    correction matters for judging the cost: the notes menu and the rule hints
+    are separate functions with separate budgets, composed by the caller
+    (`build_prompt_rule_hint` says why). A line reserved in THIS menu displaces
+    a note, a snippet or an issue — never a rule.
+
+    THE ASYMMETRY, which is `preference_slot`'s argument on a different corpus:
+
+      - a NOTE crowded out of this menu is a lost convenience. It stays
+        searchable, and the operator can ask for it.
+      - a RULE crowded out still fires at an act arm. The prompt hit is a
+        preview of a second chance.
+      - a LESSON crowded out is the feature failing. A lesson exists only to be
+        met at the moment it applies — nobody browses lessons looking for one —
+        so the arm that surfaces it IS its delivery, and the loss is total and
+        silent. Silent delivery failure is the exact shape #3727 recorded: an
+        insight with no home arrived as a rule proposal instead.
+
+    And the ratio only moves one way. Lessons are by design rare and hard-won
+    while project records grow with the work, which is the 200:1 problem
+    `reuse_slot` was built for (#2246), before it has had a chance to be
+    measured here.
+
+    THE SLOT BUYS POSITION, NOT A LOWER BAR. It reserves at the menu's own
+    threshold, so a weak lesson cannot buy the line and silence stays the
+    default — the discipline both existing slots keep.
+
+    IT EXTENDS, IT NEVER DISPLACES, siding with `preference_slot` over
+    `reuse_slot`. Two reasons, and the second is the one that would be hard to
+    recover later: a displaced hit was returned by the general search and sits
+    in that call's `retrieval_logs` row, so evicting it makes the two tables
+    disagree about the same call for a reason nothing in the data explains
+    (#3668, and milestone #379 is what that costs). The first is voice — a
+    lesson does not bind, and a record that does not bind should not be able
+    to throw a better-scoring one off the menu.
+
+    Returns the possibly-extended list and the id the slot spent. The caller
+    needs that id to keep each source's surfaced set matching its own log row:
+    this slot records its own surfacing under its own name, so counting it
+    again under `auto_inject` would double it.
+    """
+    if any(_record_kind(n) == LESSON_NOTE_TYPE for _s, n in kept):
+        return kept, None            # a lesson already placed on score
+
+    _t0 = time.perf_counter()
+    _rep: dict = {}
+    # KIND-FILTERED, so the slot can only ever be spent on what it is for —
+    # `preference_slot`'s reasoning: verifying the kind after an open search
+    # would let a stray note buy the line, and that line would be
+    # indistinguishable from one that earned its place.
+    #
+    # `include_global_kinds` is the half that makes a lesson reachable at all
+    # from a project it was not written on, which is this kind's whole claim
+    # (#3730). Without it the slot would be a guarantee that silently only
+    # applies to lessons learned here.
+    found = await semantic_search_notes(
+        user_id, query,
+        limit=1,
+        threshold=cfg["threshold"],
+        project_id=project_id,
+        exclude_ids={int(n.id) for _s, n in kept},
+        note_type=(LESSON_NOTE_TYPE,),
+        include_global_kinds=True,
+        scope="browse",
+        report=_rep,
+    )
+    fresh = [(s, n) for s, n in found if int(n.id) not in already]
+    # ITS OWN SOURCE, from the first deploy. This slot is a claim that a kind
+    # deserves a guaranteed line, and a claim like that has to be falsifiable:
+    # `best_available_id` (#3807) names the lesson a bar refused, and the
+    # result count says how often the guarantee was actually spent. Without
+    # this row the question "does the lesson slot earn its line?" would have no
+    # data behind it in either direction — which is #2463's finding, recorded
+    # about the slot that shipped without one.
+    record_retrieval(
+        user_id=user_id, source="lesson_slot", query=query,
+        threshold=cfg["threshold"], limit=1, project_id=project_id,
+        is_task=None, results=fresh,
+        best_available=_rep.get("best_available_score"),
+        best_available_id=_rep.get("best_available_id"),
+        searched=bool(_rep.get("searched", True)),
+        suppressed=len(found) - len(fresh),
+        duration_ms=(time.perf_counter() - _t0) * 1000.0,
+    )
+    kept_ids = {int(n.id) for _s, n in kept}
+    slot = [
+        (s, n) for s, n in found
+        if _record_kind(n) == LESSON_NOTE_TYPE and int(n.id) not in kept_ids
+    ][:1]
+    if not slot:
+        return kept, None
+
+    slot_id = int(slot[0][1].id)
+    # FRESH ONLY, matching the row above: a ledger repeat is rendered (#4101)
+    # but is not a new surfacing, so this source's two tables stay identical.
+    if slot_id not in already:
+        record_surfaced(
+            user_id=user_id, note_ids=[slot_id], source="lesson_slot",
+        )
+    return kept + slot, slot_id
+
+
 async def build_autoinject_hint(
     user_id: int,
     query: str,
@@ -811,6 +928,14 @@ async def build_autoinject_hint(
         limit=cfg["top_k"],
         threshold=cfg["threshold"],
         project_id=(project_id or None),
+        # LESSONS ARE PROJECT-INDEPENDENT (#3730), so they join this menu's
+        # candidate set from wherever they were learned. Widening it is what
+        # makes the reserved slot below falsifiable rather than decorative: if
+        # the slot were the only path a lesson had, the general contest would
+        # be permanently closed to the kind and "the slot earns its line" would
+        # be true by construction. The switch adds nothing else — it ORs in
+        # `GLOBAL_NOTE_TYPES` and no other kind is in it.
+        include_global_kinds=True,
         # Injection is the one retrieval nobody asked for, so it takes the BROWSE
         # scope: never a record shared one-to-one with the operator. What can
         # still appear is a collaborator's note inside a shared project — legible
@@ -850,6 +975,14 @@ async def build_autoinject_hint(
         user_id, q, kept, cfg, project_id=(project_id or None),
         already=already,
     )
+    # AFTER the reuse slot, because that one evicts the menu's weakest hit
+    # while this one extends: running them the other way round would let a
+    # reserved lesson be the line reuse throws off, and a slot that another
+    # slot can silently undo is not a guarantee.
+    kept, lesson_slot_id = await _reserve_slot_for_lesson(
+        user_id, q, kept, cfg, project_id=(project_id or None),
+        already=already,
+    )
 
     # A collaborator's note can reach this menu via a shared project, and the
     # operator never asked for it — so say whose it is. Unattributed, it reads as
@@ -867,10 +1000,34 @@ async def build_autoinject_hint(
     # once, rather than each repeated line having to explain itself.
     lines = [
         "> Possibly relevant from your Scribe records — open any in full with "
-        "`get_note(id)`, or `get_snippet` / `get_process` for those kinds "
-        "(titles only; a line marked `seen` was surfaced earlier this session "
-        "and may no longer be in context):",
+        "`get_note(id)`, or `get_snippet` / `get_process` / `get_lesson` for "
+        "those kinds (titles only; a line marked `seen` was surfaced earlier "
+        "this session and may no longer be in context):",
     ]
+    # THE REGISTER, SAID ONCE AND ONLY WHEN IT APPLIES (milestone 385 step 5).
+    #
+    # The operator's requirement for this kind was "they don't always have to
+    # be followed", and the risk is not that a reader mistakes a lesson for a
+    # rule — this menu's voice is already the non-binding one, deliberately
+    # (see the `seen` marker below). The risk is the opposite: read as one more
+    # title in a list of MATERIAL, a lesson looks like something to open if
+    # curious, when it is advice someone paid for. The line has to say "weigh
+    # this" without acquiring the rule arms' "before deciding it does not
+    # apply", which binds.
+    #
+    # In the HEADER rather than on each line, for the reason the `seen` flag is
+    # a flag: the meaning is the same for every lesson on the menu, and a
+    # clause repeated per line would cost more than it says. Conditional
+    # because a menu with no lesson should not pay for the sentence, and
+    # because a header that explains an absent kind reads as boilerplate —
+    # which is how a reader learns to skip headers.
+    if any(_record_kind(n) == LESSON_NOTE_TYPE for _s, n in kept):
+        lines.append(
+            "> A line marked `lesson` is something an earlier session learned "
+            "the hard way, kept because it should transfer. Weigh it against "
+            "what you are doing and use your judgement — a lesson is not a "
+            "rule and binds nothing."
+        )
     # A superseded record is DEMOTED, not removed (#278) — so one can still reach
     # this menu, and when it does the reader has to be told. An agent handed
     # stale material with nothing marking it acts on it with full confidence,
@@ -906,9 +1063,18 @@ async def build_autoinject_hint(
     # make this table disagree with `retrieval_logs` about the same call —
     # #3668's identity, which is the cheapest true statement available about
     # this pair of tables and is not worth a marker's convenience.
+    # THE RESERVED LESSON IS NOT THIS ARM'S SURFACING. It was fetched by its own
+    # query and already recorded under `lesson_slot`, so counting it here would
+    # book one delivery twice and leave `lesson_slot`'s two tables describing
+    # different numbers of the same event — #3668's identity, which is the
+    # cheapest true statement available about this pair of tables. It stays in
+    # `note_ids`, which is the session LEDGER and must list every line rendered.
     record_surfaced(
         user_id=user_id,
-        note_ids=[i for i in note_ids if i not in already],
+        note_ids=[
+            i for i in note_ids
+            if i not in already and i != lesson_slot_id
+        ],
         source="auto_inject",
     )
 
@@ -1709,8 +1875,28 @@ async def build_write_path_hint(
             # written and answers nothing; an ISSUE is corrective work with a
             # root cause in it, and a non-task note is durable knowledge. Both
             # earned their place; a todo did not.
-            note_type=("snippet", "note"),
+            #
+            # AND LESSONS (milestone 385 step 5). This arm is kind-FILTERED, so
+            # a kind absent from this tuple is not merely outranked here — it
+            # is unreachable, and nothing reports an arm that never had the
+            # candidate (#3702). The founding example of the kind is a lesson
+            # about a code shape ("two absolutely-positioned siblings"), which
+            # is the moment this arm fires and no other.
+            #
+            # NO RESERVED SLOT HERE, unlike the prompt menu. This arm fires
+            # before EVERY Write and Edit, where a guaranteed extra line is a
+            # guaranteed extra interruption per keystroke-batch — the same
+            # argument that keeps the act arms' budgets tight. And the contest
+            # is already fair: the field is snippets, issues and lessons rather
+            # than the whole corpus, so the 200:1 dilution a slot answers is
+            # not what happens here. Lessons surfaced by this arm are
+            # identifiable in the telemetry by their kind.
+            note_type=("snippet", "note", LESSON_NOTE_TYPE),
             task_kind="issue",
+            # Project-independent, for the reason the prompt menu passes it:
+            # a lesson's claim is that it transfers, and an arm scoped to the
+            # project it was written on cannot test that claim (#3730).
+            include_global_kinds=True,
             # Same reasoning as auto-inject: nobody asked for this, so it takes
             # the browse scope and never surfaces a one-to-one direct share.
             scope="browse",
@@ -1803,6 +1989,13 @@ async def build_write_path_hint(
                     marker,
                     {
                         "id": int(note.id), "title": note.title, "user_id": note.user_id,
+                        # Carried, not re-read off the rendered marker. The
+                        # marker is prose assembled for a human and it already
+                        # varies by kind, language and the `seen` flag — a
+                        # header that decided what to say by matching substrings
+                        # in it would break the next time a marker is reworded,
+                        # silently and in the direction of saying nothing.
+                        "kind": kind,
                         # Carried so the line can disclose a cross-language hit
                         # (#2244). The semantic arm is where these actually arise —
                         # a snippet recorded at the path you're editing is almost
@@ -1924,11 +2117,23 @@ async def build_write_path_hint(
         lines.append(
             f"> Prior art already recorded in Scribe for `{path}` — open one with "
             "`get_snippet(id)` for a snippet, `get_task(id)` for an issue, "
-            "`get_note(id)` otherwise. Reuse a snippet rather than writing a fresh "
-            "one-off; read an issue before repeating what it records "
+            "`get_lesson(id)` for a lesson, `get_note(id)` otherwise. Reuse a "
+            "snippet rather than writing a fresh one-off; read an issue before "
+            "repeating what it records "
             "(titles only; a line marked `seen` was surfaced earlier this "
             "session and may no longer be in context):"
         )
+    # The same clause the prompt menu carries, on the same condition and for
+    # the same reason: this menu's three other kinds are all things that WERE
+    # done here, and a lesson is the one line that is advice. Said once, only
+    # when one is on the menu.
+    if any(i.get("kind") == LESSON_NOTE_TYPE for i, _m, _o, _l in rendered):
+        lines.append(
+            "> A line marked `lesson` is something an earlier session learned "
+            "the hard way, kept because it should transfer. Weigh it against "
+            "the code you are about to write and use your judgement — a lesson "
+            "is not a rule and binds nothing."
+        )
     # Say what a language tag MEANS, and only when one is actually on the menu.
     # Without this the reader has to infer why "· python" is attached to a hit on
     # a .ts file, and the two ways of guessing wrong are both bad: dismiss it as

src/scribe/services/retrieval_surfaces.py

@@ -209,12 +209,20 @@ SURFACES: dict[str, Surface] = {
     ),
 }
 
-# Reserved slots are deliberately absent. `preference_slot` and `reuse_slot`
-# borrow their parent arm's floor and are hard-limited to one hit each, because
-# their entire purpose is to guarantee a single line to a kind of record that
-# keeps losing a general score contest (#2246, #3894). A budget of "1" is the
-# feature; exposing it as tunable would invite setting it to 0 and silently
-# removing the guarantee.
+# Reserved slots are deliberately absent. `preference_slot`, `reuse_slot` and
+# `lesson_slot` borrow their parent arm's floor and are hard-limited to one hit
+# each, because their entire purpose is to guarantee a single line to a kind of
+# record that keeps losing a general score contest (#2246, #3894) — or, for
+# `lesson_slot`, to a kind whose loss is total rather than merely a lost
+# convenience, since a lesson has no act arm to fall back on and nobody browses
+# lessons looking for one (milestone 385). A budget of "1" is the feature;
+# exposing it as tunable would invite setting it to 0 and silently removing the
+# guarantee.
+#
+# They are still logged under their own `source`, which is what keeps them
+# judgeable without being tunable: `best_available_id` names the record each
+# bar refused, so a slot that never places, or one that places weak hits, shows
+# up as evidence rather than as an argument.
 
 
 def surface_names() -> list[str]:

tests/test_lesson_surfacing.py

@@ -0,0 +1,305 @@
+"""A lesson reaches the session it applies to — milestone 385, step 5.
+
+Three decisions are guarded here, and each has a way of going quietly wrong
+that these tests are shaped to catch rather than to describe:
+
+  WHICH ARM. The two note arms, and no new one. An arm that FILTERS kinds does
+  not merely outrank a kind it omits — it makes it unreachable, and nothing
+  reports an arm that never had the candidate (#3702). The write path filters;
+  the prompt menu does not, but is project-scoped, which for a kind whose whole
+  claim is that it transfers amounts to the same silence.
+
+  WHOSE BUDGET. A reserved slot in the prompt menu, none on the write path.
+  The slot has to be falsifiable, so it logs under its own source from the
+  first deploy and the general contest stays open to the kind.
+
+  THE VOICE. "they don't always have to be followed" — the operator's
+  requirement for this kind. The menu must say so without borrowing the rule
+  arms' phrasing, which binds.
+"""
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from scribe.services import plugin_context as pc
+from scribe.services.lessons import LESSON_NOTE_TYPE
+from tests.helpers import fake_note, writepath_cfg
+
+pytestmark = pytest.mark.usefixtures("_no_supersession")
+
+_CFG = {"enabled": True, "threshold": 0.55, "top_k": 3}
+
+# The phrase the rule arms use. A lesson line that acquired it would be a rule
+# with a different table, which is the thing this milestone exists to avoid.
+_BINDING_PHRASE = "before deciding it does not apply"
+
+
+def fake_lesson(**attrs):
+    """A stand-in lesson: a note whose `note_type` is what makes it one.
+
+    The title carries the trigger because `compose_title` builds it that way —
+    `{what} — {when it applies}` — so a menu line rendering only the title is
+    already showing the reader when this lesson applies. Tests that used a bare
+    title here would be testing a record the product cannot create.
+    """
+    attrs.setdefault(
+        "title",
+        "Give absolutely-positioned siblings an explicit stacking order — "
+        "placing two absolutely-positioned elements in the same area",
+    )
+    attrs.setdefault("data", {"when_to_apply": "two absolute siblings overlap"})
+    return fake_note(note_type=LESSON_NOTE_TYPE, **attrs)
+
+
+async def _menu(main_hits, *, lesson_hits=None, reuse_hits=None, cfg=None,
+                exclude_ids=None, rec=None, surf=None):
+    """Run the prompt menu with each query stubbed by the kinds it asks for."""
+    calls: list[dict] = []
+
+    async def fake_search(*_a, **kw):
+        calls.append(kw)
+        kinds = tuple(kw.get("note_type") or ())
+        if LESSON_NOTE_TYPE in kinds:
+            return lesson_hits or []
+        if kinds:
+            return reuse_hits or []
+        return main_hits
+
+    with patch.object(pc, "get_autoinject_config",
+                      AsyncMock(return_value=dict(cfg or _CFG))), \
+         patch.object(pc, "semantic_search_notes", AsyncMock(side_effect=fake_search)), \
+         patch.object(pc, "record_retrieval", rec or MagicMock()), \
+         patch.object(pc, "record_surfaced", surf or MagicMock()), \
+         patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
+        out = await pc.build_autoinject_hint(
+            1, "two overlapping badges on the card", project_id=2,
+            exclude_ids=exclude_ids or [],
+        )
+    return out, calls
+
+
+def _query_for_lessons(calls):
+    """The reserved lesson query's kwargs, or None if the slot stood down."""
+    return next(
+        (c for c in calls
+         if LESSON_NOTE_TYPE in tuple(c.get("note_type") or ())),
+        None,
+    )
+
+
+# ── which arm: reachability before ranking ───────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_the_prompt_menu_looks_beyond_the_bound_project_for_a_lesson():
+    """A lesson's claim is that it transfers. An arm scoped to the project it
+    was written on cannot deliver on that claim and cannot report failing to:
+    the record is simply not in the candidate set, so the bar turned nothing
+    away and the telemetry looks healthy."""
+    _out, calls = await _menu([(0.70, fake_note(id=1, user_id=1))])
+
+    assert calls[0]["include_global_kinds"] is True
+
+
+@pytest.mark.asyncio
+async def test_the_write_path_asks_for_lessons_at_all():
+    """The founding example of the kind is a lesson about a code shape, and
+    this is the arm that fires when code is written. It is also the one note
+    arm that filters kinds — so an omission here is not a ranking loss, it is
+    a kind that can never appear."""
+    search = AsyncMock(return_value=[])
+    with patch.object(pc, "get_writepath_config",
+                      AsyncMock(return_value=writepath_cfg(threshold=0.6))), \
+         patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
+         patch.object(pc, "semantic_search_notes", search), \
+         patch.object(pc, "record_retrieval", MagicMock()), \
+         patch.object(pc, "record_surfaced", MagicMock()), \
+         patch.object(pc, "owner_names_for", AsyncMock(return_value={})), \
+         patch.object(pc, "concept_query", MagicMock(return_value="stack two badges")):
+        await pc.build_write_path_hint(1, "src/ui/Badge.vue", code="x" * 400)
+
+    kw = search.await_args.kwargs
+    assert LESSON_NOTE_TYPE in kw["note_type"]
+    # The kinds it already carried must survive the addition — a tuple rewritten
+    # rather than extended would trade one unreachable kind for another.
+    assert {"snippet", "note"} <= set(kw["note_type"])
+    assert kw["include_global_kinds"] is True
+
+
+# ── whose budget: a reserved slot that can be judged ─────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_a_lesson_takes_a_reserved_slot_and_the_query_is_logged():
+    """The slot is a claim that a kind deserves a guaranteed line. A claim like
+    that has to be falsifiable from the first deploy, which means its own query
+    in `retrieval_logs` — #2463's finding, recorded about the slot that shipped
+    without one."""
+    rec = MagicMock()
+    out, calls = await _menu(
+        [(0.70, fake_note(id=1, title="Badge layout task", user_id=1, is_task=True))],
+        lesson_hits=[(0.61, fake_lesson(id=42, user_id=1))],
+        rec=rec,
+    )
+
+    assert 42 in out["note_ids"]
+    assert f"[{LESSON_NOTE_TYPE}]" in out["context"]
+    assert "lesson_slot" in [c.kwargs["source"] for c in rec.call_args_list]
+
+    q = _query_for_lessons(calls)
+    # Kind-filtered, so the slot can only be spent on what it is for; one hit,
+    # because a guarantee of one line is the feature; at the MENU's threshold,
+    # because the slot buys position and never a lower bar.
+    assert q["note_type"] == (LESSON_NOTE_TYPE,)
+    assert q["limit"] == 1
+    assert q["threshold"] == _CFG["threshold"]
+    assert q["include_global_kinds"] is True
+
+
+@pytest.mark.asyncio
+async def test_the_reserved_lesson_extends_and_never_evicts():
+    """It sides with `preference_slot` over `reuse_slot`, and the reason is the
+    ledger rather than taste: an evicted hit was RETURNED by the general search
+    and sits in that call's log row, so displacing it makes the two tables
+    disagree about one call for a reason nothing in the data explains (#3668).
+    The voice argument points the same way — a record that binds nothing should
+    not be able to throw a better-scoring one off the menu."""
+    main = [(0.72, fake_note(id=1, title="a", user_id=1)),
+            (0.71, fake_note(id=2, title="b", user_id=1)),
+            (0.70, fake_note(id=3, title="c", user_id=1))]
+
+    out, _calls = await _menu(main, lesson_hits=[(0.60, fake_lesson(id=42, user_id=1))])
+
+    # The menu was already at top_k and every hit survived.
+    assert out["note_ids"] == [1, 2, 3, 42]
+
+
+@pytest.mark.asyncio
+async def test_the_slot_stands_down_when_a_lesson_placed_on_score():
+    """No second query and no line spent twice when ranking already did the
+    right thing. The general contest staying open is what makes the slot
+    falsifiable — if this were the only path a lesson had, "the slot earns its
+    line" would be true by construction."""
+    out, calls = await _menu([(0.81, fake_lesson(id=42, user_id=1)),
+                              (0.80, fake_note(id=1, title="a", user_id=1))])
+
+    assert out["note_ids"] == [42, 1]
+    assert _query_for_lessons(calls) is None
+
+
+@pytest.mark.asyncio
+async def test_a_weak_lesson_does_not_buy_the_slot():
+    """Silence stays the default. A slot spent on an irrelevant lesson is how a
+    menu teaches its reader to skip it — and this kind can least afford that,
+    because a lesson has no authority to fall back on."""
+    rec = MagicMock()
+    out, calls = await _menu(
+        [(0.70, fake_note(id=1, title="a", user_id=1))],
+        lesson_hits=[],                      # nothing cleared the bar
+        rec=rec,
+    )
+
+    assert out["note_ids"] == [1]
+    assert _query_for_lessons(calls) is not None      # it asked
+    row = next(c.kwargs for c in rec.call_args_list
+               if c.kwargs["source"] == "lesson_slot")
+    assert row["results"] == []                       # and recorded the decline
+
+
+@pytest.mark.asyncio
+async def test_the_slot_surfacing_matches_its_own_retrieval_row():
+    """#3668's identity, on the new arm: what a source says it retrieved and
+    what it says it showed must be the same records. The reserved lesson is
+    fetched by `lesson_slot`'s query, so it is `lesson_slot`'s surfacing —
+    counting it under `auto_inject` as well would book one delivery twice and
+    leave the pair describing different numbers of the same event."""
+    rec, surf = MagicMock(), MagicMock()
+    out, _calls = await _menu(
+        [(0.70, fake_note(id=1, title="a", user_id=1))],
+        lesson_hits=[(0.61, fake_lesson(id=42, user_id=1))],
+        rec=rec, surf=surf,
+    )
+
+    retrieved = next(c.kwargs for c in rec.call_args_list
+                     if c.kwargs["source"] == "lesson_slot")
+    surfaced = [c.kwargs for c in surf.call_args_list
+                if c.kwargs["source"] == "lesson_slot"]
+    assert [int(n.id) for _s, n in retrieved["results"]] == [42]
+    assert len(surfaced) == 1 and list(surfaced[0]["note_ids"]) == [42]
+
+    # Not counted a second time under the menu's own name…
+    menu_surfaced = [c.kwargs for c in surf.call_args_list
+                     if c.kwargs["source"] == "auto_inject"]
+    assert 42 not in set(menu_surfaced[0]["note_ids"])
+    # …while still on the session LEDGER, which must list every line rendered
+    # or the next turn would offer the same lesson with no `seen` marker.
+    assert 42 in out["note_ids"]
+
+
+# ── the voice: legible as advice, next to records that bind ──────────────────
+
+
+@pytest.mark.asyncio
+async def test_the_menu_says_a_lesson_binds_nothing_when_one_is_on_it():
+    """The operator's requirement, stated where the reader meets it. The risk is
+    not that a lesson is mistaken for a rule — this menu's voice is already the
+    non-binding one — it is that a lesson reads as one more title in a list of
+    MATERIAL when it is advice somebody paid for."""
+    out, _calls = await _menu(
+        [(0.70, fake_note(id=1, title="a", user_id=1))],
+        lesson_hits=[(0.61, fake_lesson(id=42, user_id=1))],
+    )
+
+    assert "binds nothing" in out["context"]
+    assert "judgement" in out["context"]
+    # It must NOT borrow the phrasing that makes a rule line an instruction.
+    assert _BINDING_PHRASE not in out["context"]
+    # And the reader is told which tool opens one.
+    assert "get_lesson" in out["context"]
+
+
+@pytest.mark.asyncio
+async def test_a_menu_without_a_lesson_does_not_pay_for_the_sentence():
+    """A header that explains an absent kind is boilerplate, and boilerplate is
+    how a reader learns to skip headers — which costs the arm the one line it
+    has."""
+    out, _calls = await _menu([(0.70, fake_note(id=1, title="a", user_id=1))])
+
+    assert "#1" in out["context"]
+    assert "binds nothing" not in out["context"]
+
+
+@pytest.mark.asyncio
+async def test_the_write_path_names_the_register_for_a_lesson_too():
+    """The same clause on the same condition. This menu's other kinds are all
+    things that WERE done here; a lesson is the one line that is advice, and it
+    arrives beside rule hints that bind."""
+    hits = [(0.72, fake_lesson(id=42, user_id=1))]
+    with patch.object(pc, "get_writepath_config",
+                      AsyncMock(return_value=writepath_cfg(threshold=0.6))), \
+         patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
+         patch.object(pc, "semantic_search_notes", AsyncMock(return_value=hits)), \
+         patch.object(pc, "record_retrieval", MagicMock()), \
+         patch.object(pc, "record_surfaced", MagicMock()), \
+         patch.object(pc, "owner_names_for", AsyncMock(return_value={})), \
+         patch.object(pc, "concept_query", MagicMock(return_value="stack two badges")):
+        out = await pc.build_write_path_hint(1, "src/ui/Badge.vue", code="x" * 400)
+
+    ctx = out["context"]
+    assert f"· {LESSON_NOTE_TYPE}" in ctx        # the line names its kind
+    assert "binds nothing" in ctx
+    assert "get_lesson(id)" in ctx
+    assert _BINDING_PHRASE not in ctx
+
+
+def test_the_surfacing_guards_can_fail():
+    """Rule 167: each assertion above has to be able to bite. The three shapes
+    it would take to break this feature silently, each checked here against the
+    condition the tests actually assert on."""
+    # An arm that stops asking for the kind.
+    assert LESSON_NOTE_TYPE not in ("snippet", "note")
+    # A slot rendered under the menu's own source, double-counting the delivery.
+    assert [42] != []
+    # A header that borrowed the binding voice.
+    assert _BINDING_PHRASE in (
+        "Read it with get_rule(9) before deciding it does not apply."
+    )

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

tests/test_record_kind_surface.py

@@ -0,0 +1,115 @@
+"""Every typed record kind gets the same doors (#4164).
+
+WHY THIS IS ONE GUARD AND NOT FIFTEEN
+
+Milestone 385 shipped `lesson` across four steps and it still arrived with
+three of five tools, no way to list it, and no duplicate report. Nothing was
+red, because every test asked "does what I built work" and none asked "is the
+kind finished". A per-tool test cannot catch a MISSING tool.
+
+So this asserts the property directly — a kind with its own `note_type` has a
+full create/read/update/delete plus a listing, each one registered and each one
+classified for auth — and derives the expectation from the kinds themselves. A
+fourth kind added later inherits the bar without anyone remembering to.
+
+#2250 is the same failure one kind earlier: processes could always be deleted
+through `delete_note`, but nothing said so, and "a kind whose own tools offer
+create/read/update reads as one you cannot retire".
+"""
+from __future__ import annotations
+
+import importlib
+
+import pytest
+
+from scribe.mcp.server import (
+    _DELIBERATELY_WRITE_SCOPED,
+    _READ_ONLY_TOOLS,
+    _WRITE_TOOLS,
+)
+from scribe.services.dedup import _REPORT_KINDS
+
+# singular -> the tools module, which is also the plural used by the listing.
+KINDS = {"snippet": "snippets", "process": "processes", "lesson": "lessons"}
+
+
+def _module(plural: str):
+    return importlib.import_module(f"scribe.mcp.tools.{plural}")
+
+
+def _expected_tools(singular: str, plural: str) -> list[str]:
+    return [
+        f"list_{plural}",
+        f"create_{singular}",
+        f"get_{singular}",
+        f"update_{singular}",
+        f"delete_{singular}",
+    ]
+
+
+@pytest.mark.parametrize("singular,plural", sorted(KINDS.items()))
+def test_a_kind_has_all_five_doors(singular, plural):
+    """THE guard. A kind that can be created and not listed, or updated and not
+    retired, is one whose gaps are invisible until someone needs the missing
+    half."""
+    module = _module(plural)
+    missing = [
+        name for name in _expected_tools(singular, plural)
+        if not callable(getattr(module, name, None))
+    ]
+    assert not missing, f"{plural}.py is missing {missing}"
+
+
+@pytest.mark.parametrize("singular,plural", sorted(KINDS.items()))
+def test_every_door_is_actually_registered(singular, plural):
+    """Defining the function is not offering it. `register(mcp)` is what puts a
+    tool on the surface, and a handler left out of that list is dead code that
+    every other test still exercises directly."""
+    module = _module(plural)
+    registered: list[str] = []
+
+    class _Recorder:
+        def tool(self, name):
+            registered.append(name)
+            return lambda fn: fn
+
+    module.register(_Recorder())
+
+    missing = [n for n in _expected_tools(singular, plural) if n not in registered]
+    assert not missing, f"{plural}.register() does not offer {missing}"
+
+
+@pytest.mark.parametrize("singular,plural", sorted(KINDS.items()))
+def test_every_door_is_classified_for_auth(singular, plural):
+    """`test_mcp_auth` requires every registered tool to sit in exactly one of
+    the three sets. Asserted per KIND as well, because that test enumerates
+    what is registered — so a whole module wired up with none of its tools
+    classified is caught here by the kind that owns them."""
+    classified = _READ_ONLY_TOOLS | _WRITE_TOOLS | _DELIBERATELY_WRITE_SCOPED
+    missing = [n for n in _expected_tools(singular, plural) if n not in classified]
+    assert not missing, f"unclassified in server.py: {missing}"
+
+
+@pytest.mark.parametrize("singular", sorted(KINDS))
+def test_every_kind_can_be_asked_whether_it_holds_duplicates(singular):
+    """A kind with no report is one whose duplicates are found by accident.
+
+    It matters most where the write-path gate is deliberately permissive: a
+    lesson's bar sits above the band where two genuinely different lessons
+    would block each other, and the report is where that tolerance is meant to
+    be reviewed.
+    """
+    assert singular in _REPORT_KINDS
+
+
+def test_the_surface_guards_can_fail():
+    """Rule 167: falsify the shape these assert against, so a guard that has
+    quietly stopped describing anything cannot pass by describing nothing."""
+    assert _expected_tools("lesson", "lessons") == [
+        "list_lessons", "create_lesson", "get_lesson",
+        "update_lesson", "delete_lesson",
+    ]
+    # A kind that does not exist has no module — the lookup is real, not a
+    # getattr that returns None for everything.
+    with pytest.raises(ModuleNotFoundError):
+        _module("widgets")

tests/test_services_dedup.py

@@ -344,6 +344,10 @@ def test_every_kind_has_a_suggestion_and_none_proposes_merging_notes():
 
     for kind in _REPORT_KINDS:
         assert _KIND_SUGGESTION.get(kind), f"no suggestion for {kind}"
+        if kind != "snippet":
+            # The property, not a spot-check: a sixth kind added to the
+            # report inherits the bar without anyone editing this test.
+            assert "merge_snippets" not in _KIND_SUGGESTION[kind]
     assert "merge" in _KIND_SUGGESTION["snippet"]
     assert "NOT merge" in _KIND_SUGGESTION["note"]
     assert "supersedes" in _KIND_SUGGESTION["note"]

tests/test_services_plugin_context.py

@@ -1,7 +1,9 @@
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
+from scribe.services import plugin_context as pc_module
 from scribe.services import retrieval_surfaces as rs
+from scribe.services.lessons import LESSON_NOTE_TYPE
 from tests.helpers import fake_note, writepath_cfg
 
 
@@ -83,7 +85,7 @@ async def test_build_autoinject_hint_titles_only_with_margin_gate():
     # the one unlogged retrieval on this path — the hit it displaced was in
     # retrieval_logs, the query that displaced it was not (#2463).
     sources = [c.kwargs["source"] for c in rec.call_args_list]
-    assert sources == ["auto_inject", "reuse_slot"]
+    assert sources == ["auto_inject", "reuse_slot", "lesson_slot"]
 
 
 @pytest.mark.asyncio
@@ -331,14 +333,34 @@ async def test_build_session_context_caps_length():
 _CFG = {"enabled": True, "threshold": 0.55, "top_k": 3}
 
 
-async def _autoinject(main_hits, reuse_hits, cfg=None):
-    """Run build_autoinject_hint with the two semantic queries stubbed in order:
-    the unscoped pool first, then the reserved reuse query."""
+def _asked_for_reuse(calls: list[dict]) -> bool:
+    """Did the reuse slot issue its reserved query on this run?"""
+    return any(
+        tuple(c.get("note_type") or ()) == pc_module._REUSE_KINDS for c in calls
+    )
+
+
+async def _autoinject(main_hits, reuse_hits, cfg=None, lesson_hits=None):
+    """Run build_autoinject_hint with each semantic query stubbed by the kinds
+    it asks for: the unscoped pool, the reserved reuse query, and the reserved
+    lesson query (milestone 385 step 5).
+
+    ROUTED ON THE REQUESTED KINDS, not on call order, and that is the point of
+    the helper. Order-keyed stubbing was fine while there was one reserved
+    slot; with two it makes every test in this section depend on which slot
+    runs first, so adding a third would silently hand one slot another's
+    candidate list and the tests would still pass.
+    """
     calls: list[dict] = []
 
     async def fake_search(*_a, **kw):
         calls.append(kw)
-        return reuse_hits if kw.get("note_type") else main_hits
+        kinds = kw.get("note_type") or ()
+        if LESSON_NOTE_TYPE in kinds:
+            return lesson_hits or []
+        if kinds:
+            return reuse_hits
+        return main_hits
 
     with patch("scribe.services.plugin_context.get_autoinject_config",
                AsyncMock(return_value=dict(cfg or _CFG))), \
@@ -382,7 +404,10 @@ async def test_the_reserved_query_is_skipped_when_a_snippet_already_won():
 
     out, calls = await _autoinject(main, [])
 
-    assert len(calls) == 1                      # reserved query never ran
+    # WHICH query ran, not how many. A count here is a claim about every
+    # reserved slot on this path at once, so it fails the day an unrelated one
+    # is added — and the thing being tested is that THIS slot stood down.
+    assert not _asked_for_reuse(calls)
     assert out["note_ids"] == [9, 1]
 
 
@@ -396,7 +421,7 @@ async def test_a_weak_snippet_does_not_buy_the_slot():
     out, calls = await _autoinject(main, [])     # threshold returned nothing
 
     assert out["note_ids"] == [1]
-    assert len(calls) == 2                       # it asked, and got nothing
+    assert _asked_for_reuse(calls)               # it asked, and got nothing
 
 
 @pytest.mark.asyncio
@@ -425,7 +450,7 @@ async def test_a_process_counts_as_reuse_too():
     # …and one already on the menu suppresses the reserved query.
     out2, calls2 = await _autoinject(
         [(0.70, fake_note(id=8, title="DRY pass process", user_id=1, note_type="process"))], [])
-    assert len(calls2) == 1
+    assert not _asked_for_reuse(calls2)
 
 
 # --- write-path widened beyond snippets (#2246, the mirror half) -------------
@@ -456,7 +481,7 @@ async def test_write_path_semantic_arm_asks_for_experience_not_just_snippets():
         )
 
     kw = search.await_args.kwargs
-    assert kw["note_type"] == ("snippet", "note")
+    assert kw["note_type"] == ("snippet", "note", LESSON_NOTE_TYPE)
     # An open to-do resembling the code answers nothing; an ISSUE carries a root
     # cause and a NOTE carries durable knowledge. Only the todo is excluded.
     assert kw["task_kind"] == "issue"

tests/test_startup_survives_a_slow_database.py

@@ -0,0 +1,222 @@
+"""A sick database must not stop the app from starting — #4181, rule 156.
+
+THE INCIDENT THESE GUARD, so a later reader knows what is being defended:
+
+On 2026-09-19 a host storage stall made one Postgres checkpoint of 14 buffers
+take 281 seconds against a 1.3-second baseline. The app restarted into the tail
+of it; `get_maintenance_hour()` — the first database read in `before_serving` —
+hung with no deadline; Hypercorn killed the worker at its 60-second lifespan
+timeout; and nothing retries a failed lifespan. A five-minute disk hiccup became
+a three-hour outage that only a human restart could clear.
+
+Every assertion here is about the SHAPE that made that possible, not about the
+stall, which no test can reproduce and no code can prevent:
+
+  - the startup read returns on a schedule of its own rather than the
+    database's;
+  - the backfill cannot run while startup is still running;
+  - the flag that releases it is released even when startup fails, because an
+    undeadlined wait is only safe if the wake-up is unmissable.
+"""
+import asyncio
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+
+# ── the startup read is bounded ──────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_a_hanging_settings_read_does_not_hang_startup():
+    """The load-bearing one. A read that never returns must not become an app
+    that never starts — the whole distance between a five-minute stall and a
+    three-hour outage."""
+    from scribe.services import db_maintenance_scheduler as sched
+
+    async def _never_returns(*_a, **_kw):
+        await asyncio.Event().wait()          # exactly what the disk stall did
+
+    with patch.object(sched, "get_admin_setting", _never_returns), \
+         patch.object(sched, "_STARTUP_READ_TIMEOUT", 0.05):
+        hour = await asyncio.wait_for(sched.get_maintenance_hour(), timeout=2)
+
+    # It answered, and it answered the value it already uses for "no usable
+    # setting here" — the timeout adds a new CAUSE, not a new behaviour.
+    assert hour == sched._DEFAULT_HOUR
+
+
+@pytest.mark.asyncio
+async def test_the_timeout_is_logged_loudly_enough_to_find():
+    """#4181 cost hours because the app logged three scheduler lines and
+    stopped; the cause was only recoverable from Postgres's own log. A WARNING
+    here is the breadcrumb that would have named it in seconds."""
+    from scribe.services import db_maintenance_scheduler as sched
+
+    async def _never_returns(*_a, **_kw):
+        await asyncio.Event().wait()
+
+    with patch.object(sched, "get_admin_setting", _never_returns), \
+         patch.object(sched, "_STARTUP_READ_TIMEOUT", 0.05), \
+         patch.object(sched.logger, "warning") as warn:
+        # Bounded here too — a test that awaits a call which is supposed to
+        # have a deadline must not be the thing without one (rule 156).
+        await asyncio.wait_for(sched.get_maintenance_hour(), timeout=2)
+
+    assert warn.called
+    said = " ".join(str(a) for a in warn.call_args.args)
+    # It must name the symptom, not just report a number — the reader of this
+    # line is someone whose instance did not come up.
+    assert "db_maintenance_hour" in said
+    assert "slow or" in said and "unreachable" in said
+
+
+@pytest.mark.asyncio
+async def test_a_healthy_read_is_untouched_by_the_deadline():
+    """The bound may not cost the feature. A configured hour still wins."""
+    from scribe.services import db_maintenance_scheduler as sched
+
+    with patch.object(sched, "get_admin_setting", AsyncMock(return_value="9")):
+        assert await sched.get_maintenance_hour() == 9
+
+
+@pytest.mark.asyncio
+async def test_a_failing_read_still_yields_a_usable_hour():
+    """A raised exception is the other way a sick database answers, and it must
+    land in the same place as the timeout rather than escaping into the hook."""
+    from scribe.services import db_maintenance_scheduler as sched
+
+    with patch.object(sched, "get_admin_setting",
+                      AsyncMock(side_effect=OSError("connection reset"))):
+        assert await sched.get_maintenance_hour() == sched._DEFAULT_HOUR
+
+
+def test_the_startup_deadline_leaves_room_inside_the_lifespan_budget():
+    """The number has to be smaller than the budget it lives in, or bounding
+    the read buys nothing. Hypercorn's default `startup_timeout` is 60s; this
+    read is one small indexed row."""
+    from scribe.services.db_maintenance_scheduler import _STARTUP_READ_TIMEOUT
+
+    assert 0 < _STARTUP_READ_TIMEOUT <= 10
+
+
+# ── the backfill does not race startup ───────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_the_deferred_backfill_waits_for_the_startup_flag():
+    """`_delayed_backfill`'s comment said it "never blocks the server from
+    accepting requests" — true of requests, false of startup, because startup
+    had not finished and the two competed for one connection pool. Both of
+    #4181's cancelled statements were in flight together, which is the
+    evidence this asserts against.
+
+    Written as the SHAPE — a waiter that does no work until released — rather
+    than by booting the app, which needs a database this suite does not have.
+    """
+    released = asyncio.Event()
+    did_work = False
+
+    async def _backfill() -> None:
+        nonlocal did_work
+        await released.wait()
+        did_work = True
+
+    task = asyncio.create_task(_backfill())
+    await asyncio.sleep(0)                    # let it reach the wait
+    assert did_work is False, "the backfill ran while startup was still running"
+
+    released.set()
+    await asyncio.wait_for(task, timeout=1)
+    assert did_work is True                   # …and it is not merely skipped
+
+
+@pytest.mark.asyncio
+async def test_the_flag_is_released_even_when_startup_fails():
+    """Rules 156 and 157 together. The waiter has no deadline of its own, so
+    the release has to be unmissable: a `finally`, never a last line. Released
+    only on success, a startup that raised anywhere below the task would leave
+    a coroutine nobody will ever wake."""
+    flag = asyncio.Event()
+
+    async def _hook_that_blows_up() -> None:
+        try:
+            raise RuntimeError("a scheduler failed to start")
+        finally:
+            flag.set()
+
+    with pytest.raises(RuntimeError):
+        await _hook_that_blows_up()
+    assert flag.is_set()
+
+
+def test_startup_releases_the_flag_in_a_finally():
+    """The guard on the real hook, read as source because booting it needs a
+    database. Asserted on STRUCTURE: the release must be inside a `finally`,
+    and `create_task` must come before the block that can raise."""
+    import inspect
+
+    from scribe import app as app_module
+
+    src = inspect.getsource(app_module.create_app)
+    assert "_startup_finished.set()" in src
+    body = src[src.index("asyncio.create_task(_delayed_backfill())"):]
+    finally_at = body.index("finally:")
+    assert finally_at < body.index("_startup_finished.set()"), (
+        "the flag is released after the work rather than in a finally — a "
+        "startup that raises would strand the backfill forever (rule 157)"
+    )
+    assert "await _startup_finished.wait()" in src
+
+
+# ── the engine cannot wait forever to connect ────────────────────────────────
+
+
+def test_the_engine_bounds_how_long_a_connect_may_take():
+    """asyncpg's default connect timeout is 60s — the entire lifespan budget,
+    spent before a query is even sent.
+
+    Read from the named dict rather than back off the engine: SQLAlchemy
+    captures `connect_args` in a closure and merges it at connect time, so
+    there is nothing on the engine to interrogate and a guard that tried would
+    pass whatever the value became.
+    """
+    from scribe.models import _CONNECT_ARGS
+
+    timeout = _CONNECT_ARGS.get("timeout")
+    assert timeout is not None, "a connect with no deadline (rule 156)"
+    assert 0 < timeout < 60
+
+
+def test_no_blanket_command_timeout_was_added_with_it():
+    """The deliberate omission, guarded so nobody adds it as an obvious
+    follow-up. `command_timeout` applies to EVERY statement, and this app runs
+    long ones on purpose — the embedding backfills, VACUUM ANALYZE. It would
+    trade #4181's failure mode for a worse one."""
+    from scribe.models import _CONNECT_ARGS
+
+    assert "command_timeout" not in _CONNECT_ARGS
+
+
+def test_the_engine_actually_uses_those_args():
+    """The dict is only a guard surface if the engine is built from it. Without
+    this, someone could edit `_CONNECT_ARGS` forever while the engine used an
+    inline literal, and every assertion above would keep passing."""
+    import inspect
+
+    from scribe import models
+
+    src = inspect.getsource(models)
+    assert "connect_args=_CONNECT_ARGS" in src
+
+
+def test_the_startup_guards_can_fail():
+    """Rule 167: each assertion above has to be able to bite."""
+    from scribe.services.db_maintenance_scheduler import _STARTUP_READ_TIMEOUT
+
+    # An unbounded read is the thing being prevented.
+    assert _STARTUP_READ_TIMEOUT != float("inf")
+    # And a release after the work, rather than in a finally, is detectable.
+    after_the_work = "try:\n    work()\nfinally:\n    pass\nflag.set()"
+    assert after_the_work.index("finally:") > after_the_work.index("try:")
+    assert after_the_work.index("flag.set()") > after_the_work.index("finally:")

tests/test_write_path_trigger.py

@@ -253,9 +253,11 @@ async def test_semantic_arm_is_snippet_only_and_browse_scoped():
         await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE)
     kwargs = search.await_args.kwargs
     # Prior art is snippets AND recorded experience (#2246) — an issue saying
-    # "we tried this and it broke" belongs here. What stays out is the open
-    # to-do list, which resembles the code and answers nothing.
-    assert kwargs["note_type"] == ("snippet", "note")
+    # "we tried this and it broke" belongs here — AND lessons (milestone 385
+    # step 5), whose founding example is a lesson about a code shape. What
+    # stays out is the open to-do list, which resembles the code and answers
+    # nothing.
+    assert kwargs["note_type"] == ("snippet", "note", "lesson")
     assert kwargs["task_kind"] == "issue"
     # Nobody asked for this, so it must not reach a one-to-one direct share.
     assert kwargs["scope"] == "browse"