Merge pull request 'Lessons point at rules (milestone 440 steps 1–2) + the away-from-home usage readout' (#189) from dev into main

alembic/versions/0111_lesson_rule_links.py

@@ -0,0 +1,54 @@
+"""lesson_rule_links — a lesson points at the rule it is an instance of
+(milestone 440 step 1, #4196)
+
+Revision ID: 0111
+Revises: 0110
+Create Date: 2026-10-01
+
+One row per (lesson, rule) pair with a state: `suggested` while evidence
+accumulates, `confirmed` or `rejected` once a judgment is made. Only a
+confirmed link changes what surfaces; a rejected one is kept so the pair is not
+proposed again. CASCADE on both ends — a link to a record that no longer exists
+says nothing. No backfill: no lesson has ever been linked, and inventing a link
+would assert a judgment nobody made.
+"""
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+from alembic import op
+
+revision = "0111"
+down_revision = "0110"
+branch_labels = None
+depends_on = None
+
+# One place, so the CHECK and the model's LINK_STATES cannot drift (rule 36:
+# a new value later means DROP + ADD CONSTRAINT in the same migration).
+_STATES = ("suggested", "confirmed", "rejected")
+
+
+def upgrade() -> None:
+    op.create_table(
+        "lesson_rule_links",
+        sa.Column("id", sa.BigInteger(), primary_key=True),
+        sa.Column("lesson_id", sa.Integer(), sa.ForeignKey("notes.id", ondelete="CASCADE"), nullable=False),
+        sa.Column("rule_id", sa.BigInteger(), sa.ForeignKey("rules.id", ondelete="CASCADE"), nullable=False),
+        sa.Column("state", sa.Text(), nullable=False, server_default="suggested"),
+        sa.Column("note", sa.Text(), nullable=True),
+        sa.Column("evidence", postgresql.JSONB(), nullable=True),
+        sa.Column("judged_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()")),
+        sa.UniqueConstraint("lesson_id", "rule_id", name="uq_lesson_rule_links_pair"),
+    )
+    op.create_check_constraint(
+        "ck_lesson_rule_links_state", "lesson_rule_links",
+        "state IN (" + ", ".join(f"'{s}'" for s in _STATES) + ")",
+    )
+    op.create_index("ix_lesson_rule_links_lesson_id", "lesson_rule_links", ["lesson_id"])
+    op.create_index("ix_lesson_rule_links_rule_id", "lesson_rule_links", ["rule_id"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_lesson_rule_links_rule_id", table_name="lesson_rule_links")
+    op.drop_index("ix_lesson_rule_links_lesson_id", table_name="lesson_rule_links")
+    op.drop_constraint("ck_lesson_rule_links_state", "lesson_rule_links", type_="check")
+    op.drop_table("lesson_rule_links")

alembic/versions/0112_lesson_no_rule.py

@@ -0,0 +1,33 @@
+"""lesson_no_rule — "no rule fits" is a recorded answer (milestone 440 step 2,
+#4631)
+
+Revision ID: 0112
+Revises: 0111
+Create Date: 2026-10-01
+
+One row per lesson that was judged to fall under no rule, with the reason. A
+lesson with neither a confirmed link nor this row is UNJUDGED, which is what
+`list_lessons(unjudged=true)` lists. CASCADE from the lesson. No backfill: no
+lesson has been judged yet, and writing "no rule fits" for one nobody looked
+at would assert a judgment nobody made.
+"""
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0112"
+down_revision = "0111"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "lesson_no_rule",
+        sa.Column("lesson_id", sa.Integer(), sa.ForeignKey("notes.id", ondelete="CASCADE"), primary_key=True),
+        sa.Column("why", sa.Text(), nullable=False),
+        sa.Column("judged_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()")),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("lesson_no_rule")

frontend/src/components/UsageBadge.vue

@@ -42,9 +42,15 @@ const title = () => {
     ? `Last opened ${new Date(u.last_pulled_at).toLocaleDateString()}.`
     : "Never opened.";
   const verdict = isDeadWeight() ? ` ${props.deadWeightAdvice}` : "";
+  // Said only when it happened: "0× elsewhere" on every row would read as a
+  // finding about records that simply have not been near another project.
+  const awaySurfaced = u.surfaced_away_count ?? 0;
+  const away = awaySurfaced
+    ? ` On other projects: surfaced ${awaySurfaced}×, opened ${u.pulled_away_count ?? 0}×.`
+    : "";
   return (
     `Surfaced to an agent ${u.surfaced_count}×, opened in full ` +
-    `${u.pull_count}×. ${last}${verdict}`
+    `${u.pull_count}×.${away} ${last}${verdict}`
   );
 };
 </script>

frontend/src/types/usage.ts

@@ -20,4 +20,9 @@ export interface RecordUsage {
   pull_count: number;
   last_surfaced_at: string | null;
   last_pulled_at: string | null;
+  /** The subsets that happened on a project other than the record's own —
+   *  the evidence it transferred (milestone 385). Notes and snippets carry
+   *  them; rules are global, so their readout does not. */
+  surfaced_away_count?: number;
+  pulled_away_count?: number;
 }

plugin/.claude-plugin/plugin.json

@@ -1,7 +1,7 @@
 {
   "name": "scribe",
   "description": "Scribe for Claude Code: connects the scribe MCP server, adds the hooks that deliver live project state and relevant records at the right moment, ships the shared client-neutral Scribe skills (using-scribe, writing-plans, reporting-back, systematic-debugging, verification, brainstorming, reusing-code, shape-accounting), and syncs your saved Scribe Processes as skills (/scribe:sync).",
-  "version": "2026.10.01.1246",
+  "version": "2026.10.01.1724",
   "author": {
     "name": "Bryan Van Deusen"
   },

plugin/skills/using-scribe/writing-records.md

@@ -8,6 +8,7 @@ arrives that names the situation you are actually in; and before filling
 ## Contents
 - Where a new rule goes — its home, its trigger, what already covers the moment
 - A lesson grows each time it proves itself
+- A lesson names the rule it is an instance of
 - A note that asserts a fact can carry its own check
 
 ## Where a new rule goes
@@ -76,6 +77,29 @@ waiting in a situation nobody is in. One claim that has met the same failure
 four times is worth more than four claims that each met it once, so when a
 near-duplicate create hands back an existing id, that is the record to grow.
 
+## A lesson names the rule it is an instance of
+
+A lesson records one situation; a rule records the binding choice for a class
+of them. Linking the two is what lets a rule be reached through the situations
+that keep proving it — moments that resemble why it was written more closely
+than its own wording does. A lesson never becomes a rule. It points at one.
+
+The writer is best placed to say which rule, so answer while writing, in
+`create_lesson` itself or in the `update_lesson` right after it:
+
+- **`rule_ids=[…]`** — the rule or preference whose situation this is an
+  instance of.
+- **`no_rule="why"`** — no rule governs it, in a line. That is a full answer,
+  not a gap: lessons that stand alone and keep landing in one situation are
+  what a missing rule looks like, and the reason is what lets the answer be
+  re-judged once a rule exists.
+
+A lesson created with neither comes back with `rule_candidates`, the rules it
+most resembles, each with its trigger. Read them against the situation the
+lesson describes, not its topic — a rule about the same subsystem that governs
+a different moment is not its rule — and give one of the two answers.
+`list_lessons(unjudged=true)` gathers the ones still open.
+
 ## A note that asserts a fact can carry its own check
 
 **A few notes assert a FACT, and those can carry their own check.**

src/scribe/mcp/server.py

@@ -161,7 +161,7 @@ _READ_ONLY_TOOLS = frozenset({
 _WRITE_TOOLS = frozenset({
     # notes, tasks, planning
     "create_note", "update_note", "delete_note",
-    "create_lesson", "update_lesson", "delete_lesson",
+    "create_lesson", "update_lesson", "delete_lesson", "judge_lesson_link",
     "create_task", "update_task", "delete_task", "add_task_log",
     "create_records", "start_planning",
     "create_milestone", "update_milestone", "delete_milestone",

src/scribe/mcp/tools/lessons.py

@@ -13,6 +13,7 @@ from scribe.mcp._context import current_user_id
 from scribe.services import access as access_svc
 from scribe.services import dedup as dedup_svc
 from scribe.services import knowledge as knowledge_svc
+from scribe.services import lesson_rules as lesson_rules_svc
 from scribe.services import lessons as lessons_svc
 from scribe.services import systems as systems_svc
 from scribe.services import trash as trash_svc
@@ -28,7 +29,7 @@ _to_dict = lessons_svc.lesson_to_dict
 
 async def list_lessons(
     q: str = "", tag: str = "", limit: int = 50, offset: int = 0,
-    project_id: int = 0,
+    project_id: int = 0, unjudged: bool = False,
 ) -> dict:
     """List lessons — the kind enumerated, rather than only what a query
     resembles.
@@ -50,10 +51,26 @@ async def list_lessons(
         project_id: Narrow to where a lesson was WRITTEN. 0 = every project.
             A lesson is retrievable from anywhere regardless (step 3); this
             filters the listing, not the reach.
+        unjudged: List only the lessons nobody has answered "which rule is
+            this an instance of?" for — no rule named, and no "no rule fits"
+            recorded. Work down this list with `update_lesson(rule_ids=…)` or
+            `update_lesson(no_rule="why")`. A listing rather than a search, so
+            it takes `tag` and `project_id` but not `q`.
 
     Returns {"lessons": [{id, title, when_to_apply, tags, preview}], "total"}.
     """
     uid = current_user_id()
+    if unjudged:
+        if q:
+            raise ValueError(
+                "unjudged lists every unjudged lesson rather than searching "
+                "them — call it with tag/project_id and without q."
+            )
+        items, total = await lesson_rules_svc.list_unjudged(
+            uid, tag=tag, project_id=project_id or None,
+            limit=max(1, min(limit, 100)), offset=max(0, offset),
+        )
+    else:
         items, total = await knowledge_svc.query_knowledge(
             user_id=uid, note_type=lessons_svc.LESSON_NOTE_TYPE,
             tags=[tag] if tag else [], sort="modified", q=q or None,
@@ -89,11 +106,30 @@ async def create_lesson(
     tags: list[str] | None = None,
     project_id: int = 0,
     system_ids: list[int] | None = None,
+    rule_ids: list[int] | None = None,
+    no_rule: str = "",
     force: bool = False,
 ) -> dict:
     """Record something you LEARNED, so a later session meets it at the moment
     it applies — on this project or any other.
 
+    A LESSON POINTS AT THE RULE IT IS AN INSTANCE OF. While you write it, you
+    know the situation better than anyone will again, so that is when to say
+    which binding choice it falls under. Answer one of two ways, in this call
+    or right after it with `update_lesson`:
+
+    - `rule_ids=[…]` — the rule(s) or preference(s) it is an instance of. The
+      rule then becomes reachable through the situation the lesson describes,
+      which is closer to why it was written than its own wording often is.
+    - `no_rule="why"` — no rule governs this situation, in a line. That is a
+      real answer: a lesson that stands alone is what a rule nobody has
+      written yet is made from, and the reason lets it be re-judged later.
+
+    A lesson created with neither comes back with `rule_candidates` — the
+    rules it most resembles, each with its trigger — and `rule_judgment:
+    "unjudged"`. Read them against the lesson and answer; `list_lessons(
+    unjudged=true)` gathers any left open.
+
     A LESSON OR A RULE? The difference is FORCE, not importance. A rule is
     something that must be followed; a lesson is something worth knowing. If
     ignoring it would be a mistake, it is a rule (create_rule) and needs the
@@ -145,9 +181,18 @@ async def create_lesson(
             reach: a lesson is retrievable from every project (that is the
             point of the kind). 0 = none.
         system_ids: Systems (subsystems/areas) to file it under.
+        rule_ids: The rule(s) or preference(s) this lesson is an instance of —
+            the binding choice its situation falls under. A lesson never
+            becomes a rule; it points at the one that governs it, and the rule
+            is then reachable through the situation the lesson describes
+            (milestone 440). Naming a rule here confirms the link.
+        no_rule: The reason no rule governs this situation, in a line — the
+            other answer to "which rule?". Give one or the other, not both.
         force: Create even if a near-duplicate exists.
 
-    Returns the created lesson. On a near-duplicate, returns the existing id
+    Returns the created lesson with its `rules` and `rule_judgment`, plus
+    `rule_candidates` when it is unjudged (absent when the rule search could
+    not run). On a near-duplicate, returns the existing id
     instead of creating — two lessons about one failure class want to be one
     lesson, so update that one rather than adding a second.
     """
@@ -161,6 +206,10 @@ async def create_lesson(
         )
 
     sources = lessons_svc.normalize_sources(learned_from)
+    # Validated before anything is written, so a lesson naming a rule the
+    # caller cannot read fails whole rather than saving half-linked.
+    linked = await lesson_rules_svc.require_rules(uid, rule_ids)
+    lesson_rules_svc.require_one_answer(linked, no_rule)
     title, body = lessons_svc.lesson_document(
         what, when_to_apply, insight, sources,
     )
@@ -179,11 +228,36 @@ async def create_lesson(
     )
     if system_ids:
         await systems_svc.set_record_systems(uid, note.id, system_ids)
+    if linked:
+        await lesson_rules_svc.set_lesson_rules(uid, note.id, linked)
+    elif no_rule.strip():
+        await lesson_rules_svc.set_no_rule(uid, note.id, no_rule)
     data = _to_dict(note)
     await systems_tools.attach_systems(uid, uid, data, note.id, project_id or None)
+    await lesson_rules_svc.attach_lesson_rules(uid, [data])
+    if not linked and not no_rule.strip():
+        await _offer_candidates(uid, data, what, when_to_apply, project_id)
     return data
 
 
+async def _offer_candidates(uid: int, data: dict, what: str, trigger: str, project_id: int) -> None:
+    """Put the rules an unjudged lesson resembles in front of its writer.
+
+    `rule_judgment` is set here too, because the attach above is fail-open and
+    may have left it off; a lesson just created with no answer IS unjudged.
+    """
+    data["rule_judgment"] = lesson_rules_svc.UNJUDGED
+    candidates = await lesson_rules_svc.rule_candidates(uid, what, trigger, project_id or None)
+    if candidates is not None:
+        data["rule_candidates"] = candidates
+    data["rule_hint"] = (
+        "Which rule is this lesson an instance of? If one of rule_candidates "
+        "governs its situation, update_lesson(lesson_id, rule_ids=[…]) links "
+        "it; if none does, update_lesson(lesson_id, no_rule=\"why\") records "
+        "that it stands alone."
+    )
+
+
 async def get_lesson(lesson_id: int, project_id: int = 0) -> dict:
     """Fetch one lesson by id, with its trigger and sources read back out.
 
@@ -214,6 +288,7 @@ async def get_lesson(lesson_id: int, project_id: int = 0) -> dict:
     # one that was true when it asked — otherwise every first read of a lesson
     # reports a pull that is its own.
     await attach_usage([out])
+    await lesson_rules_svc.attach_lesson_rules(uid, [out])
     record_pulled(
         user_id=uid, note_id=int(note.id),
         source="mcp_get_lesson", project_id=project_id,
@@ -229,6 +304,8 @@ async def update_lesson(
     learned_from: list[int] | None = None,
     tags: list[str] | None = None,
     system_ids: list[int] | None = None,
+    rule_ids: list[int] | None = None,
+    no_rule: str = "",
 ) -> dict:
     """Update a lesson. Empty fields are left unchanged.
 
@@ -258,8 +335,22 @@ async def update_lesson(
             when that argument gets dropped (#4249). A System tag is how
             `list_system_records` gathers an area's pile, so an untagged
             lesson is reachable by search and by nothing else.
+        rule_ids: Replace the rule(s) this lesson is an instance of. None
+            leaves unchanged; pass the FULL list. A rule that was linked and
+            is left out is recorded as REJECTED — "not an instance of this
+            one" — so the pair is not proposed again; `[]` rejects them all.
+            Naming a rule replaces any "no rule fits" answer.
+        no_rule: Record that no rule governs this lesson's situation, with the
+            reason in a line. Any rule still linked is rejected with that
+            reason. Empty leaves the answer unchanged; give this or a
+            non-empty `rule_ids`, not both.
     """
     uid = current_user_id()
+    linked = (
+        await lesson_rules_svc.require_rules(uid, rule_ids)
+        if rule_ids is not None else None
+    )
+    lesson_rules_svc.require_one_answer(linked, no_rule)
     note = await lessons_svc.update_lesson(
         uid, lesson_id,
         what=what or None,
@@ -276,7 +367,32 @@ async def update_lesson(
     if system_ids is not None:
         await systems_svc.set_record_systems(uid, lesson_id, system_ids)
         note = await lessons_svc.get_lesson(uid, lesson_id) or note
-    return _to_dict(note)
+    if linked is not None:
+        await lesson_rules_svc.set_lesson_rules(uid, lesson_id, linked)
+    if no_rule.strip():
+        await lesson_rules_svc.set_no_rule(uid, lesson_id, no_rule)
+    out = _to_dict(note)
+    await lesson_rules_svc.attach_lesson_rules(uid, [out])
+    return out
+
+
+async def judge_lesson_link(
+    lesson_id: int, rule_id: int, verdict: str, note: str = "",
+) -> dict:
+    """Say whether a lesson is an instance of a rule: `verdict` is "confirm"
+    or "reject".
+
+    Reach for this when Scribe proposes a pair — a lesson and a rule that keep
+    arriving together in different situations — or whenever you are reading a
+    lesson and recognise the rule it falls under. Confirming makes the rule
+    reachable through the situation the lesson describes; rejecting records
+    that it is not, so the pair is not proposed again. `note` is the why, and
+    the next reader judges the link by it.
+
+    Returns the link: {lesson_id, rule_id, state, note, evidence, judged_at}.
+    """
+    uid = current_user_id()
+    return await lesson_rules_svc.judge_link(uid, lesson_id, rule_id, verdict, note)
 
 
 async def delete_lesson(lesson_id: int) -> dict:
@@ -312,5 +428,5 @@ async def delete_lesson(lesson_id: int) -> dict:
 
 def register(mcp) -> None:
     for fn in (list_lessons, create_lesson, get_lesson, update_lesson,
-               delete_lesson):
+               delete_lesson, judge_lesson_link):
         mcp.tool(name=fn.__name__)(fn)

src/scribe/mcp/tools/rulebooks.py

@@ -393,6 +393,14 @@ async def create_rule(
     observations belong, and the reason the kind question is worth asking
     out loud rather than settled silently before the proposal.
 
+    A RULE WRITTEN FOR A SITUATION LESSONS ALREADY RECORD is linked from
+    them once it exists: `update_lesson(lesson_id, rule_ids=[…, <new id>])`
+    for each lesson it governs. Those lessons are the concrete situations
+    that proved it, and linked, they are how the rule is reached in moments
+    closer to why it was written than its own wording. A lesson that
+    answered "no rule fits" before this rule existed is the first to
+    re-judge; `get_lesson` shows the answer and its reason.
+
     Where the interface offers structured choices, ask it that way — a
     question with named options is answered in a click, while the same
     question inside a paragraph is answered by scrolling past. Where it does

src/scribe/models/__init__.py

@@ -77,6 +77,8 @@ from scribe.models.canonical_system import CanonicalSystem  # noqa: E402, F401
 from scribe.models.rulebook import (  # noqa: E402, F401
     Rulebook, RulebookTopic, Rule, RuleRelation, rule_systems,
 )
+# After notes and rules: it foreign-keys both (milestone 440).
+from scribe.models.lesson_rule_link import LessonNoRule, LessonRuleLink  # noqa: E402, F401
 from scribe.models.repo_binding import RepoBinding  # noqa: E402, F401
 from scribe.models.forge_connection import ForgeConnection  # noqa: E402, F401
 from scribe.models.code_shape import CodeShape, CodeShapeConsumer, CodeShapeEvent, CodeShapeUse  # noqa: E402, F401

src/scribe/models/lesson_rule_link.py

@@ -0,0 +1,112 @@
+from datetime import datetime
+
+from sqlalchemy import BigInteger, DateTime, ForeignKey, Integer, Text, UniqueConstraint, func
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.orm import Mapped, mapped_column
+
+from scribe.models import Base
+from scribe.models.base import CreatedAtMixin, iso
+
+SUGGESTED = "suggested"
+CONFIRMED = "confirmed"
+REJECTED = "rejected"
+# CHECK ck_lesson_rule_links_state (migration 0111, rule 36).
+LINK_STATES = (SUGGESTED, CONFIRMED, REJECTED)
+
+
+class LessonRuleLink(Base, CreatedAtMixin):
+    """A lesson says it is an instance of a rule (milestone 440, #4196).
+
+    A lesson records one concrete situation; a rule records the binding choice
+    for a class of them. The link lets the rule be reached through the
+    situations that keep proving it, and lets a situation with lessons and no
+    rule be noticed. A lesson never BECOMES a rule — it points at one.
+
+    ONE ROW PER PAIR, WITH A STATE, because the operator's design forms a link
+    in two stages ("a sort of soft and then hard link once it's been proven"):
+
+    - ``suggested`` — evidence is accumulating that the two belong together
+      (they keep arriving in the same request, in distinct situations). It
+      carries nothing in retrieval: a suggested link that brought its rule
+      along would manufacture the co-surfacing it counts, and prove itself.
+    - ``confirmed`` — a judgment said the lesson is an instance of the rule.
+      The only state that changes what surfaces.
+    - ``rejected`` — a judgment said it is not. Kept, evidence and all, so the
+      pair is never proposed again and the reason stays readable.
+
+    A table rather than a list on the lesson's `data`: the link needs a foreign
+    key on both ends (a deleted rule must not leave a lesson pointing at
+    nothing), a reverse index (a rule lists its lessons), and an id remap at
+    restore — none of which a JSON list gives.
+    """
+
+    __tablename__ = "lesson_rule_links"
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
+    lesson_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("notes.id", ondelete="CASCADE"), index=True,
+    )
+    rule_id: Mapped[int] = mapped_column(
+        BigInteger, ForeignKey("rules.id", ondelete="CASCADE"), index=True,
+    )
+    state: Mapped[str] = mapped_column(Text, default=SUGGESTED, server_default=SUGGESTED)
+    # Why it was confirmed or rejected — the reasoning a later reader needs to
+    # decide whether it still holds, as a rule relation's `note` is.
+    note: Mapped[str | None] = mapped_column(Text, nullable=True)
+    # What the suggestion rests on (filled by the co-surfacing recorder, #4637).
+    # Kept after a judgment so a confirmation can be read beside its evidence.
+    evidence: Mapped[dict | None] = mapped_column(JSONB, nullable=True)
+    # When a judgment moved it out of `suggested`. Null while suggested.
+    judged_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True,
+    )
+
+    __table_args__ = (
+        UniqueConstraint("lesson_id", "rule_id", name="uq_lesson_rule_links_pair"),
+    )
+
+    def to_dict(self) -> dict:
+        return {
+            "lesson_id": self.lesson_id,
+            "rule_id": self.rule_id,
+            "state": self.state,
+            "note": self.note or "",
+            "evidence": self.evidence or {},
+            "judged_at": iso(self.judged_at),
+            "created_at": iso(self.created_at),
+        }
+
+
+class LessonNoRule(Base):
+    """A judgment that no rule governs a lesson's situation (#4631).
+
+    The third answer to "which rule is this an instance of?", beside naming
+    one and leaving it open. It is what lets an UNJUDGED lesson be told apart
+    from one that was looked at and found to stand alone — and a lesson with
+    no rule is the raw material #4634 reads for convergence, so the
+    difference is the whole signal.
+
+    A row of its own rather than a key in the lesson's `data`: that column is
+    a mirror re-composed from the body on every edit (services/lessons.py),
+    so anything kept there that the body does not render is erased by the
+    next rewording — and the lesson would silently read as unjudged again.
+
+    One row per lesson. Naming a rule later deletes it: "no rule fits" and
+    "this rule fits" cannot both be the current answer.
+    """
+
+    __tablename__ = "lesson_no_rule"
+
+    lesson_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("notes.id", ondelete="CASCADE"), primary_key=True,
+    )
+    # Why none fits — required, because "no rule" with no reason cannot be
+    # re-judged when a rule is later written for the situation.
+    why: Mapped[str] = mapped_column(Text)
+    judged_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(),
+    )
+
+    def to_dict(self) -> dict:
+        return {"why": self.why, "judged_at": iso(self.judged_at)}
+

src/scribe/routes/lessons.py

@@ -28,6 +28,7 @@ from scribe.auth import get_current_user_id, login_required
 from scribe.routes.utils import not_found, parse_pagination
 from scribe.services import dedup as dedup_svc
 from scribe.services import knowledge as knowledge_svc
+from scribe.services import lesson_rules as lesson_rules_svc
 from scribe.services import lessons as lessons_svc
 from scribe.services import systems as systems_svc
 from scribe.services import trash as trash_svc
@@ -62,6 +63,15 @@ async def list_lessons_route():
         project_id = None
     limit, offset = parse_pagination(default_limit=24, max_limit=100)
 
+    # The lessons nobody has answered "which rule?" for (#4631). A listing,
+    # so it does not combine with a search — the MCP door says the same.
+    if request.args.get("unjudged") in ("1", "true"):
+        if q:
+            return jsonify({"error": "unjudged is a listing; drop q"}), 400
+        items, total = await lesson_rules_svc.list_unjudged(
+            uid, tag=tag, project_id=project_id, limit=limit, offset=offset,
+        )
+    else:
         items, total = await knowledge_svc.query_knowledge(
             user_id=uid,
             note_type=lessons_svc.LESSON_NOTE_TYPE,
@@ -136,6 +146,14 @@ async def create_lesson_route():
 
     project_id = data.get("project_id") or None
     learned_from = data.get("learned_from") or []
+    # The rules this lesson is an instance of (milestone 440). Validated before
+    # anything is written, as the MCP door does, so a bad id saves nothing.
+    no_rule = (data.get("no_rule") or "").strip()
+    try:
+        linked = await lesson_rules_svc.require_rules(uid, data.get("rule_ids") or [])
+        lesson_rules_svc.require_one_answer(linked, no_rule)
+    except ValueError as exc:
+        return jsonify({"error": str(exc)}), 400
 
     # The same near-duplicate gate the MCP create path applies. Two lessons
     # under one trigger compete in a single ranked list for one reserved slot,
@@ -165,10 +183,23 @@ async def create_lesson_route():
     )
     if data.get("system_ids") is not None:
         await systems_svc.set_record_systems(uid, note.id, data["system_ids"])
+    if linked:
+        await lesson_rules_svc.set_lesson_rules(uid, note.id, linked)
+    elif no_rule:
+        await lesson_rules_svc.set_no_rule(uid, note.id, no_rule)
     out = lessons_svc.lesson_to_dict(note)
     out["systems"] = [
         s.to_dict() for s in await systems_svc.list_record_systems(uid, note.id)
     ]
+    await lesson_rules_svc.attach_lesson_rules(uid, [out])
+    # The same offer the MCP door makes, so the form can show the rules to
+    # judge against while the writer still has the situation in mind.
+    if not linked and not no_rule:
+        candidates = await lesson_rules_svc.rule_candidates(
+            uid, what, when_to_apply, project_id,
+        )
+        if candidates is not None:
+            out["rule_candidates"] = candidates
     return jsonify(out), 201
 
 
@@ -194,6 +225,7 @@ async def get_lesson_route(lesson_id: int):
     )
     out.update(await describe_provenance(uid, note))
     await attach_usage([out])
+    await lesson_rules_svc.attach_lesson_rules(uid, [out])
     # Opening the detail view IS a pull — the operator chose to look. Tagged
     # apart from the MCP sources so "an agent was handed it" and "a human read
     # it" stay distinguishable; they mean different things for pruning (#2085).
@@ -233,9 +265,24 @@ async def update_lesson_route(lesson_id: int):
             ),
         }), 400
 
+    linked = None
+    no_rule = (data.get("no_rule") or "").strip()
+    try:
+        if data.get("rule_ids") is not None:
+            linked = await lesson_rules_svc.require_rules(uid, data["rule_ids"])
+        lesson_rules_svc.require_one_answer(linked, no_rule)
+    except ValueError as exc:
+        return jsonify({"error": str(exc)}), 400
+
     updated = await lessons_svc.update_lesson(owner_uid, lesson_id, **kwargs)
     if updated is None:
         return not_found("Lesson")
+    if linked is not None:
+        # The CALLER, for the reason system_ids below uses it: the rules named
+        # must be ones the person making the edit can read.
+        await lesson_rules_svc.set_lesson_rules(uid, lesson_id, linked)
+    if no_rule:
+        await lesson_rules_svc.set_no_rule(uid, lesson_id, no_rule)
     if data.get("system_ids") is not None:
         # The CALLER, not owner_uid (#4249). `set_record_systems` runs its own
         # `can_write_note` and links only Systems the acting user can read;
@@ -249,9 +296,28 @@ async def update_lesson_route(lesson_id: int):
         s.to_dict()
         for s in await systems_svc.list_record_systems(owner_uid, lesson_id)
     ]
+    await lesson_rules_svc.attach_lesson_rules(uid, [out])
     return jsonify(out)
 
 
+@lessons_bp.route("/<int:lesson_id>/rules/<int:rule_id>", methods=["PUT"])
+@login_required
+async def judge_lesson_link_route(lesson_id: int, rule_id: int):
+    """Confirm or reject one lesson→rule link — the REST twin of the MCP
+    `judge_lesson_link`. Body: {"verdict": "confirm" | "reject", "note": "…"}."""
+    uid = get_current_user_id()
+    data = await request.get_json() or {}
+    try:
+        link = await lesson_rules_svc.judge_link(
+            uid, lesson_id, rule_id, data.get("verdict", ""), data.get("note", ""),
+        )
+    except PermissionError:
+        return jsonify({"error": "Permission denied"}), 403
+    except ValueError as exc:
+        return jsonify({"error": str(exc)}), 400
+    return jsonify(link)
+
+
 @lessons_bp.route("/<int:lesson_id>", methods=["DELETE"])
 @login_required
 async def delete_lesson_route(lesson_id: int):

src/scribe/services/backup.py

@@ -16,6 +16,7 @@ from scribe.models.rule_usage import RuleUsageEvent
 from scribe.models.retrieval_tuning import RetrievalTuningEvent
 from scribe.models.canonical_system import CanonicalSystem
 from scribe.models.rulebook import RuleRelation, rule_systems as rule_systems_t
+from scribe.models.lesson_rule_link import LessonNoRule, LessonRuleLink
 from scribe.models.code_shape import CodeShape, CodeShapeEvent, CodeShapeUse
 from scribe.models.project import Project
 from scribe.models.repo_binding import RepoBinding
@@ -80,8 +81,15 @@ logger = logging.getLogger(__name__)
 # say whether it still measures anything. Both travel NULLABLE and unfilled —
 # a row written before the stamp existed restores unstamped, because inventing
 # the model it was measured under would turn "unknown" into a stated fact.
+# v18 (2026-10) added lesson_rule_links (milestone 440): which rule each lesson
+# is an instance of, and the judgments that confirmed or rejected each pair. A
+# confirmed link is a judgment nothing else records, and a rejected one is
+# what stops the pair being proposed again — losing either undoes work.
+# v19 (2026-10) added lesson_no_rule (#4631): the "no rule fits" answer and its
+# reason. Without it a restored lesson that was judged to stand alone reads as
+# never judged, and lands back on the unjudged list.
 # Bump when the serialized schema changes.
-BACKUP_VERSION = 17
+BACKUP_VERSION = 19
 
 # Every table this backup carries, by its REAL name. Paired with _NOT_INCLUDED
 # below, these two lists must together account for the entire schema — which is
@@ -119,6 +127,10 @@ _BACKED_UP = [
     # on the operator's behalf, a restore that kept the numbers and dropped the
     # reasons would leave an install tuned by nobody it can name.
     "retrieval_tuning_events",
+    # v18 (2026-10): lesson → rule links and their judgments (milestone 440).
+    "lesson_rule_links",
+    # v19 (2026-10): "no rule fits" answers (#4631).
+    "lesson_no_rule",
 ]
 
 # Tables intentionally NOT in the backup, surfaced in the payload so the gap is
@@ -212,6 +224,10 @@ _COLUMN_EXCLUSIONS: dict[str, set[str]] = {
     "record_systems": {"id", "created_at"},
     "note_supersessions": {"id", "created_at"},
     "rule_relations": {"id", "created_at"},
+    # The pair is the row; everything else is the judgment and its evidence.
+    "lesson_rule_links": {"id"},
+    # Keyed by the lesson itself; nothing to exclude, every column travels.
+    "lesson_no_rule": set(),
     "note_usage_events": {"id"},
     # Same as the note twin: the surrogate key is re-issued on insert.
     "rule_usage_events": {"id"},
@@ -299,6 +315,8 @@ _IMPORT_COLUMN_EXCLUSIONS: dict[str, set[str]] = {
     "record_systems": {"id", "created_at"},
     "note_supersessions": {"id", "created_at"},
     "rule_relations": {"id", "created_at"},
+    "lesson_rule_links": {"id"},
+    "lesson_no_rule": set(),
     "note_usage_events": {"id"},
     "rule_usage_events": {"id"},
     "retrieval_tuning_events": {"id"},
@@ -692,6 +710,32 @@ def _rule_relation_rows(rows) -> list[dict]:
     ]
 
 
+def _lesson_rule_link_rows(rows) -> list[dict]:
+    """Which rule each lesson is an instance of, with the judgment's state,
+    reason and evidence (milestone 440). Ids are SOURCE ids, remapped through
+    the note and rule maps at restore."""
+    return [
+        {
+            "lesson_id": r.lesson_id, "rule_id": r.rule_id, "state": r.state,
+            "note": r.note, "evidence": r.evidence,
+            "judged_at": r.judged_at.isoformat() if r.judged_at else None,
+            "created_at": r.created_at.isoformat() if r.created_at else None,
+        }
+        for r in rows
+    ]
+
+
+def _lesson_no_rule_rows(rows) -> list[dict]:
+    """The "no rule fits" answers, by SOURCE lesson id (#4631)."""
+    return [
+        {
+            "lesson_id": r.lesson_id, "why": r.why,
+            "judged_at": r.judged_at.isoformat() if r.judged_at else None,
+        }
+        for r in rows
+    ]
+
+
 def _rule_rows(rows) -> list[dict]:
     return [
         {
@@ -739,6 +783,8 @@ async def export_full_backup() -> dict:
             .join(CanonicalSystem, CanonicalSystem.id == rule_systems_t.c.canonical_id)
         )).all()
         rule_relations = (await session.execute(select(RuleRelation))).scalars().all()
+        lesson_rule_links = (await session.execute(select(LessonRuleLink))).scalars().all()
+        lesson_no_rule = (await session.execute(select(LessonNoRule))).scalars().all()
         record_systems = (await session.execute(select(RecordSystem))).scalars().all()
         supersessions = (
             await session.execute(select(NoteSupersession))
@@ -797,6 +843,8 @@ async def export_full_backup() -> dict:
         "canonical_systems": _canonical_system_rows(canonical_systems),
         "rule_systems": _rule_system_rows(rule_system_rows),
         "rule_relations": _rule_relation_rows(rule_relations),
+        "lesson_rule_links": _lesson_rule_link_rows(lesson_rule_links),
+        "lesson_no_rule": _lesson_no_rule_rows(lesson_no_rule),
         "systems": _system_rows(
             systems, {c.id: c.slug for c in canonical_systems}
         ),
@@ -957,6 +1005,17 @@ async def export_user_backup(user_id: int) -> dict:
                 RuleRelation.to_rule_id.in_(_rule_ids),
             )
         )).scalars().all() if _rule_ids else []
+        # Both ends in THIS user's export, for the supersession reason: a link
+        # to a lesson or rule the import will not create restores as nothing.
+        lesson_rule_links = (await session.execute(
+            select(LessonRuleLink).where(
+                LessonRuleLink.lesson_id.in_(note_ids),
+                LessonRuleLink.rule_id.in_(_rule_ids),
+            )
+        )).scalars().all() if (_rule_ids and note_ids) else []
+        lesson_no_rule = (await session.execute(
+            select(LessonNoRule).where(LessonNoRule.lesson_id.in_(note_ids))
+        )).scalars().all() if note_ids else []
 
     return {
         "version": BACKUP_VERSION,
@@ -984,6 +1043,8 @@ async def export_user_backup(user_id: int) -> dict:
         "canonical_systems": _canonical_system_rows(canonical_systems),
         "rule_systems": _rule_system_rows(rule_system_rows),
         "rule_relations": _rule_relation_rows(rule_relations),
+        "lesson_rule_links": _lesson_rule_link_rows(lesson_rule_links),
+        "lesson_no_rule": _lesson_no_rule_rows(lesson_no_rule),
         "systems": _system_rows(
             systems, {c.id: c.slug for c in canonical_systems}
         ),
@@ -1329,6 +1390,37 @@ def _build_rule_relation(row: dict, maps: _Maps) -> RuleRelation | None:
     )
 
 
+def _build_lesson_no_rule(row: dict, maps: _Maps) -> LessonNoRule | None:
+    """Skipped when its lesson did not restore — the answer is about that
+    lesson and no other."""
+    lesson = maps.notes.get(row.get("lesson_id", 0))
+    if lesson is None or not (row.get("why") or "").strip():
+        return None
+    return LessonNoRule(
+        lesson_id=lesson, why=row["why"], judged_at=_dt(row.get("judged_at")),
+    )
+
+
+def _build_lesson_rule_link(row: dict, maps: _Maps) -> LessonRuleLink | None:
+    """Both ends must map: a link whose lesson or rule did not restore points
+    at whatever took that number in the destination."""
+    lesson = maps.notes.get(row.get("lesson_id", 0))
+    rule = maps.rules.get(row.get("rule_id", 0))
+    if lesson is None or rule is None:
+        return None
+    return LessonRuleLink(
+        lesson_id=lesson,
+        rule_id=rule,
+        state=row.get("state") or "suggested",
+        note=row.get("note") or None,
+        evidence=row.get("evidence"),
+        # Kept absent when absent: a suggested link was never judged, and
+        # stamping it with the restore time would say it was.
+        judged_at=_dt_or_none(row.get("judged_at")),
+        created_at=_dt(row.get("created_at")),
+    )
+
+
 def _build_rule_version(row: dict, maps: _Maps) -> RuleVersion | None:
     rid = maps.rules.get(row.get("rule_id", 0))
     if rid is None:
@@ -1725,7 +1817,8 @@ async def _restore_v2(data: dict) -> dict:
         "note_supersessions": 0, "code_shapes": 0, "code_shape_events": 0,
         "code_shape_uses": 0, "canonical_systems": 0,
         "rule_systems": 0, "rule_relations": 0, "rule_versions": 0,
-        "retrieval_tuning_events": 0,
+        "retrieval_tuning_events": 0, "lesson_rule_links": 0,
+        "lesson_no_rule": 0,
     }
 
     async with async_session() as session:
@@ -1920,6 +2013,23 @@ async def _restore_v2(data: dict) -> dict:
             session.add(relation)
             stats["rule_relations"] += 1
 
+        # Lesson → rule links (milestone 440): after both notes and rules are
+        # mapped, which is why they sit beside the rule edges. Archives before
+        # v18 carry no section and restore with none.
+        for lr in data.get("lesson_rule_links", []):
+            link = _build_lesson_rule_link(lr, maps)
+            if link is None:
+                continue
+            session.add(link)
+            stats["lesson_rule_links"] += 1
+        # "No rule fits" answers (#4631); archives before v19 carry none.
+        for nr in data.get("lesson_no_rule", []):
+            answer = _build_lesson_no_rule(nr, maps)
+            if answer is None:
+                continue
+            session.add(answer)
+            stats["lesson_no_rule"] += 1
+
         # A rule's edit history (milestone 323). Must come after the rules
         # themselves — the rule map is only populated above — and both ids are
         # ids in the SOURCE database, which is #3182's arose_from_id trap.

src/scribe/services/lesson_rules.py

@@ -0,0 +1,447 @@
+"""Lessons point at rules (milestone 440, #4196).
+
+A lesson is a non-binding record of one situation; a rule is the binding
+choice for a class of them. This service owns the link between the two — which
+rule a lesson is an instance of — and both directions of reading it.
+
+THE STATES (models/lesson_rule_link.py says why each exists):
+`suggested` while evidence accumulates, `confirmed` or `rejected` once a
+judgment is made. Every write here is a JUDGMENT, so every write lands as
+confirmed or rejected; `suggested` rows come from the co-surfacing recorder
+(#4637), never from a caller naming a rule.
+
+THE THIRD ANSWER (#4631). "Which rule is this an instance of?" has three
+answers, and only two of them are links: a rule named (confirmed), or NO RULE
+FITS — a `lesson_no_rule` row carrying the reason. A lesson with neither is
+UNJUDGED, and `list_unjudged` lists those. The three are kept apart because a
+lesson that stands alone is the raw material for a rule nobody has written yet
+(#4634), and that is unreadable if "nobody looked" means the same thing.
+
+ACL (rule 78). Linking changes what a lesson says about itself, so it needs
+WRITE on the lesson (`access.can_write_note`, share-aware). It names a rule, so
+it needs the rule to be one the caller may read — rules are owner-scoped, and
+`rulebooks._fetch_owned_rule` / `_owned_rules_clause` are that check's two
+forms. A read shows only the rules the READER owns: a lesson shared with
+someone must not hand them the titles of its owner's private rules.
+"""
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timezone
+
+from sqlalchemy import and_, delete, exists, func, not_, select
+
+from scribe.models import async_session
+from scribe.models.lesson_rule_link import (
+    CONFIRMED, REJECTED, SUGGESTED, LessonNoRule, LessonRuleLink,
+)
+from scribe.models.note import Note
+from scribe.models.rulebook import Rule
+
+logger = logging.getLogger(__name__)
+
+# What a caller says to judge one pair. Words rather than the stored states,
+# because "confirm" and "reject" are acts and the states are their results.
+VERDICTS = {"confirm": CONFIRMED, "reject": REJECTED}
+
+# Recorded on a link that set-semantics removed, so a reader of the rejected
+# row can tell an explicit "not this rule" from a link dropped by a rewrite.
+_REMOVED_NOTE = "removed from the lesson's rules by an update"
+
+# Recorded on a confirmed link that a "no rule fits" answer overturned, ahead
+# of that answer's reason.
+_NO_RULE_NOTE = "no rule fits: "
+
+# What `rule_judgment` reads on a lesson payload — the three answers.
+LINKED, NO_RULE, UNJUDGED = "linked", "no_rule", "unjudged"
+
+# How many rules a new lesson is offered to judge against. A handful, not the
+# fifty `what_might_apply` returns: this is read in the create response, at
+# the moment of writing, and a long tail there buries the one that fits.
+CANDIDATE_LIMIT = 5
+
+
+def _ids(values) -> list[int]:
+    """Positive ints, de-duplicated, in the order given."""
+    out: list[int] = []
+    for v in values or []:
+        try:
+            i = int(v)
+        except (TypeError, ValueError):
+            raise ValueError(f"rule id {v!r} is not an integer")
+        if i > 0 and i not in out:
+            out.append(i)
+    return out
+
+
+async def _require_lesson_writable(user_id: int, lesson_id: int) -> None:
+    from scribe.services import access
+    from scribe.services import lessons as lessons_svc
+
+    note = await lessons_svc.get_lesson(user_id, lesson_id)
+    if note is None:
+        raise ValueError(f"lesson {lesson_id} not found")
+    if not await access.can_write_note(user_id, lesson_id):
+        raise PermissionError(f"lesson {lesson_id} is not yours to change")
+
+
+async def require_rules(user_id: int, rule_ids) -> list[int]:
+    """The ids, validated as rules the caller owns — all of them or none.
+
+    Checked BEFORE anything is written, by every caller, so a lesson create
+    that names a rule it cannot see fails without leaving a half-linked
+    lesson behind.
+    """
+    from scribe.services import rulebooks as rulebooks_svc
+
+    wanted = _ids(rule_ids)
+    missing = [rid for rid in wanted if await rulebooks_svc.get_rule(rid, user_id) is None]
+    if missing:
+        raise ValueError(
+            f"rule(s) {missing} not found — a lesson can point only at a rule "
+            "you can read. Nothing was linked."
+        )
+    return wanted
+
+
+def require_one_answer(linked, no_rule: str) -> None:
+    """Refuse "these rules fit" and "no rule fits" together, before any write
+    — they contradict, and either order of applying them loses one."""
+    if linked and (no_rule or "").strip():
+        raise ValueError(
+            "rule_ids and no_rule are the two answers to \"which rule is this "
+            "an instance of?\" — give the one that holds. Nothing was written."
+        )
+
+
+async def _upsert(session, lesson_id: int, rule_id: int, state: str, note: str) -> None:
+    now = datetime.now(timezone.utc)
+    row = (await session.execute(
+        select(LessonRuleLink).where(
+            LessonRuleLink.lesson_id == lesson_id,
+            LessonRuleLink.rule_id == rule_id,
+        )
+    )).scalar_one_or_none()
+    if row is None:
+        session.add(LessonRuleLink(
+            lesson_id=lesson_id, rule_id=rule_id, state=state,
+            note=note or None, judged_at=now,
+        ))
+        return
+    # Evidence is kept across a judgment: a confirmation reads best beside
+    # what it rested on.
+    row.state = state
+    row.note = note or row.note
+    row.judged_at = now
+
+
+async def _clear_no_rule(session, lesson_id: int) -> None:
+    """A confirmed rule overturns "no rule fits" — both cannot be the answer."""
+    await session.execute(delete(LessonNoRule).where(LessonNoRule.lesson_id == lesson_id))
+
+
+async def set_lesson_rules(
+    user_id: int, lesson_id: int, rule_ids, *, note: str = "",
+) -> None:
+    """Make the lesson's CONFIRMED rules exactly `rule_ids` (set-semantics).
+
+    A rule named here is confirmed, whatever state it was in: the writer of a
+    lesson saying "this is an instance of rule N" is the judgment the
+    suggested state waits for, so it needs no evidence bar.
+
+    A rule that WAS confirmed and is no longer named becomes `rejected`, not
+    deleted. Dropping it is a judgment that the lesson is not an instance of
+    that rule, and a deleted row would let the co-surfacing recorder propose
+    the same pair again.
+    """
+    await _require_lesson_writable(user_id, lesson_id)
+    wanted = await require_rules(user_id, rule_ids)
+    async with async_session() as session:
+        current = (await session.execute(
+            select(LessonRuleLink).where(
+                LessonRuleLink.lesson_id == lesson_id,
+                LessonRuleLink.state == CONFIRMED,
+            )
+        )).scalars().all()
+        for row in current:
+            if row.rule_id not in wanted:
+                row.state = REJECTED
+                row.note = _REMOVED_NOTE
+                row.judged_at = datetime.now(timezone.utc)
+        for rid in wanted:
+            await _upsert(session, lesson_id, rid, CONFIRMED, note)
+        if wanted:
+            await _clear_no_rule(session, lesson_id)
+        await session.commit()
+
+
+async def set_no_rule(user_id: int, lesson_id: int, why: str) -> dict:
+    """Record that no rule governs this lesson's situation, and why.
+
+    A confirmed link the lesson still carries becomes `rejected`, with the
+    reason: saying none fits is saying the linked one does not either. A
+    second answer replaces the first, so the reason on file is the latest.
+    """
+    why = (why or "").strip()
+    if not why:
+        raise ValueError(
+            "no_rule takes the reason no rule fits, in a line — it is what "
+            "lets the answer be re-judged when a rule is later written for "
+            "this situation."
+        )
+    await _require_lesson_writable(user_id, lesson_id)
+    now = datetime.now(timezone.utc)
+    async with async_session() as session:
+        confirmed = (await session.execute(
+            select(LessonRuleLink).where(
+                LessonRuleLink.lesson_id == lesson_id,
+                LessonRuleLink.state == CONFIRMED,
+            )
+        )).scalars().all()
+        for row in confirmed:
+            row.state = REJECTED
+            row.note = _NO_RULE_NOTE + why
+            row.judged_at = now
+        answer = await session.get(LessonNoRule, lesson_id)
+        if answer is None:
+            session.add(LessonNoRule(lesson_id=lesson_id, why=why, judged_at=now))
+        else:
+            answer.why = why
+            answer.judged_at = now
+        await session.commit()
+    return {"why": why, "judged_at": now.isoformat()}
+
+
+async def judge_link(
+    user_id: int, lesson_id: int, rule_id: int, verdict: str, note: str = "",
+) -> dict:
+    """Confirm or reject one (lesson, rule) pair, suggested or not."""
+    state = VERDICTS.get((verdict or "").strip().lower())
+    if state is None:
+        raise ValueError(f"verdict must be one of {sorted(VERDICTS)}, got {verdict!r}")
+    await _require_lesson_writable(user_id, lesson_id)
+    [rid] = await require_rules(user_id, [rule_id])
+    async with async_session() as session:
+        await _upsert(session, lesson_id, rid, state, note)
+        if state == CONFIRMED:
+            await _clear_no_rule(session, lesson_id)
+        await session.commit()
+        row = (await session.execute(
+            select(LessonRuleLink).where(
+                LessonRuleLink.lesson_id == lesson_id,
+                LessonRuleLink.rule_id == rid,
+            )
+        )).scalar_one()
+        return row.to_dict()
+
+
+async def rules_for_lessons(user_id: int, lesson_ids) -> dict[int, list[dict]]:
+    """{lesson_id: [{id, title, kind, state, note}]}, for rules the READER owns.
+
+    One query for the whole set — a lesson list would otherwise be N+1.
+    Confirmed first, then suggested, then rejected: the order a reader cares
+    about them in.
+    """
+    from scribe.services.rulebooks import _owned_rules_clause
+
+    ids = [int(i) for i in lesson_ids or []]
+    out: dict[int, list[dict]] = {i: [] for i in ids}
+    if not ids:
+        return out
+    order = {s: n for n, s in enumerate((CONFIRMED, SUGGESTED, REJECTED))}
+    async with async_session() as session:
+        rows = (await session.execute(
+            select(LessonRuleLink, Rule)
+            .join(Rule, Rule.id == LessonRuleLink.rule_id)
+            .where(LessonRuleLink.lesson_id.in_(ids))
+            .where(_owned_rules_clause(user_id))
+        )).all()
+    for link, rule in sorted(rows, key=lambda r: (order.get(r[0].state, 9), r[1].id)):
+        out[link.lesson_id].append({
+            "id": rule.id, "title": rule.title, "kind": rule.kind,
+            "state": link.state, "note": link.note or "",
+        })
+    return out
+
+
+async def lessons_for_rule(user_id: int, rule_id: int) -> list[dict]:
+    """The lessons that point at one rule, readable by the caller (share-aware).
+
+    The reverse direction, and the one a rule's page needs: the concrete
+    situations that have been judged instances of it.
+    """
+    from scribe.services.access import readable_notes_clause
+
+    async with async_session() as session:
+        rows = (await session.execute(
+            select(LessonRuleLink, Note)
+            .join(Note, Note.id == LessonRuleLink.lesson_id)
+            .where(LessonRuleLink.rule_id == int(rule_id))
+            .where(Note.deleted_at.is_(None))
+            .where(readable_notes_clause(user_id))
+        )).all()
+    order = {s: n for n, s in enumerate((CONFIRMED, SUGGESTED, REJECTED))}
+    return [
+        {"id": note.id, "title": note.title, "state": link.state, "note": link.note or ""}
+        for link, note in sorted(rows, key=lambda r: (order.get(r[0].state, 9), r[1].id))
+    ]
+
+
+async def judgments_for_lessons(lesson_ids) -> dict[int, dict]:
+    """{lesson_id: {"rule_judgment": linked|no_rule|unjudged, "no_rule"?}}.
+
+    Read from the link and answer tables directly, NOT from the reader's view
+    of the rules: a lesson shared with someone who cannot see its owner's rule
+    is still a judged lesson, and calling it unjudged would invite them to
+    judge it again.
+    """
+    ids = [int(i) for i in lesson_ids or []]
+    if not ids:
+        return {}
+    async with async_session() as session:
+        linked = set((await session.execute(
+            select(LessonRuleLink.lesson_id).where(
+                LessonRuleLink.lesson_id.in_(ids),
+                LessonRuleLink.state == CONFIRMED,
+            )
+        )).scalars().all())
+        answers = {
+            a.lesson_id: a for a in (await session.execute(
+                select(LessonNoRule).where(LessonNoRule.lesson_id.in_(ids))
+            )).scalars().all()
+        }
+    out: dict[int, dict] = {}
+    for i in ids:
+        if i in linked:
+            out[i] = {"rule_judgment": LINKED}
+        elif i in answers:
+            out[i] = {"rule_judgment": NO_RULE, "no_rule": answers[i].to_dict()}
+        else:
+            out[i] = {"rule_judgment": UNJUDGED}
+    return out
+
+
+async def attach_lesson_rules(user_id: int, rows: list[dict], *, key: str = "id") -> None:
+    """Add `rules` and `rule_judgment` (plus `no_rule` when that is the answer)
+    to each lesson payload row, in place — a fixed number of queries per page.
+
+    Fail-open (snippet #4286): this decorates a lesson the caller already has,
+    so a failed lookup leaves the keys off and logs, rather than refusing the
+    lesson. An absent key reads as "not attached", never as "no rule".
+    """
+    ids = [int(r[key]) for r in rows if isinstance(r.get(key), int)]
+    try:
+        found = await rules_for_lessons(user_id, ids)
+        judged = await judgments_for_lessons(ids)
+    except Exception:
+        logger.warning("lesson→rule links could not be read", exc_info=True)
+        return
+    for r in rows:
+        if isinstance(r.get(key), int):
+            r["rules"] = found.get(r[key], [])
+            r.update(judged.get(r[key], {}))
+
+
+def unjudged_clause():
+    """SQL: a lesson (a `notes` row) with no confirmed rule and no "no rule
+    fits" answer. A rejected link alone leaves it unjudged — "not that rule"
+    does not say whether another one fits."""
+    from scribe.models.note import Note
+
+    return and_(
+        not_(exists().where(and_(
+            LessonRuleLink.lesson_id == Note.id,
+            LessonRuleLink.state == CONFIRMED,
+        ))),
+        not_(exists().where(LessonNoRule.lesson_id == Note.id)),
+    )
+
+
+async def list_unjudged(
+    user_id: int, *, tag: str = "", project_id: int | None = None,
+    limit: int = 50, offset: int = 0,
+) -> tuple[list[dict], int]:
+    """The lessons nobody has answered "which rule?" for, newest edit first.
+
+    The browse scope (`browsable_notes_clause`), the list it sits beside uses:
+    a lesson shared directly with the caller is search-only and stays out of
+    an ambient list. Rows come back in the knowledge list's item shape.
+    """
+    from scribe.models.note import Note
+    from scribe.services.access import browsable_notes_clause
+    from scribe.services.knowledge import _note_to_item
+    from scribe.services.lessons import LESSON_NOTE_TYPE
+
+    base = (
+        select(Note)
+        .where(browsable_notes_clause(user_id))
+        .where(Note.note_type == LESSON_NOTE_TYPE)
+        .where(Note.deleted_at.is_(None))
+        .where(unjudged_clause())
+    )
+    if tag:
+        base = base.where(Note.tags.contains([tag]))
+    if project_id is not None:
+        base = base.where(Note.project_id == project_id)
+    async with async_session() as session:
+        total = (await session.execute(
+            select(func.count()).select_from(base.subquery())
+        )).scalar_one()
+        rows = (await session.execute(
+            base.order_by(Note.updated_at.desc()).limit(limit).offset(offset)
+        )).scalars().all()
+    return [_note_to_item(n) for n in rows], int(total)
+
+
+async def rule_candidates(
+    user_id: int, what: str, when_to_apply: str, project_id: int | None = None,
+) -> list[dict] | None:
+    """The rules a lesson most resembles, for the writer to judge against.
+
+    Searched with the lesson's claim and its trigger — the same two halves a
+    rule's own document leads with (`trigger_title`), so like is
+    compared with like; the insight is the story and would dilute the vector.
+    Scoped as any project read is: global rules plus the lesson's project's
+    own. The bar is the explicit rule search's, since this is an explicit
+    question rather than an injection.
+
+    None when the search could not run (no embedder, a failed query), so a
+    caller can say "candidates unavailable" rather than "nothing resembles".
+    """
+    from scribe.services.embeddings import (
+        DEFAULT_SIMILARITY_THRESHOLD, semantic_search_rules, trigger_title,
+    )
+
+    # The rule document's own title join, so like is compared with like.
+    query = trigger_title(what, when_to_apply)
+    report: dict = {}
+    found = await semantic_search_rules(
+        user_id, query, limit=CANDIDATE_LIMIT,
+        threshold=DEFAULT_SIMILARITY_THRESHOLD, report=report,
+        project_id=project_id or None,
+    )
+    if not report.get("searched"):
+        return None
+    return [
+        {
+            "id": rule.id, "title": rule.title, "kind": rule.kind,
+            "when_to_apply": rule.when_to_apply or "", "score": round(score, 3),
+        }
+        for score, rule in found
+    ]
+
+
+async def attach_rule_lessons(user_id: int, data: dict, rule_id: int) -> None:
+    """Add `lessons` to a rule payload, in place, when there are any.
+
+    Present-only, as a rule's `systems` and `relations` are (#2483), and
+    fail-open for the reason `attach_lesson_rules` gives.
+    """
+    try:
+        lessons = await lessons_for_rule(user_id, rule_id)
+    except Exception:
+        logger.warning("rule→lesson links could not be read", exc_info=True)
+        return
+    if lessons:
+        data["lessons"] = lessons
+

src/scribe/services/note_usage.py

@@ -35,6 +35,7 @@ from collections.abc import Sequence
 from sqlalchemy import case, func, select
 
 from scribe.models import async_session
+from scribe.models.note import Note
 from scribe.models.note_usage import PULLED, SURFACED, NoteUsageEvent
 from scribe.models.base import iso
 from scribe.services.background import report_telemetry_failure
@@ -193,6 +194,13 @@ def empty_usage() -> dict:
     record. `ambient_count` is the rest (see AMBIENT_SOURCES). The split is the
     readout half of #2477: the "high surfaced, zero pulls → dead weight"
     reading is only valid over surfacings that were choices.
+
+    `surfaced_away_count` / `pulled_away_count` are the subsets that happened
+    on a project other than the one the record was written on — the evidence
+    that a record TRANSFERRED, which is the claim the lesson kind rests on
+    (milestone 385, #3735). Counted only where both projects are known: an
+    event with no reader project, or a record with no project of its own,
+    cannot speak to "away" and is left out rather than guessed.
     """
     return {
         "surfaced_count": 0,
@@ -200,6 +208,8 @@ def empty_usage() -> dict:
         "pull_count": 0,
         "last_surfaced_at": None,
         "last_pulled_at": None,
+        "surfaced_away_count": 0,
+        "pulled_away_count": 0,
     }
 
 
@@ -247,6 +257,30 @@ async def usage_for_notes(note_ids: list[int]) -> dict[int, dict]:
                     )
                 )
             ).all()
+            # Away from home: the reader's project against the record's own.
+            # A second aggregate rather than a column on the first, because
+            # it needs the join to `notes` and the first must keep counting
+            # events whose note has since been deleted (the table is FK-free
+            # so that evidence outlives the row). Ranked surfacings only, for
+            # the same reason surfaced_count is (#2477).
+            away_rows = (
+                await session.execute(
+                    select(
+                        NoteUsageEvent.note_id,
+                        NoteUsageEvent.event,
+                        func.count().label("n"),
+                    )
+                    .join(Note, Note.id == NoteUsageEvent.note_id)
+                    .where(
+                        NoteUsageEvent.note_id.in_(ids),
+                        NoteUsageEvent.project_id.is_not(None),
+                        Note.project_id.is_not(None),
+                        NoteUsageEvent.project_id != Note.project_id,
+                        NoteUsageEvent.source.not_in(AMBIENT_SOURCES),
+                    )
+                    .group_by(NoteUsageEvent.note_id, NoteUsageEvent.event)
+                )
+            ).all()
     except Exception:
         # A telemetry readout must not be able to break the list it decorates —
         # but it must say it failed, or a broken readout is indistinguishable
@@ -271,6 +305,14 @@ async def usage_for_notes(note_ids: list[int]) -> dict[int, dict]:
             latest = iso(last_at)
             if latest and (slot["last_pulled_at"] or "") < latest:
                 slot["last_pulled_at"] = latest
+    for note_id, event, n in away_rows:
+        slot = out.get(int(note_id))
+        if slot is None:
+            continue
+        if event == SURFACED:
+            slot["surfaced_away_count"] = int(n)
+        elif event == PULLED:
+            slot["pulled_away_count"] = int(n)
     return out
 
 

src/scribe/services/rulebooks.py

@@ -455,6 +455,11 @@ async def rule_detail(user_id: int, rule: Rule, system_ids: list[int] | None = N
         data["systems"] = systems
     if relations:
         data["relations"] = relations
+    # The concrete situations judged (or proposed) to be instances of this
+    # rule — milestone 440. Same present-only convention as the two above.
+    from scribe.services.lesson_rules import attach_rule_lessons
+
+    await attach_rule_lessons(user_id, data, rule.id)
     return data
 
 

tests/conftest.py

@@ -142,6 +142,32 @@ def _no_task_log_arm():
         yield
 
 
+@pytest.fixture(autouse=True)
+def _no_lesson_rule_links(request):
+    """Stub the lesson↔rule link decorations (milestone 440).
+
+    Every door that returns a lesson or a rule now attaches its links, and the
+    read is a real database call — so every unit test that opens either would
+    otherwise reach for the fake DATABASE_URL to learn that nothing is linked.
+    Both decorations are fail-open, so the cost would be a slow failed connect
+    per test rather than a failure, which is worse: it would never be noticed.
+
+    The candidate search a new lesson runs (#4631) is stubbed to "nothing
+    resembles it" for the same reason: it embeds, and the model load would
+    otherwise ride along with every unit test that creates a lesson.
+
+    Skipped for integration tests, which exercise the real links against
+    Postgres (tests/test_integration_lesson_rule_links.py).
+    """
+    if request.node.get_closest_marker("integration"):
+        yield
+        return
+    with patch("scribe.services.lesson_rules.attach_lesson_rules", AsyncMock()), \
+         patch("scribe.services.lesson_rules.attach_rule_lessons", AsyncMock()), \
+         patch("scribe.services.lesson_rules.rule_candidates", AsyncMock(return_value=[])):
+        yield
+
+
 @pytest.fixture(autouse=True)
 def _no_rule_arm():
     """Stub the write-path hint's standing-RULES arm (milestone 307).

tests/test_guidance_ownership.py

@@ -145,6 +145,15 @@ TOPICS: tuple[Topic, ...] = (
           # one fires mid-work, the moment a lesson arrives in a situation it
           # actually names. `lessons inform` already points at the kind.
           shared_with=("docstrings",)),
+    Topic("a lesson names the rule it is an instance of", U,
+          ("rule_ids", "no_rule", "unjudged=true"),
+          "a rule about the same subsystem that governs a different moment "
+          "is not its rule",
+          # Mid-work, like the topic above: it fires when a lesson is written,
+          # so no index marker. create_lesson's docstring states the two
+          # answers at the moment they are given, and may — tool contracts
+          # are not copy-scanned.
+          shared_with=("docstrings",)),
     Topic("preferences shape how work is done, never what is recorded", U,
           ("never what gets recorded",),
           "a preference never makes a task into a note"),

tests/test_integration_lesson_rule_links.py

@@ -0,0 +1,243 @@
+"""Real-Postgres tests for lesson → rule links (milestone 440 step 1, #4630).
+
+What a mock cannot show: that the CHECK holds the three states, that deleting
+either end takes the link with it, that a rule the caller does not own cannot
+be linked (and nothing is half-written when one id is bad), and that each side
+reads only what its reader may see.
+"""
+import uuid
+from unittest.mock import MagicMock, patch
+
+import pytest
+import pytest_asyncio
+from sqlalchemy import delete, select
+from sqlalchemy.exc import IntegrityError
+
+from scribe.models import async_session
+from scribe.models.lesson_rule_link import LessonRuleLink
+from scribe.models.note import Note
+from scribe.models.project import Project
+from scribe.models.rulebook import Rule
+from scribe.services import lesson_rules as links_svc
+from scribe.services import lessons as lessons_svc
+from scribe.services import rulebooks as rulebooks_svc
+from tests.helpers import ensure_user
+
+pytestmark = [
+    pytest.mark.integration,
+    pytest.mark.usefixtures("_dispose_engine", "_no_embedding"),
+]
+
+
+@pytest.fixture(autouse=True)
+def _no_reindex():
+    """Rule writes detach an embedding refresh that outlives the test's loop."""
+    with patch("scribe.services.rulebooks._refresh_rule_embedding", MagicMock()):
+        yield
+
+
+@pytest_asyncio.fixture
+async def world():
+    """An owner with a lesson and two rules; a stranger with a rule of their own."""
+    tag = uuid.uuid4().hex[:8]
+    async with async_session() as s:
+        owner = await ensure_user(s, f"lrl_owner_{tag}")
+        stranger = await ensure_user(s, f"lrl_stranger_{tag}")
+        mine = Project(user_id=owner.id, title="Mine")
+        theirs = Project(user_id=stranger.id, title="Theirs")
+        s.add_all([mine, theirs])
+        await s.flush()
+        ids = {"owner": owner.id, "stranger": stranger.id,
+               "mine": mine.id, "theirs": theirs.id}
+        await s.commit()
+
+    owner = ids["owner"]
+    r1 = await rulebooks_svc.create_project_rule(
+        ids["mine"], owner, "Read the job log first", "Before waiting longer.",
+        when_to_apply="a CI run has overrun its usual duration",
+    )
+    r2 = await rulebooks_svc.create_project_rule(
+        ids["mine"], owner, "Probe the system itself", "Not a proxy for it.",
+        when_to_apply="about to state what version is deployed",
+    )
+    foreign = await rulebooks_svc.create_project_rule(
+        ids["theirs"], ids["stranger"], "Their rule", "Not yours.",
+        when_to_apply="something only they do",
+    )
+    lesson = await lessons_svc.create_lesson(
+        owner, what="An overrun run usually failed early",
+        when_to_apply="a CI run is still in_progress far past its usual time",
+        insight="Read the log; the failure is often minutes old.",
+        project_id=ids["mine"],
+    )
+    ids.update(r1=r1.id, r2=r2.id, foreign=foreign.id, lesson=lesson.id)
+    return ids
+
+
+async def _states(lesson_id: int) -> dict[int, str]:
+    async with async_session() as s:
+        rows = (await s.execute(
+            select(LessonRuleLink).where(LessonRuleLink.lesson_id == lesson_id)
+        )).scalars().all()
+    return {r.rule_id: r.state for r in rows}
+
+
+async def test_naming_rules_confirms_them_and_both_sides_read_the_link(world):
+    owner = world["owner"]
+    await links_svc.set_lesson_rules(owner, world["lesson"], [world["r1"], world["r2"]])
+
+    assert await _states(world["lesson"]) == {world["r1"]: "confirmed", world["r2"]: "confirmed"}
+    rules = (await links_svc.rules_for_lessons(owner, [world["lesson"]]))[world["lesson"]]
+    assert {r["id"] for r in rules} == {world["r1"], world["r2"]}
+    lessons = await links_svc.lessons_for_rule(owner, world["r1"])
+    assert [(l["id"], l["state"]) for l in lessons] == [(world["lesson"], "confirmed")]
+
+
+async def test_a_rule_left_out_of_the_set_is_rejected_not_forgotten(world):
+    owner = world["owner"]
+    await links_svc.set_lesson_rules(owner, world["lesson"], [world["r1"], world["r2"]])
+    await links_svc.set_lesson_rules(owner, world["lesson"], [world["r1"]])
+
+    assert await _states(world["lesson"]) == {world["r1"]: "confirmed", world["r2"]: "rejected"}
+    rules = (await links_svc.rules_for_lessons(owner, [world["lesson"]]))[world["lesson"]]
+    # Confirmed reads first; the rejection keeps its reason.
+    assert [r["state"] for r in rules] == ["confirmed", "rejected"]
+    assert rules[1]["note"] == links_svc._REMOVED_NOTE
+
+
+async def test_judge_moves_a_pair_both_ways_and_refuses_a_nonsense_verdict(world):
+    owner = world["owner"]
+    out = await links_svc.judge_link(owner, world["lesson"], world["r1"], "reject", "different failure")
+    assert (out["state"], out["note"]) == ("rejected", "different failure")
+    assert out["judged_at"] is not None
+    out = await links_svc.judge_link(owner, world["lesson"], world["r1"], "confirm")
+    assert out["state"] == "confirmed"
+    with pytest.raises(ValueError):
+        await links_svc.judge_link(owner, world["lesson"], world["r1"], "maybe")
+
+
+async def test_a_rule_the_caller_cannot_read_links_nothing_at_all(world):
+    """All-or-nothing: one unreadable id refuses the whole set, so the readable
+    one beside it is not linked either."""
+    with pytest.raises(ValueError):
+        await links_svc.set_lesson_rules(
+            world["owner"], world["lesson"], [world["r1"], world["foreign"]],
+        )
+    assert await _states(world["lesson"]) == {}
+
+
+async def test_a_stranger_cannot_link_someone_elses_lesson(world):
+    with pytest.raises((ValueError, PermissionError)):
+        await links_svc.set_lesson_rules(world["stranger"], world["lesson"], [world["foreign"]])
+    assert await _states(world["lesson"]) == {}
+
+
+async def test_deleting_either_end_takes_the_link_with_it(world):
+    owner = world["owner"]
+    await links_svc.set_lesson_rules(owner, world["lesson"], [world["r1"], world["r2"]])
+    async with async_session() as s:
+        await s.execute(delete(Rule).where(Rule.id == world["r1"]))
+        await s.commit()
+    assert await _states(world["lesson"]) == {world["r2"]: "confirmed"}
+    async with async_session() as s:
+        await s.execute(delete(Note).where(Note.id == world["lesson"]))
+        await s.commit()
+    assert await _states(world["lesson"]) == {}
+
+
+async def test_the_check_holds_the_three_states(world):
+    async with async_session() as s:
+        s.add(LessonRuleLink(lesson_id=world["lesson"], rule_id=world["r1"], state="maybe"))
+        with pytest.raises(IntegrityError):
+            await s.commit()
+
+
+async def test_one_row_per_pair(world):
+    async with async_session() as s:
+        s.add_all([
+            LessonRuleLink(lesson_id=world["lesson"], rule_id=world["r1"], state="suggested"),
+            LessonRuleLink(lesson_id=world["lesson"], rule_id=world["r1"], state="confirmed"),
+        ])
+        with pytest.raises(IntegrityError):
+            await s.commit()
+
+
+# ── #4631: "no rule fits" and the unjudged list ─────────────────────────────
+
+
+async def _judgment(lesson_id: int) -> dict:
+    return (await links_svc.judgments_for_lessons([lesson_id]))[lesson_id]
+
+
+async def _unjudged_ids(user_id: int) -> set[int]:
+    items, _total = await links_svc.list_unjudged(user_id, limit=100)
+    return {i["id"] for i in items}
+
+
+async def test_a_new_lesson_is_unjudged_until_it_is_answered_either_way(world):
+    owner, lesson = world["owner"], world["lesson"]
+    assert (await _judgment(lesson))["rule_judgment"] == "unjudged"
+    assert lesson in await _unjudged_ids(owner)
+
+    await links_svc.set_no_rule(owner, lesson, "specific to this CI host")
+    judged = await _judgment(lesson)
+    assert judged["rule_judgment"] == "no_rule"
+    assert judged["no_rule"]["why"] == "specific to this CI host"
+    assert lesson not in await _unjudged_ids(owner)
+
+    # Naming a rule overturns the answer — both cannot stand.
+    await links_svc.set_lesson_rules(owner, lesson, [world["r1"]])
+    judged = await _judgment(lesson)
+    assert judged == {"rule_judgment": "linked"}
+    assert lesson not in await _unjudged_ids(owner)
+
+
+async def test_no_rule_rejects_a_confirmed_link_with_its_reason(world):
+    owner, lesson = world["owner"], world["lesson"]
+    await links_svc.set_lesson_rules(owner, lesson, [world["r1"]])
+    await links_svc.set_no_rule(owner, lesson, "the log was not the issue")
+
+    assert await _states(lesson) == {world["r1"]: "rejected"}
+    rules = (await links_svc.rules_for_lessons(owner, [lesson]))[lesson]
+    assert rules[0]["note"].endswith("the log was not the issue")
+    assert (await _judgment(lesson))["rule_judgment"] == "no_rule"
+
+
+async def test_a_confirmation_by_judgment_also_clears_no_rule(world):
+    owner, lesson = world["owner"], world["lesson"]
+    await links_svc.set_no_rule(owner, lesson, "nothing yet")
+    await links_svc.judge_link(owner, lesson, world["r2"], "confirm")
+    assert (await _judgment(lesson))["rule_judgment"] == "linked"
+
+
+async def test_a_rejection_alone_leaves_the_lesson_unjudged(world):
+    """"Not that rule" says nothing about whether another one fits."""
+    owner, lesson = world["owner"], world["lesson"]
+    await links_svc.judge_link(owner, lesson, world["r1"], "reject", "different failure")
+    assert (await _judgment(lesson))["rule_judgment"] == "unjudged"
+    assert lesson in await _unjudged_ids(owner)
+
+
+async def test_no_rule_needs_a_reason_and_a_writer(world):
+    with pytest.raises(ValueError):
+        await links_svc.set_no_rule(world["owner"], world["lesson"], "  ")
+    with pytest.raises((ValueError, PermissionError)):
+        await links_svc.set_no_rule(world["stranger"], world["lesson"], "mine now")
+    assert (await _judgment(world["lesson"]))["rule_judgment"] == "unjudged"
+
+
+async def test_the_unjudged_list_is_scoped_to_the_reader(world):
+    assert world["lesson"] not in await _unjudged_ids(world["stranger"])
+
+
+async def test_deleting_the_lesson_takes_its_answer_with_it(world):
+    from scribe.models.lesson_rule_link import LessonNoRule
+
+    await links_svc.set_no_rule(world["owner"], world["lesson"], "alone")
+    async with async_session() as s:
+        await s.execute(delete(Note).where(Note.id == world["lesson"]))
+        await s.commit()
+        left = (await s.execute(
+            select(LessonNoRule).where(LessonNoRule.lesson_id == world["lesson"])
+        )).scalars().all()
+    assert left == []

tests/test_lesson_rule_link_doors.py

@@ -0,0 +1,265 @@
+"""The lesson → rule link at its doors (milestone 440 step 1, #4630).
+
+The link's own behaviour — states, cascade, ACL — is tested against Postgres in
+tests/test_integration_lesson_rule_links.py. These pin the wiring a mock CAN
+see: that a bad rule id stops a lesson create before anything is written, that
+the doors pass the set through with the semantics their docstrings promise,
+that the judge tool is registered, that backup skips a link it cannot map, and
+that the migration and the model agree about the states.
+"""
+from __future__ import annotations
+
+import importlib.util
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from scribe.mcp._context import _user_id_ctx
+from scribe.mcp.tools import lessons as lesson_tools
+from scribe.models.lesson_rule_link import LINK_STATES
+from scribe.services import backup
+from scribe.services import lesson_rules as links_svc
+from scribe.services import lessons as lessons_svc
+
+TRIGGER = "a CI run is still in_progress far past its usual time"
+
+
+def _stub_note(**kw):
+    base = dict(
+        id=41, 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)
+
+
+def _create_patches(created):
+    return (
+        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()),
+    )
+
+
+@pytest.mark.asyncio
+async def test_an_unreadable_rule_stops_the_lesson_before_it_is_written():
+    """All-or-nothing at the door: the rule ids are validated BEFORE the
+    lesson exists, so a bad id cannot leave a lesson saved and unlinked."""
+    _user_id_ctx.set(7)
+    created = AsyncMock(return_value=_stub_note())
+    p1, p2, p3 = _create_patches(created)
+    with p1, p2, p3, patch.object(
+        links_svc, "require_rules", AsyncMock(side_effect=ValueError("rule(s) [9] not found")),
+    ):
+        with pytest.raises(ValueError):
+            await lesson_tools.create_lesson(what="x", when_to_apply=TRIGGER, rule_ids=[9])
+    created.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_create_links_the_named_rules_to_the_new_lesson():
+    _user_id_ctx.set(7)
+    created = AsyncMock(return_value=_stub_note(id=41))
+    linked = AsyncMock()
+    p1, p2, p3 = _create_patches(created)
+    with p1, p2, p3, \
+         patch.object(links_svc, "require_rules", AsyncMock(return_value=[5, 6])), \
+         patch.object(links_svc, "set_lesson_rules", linked):
+        await lesson_tools.create_lesson(what="x", when_to_apply=TRIGGER, rule_ids=[5, 6])
+    linked.assert_awaited_once()
+    assert linked.await_args.args[1:] == (41, [5, 6])
+
+
+@pytest.mark.asyncio
+async def test_create_without_rules_writes_no_link():
+    _user_id_ctx.set(7)
+    linked = AsyncMock()
+    p1, p2, p3 = _create_patches(AsyncMock(return_value=_stub_note()))
+    with p1, p2, p3, patch.object(links_svc, "set_lesson_rules", linked):
+        await lesson_tools.create_lesson(what="x", when_to_apply=TRIGGER)
+    linked.assert_not_awaited()
+
+
+@pytest.mark.parametrize("rule_ids, expect_call", [(None, False), ([], True), ([5], True)])
+@pytest.mark.asyncio
+async def test_update_leaves_links_alone_on_none_and_replaces_on_a_list(rule_ids, expect_call):
+    """None is "unchanged"; a list — including [] — is the full new set, which
+    the service turns into confirmations and rejections."""
+    _user_id_ctx.set(7)
+    linked = AsyncMock()
+    with patch.object(lessons_svc, "update_lesson", AsyncMock(return_value=_stub_note())), \
+         patch.object(links_svc, "require_rules", AsyncMock(side_effect=lambda uid, ids: list(ids))), \
+         patch.object(links_svc, "set_lesson_rules", linked):
+        await lesson_tools.update_lesson(lesson_id=41, rule_ids=rule_ids)
+    assert linked.await_count == (1 if expect_call else 0)
+    if expect_call:
+        assert linked.await_args.args[1:] == (41, rule_ids)
+
+
+def test_the_judge_tool_is_registered():
+    names = []
+    fake = SimpleNamespace(tool=lambda name: (lambda fn: names.append(name) or fn))
+    lesson_tools.register(fake)
+    assert "judge_lesson_link" in names
+
+
+def test_verdicts_name_real_states():
+    assert set(links_svc.VERDICTS.values()) <= set(LINK_STATES)
+
+
+def test_the_migration_check_and_the_model_agree_on_the_states():
+    """Rule 36's drift, guarded: the CHECK is written from the migration's
+    tuple and the code from the model's, so they must be the same tuple."""
+    path = Path(__file__).resolve().parents[1] / "alembic" / "versions" / "0111_lesson_rule_links.py"
+    spec = importlib.util.spec_from_file_location("m0111", path)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    assert tuple(module._STATES) == tuple(LINK_STATES)
+
+
+@pytest.mark.parametrize("lesson_mapped, rule_mapped", [(False, True), (True, False)])
+def test_backup_skips_a_link_whose_end_did_not_restore(lesson_mapped, rule_mapped):
+    maps = backup._Maps()
+    if lesson_mapped:
+        maps.notes[10] = 110
+    if rule_mapped:
+        maps.rules[20] = 120
+    row = {"lesson_id": 10, "rule_id": 20, "state": "confirmed"}
+    assert backup._build_lesson_rule_link(row, maps) is None
+
+
+def test_backup_keeps_an_unjudged_link_unjudged():
+    """A suggested link was never judged; restoring it with the restore's time
+    in judged_at would say it was."""
+    maps = backup._Maps()
+    maps.notes[10] = 110
+    maps.rules[20] = 120
+    built = backup._build_lesson_rule_link(
+        {"lesson_id": 10, "rule_id": 20, "state": "suggested", "judged_at": None}, maps,
+    )
+    assert (built.lesson_id, built.rule_id, built.state) == (110, 120, "suggested")
+    assert built.judged_at is None
+
+
+# ── #4631: judged when written ───────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_an_unanswered_lesson_comes_back_with_rules_to_judge_against():
+    """Neither answer given: the create offers the rules the lesson resembles,
+    searched with its claim and trigger and scoped to its project."""
+    _user_id_ctx.set(7)
+    offered = [{"id": 5, "title": "Read the log", "kind": "rule",
+                "when_to_apply": "a run overran", "score": 0.61}]
+    search = AsyncMock(return_value=offered)
+    p1, p2, p3 = _create_patches(AsyncMock(return_value=_stub_note()))
+    with p1, p2, p3, patch.object(links_svc, "rule_candidates", search):
+        out = await lesson_tools.create_lesson(what="x", when_to_apply=TRIGGER, project_id=2)
+    assert search.await_args.args == (7, "x", TRIGGER, 2)
+    assert out["rule_candidates"] == offered
+    assert out["rule_judgment"] == links_svc.UNJUDGED
+    assert "no_rule" in out["rule_hint"] and "rule_ids" in out["rule_hint"]
+
+
+@pytest.mark.asyncio
+async def test_a_search_that_could_not_run_offers_no_list_rather_than_an_empty_one():
+    """None means unavailable; an empty list would claim nothing resembles it."""
+    _user_id_ctx.set(7)
+    p1, p2, p3 = _create_patches(AsyncMock(return_value=_stub_note()))
+    with p1, p2, p3, patch.object(links_svc, "rule_candidates", AsyncMock(return_value=None)):
+        out = await lesson_tools.create_lesson(what="x", when_to_apply=TRIGGER)
+    assert "rule_candidates" not in out
+    assert out["rule_judgment"] == links_svc.UNJUDGED
+
+
+@pytest.mark.asyncio
+async def test_no_rule_on_create_records_the_answer_and_offers_nothing():
+    _user_id_ctx.set(7)
+    answered = AsyncMock()
+    search = AsyncMock(return_value=[])
+    p1, p2, p3 = _create_patches(AsyncMock(return_value=_stub_note(id=41)))
+    with p1, p2, p3, patch.object(links_svc, "set_no_rule", answered), \
+         patch.object(links_svc, "rule_candidates", search):
+        out = await lesson_tools.create_lesson(
+            what="x", when_to_apply=TRIGGER, no_rule="a one-off of this CI host",
+        )
+    assert answered.await_args.args[1:] == (41, "a one-off of this CI host")
+    search.assert_not_awaited()
+    assert "rule_candidates" not in out
+
+
+@pytest.mark.asyncio
+async def test_rules_and_no_rule_together_write_nothing():
+    _user_id_ctx.set(7)
+    created = AsyncMock(return_value=_stub_note())
+    p1, p2, p3 = _create_patches(created)
+    with p1, p2, p3, patch.object(links_svc, "require_rules", AsyncMock(return_value=[5])):
+        with pytest.raises(ValueError, match="no_rule"):
+            await lesson_tools.create_lesson(
+                what="x", when_to_apply=TRIGGER, rule_ids=[5], no_rule="none fits",
+            )
+    created.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_update_records_no_rule_and_refuses_it_beside_named_rules():
+    _user_id_ctx.set(7)
+    answered = AsyncMock()
+    updated = AsyncMock(return_value=_stub_note())
+    with patch.object(lessons_svc, "update_lesson", updated), \
+         patch.object(links_svc, "set_no_rule", answered):
+        await lesson_tools.update_lesson(lesson_id=41, no_rule="stands alone")
+    assert answered.await_args.args[1:] == (41, "stands alone")
+
+    updated.reset_mock()
+    with patch.object(lessons_svc, "update_lesson", updated), \
+         patch.object(links_svc, "require_rules", AsyncMock(return_value=[5])):
+        with pytest.raises(ValueError):
+            await lesson_tools.update_lesson(lesson_id=41, rule_ids=[5], no_rule="none")
+    updated.assert_not_awaited()
+
+
+def test_an_empty_rule_list_beside_no_rule_is_one_answer_not_two():
+    """`rule_ids=[]` clears; it does not name a rule, so it cannot contradict."""
+    links_svc.require_one_answer([], "none fits")
+    links_svc.require_one_answer(None, "none fits")
+    links_svc.require_one_answer([5], "")
+    with pytest.raises(ValueError):
+        links_svc.require_one_answer([5], "none fits")
+
+
+@pytest.mark.asyncio
+async def test_list_unjudged_is_a_listing_and_routes_to_its_own_query():
+    _user_id_ctx.set(7)
+    listed = AsyncMock(return_value=([], 0))
+    with patch.object(links_svc, "list_unjudged", listed), \
+         patch("scribe.mcp.tools.lessons.access_svc.label_shared_items", AsyncMock(return_value=[])), \
+         patch("scribe.mcp.tools.lessons.attach_usage", AsyncMock()):
+        out = await lesson_tools.list_lessons(unjudged=True, tag="ci", project_id=2)
+    assert out == {"lessons": [], "total": 0}
+    assert listed.await_args.kwargs["tag"] == "ci"
+    assert listed.await_args.kwargs["project_id"] == 2
+    with pytest.raises(ValueError):
+        await lesson_tools.list_lessons(unjudged=True, q="log")
+
+
+def test_backup_skips_a_no_rule_answer_whose_lesson_did_not_restore():
+    maps = backup._Maps()
+    assert backup._build_lesson_no_rule({"lesson_id": 10, "why": "alone"}, maps) is None
+    maps.notes[10] = 110
+    built = backup._build_lesson_no_rule({"lesson_id": 10, "why": "alone"}, maps)
+    assert (built.lesson_id, built.why) == (110, "alone")
+    # An answer with no reason is not an answer; it restores as unjudged.
+    assert backup._build_lesson_no_rule({"lesson_id": 10, "why": " "}, maps) is None
+
+
+def test_the_no_rule_migration_follows_the_link_migration():
+    path = Path(__file__).resolve().parents[1] / "alembic" / "versions" / "0112_lesson_no_rule.py"
+    spec = importlib.util.spec_from_file_location("m0112", path)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    assert (module.revision, module.down_revision) == ("0112", "0111")

tests/test_note_usage.py

@@ -98,9 +98,12 @@ async def test_usage_for_notes_splits_counts_by_event():
         (3, "pulled", 2, ts, False),
     ]
     session = MagicMock()
-    session.execute = AsyncMock(
-        return_value=MagicMock(all=MagicMock(return_value=rows))
-    )
+    # Two aggregates in one session: every event, then the away-from-home
+    # subset (note_id, event, count) — none here.
+    session.execute = AsyncMock(side_effect=[
+        MagicMock(all=MagicMock(return_value=rows)),
+        MagicMock(all=MagicMock(return_value=[])),
+    ])
     ctx = MagicMock()
     ctx.__aenter__ = AsyncMock(return_value=session)
     ctx.__aexit__ = AsyncMock(return_value=False)
@@ -115,6 +118,36 @@ async def test_usage_for_notes_splits_counts_by_event():
     assert out[3]["last_pulled_at"] == ts.isoformat()
 
 
+async def test_usage_for_notes_counts_what_happened_away_from_home():
+    """Surfaced or opened on a project other than the record's own is the
+    evidence a record transferred (#3735). It is a SUBSET of the totals, read
+    from the second aggregate, and it must land on the right counter."""
+    from datetime import datetime, timezone
+
+    ts = datetime(2026, 10, 1, tzinfo=timezone.utc)
+    session = MagicMock()
+    session.execute = AsyncMock(side_effect=[
+        MagicMock(all=MagicMock(return_value=[
+            (3, "surfaced", 9, ts, False),
+            (3, "pulled", 4, ts, False),
+            (4, "surfaced", 2, ts, False),
+        ])),
+        MagicMock(all=MagicMock(return_value=[
+            (3, "surfaced", 5),
+            (3, "pulled", 1),
+        ])),
+    ])
+    ctx = MagicMock()
+    ctx.__aenter__ = AsyncMock(return_value=session)
+    ctx.__aexit__ = AsyncMock(return_value=False)
+    with patch.object(note_usage, "async_session", return_value=ctx):
+        out = await usage_for_notes([3, 4])
+    assert (out[3]["surfaced_away_count"], out[3]["pulled_away_count"]) == (5, 1)
+    assert (out[3]["surfaced_count"], out[3]["pull_count"]) == (9, 4)
+    # Used only at home: the away counters stay zero, not missing.
+    assert (out[4]["surfaced_away_count"], out[4]["pulled_away_count"]) == (0, 0)
+
+
 async def test_usage_readout_failure_degrades_to_zeroes():
     """A telemetry readout must not be able to break the list it decorates."""
     with patch.object(note_usage, "async_session", side_effect=RuntimeError("boom")):
@@ -303,3 +336,53 @@ async def test_record_pulled_lands_end_to_end_from_a_running_loop(_dispose_engin
         assert out[nid]["pull_count"] == 1
     finally:
         await _purge(nid)
+
+
+@pytest.mark.integration
+async def test_away_counts_compare_the_readers_project_with_the_records(_dispose_engine):
+    """The real join: an event counts as away only when the reader's project is
+    known and differs from the record's own. Home, unreported and ambient
+    events are all left out — each would otherwise read as transfer."""
+    from sqlalchemy import delete
+
+    from scribe.models import async_session
+    from scribe.models.note import Note
+    from scribe.models.project import Project
+    from scribe.services.note_usage import _insert_events
+    from tests.helpers import ensure_user
+
+    async with async_session() as s:
+        owner = await ensure_user(s, "usage_away_owner")
+        home = Project(user_id=owner.id, title="usage home")
+        away = Project(user_id=owner.id, title="usage away")
+        s.add_all([home, away])
+        await s.flush()
+        lesson = Note(user_id=owner.id, project_id=home.id, title="a lesson",
+                      note_type="lesson")
+        s.add(lesson)
+        await s.flush()
+        nid, home_id, away_id, uid = lesson.id, home.id, away.id, owner.id
+        await s.commit()
+    try:
+        def ev(event, source, pid):
+            return {"user_id": uid, "note_id": nid, "event": event,
+                    "source": source, "project_id": pid}
+        await _insert_events([
+            ev("surfaced", "lesson_slot", away_id),     # away: counts
+            ev("surfaced", "lesson_slot", home_id),     # home
+            ev("surfaced", "lesson_slot", None),        # unreported
+            ev("surfaced", "enter_project", away_id),   # ambient
+            ev("pulled", "mcp_get_lesson", away_id),    # away: counts
+            ev("pulled", "mcp_get_lesson", None),       # unreported
+        ])
+        out = await usage_for_notes([nid])
+        assert out[nid]["surfaced_away_count"] == 1
+        assert out[nid]["pulled_away_count"] == 1
+        assert out[nid]["surfaced_count"] == 3
+        assert out[nid]["pull_count"] == 2
+    finally:
+        await _purge(nid)
+        async with async_session() as s:
+            await s.execute(delete(Note).where(Note.id == nid))
+            await s.execute(delete(Project).where(Project.id.in_([home_id, away_id])))
+            await s.commit()

tests/test_services_backup.py

@@ -27,7 +27,7 @@ def test_backup_version_is_current():
 
     (Named for the number it asserted until v10, which is exactly the drift a
     name-carrying-a-value invites; it now says what it checks.)"""
-    assert backup.BACKUP_VERSION == 17
+    assert backup.BACKUP_VERSION == 19
 
 
 def _exportable_note(**over):
@@ -131,6 +131,7 @@ def test_a_repo_binding_carries_the_branch_its_ledger_follows():
 def _column_guard_targets():
     from scribe.models.canonical_system import CanonicalSystem
     from scribe.models.code_shape import CodeShape, CodeShapeEvent, CodeShapeUse
+    from scribe.models.lesson_rule_link import LessonNoRule, LessonRuleLink
     from scribe.models.design_system import DesignSystem, DesignToken
     from scribe.models.milestone import Milestone
     from scribe.models.note import Note
@@ -167,6 +168,8 @@ def _column_guard_targets():
         "record_systems": (RecordSystem, backup._record_system_rows),
         "note_supersessions": (NoteSupersession, backup._note_supersession_rows),
         "rule_relations": (RuleRelation, backup._rule_relation_rows),
+        "lesson_rule_links": (LessonRuleLink, backup._lesson_rule_link_rows),
+        "lesson_no_rule": (LessonNoRule, backup._lesson_no_rule_rows),
         "note_usage_events": (NoteUsageEvent, backup._usage_event_rows),
         "rule_usage_events": (RuleUsageEvent, backup._rule_usage_event_rows),
         "retrieval_tuning_events": (
@@ -283,6 +286,8 @@ def _import_guard_targets():
         "record_systems": backup._build_record_system,
         "note_supersessions": backup._build_note_supersession,
         "rule_relations": backup._build_rule_relation,
+        "lesson_rule_links": backup._build_lesson_rule_link,
+        "lesson_no_rule": backup._build_lesson_no_rule,
         "note_usage_events": backup._build_usage_event,
         "rule_usage_events": backup._build_rule_usage_event,
         "retrieval_tuning_events": backup._build_retrieval_tuning_event,
@@ -534,7 +539,11 @@ async def test_export_full_backup_contains_every_declared_section():
                 "note_supersessions", "code_shapes", "code_shape_events",
                 "code_shape_uses",
                 # v16: the reasons beside the settings they explain.
-                "retrieval_tuning_events"):
+                "retrieval_tuning_events",
+                # v18: which rule each lesson is an instance of.
+                "lesson_rule_links",
+                # v19: the lessons judged to fall under no rule.
+                "lesson_no_rule"):
         assert key in out, f"missing export section: {key}"
         assert out[key] == []