plugin/.claude-plugin/plugin.json

@@ -1,13 +1,17 @@
 {
   "name": "scribe",
   "description": "Scribe system-of-record for Claude Code: MCP tools over your notes/tasks/projects/rules, a session-start push channel that surfaces your always-on rules + active-project context, process-skills (writing-plans, systematic-debugging, verification, brainstorming, reusing-code), and your saved Scribe Processes auto-surfaced as skills (/scribe:sync). Replaces superpowers + file-memory with one app-backed plugin.",
-  "version": "0.1.46",
+  "version": "0.1.47",
-  "author": { "name": "Bryan Van Deusen" },
+  "author": {
+    "name": "Bryan Van Deusen"
+  },
   "mcpServers": {
     "scribe": {
       "type": "http",
       "url": "${user_config.api_endpoint}/mcp",
-      "headers": { "Authorization": "Bearer ${user_config.api_token}" }
+      "headers": {
+        "Authorization": "Bearer ${user_config.api_token}"
+      }
     }
   },
   "userConfig": {
@@ -19,7 +23,7 @@
     "api_token": {
       "type": "string",
       "title": "Scribe API key",
-      "description": "An fmcp_ API key from Settings → API Keys (read scope is enough for the session-start hook; write scope to use the tools)",
+      "description": "An fmcp_ API key from Settings \u2192 API Keys (read scope is enough for the session-start hook; write scope to use the tools)",
       "sensitive": true
     }
   }

plugin/hooks/scribe_prior_art.sh

@@ -174,17 +174,24 @@ mkdir -p "$state_dir" 2>/dev/null || true
 # (a derive group id) or a canon elsewhere (`canon:<snippet_id>`) for the
 # shapes being written. Keyed by that token, not a note id, so it dedups on
 # its own file and a family is named once per session, not at every edit.
+#
+# A FOURTH channel (milestone 307): standing RULES the write resembles. Its own
+# file for the same reason as the others — a rule named once should not be
+# re-offered on every subsequent write in the session.
 idfile=""
 syncfile=""
 derivefile=""
+rulefile=""
 exclude_q=""
 sync_exclude_q=""
 derive_exclude_q=""
+rule_exclude_q=""
 if [ -n "$session_id" ]; then
   safe_sid=$(printf '%s' "$session_id" | tr -c 'A-Za-z0-9._-' '_')
   idfile="$state_dir/${safe_sid}.ids"
   syncfile="$state_dir/${safe_sid}.sync.ids"
   derivefile="$state_dir/${safe_sid}.derive.ids"
+  rulefile="$state_dir/${safe_sid}.rules.ids"
   if [ -f "$idfile" ]; then
     seen=$(tr '\n' ',' < "$idfile" 2>/dev/null | sed 's/,$//')
     [ -n "$seen" ] && exclude_q="&exclude_ids=${seen}"
@@ -197,6 +204,10 @@ if [ -n "$session_id" ]; then
     derive_seen=$(tr '\n' ',' < "$derivefile" 2>/dev/null | sed 's/,$//' | jq -sRr '@uri' 2>/dev/null) || derive_seen=""
     [ -n "$derive_seen" ] && derive_exclude_q="&exclude_derive=${derive_seen}"
   fi
+  if [ -f "$rulefile" ]; then
+    rule_seen=$(tr '\n' ',' < "$rulefile" 2>/dev/null | sed 's/,$//')
+    [ -n "$rule_seen" ] && rule_exclude_q="&exclude_rule_ids=${rule_seen}"
+  fi
 fi
 
 # Not `|| exit 0`: an unreachable instance must not discard a local finding
@@ -206,7 +217,7 @@ fi
 reached=1
 body=$(curl -fsS --max-time 5 \
   -H "Authorization: Bearer ${token}" \
-  "${url%/}/api/plugin/prior-art?path=${path_enc}&code=${code_enc}${repo_q}${exclude_q}${sync_exclude_q}${derive_exclude_q}${shapes_q}" 2>/dev/null) || { body=""; reached=0; }
+  "${url%/}/api/plugin/prior-art?path=${path_enc}&code=${code_enc}${repo_q}${exclude_q}${sync_exclude_q}${derive_exclude_q}${rule_exclude_q}${shapes_q}" 2>/dev/null) || { body=""; reached=0; }
 unreached_context=""
 if [ "$reached" = 1 ]; then
   scribe_reached "$state_dir" "${safe_sid:-nosession}"
@@ -227,6 +238,9 @@ if [ -n "$body" ]; then
     if [ -n "$syncfile" ]; then
       printf '%s' "$body" | jq -r '(.sync_note_ids // [])[]?' 2>/dev/null >> "$syncfile" || true
     fi
+    if [ -n "$rulefile" ]; then
+      printf '%s' "$body" | jq -r '(.rule_ids // [])[]?' 2>/dev/null >> "$rulefile" || true
+    fi
     if [ -n "$derivefile" ]; then
       printf '%s' "$body" | jq -r '(.derive_keys // [])[]?' 2>/dev/null >> "$derivefile" || true
     fi

