feat(lessons): a lesson names the rule it is an instance of — lesson_rule_links (milestone 440 step 1, #4630)

The link between a lesson (one concrete situation) and the rule that governs
it, with the operator's soft-then-hard design built into its state:
suggested while evidence accumulates, confirmed or rejected once judged. Only
confirmed will carry a rule in retrieval (#4633); rejected is kept so the pair
is never proposed again.

- models/lesson_rule_link.py + migration 0111: one row per (lesson, rule),
  CASCADE on both ends, indexed both ways, CHECK on state (rule 36), evidence
  JSONB and judged_at.
- services/lesson_rules.py: require_rules (validated before any write, so
  a bad id leaves nothing half-linked), set_lesson_rules (set-semantics;
  a dropped rule becomes rejected, not forgotten), judge_link, and the two
  reads. ACL: write on the lesson (share-aware), ownership of the rule; a
  reader sees only rules they own. Decorations are fail-open (#4286).
- MCP: create_lesson / update_lesson take rule_ids; get/create/update return
  `rules`; new judge_lesson_link tool. REST: the same on /api/lessons plus
  PUT /api/lessons/<id>/rules/<rule_id>. Rules: rule_detail carries `lessons`.
- Backup v18: export (full and user-scoped, both ends in scope), builder,
  importer; both column guards register the table.
- Tests: integration (states, set-semantics, judge, ACL all-or-nothing,
  cascade both ways, CHECK, one row per pair); unit (door wiring, judge
  registered, migration/model state agreement, backup skip and unjudged
  stays unjudged). conftest stubs the decorations for unit tests.

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

tests/test_integration_lesson_rule_links.py

@@ -0,0 +1,162 @@
+"""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()