src/scribe/services/access.py

@@ -182,7 +182,17 @@ async def can_write_note(user_id: int, note_id: int) -> bool:
 # you has not earned that standing, so it waits until you go looking for it.
 # ---------------------------------------------------------------------------
 
-async def readable_notes_clause(user_id: int):
+def _my_group_ids(user_id: int):
+    """The caller's group ids as a SUBQUERY, not a fetched list.
+
+    Keeping it in SQL is what makes the clause builders below pure functions —
+    no session, no await, nothing for a caller's unit test to mock — and it folds
+    the membership lookup into the one statement the caller was already running.
+    """
+    return select(GroupMembership.group_id).where(GroupMembership.user_id == user_id)
+
+
+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,
@@ -199,17 +209,20 @@ async def readable_notes_clause(user_id: int):
     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:
+    groups = _my_group_ids(user_id)
-        group_ids = await _user_group_ids(session, user_id)
 
-    note_share_conds = [NoteShare.shared_with_user_id == user_id]
+    shared_note_ids = select(NoteShare.note_id).where(
-    project_share_conds = [ProjectShare.shared_with_user_id == user_id]
+        or_(
-    if group_ids:
+            NoteShare.shared_with_user_id == user_id,
-        note_share_conds.append(NoteShare.shared_with_group_id.in_(group_ids))
+            NoteShare.shared_with_group_id.in_(groups),
-        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(
-    shared_project_ids = select(ProjectShare.project_id).where(or_(*project_share_conds))
+        or_(
+            ProjectShare.shared_with_user_id == user_id,
+            ProjectShare.shared_with_group_id.in_(groups),
+        )
+    )
 
     return or_(
         Note.user_id == user_id,
@@ -269,7 +282,7 @@ async def label_shared_items(user_id: int, items: list[dict]) -> list[dict]:
     return items
 
 
-async def browsable_notes_clause(user_id: int):
+def browsable_notes_clause(user_id: int):
     """A WHERE predicate for PASSIVE surfaces: the caller's own notes, plus
     notes in a project they can reach (owned or shared).
 
@@ -283,14 +296,12 @@ async def browsable_notes_clause(user_id: int):
     project that you don't own therefore never matches — `NULL IN (...)` is not
     true — which is exactly the intent.
     """
-    async with async_session() as session:
+    shared_project_ids = select(ProjectShare.project_id).where(
-        group_ids = await _user_group_ids(session, user_id)
+        or_(
-
+            ProjectShare.shared_with_user_id == user_id,
-    project_share_conds = [ProjectShare.shared_with_user_id == user_id]
+            ProjectShare.shared_with_group_id.in_(_my_group_ids(user_id)),
-    if group_ids:
+        )
-        project_share_conds.append(ProjectShare.shared_with_group_id.in_(group_ids))
+    )
-
-    shared_project_ids = select(ProjectShare.project_id).where(or_(*project_share_conds))
     owned_project_ids = select(Project.id).where(Project.user_id == user_id)
 
     return or_(

src/scribe/services/knowledge.py

@@ -94,7 +94,7 @@ async def query_knowledge(
 
     # No query = browsing. Narrower scope: a record shared directly with the
     # caller is search-only and must not appear in an ambient list.
-    visible = await browsable_notes_clause(user_id)
+    visible = browsable_notes_clause(user_id)
     async with async_session() as session:
         base = select(Note).where(visible)
 
@@ -152,7 +152,7 @@ async def _semantic_knowledge_search(
     try:
         # A typed query is an explicit act, so it reaches the caller's full read
         # scope — including records shared directly with them.
-        visible = await readable_notes_clause(user_id)
+        visible = readable_notes_clause(user_id)
         async with async_session() as session:
             pattern = f"%{q}%"
             base = (
@@ -231,7 +231,7 @@ async def get_knowledge_tags(user_id: int, note_type: str | None = None) -> list
     Follows the browse list rather than the read scope: a facet is itself a
     passive surface, and offering a tag that only a search-only record carries
     would filter the visible list down to nothing."""
-    visible = await browsable_notes_clause(user_id)
+    visible = browsable_notes_clause(user_id)
     async with async_session() as session:
         base = (
             select(func.unnest(Note.tags).label("tag"))
@@ -247,7 +247,7 @@ async def get_knowledge_counts(user_id: int, tags: list[str] | None = None) -> d
     """Per-type counts for the sidebar, over what this user can BROWSE — so the
     numbers match the list they sit beside rather than promising rows that only a
     search would surface."""
-    visible = await browsable_notes_clause(user_id)
+    visible = browsable_notes_clause(user_id)
     async with async_session() as session:
         # Count non-task types
         stmt = (
@@ -316,7 +316,7 @@ async def query_knowledge_ids(
         return [item["id"] for item in items], total
 
     # Browsing (see query_knowledge) — narrower scope.
-    visible = await browsable_notes_clause(user_id)
+    visible = browsable_notes_clause(user_id)
     async with async_session() as session:
         base = select(Note.id).where(visible)
 
@@ -347,7 +347,7 @@ async def get_knowledge_by_ids(user_id: int, ids: list[int]) -> list[dict]:
         return []
     # Fetching specific ids is explicit, so this takes the full read scope — the
     # ids came from either a browse or a search, and both must resolve.
-    visible = await readable_notes_clause(user_id)
+    visible = readable_notes_clause(user_id)
     async with async_session() as session:
         stmt = (
             select(Note)

src/scribe/services/notes.py

@@ -425,7 +425,7 @@ async def resolve_process(user_id: int, name_or_id) -> tuple[Note | None, list[d
     matches.
     """
     from scribe.services.access import readable_notes_clause
-    visible = await readable_notes_clause(user_id)
+    visible = readable_notes_clause(user_id)
     async with async_session() as session:
         base = select(Note).where(
             visible,

tests/test_mcp_tool_processes.py

@@ -13,11 +13,14 @@ def _bind_user():
     _user_id_ctx.reset(token)
 
 
-def _fake_note(id=1, title="Drift Audit", note_type="process"):
+def _fake_note(id=1, title="Drift Audit", note_type="process", user_id=7):
     n = MagicMock()
     n.id = id
     n.title = title
     n.note_type = note_type
+    # Must be a real int, not an auto-attribute: the provenance check compares it
+    # against the bound caller (7) to decide whether the record is shared.
+    n.user_id = user_id
     n.to_dict.return_value = {"id": id, "title": title, "note_type": note_type}
     return n
 
@@ -53,6 +56,26 @@ async def test_get_process_returns_body_and_candidates():
         out = await get_process("drift")
     assert out["id"] == 7
     assert out["other_matches"] == [{"id": 9, "title": "Drift Audit Notes"}]
+    # The caller's own process carries no provenance marker — absence of the flag
+    # is what makes it read as theirs.
+    assert "shared" not in out
+
+
+@pytest.mark.asyncio
+async def test_get_process_flags_another_users_process():
+    """A shared Process must arrive labelled. get_process's contract is 'follow
+    the returned body', so an unlabelled one would put someone else's procedure
+    in charge of the session."""
+    note = _fake_note(id=7, user_id=9)
+    with patch("scribe.services.notes.resolve_process",
+               AsyncMock(return_value=(note, []))), \
+         patch("scribe.services.access.describe_provenance",
+               AsyncMock(return_value={"shared": True, "owner": "alex",
+                                       "permission": "viewer"})):
+        from scribe.mcp.tools.processes import get_process
+        out = await get_process("deploy")
+    assert out["shared"] is True
+    assert out["owner"] == "alex"
 
 
 @pytest.mark.asyncio

tests/test_services_access_visibility.py

@@ -1,110 +1,104 @@
-"""`readable_notes_clause` — the set-based ACL predicate behind list queries.
+"""The two ACL predicates behind list queries.
 
-It exists because `get_note_permission` answers "may I read THIS note?" one row
+`get_note_permission` answers "may I read THIS note?" one row at a time, which a
-at a time, which a list query can't use. The two must stay in step: anything
+list query can't use. These express the same resolution as set membership. They
-readable one way has to be findable the other, or a record shared with you is
+are pure SQL builders — group membership is a subquery rather than a fetched
-openable by id but invisible in every list — the bug this predicate fixed.
+list, so they need no session and callers' unit tests need not know they exist.
 
-No DB: the group lookup is stubbed and the resulting clause is compiled to SQL
+Two scopes, deliberately different (decision note 2094):
-so the shape can be asserted directly.
+  readable_*  — everything the ACL permits, for explicit acts (a typed search, a
+                fetch by id).
+  browsable_* — owner + project access only, for passive surfaces (browse lists,
+                facet counts, the process→skill manifest).
+
+Assertions compile each clause to SQL and inspect its shape.
 """
-from unittest.mock import AsyncMock, MagicMock, patch
-
 import pytest
 
-
+from scribe.services.access import browsable_notes_clause, readable_notes_clause
-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:
+def _sql(clause) -> 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}))
 
 
-async def _compiled_browse_clause(group_ids: list[int]) -> str:
+def _read(user_id: int = 7) -> str:
-    from scribe.services.access import browsable_notes_clause
+    return _sql(readable_notes_clause(user_id))
-    with patch("scribe.services.access.async_session") as cls:
-        cls.return_value = _session_returning_groups(group_ids)
-        clause = await browsable_notes_clause(user_id=7)
-    return str(clause.compile(compile_kwargs={"literal_binds": True}))
 
 
-@pytest.mark.asyncio
+def _browse(user_id: int = 7) -> str:
-async def test_clause_covers_ownership_and_both_share_paths():
+    return _sql(browsable_notes_clause(user_id))
-    sql = await _compiled_clause(group_ids=[])
+
-    # 1. ownership
+
-    assert "notes.user_id = 7" in sql
+# --- read scope --------------------------------------------------------------
-    # 2/3. a direct or group share on the note itself
+
-    assert "note_shares" in sql
+def test_read_scope_covers_ownership_and_every_share_path():
-    assert "shared_with_user_id = 7" in sql
+    sql = _read()
-    # 4. inheritance from a shared project
+    assert "notes.user_id = 7" in sql          # 1. ownership
+    assert "note_shares" in sql                # 2/3. direct or group note share
+    assert "notes.project_id IN" in sql        # 4. inherited from a shared project
     assert "project_shares" in sql
-    assert "notes.project_id IN" in sql
 
 
-@pytest.mark.asyncio
+def test_read_scope_resolves_group_membership_in_sql():
-async def test_group_membership_widens_both_share_lookups():
+    """Group ids are a subquery, not a pre-fetched list — that's what keeps this
-    sql = await _compiled_clause(group_ids=[3, 9])
+    a pure function with no session of its own."""
-    assert "shared_with_group_id IN (3, 9)" in sql
+    sql = _read()
-    # Both the note-level and project-level share lookups gain the group arm.
+    assert "group_memberships.user_id = 7" in sql
-    assert sql.count("shared_with_group_id IN (3, 9)") == 2
+    # Both the note-level and project-level share lookups consult it. Count FROM
+    # clauses rather than bare occurrences — each rendered subquery names the
+    # table in SELECT, FROM and WHERE — and compare loosely, since the assertion
+    # is about the arms existing, not about SQLAlchemy's formatting.
+    assert sql.count("FROM group_memberships") >= 2
+    assert _browse().count("FROM group_memberships") >= 1
 
 
-@pytest.mark.asyncio
+def test_read_scope_is_never_the_whole_table():
-async def test_no_groups_means_no_group_arm():
+    """Guard against the predicate degrading to always-true, which would expose
-    """A user in no groups must not produce an empty `IN ()`, which would be
+    every user's notes to every other user."""
-    invalid SQL on some backends and always-false on others."""
+    sql = _read().lower()
-    sql = await _compiled_clause(group_ids=[])
+    assert " true" not in sql
-    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
 
 
 # --- browse scope: the trust boundary ---------------------------------------
 
-@pytest.mark.asyncio
+def test_browse_scope_excludes_direct_note_shares():
-async def test_browse_scope_excludes_direct_note_shares():
+    """The whole point of the narrower scope. If `note_shares` leaks in here, a
-    """The whole point of the narrower scope (decision note 2094): a record
+    record someone shared one-to-one with the operator lands in their own browse
-    shared directly with you must NOT appear in a passive list. If `note_shares`
+    list, facet counts and skill manifest as though they had recorded it."""
-    leaks in here, someone else's one-off lands in the operator's own browse
+    assert "note_shares" not in _browse()
-    list, facet counts, and skill manifest as though they had recorded it."""
-    sql = await _compiled_browse_clause(group_ids=[3])
-    assert "note_shares" not in sql
 
 
-@pytest.mark.asyncio
+def test_browse_scope_keeps_ownership_and_project_access():
-async def test_browse_scope_keeps_ownership_and_project_access():
+    sql = _browse()
-    sql = await _compiled_browse_clause(group_ids=[])
+    assert "notes.user_id = 7" in sql   # your own records
-    assert "notes.user_id = 7" in sql        # your own
+    assert "project_shares" in sql      # a project shared with you
-    assert "project_shares" in sql           # a project shared with you
+    assert "projects" in sql            # a project you own
-    assert "projects" in sql                 # a project you own
 
 
-@pytest.mark.asyncio
+def test_browse_scope_is_strictly_narrower_than_read_scope():
-async def test_browse_scope_is_strictly_narrower_than_read_scope():
+    """Browse must never surface something read scope wouldn't also allow, or a
-    """Browse must never surface something read scope wouldn't also allow —
+    list could show a record the caller cannot then open."""
-    otherwise a list could show a record the caller can't open."""
+    browse, read = _browse(), _read()
-    browse = await _compiled_browse_clause(group_ids=[3])
-    read = await _compiled_clause(group_ids=[3])
     assert "note_shares" in read and "note_shares" not in browse
-    for shared_arm in ("notes.user_id = 7", "project_shares"):
+    for arm in ("notes.user_id = 7", "project_shares"):
-        assert shared_arm in browse and shared_arm in read
+        assert arm in browse and arm in read
+
+
+def test_browse_scope_is_never_the_whole_table():
+    sql = _browse().lower()
+    assert " true" not in sql
+    assert "1 = 1" not in sql
+
+
+@pytest.mark.parametrize("clause_fn", [readable_notes_clause, browsable_notes_clause])
+def test_clauses_are_pure_builders(clause_fn):
+    """Synchronous and side-effect free — no coroutine, no session of their own.
+
+    This is the property that keeps them usable: when they opened their own
+    session, every unrelated service test had to know they existed and stub them,
+    and four test modules broke the moment a service started calling one."""
+    import inspect
+    assert not inspect.iscoroutinefunction(clause_fn)
+    assert _sql(clause_fn(7))  # builds without touching a database

tests/test_services_knowledge_counts.py

@@ -32,12 +32,7 @@ async def test_counts_include_process_in_facet_and_total():
         _scalar(1),  # tasks
         _scalar(0),  # plans
     ])
-    from scribe.models.note import Note
+    with patch("scribe.services.knowledge.async_session") as cls:
-    # 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,12 +68,7 @@ async def test_list_projects_excludes_trashed():
 @pytest.mark.asyncio
 async def test_query_knowledge_excludes_trashed():
     cap: list[str] = []
-    from scribe.models.note import Note
+    with patch("scribe.services.knowledge.async_session") as cls:
-    # 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)