refactor(mcp): one rules_payload() for every surface that hands rules to an agent; drop the dead bearer resolver (#2828, milestone 296 area 4)

Reading the 16 tool modules against each other: the six-key applicable-rules
block (applicable_rules, applicable_rules_truncated, subscribed_rulebooks,
project_rules, suppressed_rules, suppressed_topics) was hand-built in five
places — enter_project, get_project, get_task (legacy plans), get_milestone
(three of the six) and services/planning.start_planning. rulebooks_svc.
rules_payload() is now the one place that names them; get_milestone gains the
three it lacked, so every rules-carrying payload reads the same. list_rules /
list_always_on_rules share _rule_summary. mcp/auth.resolve_bearer_to_user_id
duplicated resolve_bearer's parsing and had no product caller (only its own
tests) — removed; the tests now exercise resolve_bearer.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/scribe/mcp/auth.py

@@ -4,20 +4,6 @@ from __future__ import annotations
 from scribe.services.api_keys import lookup_key
 
 
-async def resolve_bearer_to_user_id(auth_header: str | None) -> int | None:
-    """Parse an `Authorization: Bearer <token>` header and return the user_id.
-
-    Returns None if the header is missing, malformed, or the token is invalid
-    or revoked. The underlying lookup_key already updates last_used_at on hit.
-    """
-    if not auth_header or not auth_header.startswith("Bearer "):
-        return None
-    raw_token = auth_header[len("Bearer "):].strip()
-    if not raw_token:
-        return None
-    api_key = await lookup_key(raw_token)
-    return api_key.user_id if api_key else None
-
 
 async def resolve_bearer(auth_header: str | None) -> tuple[int, str] | None:
     """Resolve a Bearer token to (user_id, scope).

src/scribe/mcp/tools/milestones.py

@@ -57,9 +57,7 @@ async def get_milestone(milestone_id: int) -> dict:
     return {
         "milestone": out,
         "steps": [t.to_dict() for t in steps],
-        "applicable_rules": applicable["rules"],
-        "subscribed_rulebooks": applicable["subscribed_rulebooks"],
-        "applicable_rules_truncated": applicable["truncated"],
+        **rulebooks_svc.rules_payload(applicable),
     }
 
 

src/scribe/mcp/tools/projects.py

@@ -192,12 +192,7 @@ async def enter_project(project_id: int) -> dict:
         ],
         "design_system": design_system,
         "milestone_summary": milestone_summary,
-        "applicable_rules": applicable["rules"],
-        "project_rules": applicable.get("project_rules", []),
-        "suppressed_rules": applicable.get("suppressed_rules", []),
-        "suppressed_topics": applicable.get("suppressed_topics", []),
-        "subscribed_rulebooks": applicable["subscribed_rulebooks"],
-        "applicable_rules_truncated": applicable["truncated"],
+        **rulebooks_svc.rules_payload(applicable),
         "open_tasks": [
             {
                 "id": t.id, "title": t.title, "status": t.status,
@@ -239,12 +234,7 @@ async def get_project(project_id: int) -> dict:
     applicable = await rulebooks_svc.get_applicable_rules(
         project_id=project_id, user_id=uid,
     )
-    data["applicable_rules"] = applicable["rules"]
-    data["applicable_rules_truncated"] = applicable["truncated"]
-    data["subscribed_rulebooks"] = applicable["subscribed_rulebooks"]
-    data["project_rules"] = applicable.get("project_rules", [])
-    data["suppressed_rules"] = applicable.get("suppressed_rules", [])
-    data["suppressed_topics"] = applicable.get("suppressed_topics", [])
+    data.update(rulebooks_svc.rules_payload(applicable))
     return data
 
 

src/scribe/mcp/tools/rulebooks.py

@@ -193,6 +193,12 @@ async def delete_topic(topic_id: int, confirmed: bool = False) -> dict:
 
 # ── Rule CRUD ──────────────────────────────────────────────────────────
 
+def _rule_summary(r) -> dict:
+    """The list-row shape for a rule: what an agent needs to APPLY it. The
+    full record (why, how_to_apply, timestamps) is get_rule's job."""
+    return {"id": r.id, "title": r.title, "statement": r.statement, "topic_id": r.topic_id}
+
+
 async def list_rules(
     rulebook_id: int = 0, topic_id: int = 0, project_id: int = 0,
 ) -> dict:
@@ -213,16 +219,7 @@ async def list_rules(
         topic_id=topic_id or None,
         project_id=project_id or None,
     )
-    return {
-        "rules": [
-            {
-                "id": r.id, "title": r.title, "statement": r.statement,
-                "topic_id": r.topic_id,
-            }
-            for r in rows
-        ],
-        "total": len(rows),
-    }
+    return {"rules": [_rule_summary(r) for r in rows], "total": len(rows)}
 
 
 async def list_always_on_rules() -> dict:
