Merge pull request 'Milestone 440 steps 4, 5, 7: rules surface through their lessons, convergence named at the write, both records show the link' (#191) from dev into main

frontend/src/api/lessons.ts

@@ -1,6 +1,34 @@
 import type { RecordUsage } from "@/types/usage";
 
-import { apiGet, apiPost, apiPatch, apiDelete } from "@/api/client";
+import { apiGet, apiPost, apiPut, apiPatch, apiDelete } from "@/api/client";
+import type { RuleKind } from "@/api/rulebooks";
+
+/** Where a lesson stands on "which rule is this an instance of?"
+ *  (`services/lesson_rules.py`: LINKED / NO_RULE / UNJUDGED). Three answers,
+ *  not two: "looked at and found to stand alone" and "nobody has looked" are
+ *  different facts, and the second is the one worth acting on. A rejected link
+ *  alone leaves a lesson unjudged — "not that rule" does not say whether
+ *  another one fits. */
+export type RuleJudgment = "linked" | "no_rule" | "unjudged";
+
+/** A link's state (`models/lesson_rule_link.py` LINK_STATES). */
+export type LinkState = "suggested" | "confirmed" | "rejected";
+
+/** One link from a lesson to a rule, in the reader's view: only rules the
+ *  reader owns are listed (`rules_for_lessons`). */
+export interface LessonRuleLink {
+  id: number;
+  title: string;
+  kind: RuleKind;
+  /** `suggested` carries nothing in retrieval until it is judged; only
+   *  `confirmed` changes what surfaces; `rejected` is kept so the pair is
+   *  never proposed again. */
+  state: LinkState;
+  /** Why it was confirmed or rejected. */
+  note: string;
+  /** What a suggestion rests on — sent on suggested links only. */
+  evidence?: { situations: number; projects: number; co_surfaced: number };
+}
 
 /** A lesson: a transferable insight, retrievable by the SITUATION it applies
  *  to rather than by its topic.
@@ -51,6 +79,13 @@ export interface Lesson {
   updated_at: string | null;
   systems?: { id: number; name: string }[];
   usage?: RecordUsage;
+  /** The rules this lesson points at, confirmed first, then suggested, then
+   *  rejected. Absent (with `rule_judgment`) when the links could not be read
+   *  — which is "not attached", never "no rule". */
+  rules?: LessonRuleLink[];
+  rule_judgment?: RuleJudgment;
+  /** The "no rule fits" answer, present when that is the judgment. */
+  no_rule?: { why: string; judged_at: string | null };
   /** Set when another user owns this record. */
   shared?: boolean;
   owner?: string | null;
@@ -128,6 +163,17 @@ export function updateLesson(
   return apiPatch<Lesson>(`/api/lessons/${id}`, payload);
 }
 
+/** Confirm or reject one lesson→rule link. Confirming also clears a "no rule
+ *  fits" answer: the two cannot both be the current answer. */
+export function judgeLessonLink(
+  lessonId: number,
+  ruleId: number,
+  verdict: "confirm" | "reject",
+  note = "",
+): Promise<unknown> {
+  return apiPut(`/api/lessons/${lessonId}/rules/${ruleId}`, { verdict, note });
+}
+
 /** Trash, not erase — recoverable. `apiDelete` discards the body, which is the
  *  established shape here (snippets delete the same way): the batch id is in
  *  the response, but no caller has needed it and inventing a second delete

frontend/src/api/rulebooks.ts

@@ -1,6 +1,7 @@
 import type { RecordUsage } from "@/types/usage";
 
 import { apiGet, apiPost, apiPatch, apiDelete } from "@/api/client";
+import type { LinkState } from "@/api/lessons";
 
 /** How a rule reaches a session (milestone 307). */
 
@@ -82,6 +83,10 @@ export interface Rule {
   /** Present only when the rule has them (the server omits empty keys). */
   systems?: { id: number; name: string }[];
   relations?: RuleRelation[];
+  /** The lessons that point at this rule (milestone 440) — the concrete
+   *  situations judged instances of it, plus any suggested and awaiting a
+   *  judgment. Readable lessons only; omitted when there are none. */
+  lessons?: { id: number; title: string; state: LinkState; note: string }[];
 }
 
 /**

frontend/src/components/rules/RuleEditorSlideOver.vue

@@ -27,6 +27,14 @@ const expiresWhen = ref("");
 
 const relations = computed(() => store.currentRule?.relations ?? []);
 
+// The lessons that point at this rule (milestone 440): confirmed instances
+// first, then suggestions waiting on a judgment. A rejected lesson does not
+// point at the rule, so it is not listed here — it stays readable on the
+// lesson, where the judgment was made.
+const lessons = computed(() =>
+  (store.currentRule?.lessons ?? []).filter((l) => l.state !== "rejected"),
+);
+
 // The label a reader needs to judge an edge, not the stored token.
 const RELATION_LABEL: Record<string, { outgoing: string; incoming: string }> = {
   co_surfaces: { outgoing: "arrives with", incoming: "arrives with" },
@@ -279,6 +287,21 @@ watch(() => props.ruleId, load);
         </p>
       </section>
 
+      <section v-if="lessons.length" class="relations">
+        <h3>Lessons that are instances of it</h3>
+        <ul>
+          <li v-for="l in lessons" :key="l.id" class="relation">
+            <router-link :to="`/lessons/${l.id}`" class="relation-target">{{ l.title }}</router-link>
+            <span v-if="l.state === 'suggested'" class="rule-chip">suggested</span>
+            <span v-if="l.note" class="relation-note">{{ l.note }}</span>
+          </li>
+        </ul>
+        <p class="field-note">
+          The situations that keep proving this rule. A confirmed lesson brings the rule along
+          when it surfaces; a suggested one waits for a judgment on the lesson's page.
+        </p>
+      </section>
+
       <label>
         Why
         <textarea v-model="why" rows="4" placeholder="Rationale — the reason this rule exists." />
@@ -394,3 +417,8 @@ legend { padding: 0 0.35rem; font-size: 0.8rem; color: var(--fs-text-tertiary);
 .trash, .close { background: none; border: none; cursor: pointer; opacity: 0.6; font-size: 1.25em; }
 .trash:hover, .close:hover { opacity: 1; }
 </style>
+
+<!-- `.rule-chip` for the "suggested" marker on a lesson — the chip the rule
+     panes already use, loaded here because the slide-over can open with no
+     pane that loads it (a link straight to ?rule=N). -->
+<style src="@/assets/rules-shared.css" />

frontend/src/utils/permission.ts

@@ -0,0 +1,12 @@
+/** Whether the reader may change a record, from the `permission` its read
+ *  carried. The levels are services/access.py's PERMISSION_RANK — viewer,
+ *  editor, admin, owner — and absent means the reader owns it: the server
+ *  labels a record only when it reached the reader through a share. This only
+ *  decides which controls a page shows; the server enforces the write. Kept in
+ *  one place so every page agrees, with each other and with the server, about
+ *  who can edit. */
+export const WRITE_LEVELS: readonly string[] = ["editor", "admin", "owner"];
+
+export function canWriteRecord(permission: string | undefined): boolean {
+  return permission === undefined || WRITE_LEVELS.includes(permission);
+}

frontend/src/views/LessonDetailView.vue

@@ -16,18 +16,31 @@
  * difference between this kind and a rule, and it is the reason the kind
  * exists (#3727) — sessions were proposing rules for things that should never
  * have bound anyone.
+ *
+ * BUT IT CAN POINT AT ONE (milestone 440). A lesson that is an instance of a
+ * rule says so, and the page shows which of the three answers it holds: the
+ * rule(s) it was judged an instance of, "no rule fits" with the reason, or
+ * not yet judged. The third is the one worth acting on, so it is stated as
+ * plainly as the other two rather than left as an empty panel.
  */
 import { computed, onMounted, ref, watch } from "vue";
 import { useRoute, useRouter } from "vue-router";
 
 import { apiErrorMessage } from "@/api/client";
-import { deleteLesson, getLesson, type Lesson } from "@/api/lessons";
+import {
+  deleteLesson,
+  getLesson,
+  judgeLessonLink,
+  type Lesson,
+  type LessonRuleLink,
+} from "@/api/lessons";
 import ConfirmDialog from "@/components/ConfirmDialog.vue";
 import TagPill from "@/components/TagPill.vue";
 import UsageBadge from "@/components/UsageBadge.vue";
 import { DEAD_WEIGHT_ADVICE } from "@/utils/deadWeight";
 import { useToastStore } from "@/stores/toast";
 import { renderMarkdown } from "@/utils/markdown";
+import { canWriteRecord } from "@/utils/permission";
 
 const route = useRoute();
 const router = useRouter();
@@ -55,6 +68,53 @@ function sourceHref(rec: { id: number; is_task?: boolean; note_type?: string })
   return `/notes/${rec.id}`;
 }
 
+const canWrite = computed(() => canWriteRecord(lesson.value?.permission));
+
+/** The links by state — each state reads differently. Confirmed links are
+ *  the answer; suggested ones are evidence waiting on a judgment; rejected
+ *  ones are kept so the pair is never proposed again, and shown so the reason
+ *  stays readable. */
+const linksIn = (state: LessonRuleLink["state"]) =>
+  computed(() => (lesson.value?.rules ?? []).filter((r) => r.state === state));
+const confirmedRules = linksIn("confirmed");
+const suggestedRules = linksIn("suggested");
+const rejectedRules = linksIn("rejected");
+
+/** A rule opens in the rules view's editor, which takes the rule alone. */
+function ruleHref(ruleId: number) {
+  return { path: "/rules", query: { rule: String(ruleId) } };
+}
+
+/** What a suggestion rests on, in the units it was proposed by: distinct
+ *  situations, not raw arrivals — one situation repeated proves nothing. */
+function evidenceLine(ev: NonNullable<LessonRuleLink["evidence"]>): string {
+  const situations = `${ev.situations} distinct situation${ev.situations === 1 ? "" : "s"}`;
+  const projects = ev.projects > 1 ? `, across ${ev.projects} projects` : "";
+  return `Arrived together in ${situations}${projects}.`;
+}
+
+const judging = ref<number | null>(null);
+
+/** Judge a suggested link from here. Re-reads the lesson rather than editing
+ *  the list in place: confirming also clears a "no rule fits" answer, and the
+ *  judgment line has to come back from the server to be right. */
+async function judge(rule: LessonRuleLink, verdict: "confirm" | "reject") {
+  judging.value = rule.id;
+  try {
+    await judgeLessonLink(lessonId.value, rule.id, verdict);
+    lesson.value = await getLesson(lessonId.value);
+    toast.show(
+      verdict === "confirm"
+        ? "Linked — the rule can now surface through this lesson"
+        : "Not an instance — this pair will not be suggested again",
+    );
+  } catch (e) {
+    toast.show(apiErrorMessage(e, "Failed to record the judgment"), "error");
+  } finally {
+    judging.value = null;
+  }
+}
+
 async function load() {
   loading.value = true;
   error.value = null;
@@ -141,6 +201,80 @@ onMounted(load);
         </ul>
       </section>
 
+      <!-- Which rule this is an instance of. Absent when the links could not
+           be read: that is "not attached", never "no rule". -->
+      <section v-if="lesson.rule_judgment" class="ld-panel">
+        <h2 class="ld-panel-label">Instance of</h2>
+
+        <template v-if="lesson.rule_judgment === 'linked'">
+          <ul v-if="confirmedRules.length" class="ld-sources">
+            <li v-for="r in confirmedRules" :key="r.id">
+              <router-link :to="ruleHref(r.id)" class="ld-link">
+                {{ r.title }}
+              </router-link>
+              <span v-if="r.kind === 'preference'" class="rule-chip rule-chip-preference">
+                preference
+              </span>
+              <p v-if="r.note" class="ld-link-note">{{ r.note }}</p>
+            </li>
+          </ul>
+          <!-- Judged by its owner against a rule this reader cannot open. -->
+          <p v-else class="ld-muted">Linked to a rule that is not shared with you.</p>
+        </template>
+        <p v-else-if="lesson.rule_judgment === 'no_rule'" class="ld-judgment">
+          No rule — {{ lesson.no_rule?.why }}
+        </p>
+        <p v-else class="ld-muted ld-empty">
+          Not yet judged — no rule has been named for this situation, and
+          nobody has said none fits.
+        </p>
+
+        <template v-if="suggestedRules.length">
+          <h3 class="ld-sub-label">Suggested</h3>
+          <ul class="ld-sources">
+            <li v-for="r in suggestedRules" :key="r.id">
+              <router-link :to="ruleHref(r.id)" class="ld-link">
+                {{ r.title }}
+              </router-link>
+              <span v-if="r.kind === 'preference'" class="rule-chip rule-chip-preference">
+                preference
+              </span>
+              <p v-if="r.evidence" class="ld-link-note">{{ evidenceLine(r.evidence) }}</p>
+              <div v-if="canWrite" class="ld-judge">
+                <button
+                  type="button"
+                  class="ld-ghost ld-small"
+                  :disabled="judging === r.id"
+                  @click="judge(r, 'confirm')"
+                >
+                  Confirm
+                </button>
+                <button
+                  type="button"
+                  class="ld-ghost ld-small"
+                  :disabled="judging === r.id"
+                  @click="judge(r, 'reject')"
+                >
+                  Not an instance
+                </button>
+              </div>
+            </li>
+          </ul>
+        </template>
+
+        <template v-if="rejectedRules.length">
+          <h3 class="ld-sub-label">Judged not an instance</h3>
+          <ul class="ld-sources ld-rejected">
+            <li v-for="r in rejectedRules" :key="r.id">
+              <router-link :to="ruleHref(r.id)" class="ld-link">
+                {{ r.title }}
+              </router-link>
+              <p v-if="r.note" class="ld-link-note">{{ r.note }}</p>
+            </li>
+          </ul>
+        </template>
+      </section>
+
       <section v-if="lesson.tags?.length" class="ld-tags">
         <TagPill v-for="t in lesson.tags" :key="t" :tag="t" />
       </section>
@@ -263,6 +397,24 @@ onMounted(load);
   font-size: 0.78rem;
 }
 
+.ld-link-note {
+  margin: var(--fs-space-1) 0 0;
+  color: var(--fs-text-secondary);
+  font-size: var(--fs-size-label);
+  line-height: 1.45;
+}
+.ld-judgment { margin: 0; font-size: 0.9rem; line-height: 1.5; }
+.ld-sub-label {
+  margin: var(--fs-space-3) 0 var(--fs-space-1);
+  font-size: var(--fs-size-label);
+  font-weight: var(--fs-weight-medium);
+  color: var(--fs-text-tertiary);
+}
+.ld-rejected .ld-link { color: var(--fs-text-secondary); }
+.ld-judge { display: flex; gap: var(--fs-space-2); margin-top: var(--fs-space-1); }
+.ld-small { padding: 0.15rem 0.6rem; font-size: var(--fs-size-label); }
+.ld-ghost:disabled { opacity: var(--fs-disabled-opacity); cursor: default; }
+
 .ld-tags { display: flex; flex-wrap: wrap; gap: 0.35rem; margin-top: 1.25rem; }
 
 .ld-origin {
@@ -307,3 +459,7 @@ onMounted(load);
   font-size: 0.88rem;
 }
 </style>
+
+<!-- `.rule-chip` and its preference variant: the marker a rule's kind wears
+     everywhere else, reused rather than re-spelled. -->
+<style src="@/assets/rules-shared.css" />

frontend/src/views/SnippetDetailView.vue

@@ -10,6 +10,7 @@ import {
 import { useToastStore } from "@/stores/toast";
 import ConfirmDialog from "@/components/ConfirmDialog.vue";
 import { renderMarkdown } from "@/utils/markdown";
+import { canWriteRecord } from "@/utils/permission";
 
 const route = useRoute();
 const router = useRouter();
@@ -22,10 +23,7 @@ const showDeleteConfirm = ref(false);
 
 const id = computed(() => Number(route.params.id));
 
-const canWrite = computed(() => {
+const canWrite = computed(() => canWriteRecord(snippet.value?.permission));
-  const p = snippet.value?.permission;
-  return p === undefined || p === "owner" || p === "edit" || p === "admin";
-});
 
 const locations = computed(() => snippet.value?.snippet.locations ?? []);
 

src/scribe/mcp/tools/lessons.py

@@ -188,6 +188,9 @@ async def create_lesson(
             (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.
+            When other lessons in the same situation also answered "no rule
+            fits", the response carries `convergence`: the group, and the
+            rule it may be missing.
         force: Create even if a near-duplicate exists.
 
     Returns the created lesson with its `rules` and `rule_judgment`, plus
@@ -237,9 +240,20 @@ async def create_lesson(
     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)
+    elif no_rule.strip():
+        await _name_convergence(uid, data, note.id)
     return data
 
 
+async def _name_convergence(uid: int, data: dict, lesson_id: int) -> None:
+    """A "no rule fits" answer is the moment to notice it is not the first
+    for this situation (#4634) — `convergence` names the group and the rule
+    it may be missing. Absent when there is no group."""
+    group = await lesson_rules_svc.convergence_for(uid, lesson_id)
+    if group:
+        data["convergence"] = group
+
+
 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.
 
@@ -343,7 +357,9 @@ async def update_lesson(
         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.
+            non-empty `rule_ids`, not both. When other lessons in the same
+            situation also answered "no rule fits", the response carries
+            `convergence`: the group, and the rule it may be missing.
     """
     uid = current_user_id()
     linked = (
@@ -373,6 +389,8 @@ async def update_lesson(
         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])
+    if no_rule.strip():
+        await _name_convergence(uid, out, lesson_id)
     return out
 
 

src/scribe/routes/lessons.py

@@ -188,6 +188,10 @@ async def create_lesson_route():
     elif no_rule:
         await lesson_rules_svc.set_no_rule(uid, note.id, no_rule)
     out = lessons_svc.lesson_to_dict(note)
+    if no_rule:
+        group = await lesson_rules_svc.convergence_for(uid, note.id)
+        if group:
+            out["convergence"] = group
     out["systems"] = [
         s.to_dict() for s in await systems_svc.list_record_systems(uid, note.id)
     ]
@@ -292,6 +296,12 @@ async def update_lesson_route(lesson_id: int):
         # may edit — it decides WHOSE reach the tagging uses (#47).
         await systems_svc.set_record_systems(uid, lesson_id, data["system_ids"])
     out = lessons_svc.lesson_to_dict(updated)
+    if no_rule:
+        # The same nudge the MCP door gives (#4634): this answer may complete
+        # a group of no-rule lessons in one situation.
+        group = await lesson_rules_svc.convergence_for(uid, lesson_id)
+        if group:
+            out["convergence"] = group
     out["systems"] = [
         s.to_dict()
         for s in await systems_svc.list_record_systems(owner_uid, lesson_id)

src/scribe/services/lesson_rules.py

@@ -39,7 +39,7 @@ import logging
 import re
 from datetime import datetime, timedelta, timezone
 
-from sqlalchemy import and_, delete, exists, func, not_, select
+from sqlalchemy import and_, delete, exists, func, not_, or_, select
 from sqlalchemy.exc import IntegrityError
 
 from scribe.models import async_session
@@ -650,3 +650,186 @@ async def _co_surfaced(user_id, lesson_ids, rule_ids, arm, situation, project_id
         )).scalar_one_or_none() or ""
     rule = owned[rid]
     return _proposal_line(lid, lesson_title, rid, rule.title, rule.kind or "rule", ev)
+
+
+# ── Reaching a rule through its lessons (#4633) ──────────────────────────────
+
+
+async def confirmed_lessons(user_id: int) -> set[int]:
+    """The lessons, readable by the caller, that carry a CONFIRMED link to a
+    rule the caller owns — the only lessons that can bring a rule along.
+
+    CONFIRMED ONLY, and the reason is the soft link's: a suggested link that
+    carried its rule would put the two side by side on every prompt that
+    found the lesson, which is exactly the co-arrival #4637 counts as
+    evidence — the suggestion would manufacture its own proof.
+
+    Asked first and cheaply, so an arm on an install with no confirmed links
+    — every install, until someone judges one — never pays for a search.
+    """
+    from scribe.services.access import readable_notes_clause
+    from scribe.services.rulebooks import _owned_rules_clause
+
+    async with async_session() as session:
+        rows = (await session.execute(
+            select(LessonRuleLink.lesson_id)
+            .join(Rule, Rule.id == LessonRuleLink.rule_id)
+            .join(Note, Note.id == LessonRuleLink.lesson_id)
+            .where(LessonRuleLink.state == CONFIRMED)
+            .where(Rule.deleted_at.is_(None))
+            .where(_owned_rules_clause(user_id))
+            .where(Note.deleted_at.is_(None))
+            .where(readable_notes_clause(user_id))
+            .distinct()
+        )).scalars().all()
+    return {int(i) for i in rows}
+
+
+async def confirmed_rules_in_scope(
+    user_id: int, lesson_ids, project_id: int | None,
+) -> dict[int, list]:
+    """{lesson_id: [Rule]} — each lesson's CONFIRMED rules that may surface here.
+
+    "May surface here" is the rule arms' own scope (milestone 414): a global
+    rule, or a rule of THIS project. A lesson is project-independent and
+    surfaces everywhere, so without this a lesson linked to project A's rule
+    would carry that rule into project B's sessions — the leak milestone 414
+    closed for the direct arms, reopened through a side door.
+    """
+    from scribe.services.rulebooks import _owned_rules_clause
+
+    ids = [int(i) for i in lesson_ids or []]
+    if not ids:
+        return {}
+    home = (
+        or_(Rule.topic_id.isnot(None), Rule.project_id == project_id)
+        if project_id else Rule.topic_id.isnot(None)
+    )
+    async with async_session() as session:
+        rows = (await session.execute(
+            select(LessonRuleLink.lesson_id, Rule)
+            .join(Rule, Rule.id == LessonRuleLink.rule_id)
+            .where(LessonRuleLink.lesson_id.in_(ids))
+            .where(LessonRuleLink.state == CONFIRMED)
+            .where(Rule.deleted_at.is_(None))
+            .where(_owned_rules_clause(user_id))
+            .where(home)
+            .order_by(Rule.id)
+        )).all()
+    out: dict[int, list] = {}
+    for lesson_id, rule in rows:
+        out.setdefault(int(lesson_id), []).append(rule)
+    return out
+
+
+# ── Convergence: lessons with no rule that keep landing in one place (#4634) ─
+#
+# A lesson answered "no rule fits" is a situation nothing binds. One is a
+# lesson. Several that resemble each other are what a missing rule looks like
+# from the outside — the same moment met again and again, each time written
+# down as advice. This is noticed at the WRITE, when the newest of them is
+# answered, and never by a sweep or a timer (#4183): the reader is in the
+# situation then, and a nudge arriving anywhere else is one nobody acts on.
+#
+# DEFAULTS, stated as defaults (rules 32, 115). Three lessons — the new one
+# and two it resembles — is the smallest group that is a pattern rather than
+# a pair. The similarity bar sits above the notes menu's ("worth showing")
+# and below the duplicate gate's ("the same record"): these lessons should be
+# about one situation without being one lesson written twice, which the
+# duplicate gate already catches.
+CONVERGENCE_LESSONS = 3
+CONVERGENCE_THRESHOLD = 0.65
+# Candidates fetched before keeping the no-rule ones; most lessons near a
+# situation may well have rules.
+_CONVERGENCE_FETCH = 20
+
+
+def convergence_group(members: list[dict]) -> dict | None:
+    """Decide from a candidate group — the new lesson FIRST, then those it
+    resembles — whether it names a missing rule. Pure, so the bar is testable.
+
+    Each member is {id, title, sources, project_id}. The bar counts DISTINCT
+    LESSONS, never incidents: a single broad lesson drawn from many incidents
+    is still one judgment about one situation, and incidents cannot stand in
+    for lessons. And when every member says what taught it and all of them
+    point at the same single incident, that is one event written up several
+    times, not a situation recurring — the group stays quiet.
+    """
+    seen: set[int] = set()
+    group = []
+    for m in members:
+        if m["id"] not in seen:
+            seen.add(m["id"])
+            group.append(m)
+    if len(group) < CONVERGENCE_LESSONS:
+        return None
+    incidents = sorted({s for m in group for s in (m.get("sources") or [])})
+    if all(m.get("sources") for m in group) and len(incidents) < 2:
+        return None
+    projects = sorted({m["project_id"] for m in group if m.get("project_id")})
+    named = ", ".join(f"#{m['id']} “{m['title']}”" for m in group)
+    return {
+        "lessons": [{"id": m["id"], "title": m["title"]} for m in group],
+        "incidents": incidents,
+        "projects": projects,
+        "hint": (
+            f"{len(group)} lessons answered \"no rule fits\" and keep landing "
+            f"in one situation: {named}. A situation met this often may want a "
+            "rule — the binding choice they each circle. Draft it with "
+            "create_rule (it goes to the operator, as every rule does), then "
+            "point each lesson at it with update_lesson(lesson_id, "
+            "rule_ids=[<new rule id>]). If no single choice is right every "
+            "time, the lessons are the right record and nothing more is needed."
+        ),
+    }
+
+
+async def _no_rule_ids(lesson_ids) -> set[int]:
+    ids = [int(i) for i in lesson_ids or []]
+    if not ids:
+        return set()
+    async with async_session() as session:
+        rows = (await session.execute(
+            select(LessonNoRule.lesson_id).where(LessonNoRule.lesson_id.in_(ids))
+        )).scalars().all()
+    return {int(i) for i in rows}
+
+
+async def convergence_for(user_id: int, lesson_id: int) -> dict | None:
+    """The convergence a newly answered no-rule lesson completes, or None.
+
+    Fail-open: this decorates a write that already succeeded, so a failed
+    search leaves the response as it was (snippet #4286's reasoning).
+    """
+    try:
+        return await _convergence_for(user_id, lesson_id)
+    except Exception:
+        logger.warning("lesson convergence could not be checked", exc_info=True)
+        return None
+
+
+async def _convergence_for(user_id: int, lesson_id: int) -> dict | None:
+    from scribe.services import lessons as lessons_svc
+    from scribe.services.embeddings import semantic_search_notes, trigger_title
+
+    lesson = await lessons_svc.get_lesson(user_id, lesson_id)
+    if lesson is None:
+        return None
+    data = lesson.data if isinstance(lesson.data, dict) else {}
+    query = trigger_title(data.get("what") or lesson.title, lessons_svc.lesson_trigger(lesson))
+    found = await semantic_search_notes(
+        user_id, query, exclude_ids={int(lesson.id)}, limit=_CONVERGENCE_FETCH,
+        threshold=CONVERGENCE_THRESHOLD, note_type=(lessons_svc.LESSON_NOTE_TYPE,),
+        include_global_kinds=True, scope="browse",
+    )
+    answered = await _no_rule_ids([int(n.id) for _s, n in found])
+
+    def member(note) -> dict:
+        return {
+            "id": int(note.id), "title": note.title,
+            "sources": lessons_svc.lesson_sources(note), "project_id": note.project_id,
+        }
+
+    return convergence_group(
+        [member(lesson)] + [member(n) for _s, n in found if int(n.id) in answered]
+    )

src/scribe/services/plugin_context.py

@@ -1390,6 +1390,126 @@ async def _reserve_slot_for_preference(
     return hits + slot, slot_id
 
 
+# ── A rule reached through its lessons (milestone 440, #4633) ──────────────
+#
+# A rule's own document is written in the rule's words, which are general by
+# design; the situations that keep proving it are often closer to what a
+# session is actually doing. A lesson JUDGED to be an instance of a rule (a
+# CONFIRMED link) carries that situation, so a lesson matching the moment
+# brings its rule along — in rule voice, naming the lesson that reached it.
+#
+# ITS OWN SLOT, NOT A RULE SLOT, as a stated default. The plan asked for this
+# to be settled by measurement, and there is nothing to measure until links
+# are confirmed (#4632). Until then the asymmetry decides it: a via-lesson
+# line that took a rule slot could push out a rule the ranker matched
+# DIRECTLY, a stronger claim displaced by a weaker one, while an extra line
+# costs one line. `rule_via_lesson` is logged as its own source so the
+# question can be answered from data later (#4636).
+VIA_LESSON_LIMIT = 1
+# Lessons fetched before keeping only the linked ones. The search cannot be
+# told "linked lessons only", so it overfetches and filters; on a corpus where
+# most lessons are unlinked, a fetch of one would almost always be spent on a
+# lesson that carries nothing.
+_VIA_LESSON_OVERFETCH = 10
+
+
+async def _rules_via_lessons(
+    user_id: int, query: str, *, project_id: int | None, skip: set[int],
+    held: set[int], where: str,
+) -> tuple[list[str], list[int]]:
+    """Rule lines reached through a matching lesson's CONFIRMED links.
+
+    The lesson bar is the notes menu's own threshold — the bar a lesson has to
+    clear to be shown at all — so a lesson too weak to surface cannot carry a
+    rule in. `skip` is every rule this response already names plus the
+    session ledger: suppression applies to the RULE, whichever lesson reached
+    it. Returns (lines, rule ids shown). Fails open, like every arm.
+    """
+    try:
+        confirmed = await lesson_rules_svc.confirmed_lessons(user_id)
+        if not confirmed:
+            return [], []
+        bar = (await get_autoinject_config(user_id))["threshold"]
+        t0 = time.perf_counter()
+        _rep: dict = {}
+        found = await semantic_search_notes(
+            user_id, query, limit=_VIA_LESSON_OVERFETCH, threshold=bar,
+            project_id=project_id, note_type=(LESSON_NOTE_TYPE,),
+            include_global_kinds=True, scope="browse", report=_rep,
+        )
+        matched = [(s, n) for s, n in found if int(n.id) in confirmed]
+        by_lesson = await lesson_rules_svc.confirmed_rules_in_scope(
+            user_id, [int(n.id) for _s, n in matched], project_id,
+        )
+        candidates = [
+            (s, rule, n) for s, n in matched for rule in by_lesson.get(int(n.id), [])
+        ]
+        chosen: list = []
+        taken: set[int] = set(skip)
+        for s, rule, n in candidates:
+            if rule.id in taken:
+                continue
+            taken.add(rule.id)
+            chosen.append((s, rule, n))
+            if len(chosen) >= VIA_LESSON_LIMIT:
+                break
+        # Logged whenever a search ran, results or not — the #3497 guard. The
+        # score is the LESSON's, because the lesson is what was matched; the
+        # best-available id is left off for the same reason, since the row's
+        # results are rules and an id beside them would read as a rule id.
+        record_retrieval(
+            user_id=user_id, source="rule_via_lesson", query=query,
+            threshold=bar, limit=VIA_LESSON_LIMIT, project_id=project_id,
+            is_task=None, results=[(s, rule) for s, rule, _n in chosen],
+            duration_ms=(time.perf_counter() - t0) * 1000.0,
+            best_available=_rep.get("best_available_score"),
+            searched=bool(_rep.get("searched", True)),
+            suppressed=len({r.id for _s, r, _n in candidates}) - len(chosen),
+        )
+        if not chosen:
+            return [], []
+        rule_ids = [rule.id for _s, rule, _n in chosen]
+        record_rule_surfaced(user_id=user_id, rule_ids=rule_ids, source="rule_via_lesson")
+        lines = [
+            _rule_hint_line(rule, where=where, seen=False, held=rule.id in held)
+            + f" Reached through lesson #{n.id} "
+            + f"\u201c{_menu_name(n.title, n.note_type, n.data, n.body)}\u201d, "
+            + "a recorded instance of it."
+            for _s, rule, n in chosen
+        ]
+        return lines, rule_ids
+    except Exception:
+        logger.debug("rule-via-lesson arm failed", exc_info=True)
+        return [], []
+
+
+async def _add_rules_via_lessons(
+    user_id: int, out: dict, *, project_id: int, exclude_rule_ids, held_rule_ids,
+    where: str,
+) -> dict:
+    """Run the via-lesson step after a direct rule arm, on the query that arm
+    actually searched with.
+
+    The arm leaves `_via_query` in its payload only when it RAN — enabled, and
+    with something to search — so an arm the operator switched off, or a
+    blank prompt, brings no rule in through a side door either. The key is
+    popped here; it never leaves the server.
+    """
+    query = out.pop("_via_query", "")
+    if not query:
+        return out
+    skip = (set(exclude_rule_ids or []) | set(out.get("rule_ids") or [])
+            | set(out.get("shown_rule_ids") or []))
+    lines, ids = await _rules_via_lessons(
+        user_id, query, project_id=project_id or None, skip=skip,
+        held=set(held_rule_ids or []), where=where,
+    )
+    if lines:
+        out["context"] = "\n".join(c for c in (out.get("context") or "", *lines) if c)
+        out["rule_ids"] = list(out.get("rule_ids") or []) + ids
+    return out
+
+
 async def build_prompt_rule_hint(
     user_id: int,
     query: str,
@@ -1398,6 +1518,28 @@ async def build_prompt_rule_hint(
     exclude_rule_ids: list[int] | None = None,
     held_rule_ids: list[int] | None = None,
     context: str = "",
+) -> dict:
+    """Rules and preferences that may apply to what the operator just asked —
+    matched directly (`_prompt_rule_hint`, where the design is written), then
+    reached through a linked lesson (`_rules_via_lessons`)."""
+    out = await _prompt_rule_hint(
+        user_id, query, project_id=project_id, exclude_rule_ids=exclude_rule_ids,
+        held_rule_ids=held_rule_ids, context=context,
+    )
+    return await _add_rules_via_lessons(
+        user_id, out, project_id=project_id, exclude_rule_ids=exclude_rule_ids,
+        held_rule_ids=held_rule_ids, where="to this request",
+    )
+
+
+async def _prompt_rule_hint(
+    user_id: int,
+    query: str,
+    *,
+    project_id: int = 0,
+    exclude_rule_ids: list[int] | None = None,
+    held_rule_ids: list[int] | None = None,
+    context: str = "",
 ) -> dict:
     """Rules and preferences that may apply to what the operator just asked.
 
@@ -1436,6 +1578,9 @@ async def build_prompt_rule_hint(
     # names it — and "yes, go ahead" names nothing a trigger can match, while
     # the reply it answers ("commit this to dev and push") does.
     q = _autoinject_query(q, context)
+    # The query this arm searches with, for the via-lesson step that follows
+    # it (#4633) — set only once the arm is going to run.
+    out["_via_query"] = q
 
     try:
         threshold = await floor_for(user_id, "prompt_rule")
@@ -2708,6 +2853,17 @@ async def build_write_path_hint(
     except Exception:
         logger.debug("write-path rule arm failed", exc_info=True)
 
+    # A rule reached through a linked lesson (#4633), after the direct band
+    # and never in place of it. Skips what the band already named and what
+    # the session's ledger holds — suppression is by rule.
+    via_lines, via_ids = await _rules_via_lessons(
+        user_id, code or path, project_id=project_id or None,
+        skip=set(exclude_rule_ids or []) | set(shown_rule_ids),
+        held=set(held_rule_ids or []), where="here",
+    )
+    lines.extend(via_lines)
+    rule_ids.extend(via_ids)
+
     # A lesson and a rule on this one response (#4637): the soft-link
     # recorder counts the pair, keyed on the FILE — every edit to one file is
     # one situation — and asks about it once the evidence holds. Fails open.
@@ -2745,6 +2901,28 @@ async def build_tool_rule_hint(
     project_id: int = 0,
     exclude_rule_ids: list[int] | None = None,
     held_rule_ids: list[int] | None = None,
+) -> dict:
+    """Standing rules that may apply to the action about to be taken — matched
+    directly (`_tool_rule_hint`, where the design is written), then reached
+    through a linked lesson (`_rules_via_lessons`, #4633)."""
+    out = await _tool_rule_hint(
+        user_id, tool_name, command, project_id=project_id,
+        exclude_rule_ids=exclude_rule_ids, held_rule_ids=held_rule_ids,
+    )
+    return await _add_rules_via_lessons(
+        user_id, out, project_id=project_id, exclude_rule_ids=exclude_rule_ids,
+        held_rule_ids=held_rule_ids, where=f"to this {tool_name} call",
+    )
+
+
+async def _tool_rule_hint(
+    user_id: int,
+    tool_name: str,
+    command: str,
+    *,
+    project_id: int = 0,
+    exclude_rule_ids: list[int] | None = None,
+    held_rule_ids: list[int] | None = None,
 ) -> dict:
     """Standing rules that may apply to the ACTION about to be taken (#3476).
 
@@ -2795,6 +2973,8 @@ async def build_tool_rule_hint(
         # embedding window, so it is bounded — the verb and its target sit at
         # the front, which is the part a rule is about.
         query = command[:_TOOL_QUERY_CHARS]
+        # For the via-lesson step (#4633) — set only once the arm is enabled.
+        out["_via_query"] = query
 
         t0 = time.perf_counter()
         _rep_ptr: dict = {}

src/scribe/services/retrieval_registry.py

@@ -137,6 +137,12 @@ POINTS: dict[str, Point] = dict([
        "the one line reserved for a preference at the prompt boundary"),
     _p("reuse_slot", UNBIDDEN, "the one line reserved for a reusable snippet"),
     _p("lesson_slot", UNBIDDEN, "the one line reserved for a lesson"),
+    _p("rule_via_lesson", UNBIDDEN,
+       "a rule reached through a lesson confirmed as an instance of it",
+       expects_traffic=False,
+       quiet_because="searches only once some lesson has a confirmed link to "
+                     "a rule; an install where none has been judged is "
+                     "correctly silent here"),
     # `fixed_query`: COMPLETION_QUERY is a module constant in
     # services/reply_preferences.py, so this arm's top score is the same number
     # on every call — measured at 0.791 across 45 consecutive calls, with p10,

src/scribe/services/rule_usage.py

@@ -110,6 +110,10 @@ RANKED_SOURCES = (
     # The completion-report lookup on update_task (milestone 409 step 4). It
     # runs its own query and shows only what cleared the bar — a ranker.
     "report_preference",
+    # A rule reached through a lesson confirmed as an instance of it (#4633).
+    # Its own source so its pull-through reads apart from the rule's direct
+    # match — whether the lesson route earns its line is #4636's question.
+    "rule_via_lesson",
 )
 
 

tests/conftest.py

@@ -157,7 +157,12 @@ def _no_lesson_rule_links(request):
     otherwise ride along with every unit test that creates a lesson. The
     co-surfacing recorder (#4637) is stubbed to "no proposal": every hook
     builder test that renders a lesson beside a rule would otherwise reach
-    for the database to count the pair.
+    for the database to count the pair. And the via-lesson rule step (#4633)
+    is stubbed to "no lesson brought a rule": all three rule arms run it, and
+    its first act is a database read. tests/test_rule_via_lesson.py binds the
+    real function at import time, before this patch runs. The convergence
+    check a "no rule fits" answer runs (#4634) is stubbed to "no group" for
+    the same reason; tests/test_lesson_convergence.py binds the real one.
 
     Skipped for integration tests, which exercise the real links against
     Postgres (tests/test_integration_lesson_rule_links.py).
@@ -168,7 +173,10 @@ def _no_lesson_rule_links(request):
     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=[])), \
-         patch("scribe.services.lesson_rules.co_surfaced", AsyncMock(return_value="")):
+         patch("scribe.services.lesson_rules.co_surfaced", AsyncMock(return_value="")), \
+         patch("scribe.services.plugin_context._rules_via_lessons",
+               AsyncMock(return_value=([], []))), \
+         patch("scribe.services.lesson_rules.convergence_for", AsyncMock(return_value=None)):
         yield
 
 

tests/test_integration_lesson_rule_links.py

@@ -314,3 +314,48 @@ async def test_confirming_a_suggestion_keeps_what_it_rested_on(world):
     out = await links_svc.judge_link(world["owner"], world["lesson"], world["r1"], "confirm", "same failure class")
     assert out["state"] == "confirmed"
     assert len(out["evidence"]["situations"]) == 3
+
+
+# ── #4633: which links can carry a rule ─────────────────────────────────────
+
+
+async def test_only_a_confirmed_link_can_carry_its_rule(world):
+    """A suggested link never expands: it would manufacture the co-arrival
+    that #4637 counts, and prove itself. A rejected one obviously not."""
+    owner, lesson = world["owner"], world["lesson"]
+    await links_svc.co_surfaced(owner, [lesson], [world["r1"]], arm="p", situation="one ask here")
+    await links_svc.judge_link(owner, lesson, world["r2"], "reject", "not this")
+    assert await links_svc.confirmed_lessons(owner) == set()
+    assert await links_svc.confirmed_rules_in_scope(owner, [lesson], world["mine"]) == {}
+
+    await links_svc.judge_link(owner, lesson, world["r1"], "confirm", "same failure class")
+    assert lesson in await links_svc.confirmed_lessons(owner)
+    rules = await links_svc.confirmed_rules_in_scope(owner, [lesson], world["mine"])
+    assert [r.id for r in rules[lesson]] == [world["r1"]]
+
+
+async def test_a_project_rule_stays_in_its_project_even_through_a_lesson(world):
+    """Milestone 414's scope, kept on the side door: the lesson surfaces
+    everywhere, its project rule only in its project."""
+    owner, lesson = world["owner"], world["lesson"]
+    await links_svc.set_lesson_rules(owner, lesson, [world["r1"]])
+    assert await links_svc.confirmed_rules_in_scope(owner, [lesson], world["mine"])
+    assert await links_svc.confirmed_rules_in_scope(owner, [lesson], world["theirs"]) == {}
+    assert await links_svc.confirmed_rules_in_scope(owner, [lesson], None) == {}
+
+
+async def test_a_stranger_reaches_no_rule_through_someone_elses_link(world):
+    await links_svc.set_lesson_rules(world["owner"], world["lesson"], [world["r1"]])
+    assert await links_svc.confirmed_lessons(world["stranger"]) == set()
+
+
+# ── #4634: which lessons count toward convergence ───────────────────────────
+
+
+async def test_only_answered_lessons_are_no_rule_lessons(world):
+    owner = world["owner"]
+    other = await lessons_svc.create_lesson(
+        owner, what="Another overrun", when_to_apply="a deploy job overran",
+    )
+    await links_svc.set_no_rule(owner, world["lesson"], "specific to this host")
+    assert await links_svc._no_rule_ids([world["lesson"], other.id]) == {world["lesson"]}

tests/test_lesson_convergence.py

@@ -0,0 +1,100 @@
+"""Convergence named at the write (milestone 440, #4634).
+
+When a lesson is answered "no rule fits", the response looks for other
+no-rule lessons in the same situation and, once there are enough, names the
+group and the rule it may be missing. The bar is pinned here as a pure
+function; the search around it with the database and embedder stubbed; the
+doors by what they return.
+"""
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from scribe.mcp._context import _user_id_ctx
+from scribe.mcp.tools import lessons as lesson_tools
+from scribe.services import lesson_rules as links_svc
+from scribe.services import lessons as lessons_svc
+
+# Bound before conftest's autouse stub replaces the module attribute.
+_REAL = links_svc.convergence_for
+
+
+def _m(lid, sources=(), project=None, title=None):
+    return {"id": lid, "title": title or f"lesson {lid}", "sources": list(sources),
+            "project_id": project}
+
+
+def test_below_the_group_size_nothing_is_named():
+    assert links_svc.convergence_group([_m(1), _m(2)]) is None
+
+
+def test_a_group_of_distinct_lessons_names_its_members_and_the_next_step():
+    group = links_svc.convergence_group([_m(1, [10], 2), _m(2, [11], 5), _m(3, [], 2)])
+    assert [m["id"] for m in group["lessons"]] == [1, 2, 3]
+    assert group["incidents"] == [10, 11]
+    assert group["projects"] == [2, 5]
+    assert "create_rule" in group["hint"] and "update_lesson" in group["hint"]
+    assert "#1 “lesson 1”" in group["hint"]
+
+
+def test_one_broad_lesson_cannot_reach_the_bar_alone():
+    """Many incidents behind ONE lesson is still one judgment: incidents never
+    stand in for lessons, and a repeated id is one member."""
+    broad = _m(1, [10, 11, 12, 13, 14])
+    assert links_svc.convergence_group([broad]) is None
+    assert links_svc.convergence_group([broad, broad, broad]) is None
+
+
+def test_one_incident_written_up_three_times_is_not_a_recurring_situation():
+    same = [_m(1, [10]), _m(2, [10]), _m(3, [10])]
+    assert links_svc.convergence_group(same) is None
+
+
+def _note(lid, *, title="t", project=None, sources=()):
+    data = {"what": title, "when_to_apply": "a CI run overran"}
+    if sources:
+        data["taught_by"] = list(sources)
+    return SimpleNamespace(
+        id=lid, title=title, note_type="lesson", body="", data=data,
+        project_id=project, arose_from_id=None, tags=[],
+        created_at=None, updated_at=None,
+    )
+
+
+@pytest.mark.asyncio
+async def test_only_lessons_answered_no_rule_join_the_group():
+    found = [(0.8, _note(2, sources=[11])), (0.7, _note(3, sources=[12])),
+             (0.7, _note(4, sources=[13]))]
+    with patch.object(lessons_svc, "get_lesson", AsyncMock(return_value=_note(1, sources=[10]))), \
+         patch("scribe.services.embeddings.semantic_search_notes", AsyncMock(return_value=found)), \
+         patch.object(links_svc, "_no_rule_ids", AsyncMock(return_value={2})):
+        assert await _REAL(7, 1) is None          # only #2 answered: a pair
+    with patch.object(lessons_svc, "get_lesson", AsyncMock(return_value=_note(1, sources=[10]))), \
+         patch("scribe.services.embeddings.semantic_search_notes", AsyncMock(return_value=found)), \
+         patch.object(links_svc, "_no_rule_ids", AsyncMock(return_value={2, 4})):
+        group = await _REAL(7, 1)
+    assert [m["id"] for m in group["lessons"]] == [1, 2, 4]
+
+
+@pytest.mark.asyncio
+async def test_a_failed_search_names_nothing_and_raises_nothing():
+    with patch.object(lessons_svc, "get_lesson", AsyncMock(side_effect=RuntimeError("db down"))):
+        assert await _REAL(7, 1) is None
+
+
+@pytest.mark.asyncio
+async def test_the_door_carries_convergence_only_with_a_no_rule_answer():
+    _user_id_ctx.set(7)
+    group = {"lessons": [], "incidents": [], "projects": [], "hint": "h"}
+    named = AsyncMock(return_value=group)
+    with patch.object(lessons_svc, "update_lesson", AsyncMock(return_value=_note(41))), \
+         patch.object(links_svc, "set_no_rule", AsyncMock()), \
+         patch.object(links_svc, "convergence_for", named):
+        out = await lesson_tools.update_lesson(lesson_id=41, no_rule="stands alone")
+        assert out["convergence"] == group
+        out = await lesson_tools.update_lesson(lesson_id=41, what="reworded")
+    assert "convergence" not in out
+    assert named.await_count == 1

tests/test_lesson_rule_link_ui.py

@@ -0,0 +1,117 @@
+"""Both records show the lesson → rule link in the web UI (milestone 440, #4635).
+
+The frontend has no component test runner, so these read the source — and
+each is written against something the backend owns, so a drift on either
+side fails here rather than rendering a page that quietly reads wrong:
+
+- the client's judgment and link-state unions are the service's and the
+  model's own values, so a renamed state cannot leave a branch that never
+  matches;
+- the lesson page branches on every judgment, with the unjudged answer as
+  the fall-through, so "not yet judged" is stated rather than left blank;
+- the write-level list the pages hide controls by is the server's
+  PERMISSION_RANK from editor up, which is the drift that had the snippet
+  page hiding its edit controls from shared editors (`"edit"` is not a level
+  the server sends).
+"""
+from __future__ import annotations
+
+import re
+from pathlib import Path
+
+from scribe.models.lesson_rule_link import LINK_STATES
+from scribe.services import lesson_rules as links_svc
+from scribe.services.access import PERMISSION_RANK
+
+FRONTEND = Path(__file__).resolve().parents[1] / "frontend" / "src"
+
+
+def _read(rel: str) -> str:
+    return (FRONTEND / rel).read_text()
+
+
+def _union(source: str, name: str) -> set[str]:
+    """The string members of `export type <name> = "a" | "b" …;`."""
+    m = re.search(rf"export type {name}\s*=\s*([^;]+);", source)
+    assert m, f"type {name} not found"
+    return set(re.findall(r'"([^"]+)"', m.group(1)))
+
+
+def _template(view: str) -> str:
+    return view[view.index("<template>"):view.rindex("</template>")]
+
+
+def test_the_client_judgment_union_is_the_services_three_answers():
+    assert _union(_read("api/lessons.ts"), "RuleJudgment") == {
+        links_svc.LINKED, links_svc.NO_RULE, links_svc.UNJUDGED,
+    }
+
+
+def test_the_client_link_states_are_the_models():
+    assert _union(_read("api/lessons.ts"), "LinkState") == set(LINK_STATES)
+
+
+def test_the_lesson_page_answers_all_three_judgments():
+    """`linked` and `no_rule` are tested by name; `unjudged` is the v-else
+    that closes the same chain, so a fourth state the server might add reads
+    as unjudged rather than as nothing."""
+    tpl = _template(_read("views/LessonDetailView.vue"))
+    linked = tpl.index(f"lesson.rule_judgment === '{links_svc.LINKED}'")
+    no_rule = tpl.index(f"lesson.rule_judgment === '{links_svc.NO_RULE}'")
+    assert linked < no_rule
+    after = tpl[no_rule:]
+    # The no-rule answer renders its reason, and the chain closes on a v-else.
+    para_end = after.index("</p>")
+    assert "lesson.no_rule?.why" in after[:para_end]
+    closing = after[para_end + len("</p>"):]
+    assert re.match(r"\s*<p v-else\b", closing), "the unjudged branch is not the chain's v-else"
+    assert "Not yet judged" in closing[:closing.index("</p>")]
+
+
+def test_the_panel_is_gated_on_the_judgment_being_attached():
+    """An absent `rule_judgment` is "the links could not be read" — never
+    "no rule" — so the panel must not render a state for it."""
+    tpl = _template(_read("views/LessonDetailView.vue"))
+    assert '<section v-if="lesson.rule_judgment"' in tpl
+
+
+def test_both_pages_link_through_to_the_other_record():
+    lesson_tpl = _read("views/LessonDetailView.vue")
+    assert re.search(r'path:\s*"/rules",\s*query:\s*\{\s*rule:', lesson_tpl)
+    rule_tpl = _template(_read("components/rules/RuleEditorSlideOver.vue"))
+    assert ':to="`/lessons/${l.id}`"' in rule_tpl
+
+
+def test_the_kind_and_state_markers_reuse_rule_chip():
+    """No new chip style where `.rule-chip` serves: both components load the
+    shared sheet and neither re-declares the class in its own block."""
+    for rel in ("views/LessonDetailView.vue", "components/rules/RuleEditorSlideOver.vue"):
+        src = _read(rel)
+        assert 'class="rule-chip' in _template(src), rel
+        assert '<style src="@/assets/rules-shared.css" />' in src, rel
+        scoped = src[src.index("<style scoped>"):]
+        scoped = scoped[:scoped.index("</style>")]
+        assert ".rule-chip" not in scoped, f"{rel} re-spells the chip"
+
+
+def test_the_write_levels_are_the_servers():
+    src = _read("utils/permission.ts")
+    m = re.search(r"WRITE_LEVELS[^=]*=\s*\[([^\]]*)\]", src)
+    assert m, "WRITE_LEVELS not found"
+    client = set(re.findall(r'"([^"]+)"', m.group(1)))
+    server = {p for p, rank in PERMISSION_RANK.items() if rank >= PERMISSION_RANK["editor"]}
+    assert client == server
+
+
+def test_no_page_spells_its_own_write_check():
+    """The helper exists so pages agree; a page comparing against a share
+    level by hand is the copy that drifted to `"edit"`. Matched on the names
+    only a share level uses — `admin` and `owner` are also user roles, and
+    `role === "admin"` is not this."""
+    offenders = [
+        str(p.relative_to(FRONTEND))
+        for p in [*FRONTEND.rglob("*.vue"), *FRONTEND.rglob("*.ts")]
+        if p != FRONTEND / "utils" / "permission.ts"
+        and re.search(r'[!=]==\s*"(?:viewer|editor|edit)"', p.read_text())
+    ]
+    assert offenders == []

tests/test_rule_usage_wiring.py

@@ -908,9 +908,23 @@ def test_neither_rule_arm_logs_its_call_behind_a_results_guard():
     #
     # The property is positional: between the search and the first guard that
     # can return early, the call row has already been written.
+    # The arm's BODY, which since #4633 lives in `_tool_rule_hint`; the public
+    # `build_tool_rule_hint` is a wrapper that adds the via-lesson step and
+    # searches nothing itself. Located by what it does — the function that
+    # calls the rule search under the pre_tool_rule source — so the next
+    # rename moves the guard with it instead of emptying it.
     fn = next(
         n for n in ast.walk(ast.parse(pc_src))
-        if isinstance(n, ast.AsyncFunctionDef) and n.name == "build_tool_rule_hint"
+        if isinstance(n, ast.AsyncFunctionDef)
+        and any(
+            isinstance(c, ast.Call) and getattr(c.func, "id", None) == "semantic_search_rules"
+            for c in ast.walk(n)
+        )
+        and any(
+            isinstance(k, ast.keyword) and k.arg == "source"
+            and isinstance(k.value, ast.Constant) and k.value.value == "pre_tool_rule"
+            for k in ast.walk(n)
+        )
     )
     search_at = min(
         n.lineno for n in ast.walk(fn)
@@ -929,7 +943,7 @@ def test_neither_rule_arm_logs_its_call_behind_a_results_guard():
         and any(isinstance(b, ast.Return) for b in n.body)
     ]
     assert bailouts, (
-        "no early return found after the search in build_tool_rule_hint — the "
+        "no early return found after the search in the pre-tool arm — the "
         "guard has nothing left to protect, which means this test is now "
         "passing vacuously rather than the arm being correct"
     )

tests/test_rule_via_lesson.py

@@ -0,0 +1,147 @@
+"""A rule reached through its lessons (milestone 440, #4633).
+
+The step runs after each of the three rule arms. These pin, with the database
+and the embedder stubbed: that nothing is searched when no lesson carries a
+confirmed link; that a matching linked lesson brings its rule in rule voice,
+naming the lesson; that suppression is by RULE; that the slot holds one line;
+and that the call is logged under its own source. That a SUGGESTED link never
+expands — the soft link proving itself — is pinned against Postgres in
+tests/test_integration_lesson_rule_links.py, where the state filter lives.
+"""
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from scribe.services import plugin_context as pc
+from scribe.services import rule_usage
+from scribe.services.retrieval_registry import POINTS
+
+# Bound before conftest's autouse stub replaces the module attribute.
+_REAL = pc._rules_via_lessons
+
+
+def _lesson(lid=41, title="An overrun run usually failed early"):
+    return SimpleNamespace(
+        id=lid, title=title, note_type="lesson", body="",
+        data={"what": title, "when_to_apply": "a CI run overran"},
+    )
+
+
+def _rule(rid, title="Read the job log first", kind="rule"):
+    return SimpleNamespace(id=rid, title=title, kind=kind, when_to_apply="a run overran")
+
+
+def _stubs(*, confirmed, found, by_lesson):
+    log, surfaced = MagicMock(), MagicMock()
+    stack = [
+        patch.object(pc.lesson_rules_svc, "confirmed_lessons", AsyncMock(return_value=confirmed)),
+        patch.object(pc.lesson_rules_svc, "confirmed_rules_in_scope", AsyncMock(return_value=by_lesson)),
+        patch.object(pc, "get_autoinject_config", AsyncMock(return_value={"threshold": 0.5})),
+        patch.object(pc, "semantic_search_notes", AsyncMock(return_value=found)),
+        patch.object(pc, "record_retrieval", log),
+        patch.object(pc, "record_rule_surfaced", surfaced),
+    ]
+    return stack, log, surfaced
+
+
+async def _run(stack, **kw):
+    for p in stack:
+        p.start()
+    try:
+        return await _REAL(1, "the CI run is still going", project_id=2,
+                           skip=kw.get("skip", set()), held=set(), where="to this request")
+    finally:
+        for p in stack:
+            p.stop()
+
+
+@pytest.mark.asyncio
+async def test_no_confirmed_link_means_no_search_and_no_row():
+    search = AsyncMock()
+    stack, log, _ = _stubs(confirmed=set(), found=[], by_lesson={})
+    stack[3] = patch.object(pc, "semantic_search_notes", search)
+    assert await _run(stack) == ([], [])
+    search.assert_not_awaited()
+    log.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_a_matching_linked_lesson_brings_its_rule_in_rule_voice():
+    stack, log, surfaced = _stubs(
+        confirmed={41}, found=[(0.71, _lesson())], by_lesson={41: [_rule(7)]},
+    )
+    lines, ids = await _run(stack)
+    assert ids == [7]
+    assert lines[0].startswith("Standing rule")
+    assert "Reached through lesson #41" in lines[0]
+    assert log.call_args.kwargs["source"] == "rule_via_lesson"
+    assert surfaced.call_args.kwargs == {"user_id": 1, "rule_ids": [7], "source": "rule_via_lesson"}
+
+
+@pytest.mark.asyncio
+async def test_a_lesson_without_a_confirmed_link_carries_nothing():
+    """The search found a lesson, but it is not in the confirmed set — a
+    suggested or unlinked lesson brings no rule."""
+    stack, log, surfaced = _stubs(
+        confirmed={99}, found=[(0.9, _lesson(41))], by_lesson={},
+    )
+    assert await _run(stack) == ([], [])
+    surfaced.assert_not_called()
+    assert log.call_args.kwargs["results"] == []
+
+
+@pytest.mark.asyncio
+async def test_suppression_is_by_rule_whichever_lesson_reached_it():
+    stack, log, surfaced = _stubs(
+        confirmed={41}, found=[(0.71, _lesson())], by_lesson={41: [_rule(7)]},
+    )
+    assert await _run(stack, skip={7}) == ([], [])
+    assert log.call_args.kwargs["suppressed"] == 1
+    surfaced.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_the_slot_holds_one_line():
+    stack, _log, _ = _stubs(
+        confirmed={41, 42},
+        found=[(0.8, _lesson(41)), (0.7, _lesson(42, "Another"))],
+        by_lesson={41: [_rule(7), _rule(8)], 42: [_rule(9)]},
+    )
+    lines, ids = await _run(stack)
+    assert ids == [7] and len(lines) == pc.VIA_LESSON_LIMIT == 1
+
+
+@pytest.mark.asyncio
+async def test_a_failure_brings_no_rule_and_raises_nothing():
+    stack, _log, _ = _stubs(confirmed={41}, found=[], by_lesson={})
+    stack[0] = patch.object(pc.lesson_rules_svc, "confirmed_lessons",
+                            AsyncMock(side_effect=RuntimeError("db down")))
+    assert await _run(stack) == ([], [])
+
+
+@pytest.mark.asyncio
+async def test_the_step_runs_only_when_the_arm_ran_and_never_leaks_its_key():
+    step = AsyncMock(return_value=(["Standing rule … Reached through lesson #41"], [7]))
+    with patch.object(pc, "_rules_via_lessons", step):
+        idle = await pc._add_rules_via_lessons(
+            1, {"context": "", "rule_ids": []}, project_id=2,
+            exclude_rule_ids=[3], held_rule_ids=[], where="here",
+        )
+        ran = await pc._add_rules_via_lessons(
+            1, {"context": "direct line", "rule_ids": [5], "_via_query": "q"},
+            project_id=2, exclude_rule_ids=[3], held_rule_ids=[], where="here",
+        )
+    assert idle == {"context": "", "rule_ids": []}
+    assert step.await_count == 1
+    assert step.await_args.kwargs["skip"] == {3, 5}
+    assert "_via_query" not in ran
+    assert ran["rule_ids"] == [5, 7]
+    assert ran["context"].startswith("direct line\n")
+
+
+def test_the_source_is_ranked_and_registered():
+    assert "rule_via_lesson" in rule_usage.RANKED_SOURCES
+    assert "rule_via_lesson" in POINTS