fix(mcp): make get_note / get_task share-aware

The injected menu tells the agent to open any hit with get_note(id), and
that menu can list a collaborator's record reached through a shared
project. The fetch was still owner-only, so those lines answered
"not found" — for a record the same user opens fine in the browser.
The agent path was strictly narrower than the web path for the same id.

Same boundary miss as #2093: the list side was widened for sharing, the
fetch side wasn't. Both tools now resolve through get_note_for_user,
apply the trash filter themselves (permission resolution says nothing
about liveness), and attach describe_provenance so a shared record
arrives marked as someone else's rather than passing as the caller's.

routes/tasks.py's parent-title lookup had the same narrowness: a shared
subtask rendered as an orphan when its parent was equally shared.

Four fetch tests across three files were patching notes_svc.get_note and
had to be retargeted — note 2109's third sub-case, caught by grepping
tests/ for the old name before pushing rather than by CI (#2159).

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

src/scribe/mcp/tools/notes.py

@@ -14,6 +14,7 @@ Sentinel conventions (inherited from existing fable-mcp tools):
 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 notes as notes_svc
 from scribe.services import systems as systems_svc
@@ -57,12 +58,17 @@ async def get_note(note_id: int) -> dict:
     """Fetch the full content of a single Scribe note by its ID.
 
     Returns id, title, body (markdown), tags, project_id, created_at, updated_at.
+    A note another user shared with you also carries `shared`, `owner` and
+    `permission` — read it as their suggestion, not as settled practice you set.
     """
     uid = current_user_id()
-    note = await notes_svc.get_note(uid, note_id)
-    if note is None:
+    loaded = await notes_svc.get_note_for_user(uid, note_id)
+    note = loaded[0] if loaded else None
+    if note is None or note.deleted_at is not None:
         raise ValueError(f"note {note_id} not found")
-    return note.to_dict()
+    out = note.to_dict()
+    out.update(await access_svc.describe_provenance(uid, note))
+    return out
 
 
 async def create_note(

src/scribe/mcp/tools/tasks.py

@@ -19,6 +19,7 @@ Sentinels (preserved from existing fable-mcp):
 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 notes as notes_svc
 from scribe.services import planning as planning_svc
@@ -68,17 +69,21 @@ async def get_task(task_id: int) -> dict:
     kind=plan tasks, the response also includes applicable_rules +
     subscribed_rulebooks from the task's project's rulebook subscriptions (new
     plans are milestones — use get_milestone for those).
+
+    A task another user shared with you also carries `shared`, `owner` and
+    `permission` — it's their work item, not one you took on.
     """
     uid = current_user_id()
-    note = await notes_svc.get_note(uid, task_id)
-    if note is None:
+    loaded = await notes_svc.get_note_for_user(uid, task_id)
+    note = loaded[0] if loaded else None
+    if note is None or note.deleted_at is not None:
         raise ValueError(f"task {task_id} not found")
     data = note.to_dict()
     parent_title = None
     if note.parent_id:
-        parent = await notes_svc.get_note(uid, note.parent_id)
-        if parent is not None:
-            parent_title = parent.title
+        parent_loaded = await notes_svc.get_note_for_user(uid, note.parent_id)
+        if parent_loaded is not None:
+            parent_title = parent_loaded[0].title
     data["parent_title"] = parent_title
 
     # Legacy kind=plan tasks predate milestone-as-plan; still surface their
@@ -93,6 +98,7 @@ async def get_task(task_id: int) -> dict:
         data["project_rules"] = applicable.get("project_rules", [])
         data["suppressed_rules"] = applicable.get("suppressed_rules", [])
         data["suppressed_topics"] = applicable.get("suppressed_topics", [])
+    data.update(await access_svc.describe_provenance(uid, note))
     return data
 
 

src/scribe/routes/tasks.py

@@ -11,7 +11,6 @@ from scribe.services import systems as systems_svc
 from scribe.services.embeddings import upsert_note_embedding
 from scribe.services.notes import (
     create_note,
-    get_note,
     get_note_for_user,
     list_notes,
     update_note,
@@ -187,8 +186,10 @@ async def get_task_route(task_id: int):
     data = task.to_dict()
     data["permission"] = permission
     if task.parent_id:
-        parent = await get_note(uid, task.parent_id)
-        data["parent_title"] = parent.title if parent else None
+        # Share-aware like the task itself: a shared subtask whose parent is also
+        # shared would otherwise render as an orphan with no parent title.
+        parent = await get_note_for_user(uid, task.parent_id)
+        data["parent_title"] = parent[0].title if parent else None
     data["systems"] = [s.to_dict() for s in await systems_svc.list_record_systems(uid, task_id)]
     return jsonify(data)
 

src/scribe/services/plugin_context.py

@@ -163,6 +163,22 @@ async def get_autoinject_config(user_id: int) -> dict:
     return {"enabled": enabled, "threshold": threshold, "top_k": top_k}
 
 
+def _record_kind(note) -> str:
+    """The one-word kind marker for an injected menu line.
+
+    The menu is drawn from every record that carries an embedding, so a snippet,
+    a stored process, an issue and a stray dev-log all arrive looking identical.
+    Recorded prior art only stands out if the line says what it is — and the kind
+    is also what tells the reader which tool opens it.
+
+    Task-ness wins over `note_type` because it's the more useful distinction at a
+    glance: "there's an open issue about this" beats "there's a note about this".
+    """
+    if note.is_task:
+        return "issue" if note.task_kind == "issue" else "task"
+    return note.note_type or "note"
+
+
 async def build_autoinject_hint(
     user_id: int,
     query: str,
@@ -175,7 +191,7 @@ async def build_autoinject_hint(
       1. high-confidence threshold (stricter than pull) — set per-user;
       2. margin gate — keep only hits within _AUTOINJECT_BAND of the top score;
       3. session dedup — caller passes already-injected ids as `exclude_ids`;
-      4. title-first payload — id + title + score only, never bodies.
+      4. title-first payload — id + kind + title + score only, never bodies.
     Disabled, blank-query, or nothing-clears-the-gates all return empty context,
     so most turns inject nothing.
 
@@ -222,15 +238,19 @@ async def build_autoinject_hint(
         int(n.user_id) for _s, n in kept if n.user_id != user_id
     })
 
+    # "records", not "notes" — the menu can hold snippets, processes and tasks
+    # too, and the kind marker on each line is only legible if the header doesn't
+    # already claim they're all one thing.
     lines = [
-        "> Possibly relevant from your Scribe notes — call `get_note(id)` to "
-        "open any in full (titles only; injected once per session):",
+        "> Possibly relevant from your Scribe records — open any in full with "
+        "`get_note(id)`, or `get_snippet` / `get_process` for those kinds "
+        "(titles only; injected once per session):",
     ]
     note_ids: list[int] = []
     for score, note in kept:
         note_ids.append(int(note.id))
         title = (note.title or "(untitled)").replace("\n", " ").strip()
-        line = f"> - #{note.id} \"{title}\" ({score:.2f})"
+        line = f"> - #{note.id} [{_record_kind(note)}] \"{title}\" ({score:.2f})"
         if note.user_id != user_id:
             who = owners.get(int(note.user_id)) or "another user"
             line += f" — shared by {who}, treat as a suggestion"

tests/test_mcp_tool_notes.py

@@ -17,11 +17,18 @@ def _bind_user():
     _user_id_ctx.reset(token)
 
 
-def _fake_note(**overrides) -> MagicMock:
+def _fake_note(*, user_id: int = 7, **overrides) -> MagicMock:
     note = MagicMock()
     base = {"id": 1, "title": "t", "body": "b", "tags": [], "is_task": False}
     base.update(overrides)
     note.to_dict.return_value = base
+    # Real values, not auto-attributes: get_note reads deleted_at, and compares
+    # user_id against the bound caller to decide whether to attach a shared/owner
+    # marker. A MagicMock is truthy on both, so it would read as another user's
+    # trashed note and reach for the DB (note 2109).
+    note.id = base["id"]
+    note.user_id = user_id
+    note.deleted_at = None
     return note
 
 
@@ -108,24 +115,61 @@ async def test_list_notes_limit_clamped():
 async def test_get_note_returns_dict():
     fake = _fake_note(id=5, title="found")
     with patch(
-        "scribe.mcp.tools.notes.notes_svc.get_note",
-        AsyncMock(return_value=fake),
+        "scribe.mcp.tools.notes.notes_svc.get_note_for_user",
+        AsyncMock(return_value=(fake, "owner")),
     ):
         out = await get_note(note_id=5)
     assert out["id"] == 5
     assert out["title"] == "found"
+    # Own record: no provenance noise.
+    assert "shared" not in out
+
+
+@pytest.mark.asyncio
+async def test_get_note_opens_a_shared_record_and_says_whose_it_is():
+    """The injected menu tells the agent to open any hit with get_note(id), and
+    that menu can list a collaborator's note reached through a shared project.
+    An owner-only fetch would answer "not found" for a record the agent was just
+    handed — and the web UI opens the same note fine."""
+    theirs = _fake_note(id=5, title="Their note", user_id=9)
+    with patch(
+        "scribe.mcp.tools.notes.notes_svc.get_note_for_user",
+        AsyncMock(return_value=(theirs, "viewer")),
+    ), patch(
+        "scribe.services.access.describe_provenance",
+        AsyncMock(return_value={"shared": True, "owner": "alex",
+                                "permission": "viewer"}),
+    ):
+        out = await get_note(note_id=5)
+    assert out["shared"] is True
+    assert out["owner"] == "alex"
+    assert out["permission"] == "viewer"
 
 
 @pytest.mark.asyncio
 async def test_get_note_raises_when_not_found():
     with patch(
-        "scribe.mcp.tools.notes.notes_svc.get_note",
+        "scribe.mcp.tools.notes.notes_svc.get_note_for_user",
         AsyncMock(return_value=None),
     ):
         with pytest.raises(ValueError, match="note 999 not found"):
             await get_note(note_id=999)
 
 
+@pytest.mark.asyncio
+async def test_get_note_treats_a_trashed_note_as_missing():
+    """get_note_for_user resolves permission, not liveness — the trash filter is
+    the caller's to apply."""
+    trashed = _fake_note(id=5)
+    trashed.deleted_at = "2026-07-01T00:00:00Z"
+    with patch(
+        "scribe.mcp.tools.notes.notes_svc.get_note_for_user",
+        AsyncMock(return_value=(trashed, "owner")),
+    ):
+        with pytest.raises(ValueError, match="note 5 not found"):
+            await get_note(note_id=5)
+
+
 @pytest.mark.asyncio
 async def test_create_note_passes_through():
     fake = _fake_note(id=10, title="new")

tests/test_mcp_tool_planning.py

@@ -24,16 +24,25 @@ async def test_start_planning_tool_delegates_to_service():
     assert mock.call_args.kwargs == {"user_id": 7, "project_id": 3, "title": "Plan it"}
 
 
-@pytest.mark.asyncio
-async def test_get_task_augments_plan_with_rules():
+def _plan_note(task_kind: str):
     note = MagicMock()
     note.parent_id = None
     note.project_id = 3
-    note.to_dict.return_value = {"id": 9, "task_kind": "plan", "project_id": 3}
+    note.id = 9
+    # Real values — get_task reads deleted_at and compares user_id to the bound
+    # caller, and a MagicMock is truthy on both (note 2109).
+    note.user_id = 7
+    note.deleted_at = None
+    note.to_dict.return_value = {"id": 9, "task_kind": task_kind, "project_id": 3}
+    return note
+
+
+@pytest.mark.asyncio
+async def test_get_task_augments_plan_with_rules():
     applicable = {"rules": [{"id": 1, "title": "r"}], "truncated": False,
                   "subscribed_rulebooks": [{"id": 2, "title": "rb"}]}
-    with patch("scribe.mcp.tools.tasks.notes_svc.get_note",
-               AsyncMock(return_value=note)), \
+    with patch("scribe.mcp.tools.tasks.notes_svc.get_note_for_user",
+               AsyncMock(return_value=(_plan_note("plan"), "owner"))), \
          patch("scribe.mcp.tools.tasks.rulebooks_svc.get_applicable_rules",
                AsyncMock(return_value=applicable)):
         from scribe.mcp.tools.tasks import get_task
@@ -45,12 +54,8 @@ async def test_get_task_augments_plan_with_rules():
 
 @pytest.mark.asyncio
 async def test_get_task_work_kind_has_no_rules():
-    note = MagicMock()
-    note.parent_id = None
-    note.project_id = 3
-    note.to_dict.return_value = {"id": 9, "task_kind": "work", "project_id": 3}
-    with patch("scribe.mcp.tools.tasks.notes_svc.get_note",
-               AsyncMock(return_value=note)), \
+    with patch("scribe.mcp.tools.tasks.notes_svc.get_note_for_user",
+               AsyncMock(return_value=(_plan_note("work"), "owner"))), \
          patch("scribe.mcp.tools.tasks.rulebooks_svc.get_applicable_rules",
                AsyncMock()) as mock_rules:
         from scribe.mcp.tools.tasks import get_task

tests/test_mcp_tool_tasks.py

@@ -18,7 +18,8 @@ def _bind_user():
     _user_id_ctx.reset(token)
 
 
-def _fake_task(*, parent_id: int | None = None, **overrides) -> MagicMock:
+def _fake_task(*, parent_id: int | None = None, user_id: int = 7,
+               **overrides) -> MagicMock:
     n = MagicMock()
     n.parent_id = parent_id
     base = {
@@ -29,6 +30,12 @@ def _fake_task(*, parent_id: int | None = None, **overrides) -> MagicMock:
     base.update(overrides)
     n.to_dict.return_value = base
     n.title = base["title"]
+    n.id = base["id"]
+    # Real values, not auto-attributes: get_task reads deleted_at and compares
+    # user_id against the bound caller for the shared/owner marker — a MagicMock
+    # is truthy on both (note 2109).
+    n.user_id = user_id
+    n.deleted_at = None
     return n
 
 
@@ -63,12 +70,13 @@ async def test_list_tasks_empty_status_means_no_filter():
 async def test_get_task_with_no_parent_returns_null_parent_title():
     fake = _fake_task(id=5, title="solo", parent_id=None)
     with patch(
-        "scribe.mcp.tools.tasks.notes_svc.get_note",
-        AsyncMock(return_value=fake),
+        "scribe.mcp.tools.tasks.notes_svc.get_note_for_user",
+        AsyncMock(return_value=(fake, "owner")),
     ):
         out = await get_task(task_id=5)
     assert out["parent_title"] is None
     assert out["title"] == "solo"
+    assert "shared" not in out
 
 
 @pytest.mark.asyncio
@@ -76,9 +84,9 @@ async def test_get_task_enriches_with_parent_title():
     """When parent_id is set, get_task fetches the parent and adds parent_title."""
     child = _fake_task(id=10, title="child", parent_id=5)
     parent = _fake_task(id=5, title="parent of 10", parent_id=None)
-    # get_note is called twice: once for child, once for parent
-    mock_get = AsyncMock(side_effect=[child, parent])
-    with patch("scribe.mcp.tools.tasks.notes_svc.get_note", mock_get):
+    # fetched twice: once for the child, once for the parent
+    mock_get = AsyncMock(side_effect=[(child, "owner"), (parent, "owner")])
+    with patch("scribe.mcp.tools.tasks.notes_svc.get_note_for_user", mock_get):
         out = await get_task(task_id=10)
     assert out["parent_title"] == "parent of 10"
     assert mock_get.await_count == 2
@@ -88,16 +96,34 @@ async def test_get_task_enriches_with_parent_title():
 async def test_get_task_parent_missing_returns_null():
     """If parent_id is set but the parent is gone (orphaned), parent_title is None."""
     child = _fake_task(id=10, parent_id=5)
-    mock_get = AsyncMock(side_effect=[child, None])
-    with patch("scribe.mcp.tools.tasks.notes_svc.get_note", mock_get):
+    mock_get = AsyncMock(side_effect=[(child, "owner"), None])
+    with patch("scribe.mcp.tools.tasks.notes_svc.get_note_for_user", mock_get):
         out = await get_task(task_id=10)
     assert out["parent_title"] is None
 
 
+@pytest.mark.asyncio
+async def test_get_task_opens_a_shared_task_and_says_whose_it_is():
+    """A task in a shared project opens in the web UI, so the agent path must not
+    answer "not found" for the same id — and must say it isn't the caller's."""
+    theirs = _fake_task(id=5, title="Their task", user_id=9)
+    with patch(
+        "scribe.mcp.tools.tasks.notes_svc.get_note_for_user",
+        AsyncMock(return_value=(theirs, "viewer")),
+    ), patch(
+        "scribe.services.access.describe_provenance",
+        AsyncMock(return_value={"shared": True, "owner": "alex",
+                                "permission": "viewer"}),
+    ):
+        out = await get_task(task_id=5)
+    assert out["shared"] is True
+    assert out["owner"] == "alex"
+
+
 @pytest.mark.asyncio
 async def test_get_task_raises_when_not_found():
     with patch(
-        "scribe.mcp.tools.tasks.notes_svc.get_note",
+        "scribe.mcp.tools.tasks.notes_svc.get_note_for_user",
         AsyncMock(return_value=None),
     ):
         with pytest.raises(ValueError, match="task 999 not found"):

tests/test_retrieval_scopes.py

@@ -17,15 +17,19 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 
 
-def _note(id=1, user_id=7, title="A note"):
+def _note(id=1, user_id=7, title="A note", note_type="note",
+          is_task=False, task_kind="work"):
     n = MagicMock()
     n.id = id
     n.user_id = user_id
     n.title = title
     n.body = "body"
     n.tags = []
-    n.is_task = False
-    n.note_type = "note"
+    # Real values, not auto-attributes: the menu reads these to label each line,
+    # and a MagicMock is truthy — every record would render as a task (note 2109).
+    n.is_task = is_task
+    n.task_kind = task_kind
+    n.note_type = note_type
     return n
 
 
@@ -146,3 +150,39 @@ async def test_injected_menu_attributes_another_users_note():
     assert "suggestion" in theirs_line
     # The operator's own note stays unadorned — absence of a marker is the signal.
     assert "shared by" not in mine_line
+
+
+@pytest.mark.asyncio
+async def test_injected_menu_labels_the_record_kind():
+    """Every injected line renders the same shape, so without a kind marker a
+    recorded snippet is indistinguishable from a stray dev-log — exactly where
+    prior art most needs to stand out."""
+    from scribe.services import plugin_context
+
+    hits = [
+        (0.92, _note(1, title="debounce — rate-limit a callback", note_type="snippet")),
+        (0.91, _note(2, title="Release checklist", note_type="process")),
+        (0.90, _note(3, title="Auth token expiry", is_task=True, task_kind="issue")),
+        (0.89, _note(4, title="Ship the drafter", is_task=True)),
+        (0.88, _note(5, title="Why we dropped CalDAV")),
+    ]
+    with patch.object(plugin_context, "semantic_search_notes",
+                      AsyncMock(return_value=hits)), \
+         patch.object(plugin_context, "get_autoinject_config",
+                      AsyncMock(return_value={"enabled": True, "threshold": 0.5,
+                                              "top_k": 5})), \
+         patch.object(plugin_context, "record_retrieval", MagicMock()):
+        out = await plugin_context.build_autoinject_hint(7, "anything")
+
+    lines = out["context"].splitlines()
+    by_id = {n: next(ln for ln in lines if f"#{n} " in ln) for n in (1, 2, 3, 4, 5)}
+    assert "[snippet]" in by_id[1]
+    assert "[process]" in by_id[2]
+    # Task-ness wins over note_type, and an issue says so rather than "task".
+    assert "[issue]" in by_id[3]
+    assert "[task]" in by_id[4]
+    assert "[note]" in by_id[5]
+    # Still title-first: the marker is metadata, not an excuse to carry bodies.
+    assert "body" not in out["context"]
+    # The header can't claim they're all notes when the markers say otherwise.
+    assert "Scribe records" in lines[0]

tests/test_services_plugin_context.py

@@ -10,13 +10,15 @@ def _rule(rid, title, topic_id):
     return r
 
 
-def _note(nid, title, user_id=1):
+def _note(nid, title, user_id=1, note_type="note", is_task=False, task_kind="work"):
     n = MagicMock()
     n.id, n.title = nid, title
-    # Real int, defaulting to the caller used in these tests: the injected menu
-    # compares it to decide whether the line needs a "shared by …" attribution,
-    # and an auto-MagicMock would read as another user's note.
+    # Real values, defaulting to the caller used in these tests: the injected menu
+    # compares user_id to decide whether the line needs a "shared by …"
+    # attribution, and reads is_task/task_kind/note_type for the kind marker. An
+    # auto-MagicMock is truthy, so every line would read as another user's task.
     n.user_id = user_id
+    n.note_type, n.is_task, n.task_kind = note_type, is_task, task_kind
     return n
 
 
@@ -79,7 +81,7 @@ async def test_build_autoinject_hint_titles_only_with_margin_gate():
                                              exclude_ids=[99])
     # Margin gate kept the top two, dropped the straggler.
     assert out["note_ids"] == [11, 22]
-    assert '#11 "Pool sizing decision" (0.80)' in out["context"]
+    assert '#11 [note] "Pool sizing decision" (0.80)' in out["context"]
     assert "#33" not in out["context"]
     # Title-first: no body text, ever.
     assert "get_note(id)" in out["context"]