fix(processes): the least-equipped kind is the one that gets followed

Survey pass 3 (#2250) tabulated capabilities per record kind. Processes came
out lowest on every column, and they are the kind with the most authority:
build_process_manifest turns each one into a skill file on the operator's
machine that auto-surfaces and is followed as written — its own docstring calls
it "the most consequential passive surface Scribe has."

Three gaps closed.

NO PULL TELEMETRY (#2476). get_process recorded nothing, while the auto-inject
menu header names get_process as the way to open that kind. Every note is
embedded regardless of note_type, so a Process is surfaceable — and the getter
the product points at was the one getter that recorded nothing, leaving every
Process permanently at zero pulls and looking like dead weight beside kinds
that merely had a counter.

get_note's own comment already listed processes as a reason to record pulls.
The fix for #2245 covered notes, tasks and snippets: it enumerated the kinds
someone thought of rather than the kinds that exist.

NO DEDUP GATE. create_process had no near-duplicate check and no force flag,
while notes, tasks, snippets and rules all have both. It matters more here than
elsewhere: two near-identical procedures don't just bloat the corpus, they
compete to be followed, and which one wins is decided by a slug collision.

NO DELETE. list/create/get/update, no delete — a kind that reads as one you
cannot retire. Deletion was always possible via delete_note, since a Process is
a note and the trash is kind-agnostic, so this was discoverability rather than
capability. delete_process checks note_type before trashing: the tool is
reached for by name, and letting it destroy an ordinary note whose id happened
to resolve would be a destructive action taken on a mistyped argument.

THE GUARD, which is the part that stops a fourth repeat.

tests/test_mcp_pull_telemetry.py discovers every get_* MCP tool by AST and
requires a record_pulled from any that loads a single note. Not a list of
getters — a get_<newkind> added tomorrow is covered the moment it loads a note
the way the others do. get_milestone is correctly excluded: it calls list_notes
for a milestone's steps, which is a surfacing, not an opening.

The loader NAMES are a list, and that residual weakness is pinned against a
rename rather than papered over. An earlier draft tried to discover new loaders
by return annotation and would have failed on create_note — which also returns
a Note. Readers and writers aren't distinguishable by type, so the honest
version is a pinned list, a non-empty assertion, and a docstring saying which
hole remains.

test_register_attaches_four_tools became a derived check of the module's public
coroutines, so the next tool added can't be left unregistered.

MCP _INSTRUCTIONS updated: product behaviour belongs in the instruction
surfaces, not in a rule (rule #119).

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UaYUaouG9jjhATyuxCKrQs

src/scribe/mcp/server.py

@@ -225,7 +225,11 @@ Scribe stores reusable Processes — saved prompts/workflows (note_type
 X process" or otherwise references a saved process, call list_processes() /
 get_process(name) and follow the returned prompt verbatim, including any
 "clarify first" steps it contains. Author a new one with create_process(title,
-body); edit with update_process.
+body); edit with update_process; retire one with delete_process (recoverable —
+it goes to the trash like anything else). A near-duplicate is refused at create
+time, because every Process becomes a skill file that auto-surfaces on the
+operator's machine: two near-identical procedures don't merely bloat the record,
+they compete to be followed.
 
 Scribe also stores Snippets — reusable functions/components recorded once for
 recall (note_type "snippet"): a name, language, signature, canonical location

src/scribe/mcp/tools/processes.py

@@ -8,8 +8,11 @@ from __future__ import annotations
 
 from scribe.mcp._context import current_user_id
 from scribe.services import access as access_svc
+from scribe.services import dedup as dedup_svc
 from scribe.services import knowledge as knowledge_svc
 from scribe.services import notes as notes_svc
+from scribe.services import trash as trash_svc
+from scribe.services.note_usage import record_pulled
 
 
 async def list_processes(q: str = "", tag: str = "", limit: int = 50) -> dict:
@@ -41,17 +44,38 @@ async def list_processes(q: str = "", tag: str = "", limit: int = 50) -> dict:
     return {"processes": procs, "total": total}
 
 
-async def create_process(title: str, body: str, tags: list[str] | None = None) -> dict:
+async def create_process(
+    title: str, body: str, tags: list[str] | None = None, force: bool = False,
+) -> dict:
     """Create a stored process (a reusable saved prompt).
 
     Args:
         title: Process name, e.g. "Drift Audit" (required).
         body: The full prompt to run later (markdown). Required.
         tags: Plain-string tags, no # prefix.
+        force: Bypass the near-duplicate gate. By default, if a title- or
+            meaning-similar process already exists, creation is BLOCKED and the
+            existing one's id is returned so you update it instead. Set true
+            only for a genuinely distinct procedure.
+
+    Returns the created process, OR — when a near-duplicate is found and force
+    is false — {"duplicate": true, "existing_id": ..., "message": ...} (nothing
+    created).
+
+    The gate matters more here than for other kinds: every process becomes a
+    skill file that auto-surfaces on the operator's machine, so two near-identical
+    procedures don't merely bloat the corpus — they compete to be followed, and
+    which one wins is decided by a slug.
     """
     if not (title or "").strip() or not (body or "").strip():
         raise ValueError("create_process requires a non-empty title and body")
     uid = current_user_id()
+    if not force:
+        dup = await dedup_svc.find_duplicate_note(
+            uid, title, body, is_task=False, note_type="process",
+        )
+        if dup is not None:
+            return dedup_svc.duplicate_response(dup, "process")
     note = await notes_svc.create_note(
         uid, title=title.strip(), body=body, note_type="process", tags=tags,
     )
@@ -82,6 +106,12 @@ async def get_process(name_or_id: str) -> dict:
     if candidates:
         out["other_matches"] = candidates
     out.update(await access_svc.describe_provenance(uid, note))
+    # A process is embedded like any other note, so auto-inject can surface one —
+    # and its menu header names THIS tool as the way to open that kind. Without
+    # this, the getter the product points at is the one getter that records
+    # nothing, and every process sits permanently at zero pulls looking like dead
+    # weight beside kinds that merely had a counter (#2476, the repeat of #2245).
+    record_pulled(user_id=uid, note_id=int(note.id), source="mcp_get_process")
     return out
 
 
@@ -119,6 +149,44 @@ async def update_process(process_id: int, title: str = "", body: str = "",
     return out
 
 
+async def delete_process(process_id: int) -> dict:
+    """Retire a stored process — it moves to the trash and is recoverable.
+
+    Reach for this when a procedure is wrong, superseded, or was never worth
+    keeping. A stored process is installed as a skill file on the operator's
+    machine and auto-surfaces there, so a bad one is followed rather than merely
+    ignored — it costs more than a missing one.
+
+    Deletion was always possible through `delete_note` (a process is a note, and
+    the trash is kind-agnostic), but nothing said so, and a kind whose own tools
+    offer create/read/update reads as one you cannot retire (#2250).
+    """
+    uid = current_user_id()
+    loaded = await notes_svc.get_note_for_user(uid, process_id)
+    note = loaded[0] if loaded else None
+    # Check the KIND before deleting: this tool is reached for by name, and
+    # letting it trash an ordinary note because the id happened to resolve would
+    # be a destructive action taken on a mistyped argument.
+    if note is None or note.note_type != "process" or note.deleted_at is not None:
+        raise ValueError(f"process {process_id} not found")
+    batch = await trash_svc.delete(uid, "note", process_id)
+    if batch is None:
+        raise ValueError(f"process {process_id} not found")
+    return {
+        "deleted_batch_id": batch,
+        "message": (
+            f"Process {process_id} moved to trash. Restore with restore('{batch}'). "
+            f"Its skill stub disappears on the operator's next process sync."
+        ),
+    }
+
+
 def register(mcp) -> None:
-    for fn in (list_processes, create_process, get_process, update_process):
+    for fn in (
+        list_processes,
+        create_process,
+        get_process,
+        update_process,
+        delete_process,
+    ):
         mcp.tool(name=fn.__name__)(fn)

tests/test_mcp_pull_telemetry.py

@@ -0,0 +1,113 @@
+"""Every getter that opens ONE note-backed record must record the pull.
+
+WHY THIS EXISTS
+
+`note_usage_events` answers "did anyone ever actually open this?" — the
+surfaced:pulled ratio is what makes dead weight visible and prunable. A getter
+that opens a record without recording it leaves that kind permanently at zero
+pulls, so it looks like dead weight beside kinds that merely had a counter.
+
+That has now happened twice:
+
+  #2245  `get_task` recorded nothing while auto-inject surfaced mostly tasks.
+         Fixed by adding the call to notes, tasks and snippets.
+  #2476  `get_process` recorded nothing — and the auto-inject menu header names
+         `get_process` as the way to open that kind. Processes were embedded
+         when #2245 was fixed; the fix enumerated the kinds someone thought of
+         rather than the kinds that exist.
+
+A missing call is the shape no behavioural test catches: it changes no return
+value (#2278, shape 4). Source inspection is the only thing that sees it.
+
+WHAT MAKES THIS DERIVED RATHER THAN A LIST
+
+The getters are not enumerated here. They are discovered from the tool modules
+by AST, and the ones that must record are identified by the loader they call —
+so a `get_<newkind>` added tomorrow is covered the moment it loads a note the
+way every other getter does.
+
+The loader names ARE a list, and that is the residual weakness. The second test
+pins them against a RENAME — the failure mode that would silently empty the
+candidate set and let this pass while checking nothing.
+
+It does not discover NEW loaders, and an earlier draft that tried to failed for
+the wrong reason: `create_note` and `update_note` also return a `Note`, so an
+annotation scan finds writers, not readers. Distinguishing them needs more than
+a type, so the honest position is a pinned list plus a non-empty assertion,
+and this paragraph saying so.
+"""
+from __future__ import annotations
+
+import ast
+import inspect
+import pathlib
+import pkgutil
+
+# Loaders that return ONE note-backed record in full. A getter calling any of
+# these is opening a record, which is the act `pulled` describes.
+#
+# `list_notes` is deliberately absent: `get_milestone` calls it to list a
+# milestone's steps, and that is a LIST — the milestone itself is not a note,
+# and its steps are surfaced rather than opened.
+SINGLE_NOTE_LOADERS = (
+    "get_note_for_user",
+    "resolve_process",
+    "get_snippet",
+)
+
+TOOLS_DIR = pathlib.Path(__file__).resolve().parents[1] / "src" / "scribe" / "mcp" / "tools"
+
+
+def _getters():
+    """(module name, function name, source) for every `get_*` MCP tool."""
+    for mod in pkgutil.iter_modules([str(TOOLS_DIR)]):
+        path = TOOLS_DIR / f"{mod.name}.py"
+        source = path.read_text()
+        for node in ast.parse(source).body:
+            if isinstance(node, ast.AsyncFunctionDef) and node.name.startswith("get_"):
+                yield mod.name, node.name, ast.get_source_segment(source, node) or ""
+
+
+def test_every_single_record_getter_records_a_pull():
+    missing = []
+    checked = []
+    for module, name, body in _getters():
+        if not any(loader in body for loader in SINGLE_NOTE_LOADERS):
+            continue
+        checked.append(f"{module}.{name}")
+        if "record_pulled" not in body:
+            missing.append(f"{module}.{name}")
+
+    # If this ever drops to zero the test has stopped testing anything — a
+    # renamed loader would silently empty the candidate set and pass.
+    assert checked, "found no note-backed getters; the loader names must have moved"
+    assert not missing, (
+        f"these getters open a record without recording the pull: {missing}. "
+        f"Add record_pulled(user_id=…, note_id=…, source='mcp_<tool>') before "
+        f"returning — see mcp/tools/notes.py:get_note."
+    )
+
+
+def test_every_named_loader_still_exists():
+    """Pins the hand-written list against a rename.
+
+    A renamed loader is the failure that matters: the candidate set above would
+    quietly empty and the first test would pass while checking nothing. The
+    `assert checked` there catches it too; this says WHICH name moved, which is
+    the difference between a five-minute fix and a puzzle.
+    """
+    from scribe.services import notes as notes_svc
+    from scribe.services import snippets as snippets_svc
+
+    available = {
+        name
+        for svc in (notes_svc, snippets_svc)
+        for name, obj in vars(svc).items()
+        if inspect.iscoroutinefunction(obj)
+    }
+    gone = [name for name in SINGLE_NOTE_LOADERS if name not in available]
+    assert not gone, (
+        f"SINGLE_NOTE_LOADERS names {gone} that no longer exist — they were "
+        f"renamed or moved. Update the list, or the pull check silently stops "
+        f"covering whatever used them."
+    )

tests/test_mcp_tool_processes.py

@@ -37,7 +37,9 @@ async def test_create_process_requires_title_and_body():
 @pytest.mark.asyncio
 async def test_create_process_sets_note_type():
     created = _fake_note()
-    with patch("scribe.services.notes.create_note",
+    with patch("scribe.mcp.tools.processes.dedup_svc.find_duplicate_note",
+               AsyncMock(return_value=None)), \
+         patch("scribe.services.notes.create_note",
                AsyncMock(return_value=created)) as mock_create:
         from scribe.mcp.tools.processes import create_process
         out = await create_process(title="Drift Audit", body="the prompt", tags=["audit"])
@@ -47,6 +49,39 @@ async def test_create_process_sets_note_type():
     assert mock_create.await_args.kwargs["title"] == "Drift Audit"
 
 
+@pytest.mark.asyncio
+async def test_create_process_blocks_a_near_duplicate():
+    """The gate matters more for processes than for other kinds: each one becomes
+    a skill file that auto-surfaces, so two near-identical procedures don't just
+    bloat the corpus — they compete to be followed (#2250)."""
+    from scribe.services.dedup import DuplicateMatch
+
+    # The real dataclass, not a MagicMock: a mock answers every attribute, so it
+    # would pass whatever field names this test happened to guess and prove
+    # nothing about the payload the tool actually returns.
+    match = DuplicateMatch(id=42, title="Drift Audit", similarity=0.94, reason="semantic")
+    with patch("scribe.mcp.tools.processes.dedup_svc.find_duplicate_note",
+               AsyncMock(return_value=match)), \
+         patch("scribe.services.notes.create_note", AsyncMock()) as mock_create:
+        from scribe.mcp.tools.processes import create_process
+        out = await create_process(title="Drift Audit", body="the prompt")
+    assert out["duplicate"] is True
+    assert out["existing_id"] == 42
+    mock_create.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_create_process_force_bypasses_the_gate():
+    created = _fake_note()
+    with patch("scribe.mcp.tools.processes.dedup_svc.find_duplicate_note",
+               AsyncMock()) as find_mock, \
+         patch("scribe.services.notes.create_note",
+               AsyncMock(return_value=created)):
+        from scribe.mcp.tools.processes import create_process
+        await create_process(title="Drift Audit", body="the prompt", force=True)
+    find_mock.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_get_process_returns_body_and_candidates():
     note = _fake_note(id=7)
@@ -117,7 +152,42 @@ async def test_update_process_refuses_a_read_only_share_with_the_reason():
     mock_update.assert_not_awaited()
 
 
-def test_register_attaches_four_tools():
+@pytest.mark.asyncio
+async def test_delete_process_trashes_it_recoverably():
+    proc = _fake_note(id=4)
+    proc.deleted_at = None
+    with patch("scribe.services.notes.get_note_for_user",
+               AsyncMock(return_value=(proc, "owner"))), \
+         patch("scribe.mcp.tools.processes.trash_svc.delete",
+               AsyncMock(return_value="batch-1")) as mock_delete:
+        from scribe.mcp.tools.processes import delete_process
+        out = await delete_process(process_id=4)
+    assert out["deleted_batch_id"] == "batch-1"
+    # Through the trash, not a hard delete — restorable like every other kind.
+    assert mock_delete.await_args.args[1] == "note"
+
+
+@pytest.mark.asyncio
+async def test_delete_process_refuses_a_plain_note():
+    """This tool is reached for by name. Letting it trash an ordinary note
+    because the id happened to resolve would be a destructive action taken on a
+    mistyped argument."""
+    plain = _fake_note(id=3, note_type="note")
+    plain.deleted_at = None
+    with patch("scribe.services.notes.get_note_for_user",
+               AsyncMock(return_value=(plain, "owner"))), \
+         patch("scribe.mcp.tools.processes.trash_svc.delete", AsyncMock()) as mock_delete:
+        from scribe.mcp.tools.processes import delete_process
+        with pytest.raises(ValueError):
+            await delete_process(process_id=3)
+    mock_delete.assert_not_awaited()
+
+
+def test_register_attaches_every_tool_in_the_module():
+    """Derived from the module rather than listed: a tool written but never
+    registered is invisible to an agent, and nothing else would notice."""
+    import inspect
+
     from scribe.mcp.tools import processes
     names: list[str] = []
 
@@ -129,6 +199,8 @@ def test_register_attaches_four_tools():
             return deco
 
     processes.register(FakeMcp())
-    assert set(names) == {
-        "list_processes", "create_process", "get_process", "update_process",
+    public = {
+        name for name, obj in vars(processes).items()
+        if inspect.iscoroutinefunction(obj) and not name.startswith("_")
     }
+    assert set(names) == public