@@ -235,16 +232,7 @@ async def list_always_on_rules() -> dict:
     """
     uid = current_user_id()
     rules = await rulebooks_svc.list_always_on_rules(uid)
-    return {
-        "rules": [
-            {
-                "id": r.id, "title": r.title, "statement": r.statement,
-                "topic_id": r.topic_id,
-            }
-            for r in rules
-        ],
-        "total": len(rules),
-    }
+    return {"rules": [_rule_summary(r) for r in rules], "total": len(rules)}
 
 
 async def get_rule(rule_id: int) -> dict:

src/scribe/mcp/tools/tasks.py

@@ -97,12 +97,7 @@ async def get_task(task_id: int) -> dict:
         applicable = await rulebooks_svc.get_applicable_rules(
             project_id=note.project_id, user_id=uid,
         )
-        data["applicable_rules"] = applicable["rules"]
-        data["subscribed_rulebooks"] = applicable["subscribed_rulebooks"]
-        data["applicable_rules_truncated"] = applicable["truncated"]
-        data["project_rules"] = applicable.get("project_rules", [])
-        data["suppressed_rules"] = applicable.get("suppressed_rules", [])
-        data["suppressed_topics"] = applicable.get("suppressed_topics", [])
+        data.update(rulebooks_svc.rules_payload(applicable))
     data.update(await access_svc.describe_provenance(uid, note))
     # Same reasoning as get_note's record_pulled, and this is the tool where it
     # matters MOST: auto-inject ranks kind-blind over a corpus that is

src/scribe/services/planning.py

@@ -60,12 +60,7 @@ async def start_planning(user_id: int, project_id: int, title: str) -> dict:
 
     return {
         "milestone": milestone.to_dict(),
-        "applicable_rules": applicable["rules"],
-        "subscribed_rulebooks": applicable["subscribed_rulebooks"],
-        "applicable_rules_truncated": applicable["truncated"],
-        "project_rules": applicable.get("project_rules", []),
-        "suppressed_rules": applicable.get("suppressed_rules", []),
-        "suppressed_topics": applicable.get("suppressed_topics", []),
+        **rulebooks_svc.rules_payload(applicable),
         "project_goal": getattr(project, "goal", "") or "",
         "open_task_count": open_count,
     }

src/scribe/services/rulebooks.py

@@ -779,3 +779,22 @@ async def get_applicable_rules(
         "truncated": truncated,
         "subscribed_rulebooks": subscribed_rulebooks,
     }
+
+
+def rules_payload(applicable: dict) -> dict:
+    """The caller-facing shape of a get_applicable_rules() result.
+
+    Every surface that hands rules to an agent (enter_project, get_project,
+    get_milestone, get_task for legacy plans, start_planning) carries the
+    same six keys under the same names — so a reader learns them once. One
+    place renames `rules` → `applicable_rules` and `truncated` →
+    `applicable_rules_truncated`; the tools merge this into their payloads.
+    """
+    return {
+        "applicable_rules": applicable["rules"],
+        "applicable_rules_truncated": applicable["truncated"],
+        "subscribed_rulebooks": applicable["subscribed_rulebooks"],
+        "project_rules": applicable.get("project_rules", []),
+        "suppressed_rules": applicable.get("suppressed_rules", []),
+        "suppressed_topics": applicable.get("suppressed_topics", []),
+    }

tests/test_mcp_auth.py

@@ -3,20 +3,20 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
-from scribe.mcp.auth import resolve_bearer, resolve_bearer_to_user_id
+from scribe.mcp.auth import resolve_bearer
 
 
 @pytest.mark.asyncio
 async def test_resolve_bearer_missing_header_returns_none():
-    assert await resolve_bearer_to_user_id(None) is None
+    assert await resolve_bearer(None) is None
 
 
 @pytest.mark.asyncio
 async def test_resolve_bearer_malformed_header_returns_none():
-    assert await resolve_bearer_to_user_id("Token abc") is None
-    assert await resolve_bearer_to_user_id("Bearer") is None
-    assert await resolve_bearer_to_user_id("Bearer ") is None
-    assert await resolve_bearer_to_user_id("") is None
+    assert await resolve_bearer("Token abc") is None
+    assert await resolve_bearer("Bearer") is None
+    assert await resolve_bearer("Bearer ") is None
+    assert await resolve_bearer("") is None
 
 
 @pytest.mark.asyncio
@@ -25,19 +25,20 @@ async def test_resolve_bearer_unknown_token_returns_none():
         "scribe.mcp.auth.lookup_key",
         AsyncMock(return_value=None),
     ):
-        assert await resolve_bearer_to_user_id("Bearer fmcp_doesnotexist") is None
+        assert await resolve_bearer("Bearer fmcp_doesnotexist") is None
 
 
 @pytest.mark.asyncio
 async def test_resolve_bearer_valid_token_returns_user_id():
     fake_key = MagicMock()
     fake_key.user_id = 42
+    fake_key.scope = "write"
     with patch(
         "scribe.mcp.auth.lookup_key",
         AsyncMock(return_value=fake_key),
     ):
-        uid = await resolve_bearer_to_user_id("Bearer fmcp_validkey")
-    assert uid == 42
+        uid, scope = await resolve_bearer("Bearer fmcp_validkey")
+    assert (uid, scope) == (42, "write")
 
 
 @pytest.mark.asyncio
@@ -45,13 +46,14 @@ async def test_resolve_bearer_calls_lookup_with_stripped_token():
     """The Bearer prefix and any trailing whitespace must be stripped before lookup."""
     fake_key = MagicMock()
     fake_key.user_id = 1
+    fake_key.scope = "write"
     mock_lookup = AsyncMock(return_value=fake_key)
     with patch("scribe.mcp.auth.lookup_key", mock_lookup):
-        await resolve_bearer_to_user_id("Bearer fmcp_abc123  ")
+        await resolve_bearer("Bearer fmcp_abc123  ")
     mock_lookup.assert_awaited_once_with("fmcp_abc123")
 
 
-# ── resolve_bearer (user_id + scope) ────────────────────────────────────
+# ── scope ───────────────────────────────────────────────────────────────
 
 @pytest.mark.asyncio
 async def test_resolve_bearer_returns_user_id_and_scope():