feat(lessons): soft links — a lesson and a rule arriving together in distinct situations are proposed as a link (milestone 440 step 3, #4637)

When one hook response puts a lesson and a rule in front of the reader, the
pair is recorded as evidence on a SUGGESTED lesson_rule_links row; once the
pair has arrived together in PROPOSE_SITUATIONS (3) distinct situations, the
next co-arrival carries one line asking the reader to judge it with
judge_lesson_link. Nothing about surfacing changes: a suggested link carries
no rule anywhere (that is #4633, confirmed links only).

- lesson_rules: co_surfaced (fail-open; only pairs the reader could confirm —
  a lesson they may write, a rule they own; judged pairs gather nothing; one
  proposal per response; PROPOSE_COOLDOWN 6h between asks), plus the pure
  counting rules: situation_key, add_evidence, proposal_due, evidence_summary.
  A situation is the prompt on /retrieve (word tokens, sorted and
  de-duplicated, so trivial rewordings count once) and the FILE on
  /prior-art (every edit to one file is one situation).
- rules_for_lessons shows a suggested link's evidence counts.
- plugin_context: build_autoinject_hint returns lesson_ids,
  build_prompt_rule_hint returns shown_rule_ids, build_write_path_hint records
  its own pair; routes/plugin /retrieve records the prompt pair. Shown lines,
  repeats included: relevance makes a co-arrival, not the session ledger.
- Evidence lives in the existing evidence column — no migration, and backup
  already carries it.
- Tests: counting rules (unit), the recorder against Postgres (bar, repeat,
  judged pairs, ownership, cooldown, evidence kept on confirm); conftest stubs
  co_surfaced for unit tests.

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

tests/conftest.py

@@ -154,7 +154,10 @@ def _no_lesson_rule_links(request):
 
     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.
+    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.
 
     Skipped for integration tests, which exercise the real links against
     Postgres (tests/test_integration_lesson_rule_links.py).
@@ -164,7 +167,8 @@ def _no_lesson_rule_links(request):
         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=[])):
+         patch("scribe.services.lesson_rules.rule_candidates", AsyncMock(return_value=[])), \
+         patch("scribe.services.lesson_rules.co_surfaced", AsyncMock(return_value="")):
         yield
 
 

tests/test_integration_lesson_rule_links.py

@@ -241,3 +241,76 @@ async def test_deleting_the_lesson_takes_its_answer_with_it(world):
             select(LessonNoRule).where(LessonNoRule.lesson_id == world["lesson"])
         )).scalars().all()
     assert left == []
+
+
+# ── #4637: soft links from co-surfacing ─────────────────────────────────────
+
+
+async def _link(lesson_id: int, rule_id: int):
+    async with async_session() as s:
+        return (await s.execute(
+            select(LessonRuleLink).where(
+                LessonRuleLink.lesson_id == lesson_id, LessonRuleLink.rule_id == rule_id,
+            )
+        )).scalar_one_or_none()
+
+
+async def _arrive(world, prompt: str, project: int | None = None, rules=None) -> str:
+    return await links_svc.co_surfaced(
+        world["owner"], [world["lesson"]], rules or [world["r1"]],
+        arm="p", situation=prompt, project_id=project,
+    )
+
+
+async def test_distinct_situations_reach_the_bar_and_the_proposal_fires_once(world):
+    assert await _arrive(world, "the CI run is still going after twenty minutes") == ""
+    assert await _arrive(world, "deploy job hanging on the image build step") == ""
+    line = await _arrive(world, "integration lane overran its usual duration", world["mine"])
+    assert f"judge_lesson_link({world['lesson']}, {world['r1']}" in line
+
+    link = await _link(world["lesson"], world["r1"])
+    assert link.state == "suggested"
+    assert links_svc.evidence_summary(link.evidence)["situations"] == 3
+    # Within the cooldown a fourth situation is counted but not asked again.
+    assert await _arrive(world, "pipeline stuck waiting on a runner for ages") == ""
+    link = await _link(world["lesson"], world["r1"])
+    assert links_svc.evidence_summary(link.evidence)["situations"] == 4
+
+
+async def test_the_same_prompt_repeated_never_reaches_the_bar(world):
+    for _ in range(5):
+        assert await _arrive(world, "is the CI run still going") == ""
+    ev = links_svc.evidence_summary((await _link(world["lesson"], world["r1"])).evidence)
+    assert ev == {"situations": 1, "projects": 0, "co_surfaced": 5}
+
+
+async def test_a_judged_pair_gathers_nothing_and_is_never_proposed(world):
+    owner = world["owner"]
+    await links_svc.judge_link(owner, world["lesson"], world["r1"], "reject", "different failure")
+    await links_svc.judge_link(owner, world["lesson"], world["r2"], "confirm")
+    for prompt in ("first situation here", "second situation there", "third situation elsewhere"):
+        assert await _arrive(world, prompt, rules=[world["r1"], world["r2"]]) == ""
+    assert (await _link(world["lesson"], world["r1"])).evidence is None
+    assert (await _link(world["lesson"], world["r2"])).evidence is None
+
+
+async def test_a_rule_the_reader_does_not_own_is_not_paired(world):
+    for prompt in ("first situation here", "second situation there", "third situation elsewhere"):
+        assert await _arrive(world, prompt, rules=[world["foreign"]]) == ""
+    assert await _link(world["lesson"], world["foreign"]) is None
+
+
+async def test_a_suggestion_shows_its_evidence_and_leaves_the_lesson_unjudged(world):
+    await _arrive(world, "the CI run is still going after twenty minutes", world["mine"])
+    rules = (await links_svc.rules_for_lessons(world["owner"], [world["lesson"]]))[world["lesson"]]
+    assert rules[0]["state"] == "suggested"
+    assert rules[0]["evidence"] == {"situations": 1, "projects": 1, "co_surfaced": 1}
+    assert (await _judgment(world["lesson"]))["rule_judgment"] == "unjudged"
+
+
+async def test_confirming_a_suggestion_keeps_what_it_rested_on(world):
+    for prompt in ("first situation here", "second situation there", "third situation elsewhere"):
+        await _arrive(world, prompt)
+    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

