"""Delivering mounted rules when their moment happens (milestone 458 step 4). The database is stubbed: the lookup's scoping is pinned against Postgres in tests/test_integration_rule_moments.py. These pin what is decided above it: - the line names the moment AND the act that reached it, so a misfire is visible where it lands and can be unmapped in the session; - a rule the session was already shown is cited, not repeated, and only the fresh ones are recorded, each with its moment; - every door fails open; - the plugin is told which tools can reach anything, and nothing more; - every Scribe tool the shipped mappings name carries the attach, and the hook and the route agree on every name between them. """ from __future__ import annotations import ast import inspect import json import os import subprocess from pathlib import Path from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch from quart import Quart, g from scribe.services import moment_actions from scribe.services import moment_delivery as md from scribe.services import retrieval_pipeline as rp from scribe.services import rule_usage, rulebooks from tests.helpers import fake_rule, http_sink, need_tools # Bound before conftest's autouse stub replaces the module attribute. _REAL_DELIVER = md.deliver_for_act ROOT = Path(__file__).resolve().parents[1] HOOK = ROOT / "plugin" / "hooks" / "scribe_moment.sh" def _rule(rid, title="Done means delivered and checked"): return fake_rule(id=rid, title=title, kind="rule", when_to_apply="about to say a piece of work is finished") def _moment(**kw): return rp.RuleMoment(user_id=1, query="", project_id=kw.pop("project_id", 2), **kw) PUSH = [{"moment": "work.deliver", "tool": "Bash", "match": "git push", "via": "default"}] # ── the arm ───────────────────────────────────────────────────────────── async def test_a_line_names_the_moment_and_the_act_that_reached_it(): lookup = AsyncMock(return_value=[(_rule(5), "work.deliver")]) surfaced = MagicMock() result = await rp.run_moment_arm( PUSH, _moment(), io=rp.MomentIO(lookup=lookup, record_rule_surfaced=surfaced), ) assert len(result.lines) == 1 assert "at work.deliver, reached by `git push`" in result.lines[0] assert "get_rule(5)" in result.lines[0] assert result.rule_ids == [5] lookup.assert_awaited_once_with(1, ["work.deliver"], 2) surfaced.assert_called_once() kw = surfaced.call_args.kwargs assert kw["source"] == rp.MOMENT_RULE_SOURCE assert kw["rule_ids"] == [5] assert kw["detail"] == {5: "work.deliver"} async def test_a_rule_already_shown_is_cited_after_the_fresh_ones_and_not_recorded(): lookup = AsyncMock(return_value=[(_rule(5), "work.deliver"), (_rule(9, "Other"), "work.deliver")]) surfaced = MagicMock() result = await rp.run_moment_arm( PUSH, _moment(exclude=frozenset({5})), io=rp.MomentIO(lookup=lookup, record_rule_surfaced=surfaced), ) assert result.shown_rule_ids == [9, 5] assert result.rule_ids == [9] assert result.lines[1].startswith("Also") assert surfaced.call_args.kwargs["rule_ids"] == [9] async def test_an_unbound_session_asks_for_global_rules_only(): lookup = AsyncMock(return_value=[]) await rp.run_moment_arm(PUSH, _moment(project_id=0), io=rp.MomentIO(lookup=lookup, record_rule_surfaced=MagicMock())) assert lookup.await_args.args[2] is None async def test_the_arm_fails_open(): broken = AsyncMock(side_effect=RuntimeError("db down")) result = await rp.run_moment_arm( PUSH, _moment(), io=rp.MomentIO(lookup=broken, record_rule_surfaced=MagicMock()), ) assert result.lines == [] and result.rule_ids == [] # A recorder that fails costs the telemetry, never the lines. result = await rp.run_moment_arm( PUSH, _moment(), io=rp.MomentIO(lookup=AsyncMock(return_value=[(_rule(5), "work.deliver")]), record_rule_surfaced=MagicMock(side_effect=RuntimeError("x"))), ) assert len(result.lines) == 1 async def test_nothing_reached_asks_nothing(): lookup = AsyncMock() result = await rp.run_moment_arm([], _moment(), io=rp.MomentIO(lookup=lookup, record_rule_surfaced=MagicMock())) assert result.lines == [] lookup.assert_not_awaited() # ── the act → the moments → the rules ─────────────────────────────────── def _stub_db(pairs): async def _moments_for(user_id, tool, tool_input): return moment_actions.resolve(tool, tool_input, []) return [ patch.object(md, "deliver_for_act", _REAL_DELIVER), patch.object(moment_actions, "moments_for", _moments_for), patch.object(rulebooks, "rules_on_moments", AsyncMock(return_value=pairs)), patch.object(rule_usage, "record_rule_surfaced", MagicMock()), ] async def _with(stack, coro_fn): for p in stack: p.start() try: return await coro_fn() finally: for p in reversed(stack): p.stop() async def test_a_task_closed_through_the_tool_carries_its_mounted_rules(): data = {"id": 40, "status": "done", "project_id": 2} out = await _with(_stub_db([(_rule(11), "work.finish")]), lambda: md.attach_moment_rules( 1, "update_task", {"status": "done", "project_id": 7}, data, )) assert out is data assert out["moment_rules"]["rule_ids"] == [11] assert "at work.finish, reached by `status=done`" in out["moment_rules"]["lines"][0] assert out["moment_rules"]["open_with"] == "get_rule(id)" async def test_the_records_project_wins_over_the_callers_argument(): lookup = AsyncMock(return_value=[]) stack = _stub_db([]) stack[2] = patch.object(rulebooks, "rules_on_moments", lookup) await _with(stack, lambda: md.attach_moment_rules( 1, "update_task", {"status": "done", "project_id": 7}, {"project_id": 2}, )) assert lookup.await_args.args == (1, ["work.finish"], 2) async def test_an_act_that_reaches_nothing_mounted_adds_nothing(): data = {"id": 40} out = await _with(_stub_db([]), lambda: md.attach_moment_rules( 1, "update_task", {"status": "todo"}, data, )) assert "moment_rules" not in out async def test_the_attach_fails_open_and_passes_other_payloads_through(): data = {"id": 1} with patch.object(md, "deliver_for_act", AsyncMock(side_effect=RuntimeError("x"))): assert await md.attach_moment_rules(1, "create_note", {}, data) == {"id": 1} marker = object() assert await md.attach_moment_rules(1, "start_planning", {}, marker) is marker # ── which tools the plugin needs to ask about ─────────────────────────── async def _reachable(mounted, mappings=()): with patch.object(rulebooks, "mounted_moments", AsyncMock(return_value=set(mounted))), \ patch.object(moment_actions, "list_mappings", AsyncMock(return_value=list(mappings))): return await md.reachable_tools(1) async def test_an_install_with_nothing_mounted_keeps_the_hook_off_the_wire(): listing = AsyncMock() with patch.object(rulebooks, "mounted_moments", AsyncMock(return_value=set())), \ patch.object(moment_actions, "list_mappings", listing): assert await md.reachable_tools(1) == [] listing.assert_not_awaited() async def test_only_the_tools_whose_moments_carry_a_mount_are_listed(): assert await _reachable({"work.deliver"}) == ["bash", "skill"] assert await _reachable({"work.finish"}) == ["skill", "update_milestone", "update_task"] assert await _reachable({"skill.release"}) == ["skill"] async def test_the_skill_loader_counts_whenever_anything_is_mounted(): """A stored process declares its own moments, resolved only when it loads — so a mount on any moment may be reached by loading a procedure.""" assert "skill" in await _reachable({"work.debug"}) async def test_an_installs_own_mapping_and_removal_both_count(): own = SimpleNamespace(tool="mcp__deploy__ship", match="", moment="work.deliver", effect="add") assert "ship" in await _reachable({"work.deliver"}, [own]) gone = SimpleNamespace(tool="AskUserQuestion", match="", moment="reply.ask", effect="remove") assert await _reachable({"reply.ask"}, [gone]) == [] # ── the plugin routes ─────────────────────────────────────────────────── async def test_the_route_reads_the_event_and_the_ledger(): from scribe.routes import plugin as routes deliver = AsyncMock(return_value=(PUSH, rp.RuleResult( lines=["a line"], rule_ids=[5], shown_rule_ids=[5, 9], ))) app = Quart(__name__) event = {"session_id": "s", "tool_name": "Bash", "tool_input": {"command": "git push"}} async with app.test_request_context( "/api/plugin/moment", method="POST", json=event, query_string={"project_id": "3", "exclude_rule_ids": "9", "held_rule_ids": "4"}, ): g.user = SimpleNamespace(id=7) with patch.object(routes.moment_delivery_svc, "deliver_for_act", deliver): resp = await routes.moment.__wrapped__() body = await resp.get_json() assert body == {"context": "a line", "rule_ids": [5], "moments": ["work.deliver"]} args, kw = deliver.await_args assert args == (7, "Bash", {"command": "git push"}) assert kw == {"project_id": 3, "exclude": frozenset({9}), "held": frozenset({4})} async def test_the_route_answers_an_empty_event_with_nothing(): from scribe.routes import plugin as routes app = Quart(__name__) async with app.test_request_context("/api/plugin/moment", method="POST", json={}): g.user = SimpleNamespace(id=7) resp = await routes.moment.__wrapped__() assert await resp.get_json() == {"context": "", "rule_ids": [], "moments": []} def test_both_routes_are_on_the_app(): from scribe.app import create_app rules = {str(r.rule) for r in create_app().url_map.iter_rules()} assert {"/api/plugin/moment", "/api/plugin/moment-tools"} <= rules # ── the hook ──────────────────────────────────────────────────────────── def test_the_hook_is_registered_on_every_tool(): manifest = json.loads((ROOT / "plugin" / "hooks" / "hooks.json").read_text()) entries = {m.get("matcher"): [h["command"] for h in m["hooks"]] for m in manifest["hooks"]["PreToolUse"]} assert any("scribe_moment.sh" in c for c in entries.get("*", [])) def test_the_hook_and_the_routes_agree_on_every_name(): """Rule 33 across the shell/Python seam: a renamed arg fails silently.""" src = HOOK.read_text() assert "/api/plugin/moment-tools" in src and "/api/plugin/moment" in src assert "exclude_rule_ids" in src and "scribe_held_query" in src assert "scribe_rules_live" in src and "scribe_rules_append" in src # The SHARED ledger, not one of its own. assert '"${TMPDIR:-/tmp}/scribe-priorart"' in src and ".rules.ids" in src for field in (".tools", ".rule_ids", ".context"): assert f"'{field}'" in src from scribe.routes import plugin as routes route = inspect.getsource(routes.moment) for name in ("exclude_rule_ids", "held_rule_ids", "tool_name", "tool_input"): assert name in route def test_the_hook_leaves_scribes_own_tools_to_their_responses(): assert "mcp__*scribe*__*) exit 0" in HOOK.read_text() def test_the_hook_never_returns_a_permission_decision(): """A mount is a recall aid: it informs the act and never holds it.""" code = [ln for ln in HOOK.read_text().splitlines() if not ln.lstrip().startswith("#")] assert not any("permissionDecision" in ln or "scribe_json_deny" in ln for ln in code) def _hook(tmp_path, port, tool="Bash", command="git push origin dev", session="s-moment"): need_tools("bash", "curl", "awk") env = {"PATH": os.environ["PATH"], "SCRIBE_URL": f"http://127.0.0.1:{port}", "SCRIBE_TOKEN": "t", "TMPDIR": str(tmp_path), "HOME": str(tmp_path)} out = subprocess.run( ["bash", str(HOOK)], input=json.dumps({"session_id": session, "cwd": str(tmp_path), "tool_name": tool, "tool_input": {"command": command}}), capture_output=True, text=True, env=env, timeout=30, ) assert out.returncode == 0, out.stderr return out.stdout TOOLS = {"/api/plugin/moment-tools": b'{"tools":["bash"]}'} def test_a_listed_tool_is_asked_about_and_its_rules_injected(tmp_path): replies = dict(TOOLS) replies["/api/plugin/moment"] = json.dumps( {"context": "Standing rule that may apply at work.deliver", "rule_ids": [5], "moments": ["work.deliver"]}).encode() with http_sink(by_path=replies) as (port, seen): out = _hook(tmp_path, port) # The list is read once per window, not once per call. _hook(tmp_path, port) paths = [e["_path"] for e in seen] assert paths == ["/api/plugin/moment-tools", "/api/plugin/moment", "/api/plugin/moment"] sent = json.loads(seen[1]["_body"]) assert sent["tool_input"] == {"command": "git push origin dev"} # The first call's fresh rule is on the shared ledger for the second. assert seen[2]["exclude_rule_ids"] == ["5"] ctx = json.loads(out)["hookSpecificOutput"]["additionalContext"] assert "work.deliver" in ctx def test_an_unlisted_tool_never_leaves_the_machine(tmp_path): with http_sink(by_path=TOOLS) as (port, seen): assert _hook(tmp_path, port, tool="Read") == "" assert _hook(tmp_path, port, tool="mcp__plugin_scribe_scribe__update_task") == "" assert [e["_path"] for e in seen] == ["/api/plugin/moment-tools"] def test_an_install_with_nothing_mounted_sends_one_request_per_window(tmp_path): with http_sink(by_path={"/api/plugin/moment-tools": b'{"tools":[]}'}) as (port, seen): for _ in range(3): assert _hook(tmp_path, port) == "" assert len(seen) == 1 # ── the MCP door: every shipped Scribe action carries the attach ──────── def _attached_names(path: Path) -> dict[str, set[str]]: """{function name: {tool names passed to attach_moment_rules in it}}.""" out: dict[str, set[str]] = {} for node in ast.parse(path.read_text()).body: if not isinstance(node, ast.AsyncFunctionDef): continue for call in ast.walk(node): func = call.func if isinstance(call, ast.Call) else None name = (func.id if isinstance(func, ast.Name) else func.attr if isinstance(func, ast.Attribute) else "") if (name == "attach_moment_rules" and len(call.args) > 1 and isinstance(call.args[1], ast.Constant)): out.setdefault(node.name, set()).add(call.args[1].value) return out def test_every_scribe_tool_the_defaults_name_attaches_its_moment_rules(): tools_dir = ROOT / "src" / "scribe" / "mcp" / "tools" defined: dict[str, dict[str, set[str]]] = {} scribe_tools: set[str] = set() for path in tools_dir.glob("*.py"): tree = ast.parse(path.read_text()) scribe_tools |= {n.name for n in tree.body if isinstance(n, ast.AsyncFunctionDef)} defined[path.name] = _attached_names(path) shipped = {a.tool for a in moment_actions.DEFAULT_ACTIONS} & scribe_tools assert {"update_task", "create_rule", "update_milestone"} <= shipped for tool in sorted(shipped): named = set().union(*(per_file.get(tool, set()) for per_file in defined.values())) assert tool in named, ( f"{tool} is mapped onto a moment by the shipped defaults but does not " f"attach its mounted rules — a client without the plugin would never " f"receive them" )