src/scribe/routes/plugin.py

@@ -129,6 +129,11 @@ async def write_path_prior_art():
                              surfaced. A separate channel on purpose: a reuse
                              hint shown early must not suppress the record-sync
                              nudge when the recorded file is edited later.
+        exclude_rule_ids (opt) — comma-separated RULE ids already surfaced
+                             this session. Its own channel like the three
+                             above, and for the same reason: a rule named
+                             twenty turns ago should not be re-offered on
+                             every subsequent write.
         exclude_derive (opt) — comma-separated derive keys (a derive group id
                              or `canon:<snippet_id>`) already named this
                              session by the ledger arm (#2900); its own
@@ -151,6 +156,7 @@ async def write_path_prior_art():
     exclude_derive = [
         p.strip() for p in (request.args.get("exclude_derive") or "").split(",") if p.strip()
     ]
+    exclude_rule_ids = _int_list(request.args.get("exclude_rule_ids"))
     shapes = _parse_shapes(request.args.get("shapes") or "")
     api_key = getattr(g, "api_key", None)
     may_stamp = api_key is None or getattr(api_key, "scope", "") == "write"
@@ -161,6 +167,7 @@ async def write_path_prior_art():
         stamp_shapes=shapes if may_stamp else None,
         repo_key=repo_bindings_svc.normalize_repo_key(repo) if repo else "",
         exclude_derive=exclude_derive,
+        exclude_rule_ids=exclude_rule_ids,
     )
     return jsonify(result)
 

src/scribe/services/embeddings.py

@@ -711,6 +711,7 @@ async def semantic_search_rules(
     query: str,
     limit: int = 5,
     threshold: float = _SIMILARITY_THRESHOLD,
+    tier: str | None = None,
 ) -> list[tuple[float, "Rule"]]:
     """Return up to *limit* (score, rule) pairs most relevant to *query*.
 
@@ -721,6 +722,11 @@ async def semantic_search_rules(
     is the surfacing question, and it has its own machinery
     (get_applicable_rules) rather than a second, subtly different copy here.
 
+    `tier` narrows to one tier. The write-path hint passes "conditional",
+    because an always-on rule is ALREADY in the session — surfacing it again as
+    a suggestion is pure noise, and noise on a hint that fires on every write
+    is how a hint gets ignored.
+
     Collapses to best-chunk-per-rule like the note search, so a long rule split
     across chunks competes once rather than crowding the results with itself.
 
@@ -757,6 +763,7 @@ async def semantic_search_rules(
                         Rulebook.owner_user_id == user_id,
                         Project.user_id == user_id,
                     ),
+                    *( [Rule.tier == tier] if tier else [] ),
                 )
                 # Overfetch so collapsing chunks to their best row still fills
                 # the page — the same reason the note search overfetches.

src/scribe/services/plugin_context.py

@@ -30,7 +30,7 @@ from scribe.services import rulebooks as rulebooks_svc
 from scribe.services import shape_ledger as shape_ledger_svc
 from scribe.services import snippets as snippets_svc
 from scribe.services.access import label_shared_items, owner_names_for
-from scribe.services.embeddings import semantic_search_notes
+from scribe.services.embeddings import semantic_search_notes, semantic_search_rules
 from scribe.services.note_usage import record_surfaced
 from scribe.services.supersession import superseded_ids
 from scribe.services.retrieval_telemetry import record_retrieval
@@ -707,6 +707,7 @@ async def build_write_path_hint(
     stamp_shapes: list[tuple[str, str]] | None = None,
     repo_key: str = "",
     exclude_derive: list[str] | None = None,
+    exclude_rule_ids: list[int] | None = None,
 ) -> dict:
     """Prior-art hint for the plugin's PreToolUse hook on Write/Edit.
 