tests/test_lesson_rule_soft_links.py

@@ -0,0 +1,103 @@
+"""Soft links — the counting rules (milestone 440, #4637).
+
+What a suggested link's evidence counts is a design decision with a failure
+mode on each side: count every arrival and two similar records "prove" a
+relation by resembling each other on every similar prompt; count too little
+and nothing ever reaches the bar. These pin the rules as pure functions. The
+recorder itself — ACL, judged pairs left alone, the proposal firing once —
+is tested against Postgres in tests/test_integration_lesson_rule_links.py.
+"""
+from __future__ import annotations
+
+import inspect
+from datetime import datetime, timedelta, timezone
+
+from scribe.services import lesson_rules as links_svc
+
+NOW = datetime(2026, 10, 1, 12, 0, tzinfo=timezone.utc)
+
+
+def test_the_same_ask_reworded_trivially_is_one_situation():
+    a = links_svc.situation_key("p", "Merge dev to main, please.")
+    assert a == links_svc.situation_key("p", "please   MERGE main to dev")
+    assert a != links_svc.situation_key("p", "merge the feature branch")
+
+
+def test_a_file_is_one_situation_however_its_code_differs():
+    """The write arm keys on the path, so a run of edits to one file — each
+    with different code — is one situation, not one per edit."""
+    path = "src/scribe/services/lesson_rules.py"
+    assert links_svc.situation_key("w", path) == links_svc.situation_key("w", path)
+    assert links_svc.situation_key("w", path) != links_svc.situation_key("w", "src/other.py")
+
+
+def test_the_two_arms_never_share_a_key():
+    assert links_svc.situation_key("p", "routes.py") != links_svc.situation_key("w", "routes.py")
+
+
+def test_nothing_to_key_on_is_no_situation():
+    assert links_svc.situation_key("p", "") == ""
+    assert links_svc.situation_key("p", "a is to") == ""   # grammar only
+
+
+def test_a_repeat_counts_as_an_arrival_but_not_a_situation():
+    key = links_svc.situation_key("p", "probe the deployed version")
+    ev = links_svc.add_evidence(None, key, 2, NOW)
+    ev = links_svc.add_evidence(ev, key, 2, NOW)
+    assert links_svc.evidence_summary(ev) == {"situations": 1, "projects": 1, "co_surfaced": 2}
+
+
+def test_projects_are_distinct_and_an_unbound_reader_adds_none():
+    ev = links_svc.add_evidence(None, "p:1", 2, NOW)
+    ev = links_svc.add_evidence(ev, "p:2", 5, NOW)
+    ev = links_svc.add_evidence(ev, "p:3", None, NOW)
+    assert links_svc.evidence_summary(ev)["projects"] == 2
+
+
+def test_evidence_is_a_new_dict_so_the_column_change_is_seen():
+    before = {"co_surfaced": 1, "situations": ["p:1"], "projects": []}
+    after = links_svc.add_evidence(before, "p:2", None, NOW)
+    assert after is not before
+    assert before["situations"] == ["p:1"]
+
+
+def test_the_proposal_fires_at_the_bar_and_not_before():
+    ev = None
+    for i in range(links_svc.PROPOSE_SITUATIONS - 1):
+        ev = links_svc.add_evidence(ev, f"p:{i}", None, NOW)
+    assert not links_svc.proposal_due(ev, NOW)
+    ev = links_svc.add_evidence(ev, "p:last", None, NOW)
+    assert links_svc.proposal_due(ev, NOW)
+
+
+def test_a_passed_over_proposal_waits_out_the_cooldown():
+    ev = {"situations": ["p:1", "p:2", "p:3"], "proposed_at": NOW.isoformat()}
+    assert not links_svc.proposal_due(ev, NOW + timedelta(minutes=5))
+    assert links_svc.proposal_due(ev, NOW + links_svc.PROPOSE_COOLDOWN)
+
+
+def test_the_proposal_names_the_call_that_answers_it():
+    line = links_svc._proposal_line(
+        41, "An overrun run usually failed early", 7, "Read the job log first",
+        "rule", {"situations": ["p:1", "p:2", "p:3"], "projects": [2, 5]},
+    )
+    assert 'judge_lesson_link(41, 7, "confirm"' in line
+    assert '"reject"' in line
+    assert "3 distinct situations across 2 projects" in line
+
+
+def test_evidence_summary_reads_nothing_as_zero_not_an_error():
+    assert links_svc.evidence_summary(None) == {"situations": 0, "projects": 0, "co_surfaced": 0}
+
+
+def test_both_hook_doors_count_the_pair_on_their_own_arm():
+    """The two responses that carry lessons and rules together. A door that
+    renders both and does not record the pair is the miss this step exists
+    to close, and it would produce no error anywhere — only an empty table."""
+    from scribe.routes import plugin as plugin_routes
+    from scribe.services import plugin_context
+
+    prompt_door = inspect.getsource(plugin_routes.autoinject_retrieve)
+    assert "co_surfaced(" in prompt_door and 'arm="p"' in prompt_door
+    write_door = inspect.getsource(plugin_context.build_write_path_hint)
+    assert "co_surfaced(" in write_door and 'arm="w"' in write_door