= "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(""):view.rindex("")]
+
+
+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("")
+ assert "lesson.no_rule?.why" in after[:para_end]
+ closing = after[para_end + len(""):]
+ assert re.match(r"\s*")]
+
+
+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 '' in src, rel
+ scoped = src[src.index("")]
+ 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 == []
diff --git a/tests/test_rule_usage_wiring.py b/tests/test_rule_usage_wiring.py
index 6b466ae..91dde55 100644
--- a/tests/test_rule_usage_wiring.py
+++ b/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"
)
diff --git a/tests/test_rule_via_lesson.py b/tests/test_rule_via_lesson.py
new file mode 100644
index 0000000..860e9ae
--- /dev/null
+++ b/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