fix(acl): make shared records findable, not just openable (#2079)

Option A of the #2079 fork, per the operator's call: widen the scope in
query_knowledge for every caller rather than special-casing the snippet path.

Until now the list/search queries filtered on Note.user_id alone, while
get_note_permission resolved shares properly. The result: a record shared with
you could be OPENED by id but never FOUND — invisible in Knowledge browse, in
snippet and process lists, and in the facet counts beside them.

- services/access.py gains readable_notes_clause(user_id): the same resolution
  get_note_permission does per row (ownership, direct share, group share,
  inherited project share), expressed as set membership so a list query can use
  it in one statement instead of O(n) permission round-trips.
- services/knowledge.py routes every query through it — query_knowledge, the
  keyword half of the hybrid search, query_knowledge_ids, get_knowledge_by_ids,
  get_knowledge_tags and get_knowledge_counts. The facets follow the list, or a
  tag visible in the list would filter it down to nothing.
- list items now carry user_id, since these lists can be mixed-ownership and
  the client has no other way to mark what isn't yours.

Reaches four surfaces: the Snippets list (the original report), Knowledge
browse, list_processes, and the plugin's process manifest — so a Process shared
with you now also syncs as a local skill stub, which is the point of sharing one.

NOT widened: semantic_search_notes, which scopes by NoteEmbedding.user_id and
also backs auto-inject and the search MCP tool. Widening it would put another
user's content into your agent context automatically — a product decision, not
a bug fix. Consequence until that call is made, marked at the call site: a
shared record is findable by wording but not by meaning.

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

src/scribe/services/access.py

@@ -9,7 +9,7 @@ Permission rank: owner > admin > editor > viewer
 
 import logging
 
-from sqlalchemy import select
+from sqlalchemy import or_, select
 
 from scribe.models import async_session
 from scribe.models.group import GroupMembership
@@ -161,3 +161,43 @@ async def can_read_note(user_id: int, note_id: int) -> bool:
 async def can_write_note(user_id: int, note_id: int) -> bool:
     perm = await get_note_permission(user_id, note_id)
     return perm in ("editor", "admin", "owner")
+
+
+# ---------------------------------------------------------------------------
+# Set-based visibility (for LIST queries)
+# ---------------------------------------------------------------------------
+
+async def readable_notes_clause(user_id: int):
+    """A SQLAlchemy WHERE predicate matching every note this user may READ.
+
+    `get_note_permission` answers "may I read THIS note?" one row at a time,
+    which a list query can't use — checking per row is O(n) round-trips. This is
+    the same resolution expressed as set membership so it can go straight into a
+    `select(Note).where(...)`:
+
+        1. ownership
+        2. a direct note share
+        3. a group note share
+        4. inheritance from a shared project
+
+    Keep the two in step: a rule added here belongs in `get_note_permission`
+    too, or a note becomes findable but not openable (or the reverse — which is
+    the bug this was written to fix).
+    """
+    async with async_session() as session:
+        group_ids = await _user_group_ids(session, user_id)
+
+    note_share_conds = [NoteShare.shared_with_user_id == user_id]
+    project_share_conds = [ProjectShare.shared_with_user_id == user_id]
+    if group_ids:
+        note_share_conds.append(NoteShare.shared_with_group_id.in_(group_ids))
+        project_share_conds.append(ProjectShare.shared_with_group_id.in_(group_ids))
+
+    shared_note_ids = select(NoteShare.note_id).where(or_(*note_share_conds))
+    shared_project_ids = select(ProjectShare.project_id).where(or_(*project_share_conds))
+
+    return or_(
+        Note.user_id == user_id,
+        Note.id.in_(shared_note_ids),
+        Note.project_id.in_(shared_project_ids),
+    )

src/scribe/services/knowledge.py

@@ -1,10 +1,18 @@
-"""Knowledge service — unified query across notes, tasks, plans, and processes."""
+"""Knowledge service — unified query across notes, tasks, plans, and processes.
+
+ACL (rules #47/#78): every query here is scoped to what the caller may READ —
+owned, directly shared, group-shared, or inherited from a shared project — via
+`access.readable_notes_clause`. These queries were owner-only until 2026-07-25,
+which meant a record shared with you could be opened by id but never *found*:
+invisible in browse, in search, and in the facet counts.
+"""
 import logging
 
 from sqlalchemy import func, select
 
 from scribe.models import async_session
 from scribe.models.note import Note
+from scribe.services.access import readable_notes_clause
 
 logger = logging.getLogger(__name__)
 
@@ -19,6 +27,9 @@ def _note_to_item(note: Note) -> dict:
         "snippet": (note.body or "")[:_SNIPPET_LEN],
         "tags": note.tags or [],
         "project_id": note.project_id,
+        # These lists now include records shared with the caller, so the client
+        # needs the owner to tell "mine" from "someone else's" in a mixed list.
+        "user_id": note.user_id,
         "created_at": note.created_at.isoformat(),
         "updated_at": note.updated_at.isoformat(),
     }
