feat(lessons): convergence is named at the write — no-rule lessons that keep landing in one situation suggest a rule (milestone 440 step 5, #4634)

When a lesson is answered "no rule fits" (create_lesson / update_lesson on
both doors), the response looks for other no-rule lessons it resembles and,
once there are CONVERGENCE_LESSONS (3) of them, carries `convergence`: the
members, their incidents and projects, and a hint to draft the missing rule
with create_rule (operator approval as always) and point each lesson at it —
or to leave them as lessons when no single choice is right every time.

- convergence_group is the pure bar: distinct LESSONS count, incidents never
  stand in for them (one broad lesson cannot trigger it), and a group whose
  sources all point at one incident is one event written up several times.
- convergence_for searches lessons by the new one's claim + trigger
  (trigger_title) at CONVERGENCE_THRESHOLD 0.65 — above the menu's "worth
  showing", below the duplicate gate's "same record" — then keeps the ones
  with a lesson_no_rule answer. Fail-open. No sweep, no timer (#4183).
- Defaults stated as defaults (rules 32, 115).
- Tests: the bar (pure), the search with stubs, the door, and the no-rule
  filter against Postgres; conftest stubs convergence_for for unit tests.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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

@@ -720,3 +720,116 @@ async def confirmed_rules_in_scope(
     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]
+    )

tests/conftest.py

@@ -160,7 +160,9 @@ def _no_lesson_rule_links(request):
     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.
+    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).
@@ -173,7 +175,8 @@ def _no_lesson_rule_links(request):
          patch("scribe.services.lesson_rules.rule_candidates", AsyncMock(return_value=[])), \
          patch("scribe.services.lesson_rules.co_surfaced", AsyncMock(return_value="")), \
          patch("scribe.services.plugin_context._rules_via_lessons",
-               AsyncMock(return_value=([], []))):
+               AsyncMock(return_value=([], []))), \
+         patch("scribe.services.lesson_rules.convergence_for", AsyncMock(return_value=None)):
         yield
 
 

tests/test_integration_lesson_rule_links.py

@@ -347,3 +347,15 @@ async def test_a_project_rule_stays_in_its_project_even_through_a_lesson(world):
 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