@@ -766,7 +767,8 @@ async def build_write_path_hint(
     """
     cfg = await get_writepath_config(user_id)
     empty = {"context": "", "note_ids": [], "sync_note_ids": [], "config": cfg,
-             "stamped": [], "divergence": [], "derive": [], "derive_keys": []}
+             "stamped": [], "divergence": [], "derive": [], "derive_keys": [],
+             "rule_ids": []}
     path = (path or "").strip()
     if not cfg["enabled"] or not path:
         return empty
@@ -1036,6 +1038,50 @@ async def build_write_path_hint(
     for arm, ids in by_arm.items():
         record_surfaced(user_id=user_id, note_ids=ids, source=arm)
 
+    # ── Standing rules that may apply here (milestone 307) ──────────────
+    #
+    # A SUGGESTION, not a binding surface, and the distinction is the design
+    # (D7): a rule BINDS by being tagged to an area the project works in,
+    # resolved deterministically at enter_project. This arm reaches for
+    # something weaker and still useful — a conditional rule whose trigger
+    # resembles what is being written, noticed at the moment it is relevant
+    # rather than by being resident in every session.
+    #
+    # CONDITIONAL ONLY. An always-on rule is already in the session; repeating
+    # it here would be noise, and noise on a hint that fires on every write is
+    # how a hint gets ignored.
+    #
+    # Fails open like every other arm: a rule hint must never break a write.
+    rule_ids: list[int] = []
+    try:
+        already = set(exclude_rule_ids or [])
+        hits = await semantic_search_rules(
+            user_id, code or path, limit=2,
+            threshold=cfg["threshold"], tier="conditional",
+        )
+        fresh = [(score, rule) for score, rule in hits if rule.id not in already]
+        for _score, rule in fresh:
+            trigger = (rule.when_to_apply or "").strip()
+            lines.append(
+                f"Standing rule that may apply here — \u201c{rule.title}\u201d"
+                + (f" ({trigger})" if trigger else "")
+                + f". Read it with get_rule({rule.id}) before deciding it "
+                  "does not apply; it is not in this session's loaded set."
+            )
+            rule_ids.append(rule.id)
+        if fresh:
+            # retrieval_logs, NOT note_usage_events: that table's ids are
+            # remapped on a backup restore, so a rule id there would return
+            # attached to whatever note took that number. This one is never
+            # restored, and `source` already separates the surfaces.
+            record_retrieval(
+                user_id=user_id, source="write_path_rule", query=code or path,
+                threshold=cfg["threshold"], limit=2, project_id=project_id,
+                is_task=None, results=fresh,
+            )
+    except Exception:
+        logger.debug("write-path rule arm failed", exc_info=True)
+
     return {
         "context": "\n".join(lines),
         "note_ids": note_ids,
@@ -1045,6 +1091,7 @@ async def build_write_path_hint(
         "divergence": divergence,
         "derive": derive,
         "derive_keys": [d["key"] for d in derive],
+        "rule_ids": rule_ids,
     }
 
 

src/scribe/services/retrieval_telemetry.py

@@ -17,6 +17,7 @@ from __future__ import annotations
 
 import asyncio
 import logging
+from typing import Any
 
 from datetime import datetime, timedelta, timezone
 
@@ -108,11 +109,19 @@ def record_retrieval(
     limit: int | None,
     project_id: int | None,
     is_task: bool | None,
-    results: list[tuple[float, Note]],
+    results: list[tuple[float, Any]],
     duration_ms: float | None = None,
 ) -> None:
     """Fire-and-forget: record one retrieval call.
 
+    `results` needs only `.id` on each record, which is why it is not typed to
+    Note: rules are retrieved too (milestone 307) and land here rather than in
+    note_usage_events. That table's ids are REMAPPED on a backup restore, so a
+    rule id written into it would come back attached to whatever note happened
+    to take that number — silent corruption of the very evidence this exists to
+    provide. retrieval_logs is not restored at all, so it has no such hazard,
+    and `source` already distinguishes the surfaces.
+
     Builds the payload inline (synchronously) then schedules the insert so the
     caller returns immediately. Never raises — telemetry must not affect search.
     """

tests/conftest.py

@@ -82,3 +82,24 @@ def _no_supersession():
     with patch("scribe.services.plugin_context.superseded_ids",
                AsyncMock(return_value=set())):
         yield
+
+
+@pytest.fixture(autouse=True)
+def _no_rule_arm():
+    """Stub the write-path hint's standing-RULES arm (milestone 307).
+
+    Autouse, and deliberately so. The arm calls semantic_search_rules, which
+    loads the embedding model — so every unrelated plugin-context test that
+    already stubs the NOTES search would otherwise pull a real model into a
+    unit test through the one arm it forgot to stub. The forty-odd existing
+    call sites should not each have to learn about a new arm.
+
+    The arm's own behaviour is covered where it belongs: the document shape in
+    tests/test_services_rule_embeddings.py, the surfacing rules against real
+    Postgres in tests/test_integration_rule_surfacing.py, and the hook's dedup
+    channel in tests/test_write_path_trigger.py. A test that wants the arm
+    live can re-patch it.
+    """
+    with patch("scribe.services.plugin_context.semantic_search_rules",
+               AsyncMock(return_value=[])):
+        yield

tests/test_write_path_trigger.py

@@ -1435,3 +1435,20 @@ async def test_the_write_time_divergence_check_is_named_in_band():
         out = await pc.build_write_path_hint(1, "x.py", code=REAL_CODE, stamp_shapes=[("sym", "f")])
     check.assert_not_awaited()
     assert out["divergence"] == []
+
+
+def test_hook_keeps_the_rule_channel_apart_from_the_other_three():
+    """Milestone 307's arm, pinned the way #2708's was.
+
+    Standing rules dedup on their OWN file and their OWN query parameter. One
+    shared channel is the bug #2708 already fixed once: a hint of one class
+    silencing a different class that had never been shown. A rule named early
+    must not be re-offered on every later write, and must not silence — or be
+    silenced by — a snippet suggestion.
+    """
+    src = HOOK.read_text()
+    assert ".rules.ids" in src                      # its own state file
+    assert "exclude_rule_ids=" in src               # its own query channel
+    assert "(.rule_ids // [])[]?" in src            # its own write-back
+    # And it rides the same request as the rest, not a second round trip.
+    assert "${rule_exclude_q}" in src