@@ -74,8 +85,9 @@ async def query_knowledge(
             offset=offset, project_id=project_id,
         )
 
+    visible = await readable_notes_clause(user_id)
     async with async_session() as session:
-        base = select(Note).where(Note.user_id == user_id)
+        base = select(Note).where(visible)
 
         base = _apply_type_filter(base, note_type)
         if project_id is not None:
@@ -129,11 +141,12 @@ async def _semantic_knowledge_search(
     # 1. Keyword search — title and body ILIKE
     keyword_notes: list[Note] = []
     try:
+        visible = await readable_notes_clause(user_id)
         async with async_session() as session:
             pattern = f"%{q}%"
             base = (
                 select(Note)
-                .where(Note.user_id == user_id)
+                .where(visible)
                 .where(Note.title.ilike(pattern) | Note.body.ilike(pattern))
             )
             base = _apply_type_filter(base, note_type)
@@ -150,7 +163,13 @@ async def _semantic_knowledge_search(
     except Exception:
         logger.warning("Keyword search failed", exc_info=True)
 
-    # 2. Semantic search — conceptual similarity
+    # 2. Semantic search — conceptual similarity.
+    # NOTE: this half is still OWNER-ONLY. `semantic_search_notes` scopes by
+    # `NoteEmbedding.user_id`, and it also backs auto-inject and the `search`
+    # MCP tool — so widening it would put another user's shared content into
+    # your agent context automatically. That's a product decision, not a bug
+    # fix, and it's tracked separately. Consequence until then: a shared record
+    # is findable here by WORDING (the keyword half above) but not by MEANING.
     semantic_notes: list[Note] = []
     try:
         from scribe.services.embeddings import semantic_search_notes
@@ -196,11 +215,15 @@ async def _semantic_knowledge_search(
 
 
 async def get_knowledge_tags(user_id: int, note_type: str | None = None) -> list[str]:
-    """Return all distinct tags used across knowledge objects for this user."""
+    """Return all distinct tags across the knowledge objects this user can read.
+
+    Follows the list scope: a tag facet that omitted shared records would filter
+    a list down to nothing for a tag that's plainly visible in it."""
+    visible = await readable_notes_clause(user_id)
     async with async_session() as session:
         base = (
             select(func.unnest(Note.tags).label("tag"))
-            .where(Note.user_id == user_id)
+            .where(visible)
         )
         base = _apply_type_filter(base, note_type)
         stmt = base.distinct().order_by("tag")
@@ -209,12 +232,14 @@ async def get_knowledge_tags(user_id: int, note_type: str | None = None) -> list
 
 
 async def get_knowledge_counts(user_id: int, tags: list[str] | None = None) -> dict[str, int]:
-    """Return per-type count of knowledge objects for the sidebar display."""
+    """Per-type counts for the sidebar, over what this user can read — so the
+    facet numbers match the list they sit beside."""
+    visible = await readable_notes_clause(user_id)
     async with async_session() as session:
         # Count non-task types
         stmt = (
             select(Note.note_type, func.count(Note.id))
-            .where(Note.user_id == user_id)
+            .where(visible)
             .where(Note.status.is_(None))
             .where(Note.deleted_at.is_(None))
             .where(Note.note_type.in_(["note", "process"]))
@@ -229,7 +254,7 @@ async def get_knowledge_counts(user_id: int, tags: list[str] | None = None) -> d
         # Count tasks separately (is_task = status IS NOT NULL)
         task_stmt = (
             select(func.count(Note.id))
-            .where(Note.user_id == user_id)
+            .where(visible)
             .where(Note.status.isnot(None))
             .where(Note.deleted_at.is_(None))
         )
@@ -243,7 +268,7 @@ async def get_knowledge_counts(user_id: int, tags: list[str] | None = None) -> d
         # but NOT added to total to avoid double-counting against "task".
         plan_stmt = (
             select(func.count(Note.id))
-            .where(Note.user_id == user_id)
+            .where(visible)
             .where(Note.status.isnot(None))
             .where(Note.task_kind == "plan")
             .where(Note.deleted_at.is_(None))
@@ -277,8 +302,9 @@ async def query_knowledge_ids(
         )
         return [item["id"] for item in items], total
 
+    visible = await readable_notes_clause(user_id)
     async with async_session() as session:
-        base = select(Note.id).where(Note.user_id == user_id)
+        base = select(Note.id).where(visible)
 
         base = _apply_type_filter(base, note_type)
         for tag in tags:
@@ -305,10 +331,11 @@ async def get_knowledge_by_ids(user_id: int, ids: list[int]) -> list[dict]:
     """Fetch full items for the given IDs, preserving the requested order."""
     if not ids:
         return []
+    visible = await readable_notes_clause(user_id)
     async with async_session() as session:
         stmt = (
             select(Note)
-            .where(Note.user_id == user_id)
+            .where(visible)
             .where(Note.id.in_(ids))
             .where(Note.deleted_at.is_(None))
         )

tests/test_services_access_visibility.py

@@ -0,0 +1,71 @@
+"""`readable_notes_clause` — the set-based ACL predicate behind list queries.
+
+It exists because `get_note_permission` answers "may I read THIS note?" one row
+at a time, which a list query can't use. The two must stay in step: anything
+readable one way has to be findable the other, or a record shared with you is
+openable by id but invisible in every list — the bug this predicate fixed.
+
+No DB: the group lookup is stubbed and the resulting clause is compiled to SQL
+so the shape can be asserted directly.
+"""
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+
+def _session_returning_groups(group_ids: list[int]):
+    """A stub async session whose only query — the caller's group ids — returns
+    `group_ids`."""
+    result = MagicMock()
+    result.scalars.return_value.all.return_value = group_ids
+    session = AsyncMock()
+    session.__aenter__ = AsyncMock(return_value=session)
+    session.__aexit__ = AsyncMock(return_value=False)
+    session.execute = AsyncMock(return_value=result)
+    return session
+
+
+async def _compiled_clause(group_ids: list[int]) -> str:
+    from scribe.services.access import readable_notes_clause
+    with patch("scribe.services.access.async_session") as cls:
+        cls.return_value = _session_returning_groups(group_ids)
+        clause = await readable_notes_clause(user_id=7)
+    return str(clause.compile(compile_kwargs={"literal_binds": True}))
+
+
+@pytest.mark.asyncio
+async def test_clause_covers_ownership_and_both_share_paths():
+    sql = await _compiled_clause(group_ids=[])
+    # 1. ownership
+    assert "notes.user_id = 7" in sql
+    # 2/3. a direct or group share on the note itself
+    assert "note_shares" in sql
+    assert "shared_with_user_id = 7" in sql
+    # 4. inheritance from a shared project
+    assert "project_shares" in sql
+    assert "notes.project_id IN" in sql
+
+
+@pytest.mark.asyncio
+async def test_group_membership_widens_both_share_lookups():
+    sql = await _compiled_clause(group_ids=[3, 9])
+    assert "shared_with_group_id IN (3, 9)" in sql
+    # Both the note-level and project-level share lookups gain the group arm.
+    assert sql.count("shared_with_group_id IN (3, 9)") == 2
+
+
+@pytest.mark.asyncio
+async def test_no_groups_means_no_group_arm():
+    """A user in no groups must not produce an empty `IN ()`, which would be
+    invalid SQL on some backends and always-false on others."""
+    sql = await _compiled_clause(group_ids=[])
+    assert "shared_with_group_id" not in sql
+
+
+@pytest.mark.asyncio
+async def test_owner_only_result_is_never_the_whole_table():
+    """Guard against the predicate degrading to something always-true — that
+    would expose every user's notes to every other user."""
+    sql = await _compiled_clause(group_ids=[])
+    assert "true" not in sql.lower()
+    assert "1 = 1" not in sql

tests/test_services_knowledge_counts.py

@@ -32,7 +32,12 @@ async def test_counts_include_process_in_facet_and_total():
         _scalar(1),  # tasks
         _scalar(0),  # plans
     ])
-    with patch("scribe.services.knowledge.async_session") as cls:
+    from scribe.models.note import Note
+    # The ACL predicate opens its own session; stub it so this test stays about
+    # the count facets.
+    with patch("scribe.services.knowledge.async_session") as cls, \
+         patch("scribe.services.knowledge.readable_notes_clause",
+               AsyncMock(return_value=Note.user_id == 1)):
         cls.return_value = session
         from scribe.services.knowledge import get_knowledge_counts
         counts = await get_knowledge_counts(user_id=1)

tests/test_trash_filtering.py

@@ -68,7 +68,12 @@ async def test_list_projects_excludes_trashed():
 @pytest.mark.asyncio
 async def test_query_knowledge_excludes_trashed():
     cap: list[str] = []
-    with patch("scribe.services.knowledge.async_session") as cls:
+    from scribe.models.note import Note
+    # The ACL predicate opens its own session; stub it so this test stays about
+    # trash filtering.
+    with patch("scribe.services.knowledge.async_session") as cls, \
+         patch("scribe.services.knowledge.readable_notes_clause",
+               AsyncMock(return_value=Note.user_id == 1)):
         cls.return_value = _capturing_session(cap)
         from scribe.services.knowledge import query_knowledge
         await query_knowledge(user_id=1, note_type=None, tags=[], sort="modified", q=None, limit=10, offset=0)