diff --git a/frontend/src/views/SettingsView.vue b/frontend/src/views/SettingsView.vue
index 4d3b3b0..1321cc4 100644
--- a/frontend/src/views/SettingsView.vue
+++ b/frontend/src/views/SettingsView.vue
@@ -22,6 +22,7 @@ const retentionSaved = ref(false);
const kbInjectEnabled = ref(true);
const kbInjectThreshold = ref("0.55");
const kbInjectTopK = ref("3");
+const kbWritePathEnabled = ref(true);
const savingKbInject = ref(false);
const kbInjectSaved = ref(false);
@@ -76,6 +77,9 @@ async function saveKbInject() {
kb_autoinject_enabled: kbInjectEnabled.value ? 'true' : 'false',
kb_autoinject_threshold: String(t),
kb_autoinject_top_k: String(k),
+ // Its own switch, but deliberately the same threshold/ceiling — see
+ // WRITEPATH_ENABLED_KEY in services/plugin_context.py.
+ kb_writepath_enabled: kbWritePathEnabled.value ? 'true' : 'false',
});
kbInjectSaved.value = true;
setTimeout(() => (kbInjectSaved.value = false), 2000);
@@ -459,6 +463,7 @@ onMounted(async () => {
if (allSettings.kb_autoinject_top_k !== undefined) {
kbInjectTopK.value = allSettings.kb_autoinject_top_k;
}
+ kbWritePathEnabled.value = allSettings.kb_writepath_enabled !== "false";
if (allSettings.notify_task_reminders !== undefined) {
notifyTaskReminders.value = allSettings.notify_task_reminders !== "false";
}
@@ -1193,6 +1198,19 @@ function formatUserDate(iso: string): string {
/>
Ceiling on titles surfaced at once (1–10).
+
+
+
+
+ Checks the file Claude is about to write or edit against your recorded
+ snippets — what's already kept at that path, and what resembles the code
+ being written — so a helper you already have is offered before it's
+ rewritten. Uses the same threshold and ceiling above, and never blocks the
+ edit. Off = prior art surfaces only on your own prompts.
+
+
{{ savingKbInject ? 'Saving…' : 'Save' }}
diff --git a/plugin/.claude-plugin/plugin.json b/plugin/.claude-plugin/plugin.json
index c450667..6342cbc 100644
--- a/plugin/.claude-plugin/plugin.json
+++ b/plugin/.claude-plugin/plugin.json
@@ -1,7 +1,7 @@
{
"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.17",
+ "version": "0.1.18",
"author": { "name": "Bryan Van Deusen" },
"mcpServers": {
"scribe": {
diff --git a/plugin/README.md b/plugin/README.md
index 9a76a44..6b98c20 100644
--- a/plugin/README.md
+++ b/plugin/README.md
@@ -7,6 +7,10 @@ instance into a first-class Claude Code extension:
rulebook (the `scribe` server).
- **Session-start push channel** — a `SessionStart` hook injects your always-on
rules + active-project context so Scribe surfaces *without being asked*.
+- **Prior-art recall on writes** — a `PreToolUse` hook on Write/Edit checks the
+ file about to be written against your recorded snippets (what's kept at that
+ path, and what resembles the code) and offers them before the helper is
+ rewritten. Titles only, never blocks the edit.
- **Universal process-skills** — using-scribe, writing-plans,
systematic-debugging, verification, brainstorming, reusing-code (record and
recall reusable code as snippets). Replaces superpowers.
@@ -42,6 +46,11 @@ On install you'll be asked for:
- `hooks/hooks.json` → SessionStart hook (`hooks/scribe_session_context.sh`),
**fail-open**: if Scribe is unreachable it injects nothing and never blocks
the session.
+- `hooks/hooks.json` → PreToolUse hook on `Write|Edit`
+ (`hooks/scribe_prior_art.sh`) → `GET /api/plugin/prior-art`. Returns
+ `additionalContext` with **no** permission decision, so it can inform the write
+ but never stop it; silent when nothing is recorded, which is most of the time.
+ Toggle in **Settings → Knowledge auto-inject**.
- `skills/` → the universal process-skills, surfaced by description match.
- `hooks/scribe_sync_processes.sh` (a 2nd SessionStart hook) + the `/scribe:sync`
command → generate `~/.claude/skills/scribe-proc-*` stubs from your Scribe
@@ -52,5 +61,6 @@ On install you'll be asked for:
- Set a `version` bump in `.claude-plugin/plugin.json` per release so clients
pick up changes.
-- The session-start hook needs only a **read**-scoped key; the MCP tools need
- **write** scope to create/update.
+- The session-start, auto-inject and prior-art hooks need only a **read**-scoped
+ key; the MCP tools need **write** scope to create/update. Every hook is a GET
+ for that reason — a read key cannot POST.
diff --git a/plugin/hooks/hooks.json b/plugin/hooks/hooks.json
index 3b1b802..a9cde62 100644
--- a/plugin/hooks/hooks.json
+++ b/plugin/hooks/hooks.json
@@ -23,6 +23,17 @@
}
]
}
+ ],
+ "PreToolUse": [
+ {
+ "matcher": "Write|Edit",
+ "hooks": [
+ {
+ "type": "command",
+ "command": "bash \"${CLAUDE_PLUGIN_ROOT}/hooks/scribe_prior_art.sh\""
+ }
+ ]
+ }
]
}
}
diff --git a/plugin/hooks/scribe_prior_art.sh b/plugin/hooks/scribe_prior_art.sh
new file mode 100755
index 0000000..8ae9bd1
--- /dev/null
+++ b/plugin/hooks/scribe_prior_art.sh
@@ -0,0 +1,119 @@
+#!/usr/bin/env bash
+# Scribe plugin — PreToolUse write-path trigger (prior-art recall).
+#
+# Auto-inject (scribe_autoinject.sh) fires on the operator's prompt. The moment
+# reuse is actually lost is later: when the AGENT decides mid-task to write a
+# helper. This hook fires there — on Write/Edit — and asks the operator's Scribe
+# instance what prior art is already recorded for the target file: a snippet at
+# that path or in its directory, plus snippets resembling the code about to be
+# written. Titles + ids only, never bodies.
+#
+# NEVER BLOCKS. It returns `additionalContext` with no `permissionDecision`, so
+# the write proceeds untouched and Claude sees the note beside the tool result.
+# Any failure — unconfigured, unreachable, malformed — exits 0 in silence. A
+# recall aid must not be able to stop the operator's work.
+#
+# Config (same as the other hooks), exported to the hook by Claude Code:
+# CLAUDE_PLUGIN_OPTION_api_endpoint base URL, no trailing slash
+# CLAUDE_PLUGIN_OPTION_api_token fmcp_ API key (sensitive)
+# SCRIBE_URL / SCRIBE_TOKEN override for the settings.json dogfooding path.
+set -uo pipefail
+
+command -v jq >/dev/null 2>&1 || exit 0
+command -v curl >/dev/null 2>&1 || exit 0
+
+# PreToolUse delivers { session_id, cwd, tool_name, tool_input: {...}, ... }
+event=$(cat 2>/dev/null || true)
+file_path=$(printf '%s' "$event" | jq -r '.tool_input.file_path // empty' 2>/dev/null) || exit 0
+session_id=$(printf '%s' "$event" | jq -r '.session_id // empty' 2>/dev/null) || session_id=""
+event_cwd=$(printf '%s' "$event" | jq -r '.cwd // empty' 2>/dev/null) || event_cwd=""
+
+[ -n "$file_path" ] || exit 0
+
+# The code about to be written. Write and Edit name this field differently, and
+# the names have changed across Claude Code versions — take whichever is present
+# rather than betting on one shape.
+code=$(printf '%s' "$event" | jq -r '
+ .tool_input.content // .tool_input.file_content //
+ .tool_input.new_string // .tool_input.new_str // empty' 2>/dev/null) || code=""
+
+# Skip formats that hold prose or data rather than reusable code. Purely to
+# avoid a pointless round-trip — the server would return nothing for these
+# anyway. Config formats are NOT skipped: a CI workflow or a compose file is
+# often exactly the thing worth reusing.
+case "$file_path" in
+ *.md|*.mdx|*.txt|*.rst|*.json|*.lock|*.log|*.csv|*.tsv|*.svg|*.png|*.jpg|*.jpeg|*.gif|*.ico|*.pdf)
+ exit 0 ;;
+esac
+
+url=${SCRIBE_URL:-${CLAUDE_PLUGIN_OPTION_api_endpoint:-}}
+token=${SCRIBE_TOKEN:-${CLAUDE_PLUGIN_OPTION_api_token:-}}
+# Guard against an unexpanded ${...} placeholder arriving as a literal.
+case "$url" in *'${'*) url="" ;; esac
+case "$token" in *'${'*) token="" ;; esac
+# Unconfigured install → silent. Prior-art recall is pure enrichment.
+[ -n "$url" ] && [ -n "$token" ] || exit 0
+
+# Snippet locations are recorded repo-relative, so send a repo-relative path —
+# an absolute one would simply match nothing.
+lookup_dir=$(dirname -- "$file_path" 2>/dev/null || true)
+[ -d "$lookup_dir" ] || lookup_dir=${event_cwd:-${CLAUDE_PROJECT_DIR:-$PWD}}
+repo_root=$(git -C "$lookup_dir" rev-parse --show-toplevel 2>/dev/null || true)
+rel_path="$file_path"
+if [ -n "$repo_root" ]; then
+ case "$file_path" in
+ "$repo_root"/*) rel_path="${file_path#"$repo_root"/}" ;;
+ esac
+fi
+
+# Cap the code sent as the semantic query. The embedder truncates at its own
+# token limit well before this, so a bigger slice buys no extra signal — and the
+# payload has to stay a GET (a read-scoped API key cannot POST, and every other
+# plugin hook works with a read key).
+q=$(printf '%s' "$code" | cut -c1-1200)
+
+path_enc=$(printf '%s' "$rel_path" | jq -rR '@uri' 2>/dev/null) || exit 0
+code_enc=$(printf '%s' "$q" | jq -rR '@uri' 2>/dev/null) || code_enc=""
+
+# Resolve the working repo's remote so the server can scope to the bound project.
+repo=$(git -C "$lookup_dir" remote get-url origin 2>/dev/null || true)
+repo_q=""
+if [ -n "$repo" ]; then
+ enc=$(printf '%s' "$repo" | jq -rR '@uri' 2>/dev/null) || enc=""
+ [ -n "$enc" ] && repo_q="&repo=${enc}"
+fi
+
+# Per-session dedup, in its own file rather than sharing auto-inject's. Each
+# surface shows a given snippet at most once per session, but they don't silence
+# each other: a title that flew past in a prompt menu twenty turns ago is
+# exactly what should reappear at the moment the duplicate is being written.
+state_dir="${TMPDIR:-/tmp}/scribe-priorart"
+mkdir -p "$state_dir" 2>/dev/null || true
+idfile=""
+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"
+ if [ -f "$idfile" ]; then
+ seen=$(tr '\n' ',' < "$idfile" 2>/dev/null | sed 's/,$//')
+ [ -n "$seen" ] && exclude_q="&exclude_ids=${seen}"
+ fi
+fi
+
+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}" 2>/dev/null) || exit 0
+[ -n "$body" ] || exit 0
+
+context=$(printf '%s' "$body" | jq -r '.context // empty' 2>/dev/null) || exit 0
+[ -n "$context" ] || exit 0
+
+# Remember what was surfaced so it isn't shown again this session.
+if [ -n "$idfile" ]; then
+ printf '%s' "$body" | jq -r '.note_ids[]? // empty' 2>/dev/null >> "$idfile" || true
+fi
+
+# No permissionDecision: this is a nudge, not a gate. The write goes ahead.
+jq -n --arg c "$context" \
+ '{hookSpecificOutput: {hookEventName: "PreToolUse", additionalContext: $c}}'
+exit 0
diff --git a/plugin/skills/reusing-code/SKILL.md b/plugin/skills/reusing-code/SKILL.md
index 822cb98..8767e94 100644
--- a/plugin/skills/reusing-code/SKILL.md
+++ b/plugin/skills/reusing-code/SKILL.md
@@ -32,6 +32,11 @@ through recall/auto-inject; this skill is the active reflex around that.
`location` points at the reference implementation. Adapt, don't re-derive.
- If auto-inject already surfaced a snippet title that looks relevant, that's
your cue to `get_snippet` it rather than start from scratch.
+- **Prior art offered beside a write is not noise — read it.** When Scribe notes
+ that a snippet is already recorded for the file you just wrote or edited, open
+ it before you go any further. Either it's the helper you were about to
+ duplicate — reuse it and drop yours — or it isn't, and the record needs the new
+ location adding. Both are cheaper now than after the duplicate settles in.
## The moment you build something reusable — record it
diff --git a/src/scribe/routes/plugin.py b/src/scribe/routes/plugin.py
index 1dc4498..339a9ba 100644
--- a/src/scribe/routes/plugin.py
+++ b/src/scribe/routes/plugin.py
@@ -99,6 +99,53 @@ async def autoinject_retrieve():
return jsonify(result)
+@plugin_bp.get("/prior-art")
+@login_required
+async def write_path_prior_art():
+ """Prior-art hint for the plugin's PreToolUse hook on Write/Edit.
+
+ Answers "what is already recorded for the file about to be written?" — by
+ place (a snippet at this path or in its directory) and by meaning (snippets
+ resembling the code about to be written). Titles only, gated, and empty most
+ of the time. See services.plugin_context.build_write_path_hint.
+
+ Query:
+ path (str) — the target file, REPO-RELATIVE, matching the
+ convention snippet locations are recorded in. An
+ absolute path simply won't match anything.
+ code (optional) — the code about to be written; the semantic query.
+ Omit it and only the location arm runs.
+ repo (optional) — working repo remote; resolved to the bound project
+ to scope the search, exactly as /retrieve does. It
+ is NOT used as the location `repo` filter — see the
+ service docstring for why those two differ.
+ project_id (opt) — explicit project scope override (ad-hoc/testing).
+ exclude_ids (opt) — comma-separated ids already surfaced this session.
+ """
+ path = (request.args.get("path") or "").strip()
+ code = request.args.get("code") or ""
+ try:
+ project_id = int(request.args.get("project_id", 0) or 0)
+ except (TypeError, ValueError):
+ project_id = 0
+
+ repo = (request.args.get("repo") or "").strip()
+ if repo and not project_id:
+ resolved = await repo_bindings_svc.resolve_project(g.user.id, repo)
+ if resolved:
+ project_id = resolved
+
+ exclude_ids = [
+ int(p) for p in (request.args.get("exclude_ids") or "").split(",")
+ if p.strip().isdigit()
+ ]
+
+ result = await plugin_ctx_svc.build_write_path_hint(
+ g.user.id, path, code=code, project_id=project_id, exclude_ids=exclude_ids
+ )
+ return jsonify(result)
+
+
@plugin_bp.get("/processes")
@login_required
async def process_manifest():
diff --git a/src/scribe/services/embeddings.py b/src/scribe/services/embeddings.py
index b88e2a4..b870433 100644
--- a/src/scribe/services/embeddings.py
+++ b/src/scribe/services/embeddings.py
@@ -114,6 +114,7 @@ async def semantic_search_notes(
threshold: float = _SIMILARITY_THRESHOLD,
project_id: int | None = None,
is_task: bool | None = None,
+ note_type: str | None = None,
orphan_only: bool = False,
scope: str = "own",
) -> list[tuple[float, Note]]:
@@ -122,6 +123,9 @@ async def semantic_search_notes(
Scores are cosine similarities in [-1, 1]; only notes at or above
*threshold* are returned, sorted highest-first.
+ `note_type` narrows to a single record kind (e.g. "snippet"), for callers
+ that want prior art rather than everything embedded.
+
`scope` ("own" | "browse" | "read", see access.notes_visibility_clause)
decides how far this may see. It exists because this one function serves
three different kinds of act: an explicit search, which should reach
@@ -175,6 +179,11 @@ async def semantic_search_notes(
stmt = stmt.where(Note.status.isnot(None))
elif is_task is False:
stmt = stmt.where(Note.status.is_(None))
+ # Narrow to one kind of record. Composes with is_task rather than
+ # replacing it — 'snippet' is a non-task note_type, so a caller asking
+ # for prior art gets snippets and not the dev-log that mentions them.
+ if note_type:
+ stmt = stmt.where(Note.note_type == note_type)
if exclude_ids:
stmt = stmt.where(NoteEmbedding.note_id.notin_(exclude_ids))
stmt = stmt.where(distance <= max_distance).order_by(distance.asc()).limit(limit)
diff --git a/src/scribe/services/plugin_context.py b/src/scribe/services/plugin_context.py
index 4f1b4f3..3924337 100644
--- a/src/scribe/services/plugin_context.py
+++ b/src/scribe/services/plugin_context.py
@@ -14,6 +14,7 @@ index alone already steers behavior.
"""
from __future__ import annotations
+import logging
import re
import time
@@ -25,11 +26,14 @@ from scribe.services import knowledge as knowledge_svc
from scribe.services import notes as notes_svc
from scribe.services import projects as projects_svc
from scribe.services import rulebooks as rulebooks_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.retrieval_telemetry import record_retrieval
from scribe.services.settings import get_setting
+logger = logging.getLogger(__name__)
+
# Defensive cap below Claude Code's 10k additionalContext limit.
_MAX_CHARS = 9000
@@ -51,6 +55,17 @@ AUTOINJECT_DEFAULT_ENABLED = True
AUTOINJECT_DEFAULT_THRESHOLD = 0.55
AUTOINJECT_DEFAULT_TOP_K = 3
+# The write-path trigger (#2082) gets its own on/off switch but SHARES the
+# threshold and top-k above. One knob for "how loud may Scribe be" is easier to
+# reason about than two that drift; the switch is separate because wanting prior
+# art on writes while keeping prompts quiet (or the reverse) is a real
+# preference, and it's the only part of the gate an operator can judge without
+# data. The two surfaces log under different `source` values, so if the
+# telemetry ever shows they want different thresholds, splitting them is a
+# data-backed change rather than a guess made up front.
+WRITEPATH_ENABLED_KEY = "kb_writepath_enabled"
+WRITEPATH_DEFAULT_ENABLED = True
+
# Margin gate: drop any hit more than this far below the top hit's score, so a
# single strong match doesn't drag in a wall of barely-passing neighbours.
_AUTOINJECT_BAND = 0.10
@@ -259,6 +274,174 @@ async def build_autoinject_hint(
return {"context": "\n".join(lines), "note_ids": note_ids, "config": cfg}
+# --- Write-path trigger (#2082): prior art at the moment code is written ------
+# Auto-inject above fires on the operator's prompt. The moment reuse is actually
+# lost is later — when the AGENT decides mid-task to write a helper — and nothing
+# fired there. This is that trigger: the plugin's PreToolUse hook on Write/Edit
+# asks what prior art is already recorded for the file being written.
+#
+# Two arms, deliberately different in kind:
+# - BY PLACE — a snippet recorded at this path (or in its directory) is prior
+# art by definition, not by resemblance, so it isn't scored or thresholded.
+# This is what the reverse lookup (#2083) was built to answer.
+# - BY MEANING — semantic search over snippets only, using the code about to be
+# written, under the same gates as auto-inject.
+# Place beats meaning in the menu because "there is already a canonical helper in
+# this exact file" is a stronger claim than "this resembles something".
+
+
+def _prior_art_line(item: dict, marker: str, owner: str | None) -> str:
+ """One menu line: `- #12 [here] "title"`, attributed when it isn't yours."""
+ title = (item.get("title") or "(untitled)").replace("\n", " ").strip()
+ line = f"> - #{item['id']} [{marker}] \"{title}\""
+ if owner:
+ line += f" — shared by {owner}, treat as a suggestion"
+ return line
+
+
+async def get_writepath_config(user_id: int) -> dict:
+ """Write-path trigger settings: its own `enabled`, auto-inject's gates."""
+ cfg = await get_autoinject_config(user_id)
+ enabled_raw = await get_setting(
+ user_id, WRITEPATH_ENABLED_KEY,
+ "true" if WRITEPATH_DEFAULT_ENABLED else "false",
+ )
+ return {
+ **cfg,
+ "enabled": enabled_raw.strip().lower() in ("true", "1", "yes", "on"),
+ }
+
+
+async def build_write_path_hint(
+ user_id: int,
+ path: str,
+ code: str = "",
+ project_id: int = 0,
+ exclude_ids: list[int] | None = None,
+) -> dict:
+ """Prior-art hint for the plugin's PreToolUse hook on Write/Edit.
+
+ `path` is the file about to be written, REPO-RELATIVE — matching the
+ convention snippet locations are recorded in. `code` is what's about to be
+ written, used only as the semantic query.
+
+ Same four anti-bloat gates as auto-inject (threshold, margin, session dedup
+ via `exclude_ids`, titles-never-bodies) plus the shared top-k cap across BOTH
+ arms — so a file with a lot of recorded history can't turn one edit into a
+ wall of text. Returns empty context when disabled, when there's no path, or
+ when nothing is recorded — which is the common case, and the point.
+
+ Note the repo↔project mapping is deliberately one-way: the hook sends a git
+ remote, which the ROUTE resolves to `project_id` through the repo bindings.
+ It is never used as the location `repo` filter — a snippet's `repo` is a
+ free-text label the operator typed ("Scribe"), not a remote URL, and matching
+ one against the other would silently return nothing.
+
+ Returns {"context": str, "note_ids": list[int], "config": dict}. The semantic
+ arm is logged to retrieval_logs as source='write_path' — its own source, so
+ its precision is tunable separately from auto-inject's.
+
+ KNOWN GAP: only the semantic arm is logged, matching auto-inject's convention
+ of recording the candidate set for threshold tuning. Location hits carry no
+ score, so folding them in would corrupt the score distribution the log exists
+ to capture — but it does mean a snippet surfaced BY PLACE leaves no trace. The
+ usage signal (#2085) correlates retrieval_logs against later pulls, so it will
+ need a home for un-scored surfacing before it can measure this arm.
+ """
+ cfg = await get_writepath_config(user_id)
+ empty = {"context": "", "note_ids": [], "config": cfg}
+ path = (path or "").strip()
+ if not cfg["enabled"] or not path:
+ return empty
+
+ top_k = cfg["top_k"]
+ excluded = set(exclude_ids or [])
+ scope_project = project_id or None
+
+ # --- arm 1: by place ---
+ here: list[dict] = []
+ nearby: list[dict] = []
+ try:
+ here, _ = await snippets_svc.list_snippets(
+ user_id, path=path, limit=top_k, project_id=scope_project,
+ )
+ directory = path.rsplit("/", 1)[0] if "/" in path else ""
+ if directory and len(here) < top_k:
+ nearby, _ = await snippets_svc.list_snippets(
+ user_id, path=directory, limit=top_k, project_id=scope_project,
+ )
+ except Exception:
+ logger.warning("Write-path location lookup failed", exc_info=True)
+
+ seen: set[int] = set(excluded)
+ placed: list[tuple[str, dict]] = []
+ for marker, items in (("here", here), ("nearby", nearby)):
+ for item in items:
+ nid = int(item["id"])
+ if nid in seen:
+ continue
+ seen.add(nid)
+ placed.append((marker, item))
+
+ # --- arm 2: by meaning ---
+ scored: list[tuple[str, dict]] = []
+ remaining = top_k - len(placed)
+ query = (code or "").strip()
+ if remaining > 0 and query:
+ t0 = time.perf_counter()
+ hits = await semantic_search_notes(
+ user_id, query,
+ limit=remaining,
+ threshold=cfg["threshold"],
+ project_id=scope_project,
+ exclude_ids=seen,
+ note_type="snippet",
+ # Same reasoning as auto-inject: nobody asked for this, so it takes
+ # the browse scope and never surfaces a one-to-one direct share.
+ scope="browse",
+ )
+ record_retrieval(
+ user_id=user_id, source="write_path", query=query,
+ threshold=cfg["threshold"], limit=remaining,
+ project_id=scope_project, is_task=False, results=hits,
+ duration_ms=(time.perf_counter() - t0) * 1000.0,
+ )
+ if hits:
+ top_score = hits[0][0]
+ for score, note in hits:
+ if score < top_score - _AUTOINJECT_BAND:
+ continue
+ scored.append((
+ f"similar {score:.2f}",
+ {"id": int(note.id), "title": note.title, "user_id": note.user_id},
+ ))
+
+ menu = (placed + scored)[:top_k]
+ if not menu:
+ return empty
+
+ owners = await owner_names_for({
+ int(it["user_id"]) for _m, it in menu
+ if it.get("user_id") is not None and int(it["user_id"]) != user_id
+ })
+
+ lines = [
+ f"> Prior art already recorded in Scribe for `{path}` — open one with "
+ "`get_snippet(id)` and reuse it rather than writing a fresh one-off "
+ "(titles only; shown once per session):",
+ ]
+ note_ids: list[int] = []
+ for marker, item in menu:
+ note_ids.append(int(item["id"]))
+ owner_id = item.get("user_id")
+ owner = None
+ if owner_id is not None and int(owner_id) != user_id:
+ owner = owners.get(int(owner_id)) or "another user"
+ lines.append(_prior_art_line(item, marker, owner))
+
+ return {"context": "\n".join(lines), "note_ids": note_ids, "config": cfg}
+
+
async def _topic_titles(topic_ids: set[int]) -> dict[int, str]:
"""Map topic_id -> title for the given ids (live topics only)."""
if not topic_ids:
diff --git a/tests/test_write_path_trigger.py b/tests/test_write_path_trigger.py
new file mode 100644
index 0000000..3fff4bf
--- /dev/null
+++ b/tests/test_write_path_trigger.py
@@ -0,0 +1,405 @@
+"""Tests for the write-path trigger (#2082) — prior art at the moment code is written.
+
+Covers the service (`build_write_path_hint`): the two arms and their precedence,
+the anti-bloat gates carried over from milestone 93, telemetry under its own
+source, and the two ways this must stay silent. Plus the plugin hook contract —
+a PreToolUse hook that returns a permission decision would be able to block the
+operator's edit, which this feature must never do.
+"""
+import json
+import re
+import subprocess
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+PLUGIN = Path(__file__).resolve().parents[1] / "plugin"
+HOOK = PLUGIN / "hooks" / "scribe_prior_art.sh"
+
+
+def _snippet_item(nid, title, user_id=1):
+ """A row as services.snippets.list_snippets returns it (a dict, not a Note)."""
+ return {"id": nid, "title": title, "user_id": user_id, "note_type": "snippet"}
+
+
+def _note(nid, title, user_id=1):
+ n = MagicMock()
+ n.id, n.title, n.user_id = nid, title, user_id
+ return n
+
+
+def _cfg(**over):
+ base = {"enabled": True, "threshold": 0.55, "top_k": 3}
+ base.update(over)
+ return base
+
+
+# --- silence: the common case ------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_disabled_returns_empty_and_touches_nothing():
+ from scribe.services import plugin_context as pc
+ search, listing = AsyncMock(), AsyncMock()
+ with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg(enabled=False))), \
+ patch.object(pc.snippets_svc, "list_snippets", listing), \
+ patch.object(pc, "semantic_search_notes", search):
+ out = await pc.build_write_path_hint(1, "src/x.py", code="def f(): ...")
+ assert out["context"] == "" and out["note_ids"] == []
+ listing.assert_not_called()
+ search.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_no_path_returns_empty():
+ """Nothing to look up by place, and the semantic arm alone isn't this feature."""
+ from scribe.services import plugin_context as pc
+ search = AsyncMock()
+ with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
+ patch.object(pc, "semantic_search_notes", search):
+ out = await pc.build_write_path_hint(1, " ", code="def f(): ...")
+ assert out["context"] == ""
+ search.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_nothing_recorded_is_silent():
+ from scribe.services import plugin_context as pc
+ with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
+ patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
+ patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])), \
+ patch.object(pc, "record_retrieval", MagicMock()):
+ out = await pc.build_write_path_hint(1, "src/x.py", code="def f(): ...")
+ assert out["context"] == "" and out["note_ids"] == []
+
+
+# --- arm 1: by place ---------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_snippet_at_this_path_is_surfaced_without_a_score():
+ """A snippet recorded HERE is prior art by definition, not by resemblance —
+ it must not be subject to the similarity threshold."""
+ from scribe.services import plugin_context as pc
+ with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
+ patch.object(pc.snippets_svc, "list_snippets",
+ AsyncMock(return_value=([_snippet_item(12, "debounce — rate-limit a callback")], 1))), \
+ patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])), \
+ patch.object(pc, "record_retrieval", MagicMock()), \
+ patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
+ out = await pc.build_write_path_hint(1, "src/x.py", code="def f(): ...")
+ assert out["note_ids"] == [12]
+ assert '#12 [here] "debounce — rate-limit a callback"' in out["context"]
+ assert "get_snippet(id)" in out["context"]
+ assert "src/x.py" in out["context"]
+
+
+@pytest.mark.asyncio
+async def test_directory_is_only_consulted_when_the_file_leaves_room():
+ from scribe.services import plugin_context as pc
+ calls = []
+
+ async def _listing(uid, **kw):
+ calls.append(kw["path"])
+ if kw["path"] == "src/lib/x.py":
+ return ([_snippet_item(i, f"s{i}") for i in (1, 2, 3)], 3)
+ return ([_snippet_item(9, "nearby")], 1)
+
+ with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg(top_k=3))), \
+ patch.object(pc.snippets_svc, "list_snippets", _listing), \
+ patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])), \
+ patch.object(pc, "record_retrieval", MagicMock()), \
+ patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
+ out = await pc.build_write_path_hint(1, "src/lib/x.py", code="x")
+ # The file alone filled the budget, so the directory was never queried.
+ assert calls == ["src/lib/x.py"]
+ assert out["note_ids"] == [1, 2, 3]
+
+
+@pytest.mark.asyncio
+async def test_directory_hits_are_marked_nearby_not_here():
+ from scribe.services import plugin_context as pc
+
+ async def _listing(uid, **kw):
+ if kw["path"] == "src/lib":
+ return ([_snippet_item(9, "sibling helper")], 1)
+ return ([], 0)
+
+ with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
+ patch.object(pc.snippets_svc, "list_snippets", _listing), \
+ patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])), \
+ patch.object(pc, "record_retrieval", MagicMock()), \
+ patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
+ out = await pc.build_write_path_hint(1, "src/lib/x.py", code="x")
+ assert '#9 [nearby] "sibling helper"' in out["context"]
+
+
+@pytest.mark.asyncio
+async def test_a_failing_location_lookup_does_not_sink_the_hint():
+ """The place arm is best-effort — a broken query must not cost the semantic one."""
+ from scribe.services import plugin_context as pc
+ with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
+ patch.object(pc.snippets_svc, "list_snippets", AsyncMock(side_effect=RuntimeError("boom"))), \
+ patch.object(pc, "semantic_search_notes",
+ AsyncMock(return_value=[(0.80, _note(5, "throttle — …"))])), \
+ patch.object(pc, "record_retrieval", MagicMock()), \
+ patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
+ out = await pc.build_write_path_hint(1, "src/x.py", code="def f(): ...")
+ assert out["note_ids"] == [5]
+
+
+# --- arm 2: by meaning, under the milestone-93 gates -------------------------
+
+@pytest.mark.asyncio
+async def test_semantic_arm_is_snippet_only_and_browse_scoped():
+ from scribe.services import plugin_context as pc
+ search = AsyncMock(return_value=[])
+ with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
+ patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
+ patch.object(pc, "semantic_search_notes", search), \
+ patch.object(pc, "record_retrieval", MagicMock()):
+ await pc.build_write_path_hint(1, "src/x.py", code="def f(): ...")
+ kwargs = search.await_args.kwargs
+ # Prior art means snippets — not the dev-log that happens to mention one.
+ assert kwargs["note_type"] == "snippet"
+ # Nobody asked for this, so it must not reach a one-to-one direct share.
+ assert kwargs["scope"] == "browse"
+
+
+@pytest.mark.asyncio
+async def test_margin_gate_applies_to_the_semantic_arm():
+ from scribe.services import plugin_context as pc
+ hits = [(0.80, _note(11, "near")), (0.74, _note(22, "alsoNear")), (0.61, _note(33, "far"))]
+ with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg(top_k=5))), \
+ patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
+ patch.object(pc, "semantic_search_notes", AsyncMock(return_value=hits)), \
+ patch.object(pc, "record_retrieval", MagicMock()), \
+ patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
+ out = await pc.build_write_path_hint(1, "src/x.py", code="def f(): ...")
+ assert out["note_ids"] == [11, 22]
+ assert "#33" not in out["context"]
+
+
+@pytest.mark.asyncio
+async def test_no_code_means_no_semantic_arm():
+ """An Edit whose payload we couldn't read still gets the location answer."""
+ from scribe.services import plugin_context as pc
+ search = AsyncMock()
+ with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
+ patch.object(pc.snippets_svc, "list_snippets",
+ AsyncMock(return_value=([_snippet_item(12, "here helper")], 1))), \
+ patch.object(pc, "semantic_search_notes", search), \
+ patch.object(pc, "record_retrieval", MagicMock()), \
+ patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
+ out = await pc.build_write_path_hint(1, "src/x.py", code="")
+ assert out["note_ids"] == [12]
+ search.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_place_beats_meaning_and_the_cap_covers_both_arms():
+ from scribe.services import plugin_context as pc
+
+ async def _listing(uid, **kw):
+ if kw["path"] == "src/x.py":
+ return ([_snippet_item(1, "placed")], 1)
+ return ([], 0)
+
+ with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg(top_k=2))), \
+ patch.object(pc.snippets_svc, "list_snippets", _listing), \
+ patch.object(pc, "semantic_search_notes",
+ AsyncMock(return_value=[(0.9, _note(7, "scored"))])), \
+ patch.object(pc, "record_retrieval", MagicMock()), \
+ patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
+ out = await pc.build_write_path_hint(1, "src/x.py", code="x")
+ # Location hit first; the whole menu stays within top_k.
+ assert out["note_ids"] == [1, 7]
+ assert len(out["note_ids"]) <= 2
+ assert out["context"].index("#1") < out["context"].index("#7")
+
+
+@pytest.mark.asyncio
+async def test_semantic_arm_only_asks_for_the_budget_the_place_arm_left():
+ from scribe.services import plugin_context as pc
+ search = AsyncMock(return_value=[])
+
+ async def _listing(uid, **kw):
+ if kw["path"] == "src/x.py":
+ return ([_snippet_item(1, "placed")], 1)
+ return ([], 0)
+
+ with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg(top_k=3))), \
+ patch.object(pc.snippets_svc, "list_snippets", _listing), \
+ patch.object(pc, "semantic_search_notes", search), \
+ patch.object(pc, "record_retrieval", MagicMock()):
+ await pc.build_write_path_hint(1, "src/x.py", code="x")
+ assert search.await_args.kwargs["limit"] == 2
+
+
+@pytest.mark.asyncio
+async def test_session_dedup_excludes_ids_from_both_arms():
+ from scribe.services import plugin_context as pc
+ search = AsyncMock(return_value=[])
+ with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
+ patch.object(pc.snippets_svc, "list_snippets",
+ AsyncMock(return_value=([_snippet_item(12, "already shown")], 1))), \
+ patch.object(pc, "semantic_search_notes", search), \
+ patch.object(pc, "record_retrieval", MagicMock()):
+ out = await pc.build_write_path_hint(1, "src/x.py", code="x", exclude_ids=[12])
+ # The location hit was already surfaced this session → dropped, not repeated.
+ assert out["note_ids"] == []
+ assert 12 in search.await_args.kwargs["exclude_ids"]
+
+
+# --- attribution + telemetry -------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_another_users_snippet_is_attributed():
+ """Rule #47 / the #2084 lesson: an unattributed line reads as your own record."""
+ from scribe.services import plugin_context as pc
+ with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
+ patch.object(pc.snippets_svc, "list_snippets",
+ AsyncMock(return_value=([_snippet_item(12, "theirs", user_id=42)], 1))), \
+ patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])), \
+ patch.object(pc, "record_retrieval", MagicMock()), \
+ patch.object(pc, "owner_names_for", AsyncMock(return_value={42: "alex"})):
+ out = await pc.build_write_path_hint(1, "src/x.py", code="x")
+ assert "shared by alex, treat as a suggestion" in out["context"]
+
+
+@pytest.mark.asyncio
+async def test_telemetry_uses_its_own_source():
+ """Separate source is what lets this surface be tuned apart from auto_inject."""
+ from scribe.services import plugin_context as pc
+ rec = MagicMock()
+ with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
+ patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
+ patch.object(pc, "semantic_search_notes",
+ AsyncMock(return_value=[(0.9, _note(7, "scored"))])), \
+ patch.object(pc, "record_retrieval", rec), \
+ patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
+ await pc.build_write_path_hint(1, "src/x.py", code="x", project_id=4)
+ rec.assert_called_once()
+ assert rec.call_args.kwargs["source"] == "write_path"
+ assert rec.call_args.kwargs["source"] != "auto_inject"
+ assert rec.call_args.kwargs["project_id"] == 4
+
+
+@pytest.mark.asyncio
+async def test_config_has_its_own_switch_but_shares_the_gates():
+ from scribe.services import plugin_context as pc
+ stored = {pc.WRITEPATH_ENABLED_KEY: "false"}
+ with patch.object(pc, "get_setting",
+ AsyncMock(side_effect=lambda uid, k, d: stored.get(k, d))):
+ cfg = await pc.get_writepath_config(1)
+ # Its own switch is off while auto-inject stays on...
+ assert cfg["enabled"] is False
+ # ...and the gates are auto-inject's, not a second set that could drift.
+ assert cfg["threshold"] == pc.AUTOINJECT_DEFAULT_THRESHOLD
+ assert cfg["top_k"] == pc.AUTOINJECT_DEFAULT_TOP_K
+
+
+# --- the route between them --------------------------------------------------
+
+def test_route_reads_every_arg_the_hook_sends():
+ """Rule #33: the hook's query string and the route's reader are one contract,
+ and the hook is the only caller — a rename on either side is silent."""
+ import inspect
+
+ from scribe.routes import plugin as routes
+ src = inspect.getsource(routes.write_path_prior_art)
+ for arg in ("path", "code", "repo", "project_id", "exclude_ids"):
+ assert f'request.args.get("{arg}"' in src, f"route ignores {arg}"
+
+ hook = HOOK.read_text()
+ for arg in ("path=", "code=", "repo=", "exclude_ids="):
+ assert arg in hook, f"hook never sends {arg}"
+
+
+def test_route_resolves_repo_to_a_project_not_to_a_location_filter():
+ """A snippet's `repo` is a label the operator typed; the hook sends a git
+ remote. Matching one against the other would silently return nothing, so the
+ remote may only reach resolve_project."""
+ import inspect
+
+ from scribe.routes import plugin as routes
+ src = inspect.getsource(routes.write_path_prior_art)
+ assert "resolve_project" in src
+ assert "repo=repo" not in src
+
+
+# --- the plugin surface ------------------------------------------------------
+
+def test_hook_is_registered_for_write_and_edit():
+ cfg = json.loads((PLUGIN / "hooks" / "hooks.json").read_text())
+ entries = cfg["hooks"]["PreToolUse"]
+ assert any(
+ "Write" in e.get("matcher", "") and "Edit" in e.get("matcher", "")
+ and any("scribe_prior_art.sh" in h["command"] for h in e["hooks"])
+ for e in entries
+ )
+
+
+def test_hook_never_returns_a_permission_decision():
+ """The load-bearing property: this hook informs a write, it cannot stop one.
+ A permissionDecision of deny/ask would put a recall aid in the way of the
+ operator's work."""
+ src = HOOK.read_text()
+ assert "additionalContext" in src
+ code_lines = [ln for ln in src.splitlines() if not ln.lstrip().startswith("#")]
+ assert not any("permissionDecision" in ln for ln in code_lines)
+
+
+def test_hook_is_executable_and_shell_valid():
+ assert HOOK.stat().st_mode & 0o111, "hook must be executable"
+ subprocess.run(["bash", "-n", str(HOOK)], check=True)
+
+
+def test_hook_reads_both_write_and_edit_payload_shapes():
+ """Write and Edit name the payload differently, and the names have changed
+ across Claude Code versions — read whichever is present."""
+ src = HOOK.read_text()
+ for field in ("content", "file_content", "new_string", "new_str"):
+ assert f".tool_input.{field}" in src
+ assert ".tool_input.file_path" in src
+
+
+def test_hook_stays_a_get_so_a_read_scoped_key_works():
+ """Every other plugin hook works with a read-scoped key; a POST would demand
+ write scope (see auth.py) and silently break those installs."""
+ src = HOOK.read_text()
+ assert "-X POST" not in src and "--data" not in src
+ assert "/api/plugin/prior-art?" in src
+
+
+def test_hook_exits_silently_when_unconfigured():
+ """An install with no Scribe URL/token must produce no output at all."""
+ out = subprocess.run(
+ ["bash", str(HOOK)],
+ input=json.dumps({
+ "session_id": "s1", "cwd": "/tmp", "tool_name": "Write",
+ "tool_input": {"file_path": "/tmp/x.py", "content": "def f(): ..."},
+ }),
+ capture_output=True, text=True,
+ env={"PATH": "/usr/bin:/bin", "SCRIBE_URL": "", "SCRIBE_TOKEN": ""},
+ )
+ assert out.returncode == 0
+ assert out.stdout.strip() == ""
+
+
+def test_hook_skips_prose_and_data_files():
+ """No round-trip for a markdown edit — the server would return nothing anyway."""
+ src = HOOK.read_text()
+ skip = re.search(r"case \"\$file_path\" in\n(.*?)esac", src, re.S)
+ assert skip, "expected an extension skip list"
+ for ext in ("*.md", "*.json", "*.lock", "*.png"):
+ assert ext in skip.group(1)
+ # Config formats are deliberately NOT skipped — a workflow file is reusable.
+ assert "*.yml" not in skip.group(1)
+
+
+def test_plugin_version_bumped_with_the_hook():
+ """The #1040 lesson: a plugin change clients can't see is a change that didn't
+ ship."""
+ manifest = json.loads((PLUGIN / ".claude-plugin" / "plugin.json").read_text())
+ version = tuple(int(p) for p in manifest["version"].split("."))
+ assert version >= (0, 1, 18)