tests/conftest.py

@@ -6,9 +6,19 @@ instance (e.g. a Docker service spun up by the CI job) and set DATABASE_URL
 in the environment before importing the app.
 
 For unit tests of pure functions no database is needed at all.
+
+The fixtures below are the ONE definition of three things that used to be
+copied into a dozen test modules each (#2825). They are deliberately not
+autouse: a module opts in with
+``pytestmark = pytest.mark.usefixtures("<name>")`` (or a test names the
+fixture as a parameter), so a unit test that never touches the engine or the
+MCP context pays nothing for them.
 """
 import os
+from unittest.mock import AsyncMock, patch
+
 import pytest
+import pytest_asyncio
 
 
 @pytest.fixture(autouse=True)
@@ -23,3 +33,52 @@ def _isolate_env(request, monkeypatch):
     monkeypatch.setenv("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test")
     monkeypatch.setenv("SECRET_KEY", "test-secret-key")
     monkeypatch.setenv("OLLAMA_URL", "http://localhost:11434")
+
+
+@pytest.fixture
+def _bind_user():
+    """Bind MCP caller #7 for the duration of a test.
+
+    The MCP tool layer reads the caller from a ContextVar the HTTP transport
+    sets per request; a unit test of a tool has no request, so it binds the
+    caller itself. Every tool-layer test module opts in with
+    ``pytestmark = pytest.mark.usefixtures("_bind_user")`` and builds its fakes
+    with user_id=7 so ownership checks see the caller as the owner.
+    """
+    from scribe.mcp._context import _user_id_ctx
+
+    token = _user_id_ctx.set(7)
+    yield
+    _user_id_ctx.reset(token)
+
+
+@pytest_asyncio.fixture
+async def _dispose_engine():
+    """Dispose the app's module-level engine after a test that hit Postgres.
+
+    The engine pools asyncpg connections per event loop, but pytest-asyncio
+    runs each test on a fresh loop — so without this, test 2 gets handed
+    test 1's connection bound to a now-dead loop ("Future attached to a
+    different loop"). Disposing in the test's own loop teardown clears the
+    pool cleanly. The import is deferred so merely collecting a module that
+    mixes unit and integration tests never builds an engine.
+    """
+    from scribe.models import engine
+
+    yield
+    await engine.dispose()
+
+
+@pytest.fixture
+def _no_supersession():
+    """Stub the auto-inject menu's "which lines are superseded?" lookup (#278).
+
+    That is a real database call on a path the plugin-context tests exercise
+    without one. Stubbed to "nothing superseded" — the ordinary state — rather
+    than hidden behind a try/except in the product, which would make the code
+    lie about what it does. The label's own behaviour is covered in
+    tests/test_supersession_ranking.py.
+    """
+    with patch("scribe.services.plugin_context.superseded_ids",
+               AsyncMock(return_value=set())):
+        yield

tests/helpers.py

@@ -0,0 +1,77 @@
+"""Shared test helpers — the plain functions tests call, as opposed to the
+fixtures in conftest.py.
+
+Each of these was copied into several test modules before #2825 consolidated
+them; a module imports what it needs with ``from tests.helpers import ...``.
+"""
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock
+
+
+def make_mock_session() -> AsyncMock:
+    """A stand-in for ``async_session()`` — usable as ``async with``, with the
+    commit/refresh/add surface a service touches.
+
+    ``add`` is a MagicMock because the real ``Session.add`` is synchronous;
+    an AsyncMock there would hand the service an un-awaited coroutine.
+    """
+    s = AsyncMock()
+    s.__aenter__ = AsyncMock(return_value=s)
+    s.__aexit__ = AsyncMock(return_value=False)
+    s.add = MagicMock()
+    s.commit = AsyncMock()
+    s.refresh = AsyncMock()
+    return s
+
+
+async def ensure_user(session, username: str, role: str = "user"):
+    """Get-or-create a User by username inside an open session (flushed, not
+    committed).
+
+    Integration tests share one database for the whole lane run, so a second
+    test re-creating the same username dies on the unique constraint —
+    every integration seed goes through this instead of ``User(...)`` + add.
+    """
+    from sqlalchemy import select
+
+    from scribe.models.user import User
+
+    existing = (
+        await session.execute(select(User).where(User.username == username))
+    ).scalar_one_or_none()
+    if existing is not None:
+        return existing
+    user = User(username=username, role=role)
+    session.add(user)
+    await session.flush()
+    return user
+
+
+def fake_note(**attrs) -> MagicMock:
+    """A MagicMock note with REAL values on every attribute the product reads
+    to label, scope, or render a record.
+
+    The hazard this exists for (note 2109): an auto-created MagicMock attribute
+    is truthy and has a repr. The injected menu reads ``is_task`` /
+    ``task_kind`` / ``note_type`` for its kind marker, ``user_id`` to decide
+    whether a line needs a "shared by …" attribution, ``data`` for a snippet's
+    language tag, and ``deleted_at`` to spot trash — on a bare MagicMock every
+    record renders as another user's trashed task with a mock repr for a
+    language. Defaults below are the ORDINARY state (own note, live, no
+    structured data); override what the test is about.
+
+    ``to_dict()`` returns the same values as a plain dict, so a tool that
+    repackages ``note.to_dict()`` sees keys that agree with the attributes.
+    """
+    values = {
+        "id": 1, "title": "t", "body": "", "tags": [], "user_id": 7,
+        "note_type": "note", "is_task": False, "task_kind": "work",
+        "data": None, "deleted_at": None,
+    }
+    values.update(attrs)
+    n = MagicMock()
+    for key, value in values.items():
+        setattr(n, key, value)
+    n.to_dict.return_value = dict(values)
+    return n

tests/test_forge_webhook.py

@@ -21,6 +21,7 @@ import pytest_asyncio
 
 from scribe.routes.webhooks import delivered_signature, push_facts, signature_ok
 from scribe.services.snippets import _path_touches
+from tests.helpers import ensure_user
 
 SECRET = "wh-secret"
 HEAD = "e" * 40
@@ -122,32 +123,17 @@ def test_route_is_registered_and_unauthenticated_by_design():
 
 # --- integration: the flag lands and clears on real Postgres -----------------
 
-@pytest_asyncio.fixture
-async def _dispose_engine():
-    from scribe.models import engine
-    yield
-    await engine.dispose()
-
 
 @pytest_asyncio.fixture
 async def seeded(_dispose_engine):
     """User + project + binding + two verified snippets + one unverified."""
-    from sqlalchemy import select
-
     from scribe.models import async_session
     from scribe.models.project import Project
-    from scribe.models.user import User
     from scribe.services import snippets as svc
     from scribe.services.repo_bindings import set_binding
 
     async with async_session() as s:
-        user = (
-            await s.execute(select(User).where(User.username == "webhook_itest"))
-        ).scalar_one_or_none()
-        if user is None:
-            user = User(username="webhook_itest")
-            s.add(user)
-            await s.flush()
+        user = await ensure_user(s, "webhook_itest")
         project = Project(user_id=user.id, title="Widget")
         s.add(project)
         await s.flush()

tests/test_integration_db_maintenance.py

@@ -7,29 +7,14 @@ connection path that unit mocks cannot: the un-awaited
 AttributeError, reporting 0/6) passes the unit suite but fails here.
 """
 import pytest
-import pytest_asyncio
 
-from scribe.models import engine
 from scribe.services.db_maintenance import (
     MAINTENANCE_TABLES,
     get_table_health,
     run_maintenance,
 )
 
-pytestmark = pytest.mark.integration
-
-
-@pytest_asyncio.fixture(autouse=True)
-async def _dispose_engine():
-    """Dispose the app's module-level engine after each test.
-
-    The engine pools asyncpg connections per event loop, but pytest-asyncio runs
-    each test on a fresh loop — so without this, test 2 gets handed test 1's
-    connection bound to a now-dead loop ("Future attached to a different loop").
-    Disposing in the test's own loop teardown clears the pool cleanly.
-    """
-    yield
-    await engine.dispose()
+pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
 
 
 @pytest.mark.asyncio

tests/test_integration_forge_keyring.py

@@ -13,12 +13,10 @@ from unittest.mock import patch
 
 import pytest
 import pytest_asyncio
-from sqlalchemy import select
 
 from scribe.config import Config
-from scribe.models import async_session, engine
+from scribe.models import async_session
 from scribe.models.project import Project
-from scribe.models.user import User
 from scribe.services.forge import get_forges
 from scribe.services.forge_connections import (
     create_connection,
@@ -27,39 +25,21 @@ from scribe.services.forge_connections import (
     set_project_pin,
     update_connection,
 )
+from tests.helpers import ensure_user
 
-pytestmark = pytest.mark.integration
+pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
 
 GITEA = "https://git.example.com"
 GITHUB = "https://github.com"
 
 
-@pytest_asyncio.fixture(autouse=True)
-async def _dispose_engine():
-    # Per-loop pool: dispose after each test (see test_integration_db_maintenance).
-    yield
-    await engine.dispose()
-
-
-async def _user(session, username: str, role: str = "user") -> User:
-    existing = (
-        await session.execute(select(User).where(User.username == username))
-    ).scalar_one_or_none()
-    if existing is not None:
-        return existing
-    user = User(username=username, role=role)
-    session.add(user)
-    await session.flush()
-    return user
-
-
 @pytest_asyncio.fixture
 async def seeded():
     """An owner with a project, plus an unrelated user and an admin."""
     async with async_session() as s:
-        owner = await _user(s, "keyring_owner")
-        other = await _user(s, "keyring_other")
-        admin = await _user(s, "keyring_admin", role="admin")
+        owner = await ensure_user(s, "keyring_owner")
+        other = await ensure_user(s, "keyring_other")
+        admin = await ensure_user(s, "keyring_admin", role="admin")
         project = Project(user_id=owner.id, title="Keyring project")
         s.add(project)
         await s.flush()

tests/test_integration_pgvector_search.py

@@ -16,13 +16,13 @@ import pytest
 import pytest_asyncio
 from sqlalchemy import delete
 
-from scribe.models import async_session, engine
+from scribe.models import async_session
 from scribe.models.embedding import EMBEDDING_DIM, NoteEmbedding
 from scribe.models.note import Note
 from scribe.models.user import User
 from scribe.services.embeddings import semantic_search_notes
 
-pytestmark = pytest.mark.integration
+pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
 
 
 def _vec(*nonzero_first):
@@ -45,13 +45,6 @@ def _emb(note_id, user_id, chunk_index, vec):
     )
 
 
-@pytest_asyncio.fixture(autouse=True)
-async def _dispose_engine():
-    # Per-loop pool: dispose after each test (see test_integration_db_maintenance).
-    yield
-    await engine.dispose()
-
-
 @pytest_asyncio.fixture
 async def seeded():
     """Insert a user + a near and a far note with hand-crafted embeddings.

tests/test_integration_shape_classify.py

@@ -9,7 +9,7 @@ import pytest
 import pytest_asyncio
 from sqlalchemy import select
 
-from scribe.models import async_session, engine
+from scribe.models import async_session
 from scribe.models.code_shape import CodeShape
 from scribe.models.project import Project
 from scribe.models.user import User
@@ -19,8 +19,9 @@ from scribe.services.shape_ledger import (
     snippet_consumers,
     sync_repo_shapes,
 )
+from tests.helpers import ensure_user
 
-pytestmark = pytest.mark.integration
+pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
 
 REPO = "git.example.com/alice/widget"
 SHAPES = [
@@ -31,39 +32,14 @@ SHAPES = [
 ]
 
 
-@pytest_asyncio.fixture(autouse=True)
-async def _dispose_engine():
-    """Dispose the app's module-level engine after each test.
-
-    The engine pools asyncpg connections per event loop, but pytest-asyncio runs
-    each test on a fresh loop — so without this, test 2 gets handed test 1's
-    connection bound to a now-dead loop ("Future attached to a different loop").
-    Disposing in the test's own loop teardown clears the pool cleanly.
-    """
-    yield
-    await engine.dispose()
-
-
-async def _user(session, username: str) -> User:
-    existing = (
-        await session.execute(select(User).where(User.username == username))
-    ).scalar_one_or_none()
-    if existing is not None:
-        return existing
-    user = User(username=username)
-    session.add(user)
-    await session.flush()
-    return user
-
-
 @pytest_asyncio.fixture
 async def seeded():
     """Owner + outsider, a project with a synced 4-shape ledger, one snippet."""
     from scribe.services import snippets as snippets_svc
 
     async with async_session() as s:
-        owner = await _user(s, "classify_owner")
-        other = await _user(s, "classify_other")
+        owner = await ensure_user(s, "classify_owner")
+        other = await ensure_user(s, "classify_other")
         project = Project(user_id=owner.id, title="Classify target")
         s.add(project)
         await s.flush()

tests/test_integration_snippet_locations.py

@@ -19,7 +19,7 @@ import pytest
 import pytest_asyncio
 from sqlalchemy import delete, func, select
 
-from scribe.models import async_session, engine
+from scribe.models import async_session
 from scribe.models.note import Note
 from scribe.models.user import User
 from scribe.services.knowledge import location_matches, location_parts
@@ -32,14 +32,7 @@ from scribe.services.snippets import (
     snippet_fields,
 )
 
-pytestmark = pytest.mark.integration
-
-
-@pytest_asyncio.fixture(autouse=True)
-async def _dispose_engine():
-    # Per-loop pool: dispose after each test (see test_integration_db_maintenance).
-    yield
-    await engine.dispose()
+pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
 
 
 def _loc(repo="", path="", symbol=""):

tests/test_mcp_tool_design_systems.py

@@ -9,15 +9,10 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
-from scribe.mcp._context import _user_id_ctx
 from scribe.services.design_systems import DesignSystemCycle
 
 
-@pytest.fixture(autouse=True)
-def _bind_user():
-    token = _user_id_ctx.set(7)
-    yield
-    _user_id_ctx.reset(token)
+pytestmark = pytest.mark.usefixtures("_bind_user")
 
 
 def _fake_system():

tests/test_mcp_tool_milestones.py

@@ -3,17 +3,12 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
-from scribe.mcp._context import _user_id_ctx
 from scribe.mcp.tools.milestones import (
     list_milestones, get_milestone, create_milestone, update_milestone,
 )
 
 
-@pytest.fixture(autouse=True)
-def _bind_user():
-    token = _user_id_ctx.set(7)
-    yield
-    _user_id_ctx.reset(token)
+pytestmark = pytest.mark.usefixtures("_bind_user")
 
 
 def _fake_ms(**overrides) -> MagicMock:

tests/test_mcp_tool_notes.py

@@ -1,20 +1,16 @@
 """Tests for fable_*_note tools."""
-from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import AsyncMock, patch
 
 import pytest
 
-from scribe.mcp._context import _user_id_ctx
 from scribe.mcp.tools.notes import (
     list_notes, get_note, create_note,
     update_note, delete_note,
 )
+from tests.helpers import fake_note
 
 
-@pytest.fixture(autouse=True)
-def _bind_user():
-    token = _user_id_ctx.set(7)
-    yield
-    _user_id_ctx.reset(token)
+pytestmark = pytest.mark.usefixtures("_bind_user")
 
 
 @pytest.fixture(autouse=True)
@@ -32,21 +28,6 @@ def _no_supersession():
         yield
 
 
-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
-
-
 @pytest.mark.asyncio
 async def test_create_note_blocked_by_duplicate_gate():
     from scribe.services.dedup import DuplicateMatch
@@ -67,7 +48,7 @@ async def test_create_note_force_bypasses_duplicate_gate():
     find_mock = AsyncMock()
     with patch("scribe.mcp.tools.notes.dedup_svc.find_duplicate_note", find_mock), \
          patch("scribe.mcp.tools.notes.notes_svc.create_note",
-               AsyncMock(return_value=_fake_note(id=3))):
+               AsyncMock(return_value=fake_note(id=3))):
         out = await create_note(title="dup", force=True)
     assert out["id"] == 3
     find_mock.assert_not_called()
@@ -75,7 +56,7 @@ async def test_create_note_force_bypasses_duplicate_gate():
 
 @pytest.mark.asyncio
 async def test_list_notes_repackages_tuple_into_dict():
-    rows = [_fake_note(id=1), _fake_note(id=2)]
+    rows = [fake_note(id=1), fake_note(id=2)]
     with patch(
         "scribe.mcp.tools.notes.notes_svc.list_notes",
         AsyncMock(return_value=(rows, 2)),
@@ -128,7 +109,7 @@ async def test_list_notes_limit_clamped():
 
 @pytest.mark.asyncio
 async def test_get_note_returns_dict():
-    fake = _fake_note(id=5, title="found")
+    fake = fake_note(id=5, title="found")
     with patch(
         "scribe.mcp.tools.notes.notes_svc.get_note_for_user",
         AsyncMock(return_value=(fake, "owner")),
@@ -153,7 +134,7 @@ async def test_get_note_warns_in_words_when_a_later_note_overtook_it():
     field to notice would be worse than not surfacing it, because the reader
     acts on it confidently either way.
     """
-    fake = _fake_note(id=5, title="June's answer")
+    fake = fake_note(id=5, title="June's answer")
     with patch("scribe.mcp.tools.notes.notes_svc.get_note_for_user",
                AsyncMock(return_value=(fake, "owner"))), \
          patch("scribe.mcp.tools.notes.supersession_svc.get_relations",
@@ -170,7 +151,7 @@ async def test_get_note_opens_a_shared_record_and_says_whose_it_is():
     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)
+    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")),
@@ -199,7 +180,7 @@ async def test_get_note_raises_when_not_found():
 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 = fake_note(id=5)
     trashed.deleted_at = "2026-07-01T00:00:00Z"
     with patch(
         "scribe.mcp.tools.notes.notes_svc.get_note_for_user",
@@ -211,7 +192,7 @@ async def test_get_note_treats_a_trashed_note_as_missing():
 
 @pytest.mark.asyncio
 async def test_create_note_passes_through():
-    fake = _fake_note(id=10, title="new")
+    fake = fake_note(id=10, title="new")
     mock = AsyncMock(return_value=fake)
     with patch("scribe.mcp.tools.notes.notes_svc.create_note", mock):
         out = await create_note(title="new", body="x", tags=["a"], project_id=5)
@@ -223,7 +204,7 @@ async def test_create_note_passes_through():
 @pytest.mark.asyncio
 async def test_create_note_project_zero_becomes_none():
     """project_id=0 sentinel must become None at the service layer (orphan note)."""
-    fake = _fake_note()
+    fake = fake_note()
     mock = AsyncMock(return_value=fake)
     with patch("scribe.mcp.tools.notes.notes_svc.create_note", mock):
         await create_note(title="t", project_id=0)
@@ -234,7 +215,7 @@ async def test_create_note_project_zero_becomes_none():
 async def test_update_note_only_sends_non_default_fields():
     """Omitted (default) fields must NOT reach the service — otherwise they'd
     overwrite real data with empty strings."""
-    fake = _fake_note()
+    fake = fake_note()
     mock = AsyncMock(return_value=fake)
     with patch("scribe.mcp.tools.notes.notes_svc.update_note", mock):
         await update_note(note_id=1, title="new title")
@@ -247,7 +228,7 @@ async def test_update_note_only_sends_non_default_fields():
 @pytest.mark.asyncio
 async def test_update_note_empty_tags_clears_explicitly():
     """tags=[] is an explicit clear, distinct from tags=None (omit)."""
-    fake = _fake_note()
+    fake = fake_note()
     mock = AsyncMock(return_value=fake)
     with patch("scribe.mcp.tools.notes.notes_svc.update_note", mock):
         await update_note(note_id=1, tags=[])
@@ -256,7 +237,7 @@ async def test_update_note_empty_tags_clears_explicitly():
 
 @pytest.mark.asyncio
 async def test_update_note_tags_none_means_omit():
-    fake = _fake_note()
+    fake = fake_note()
     mock = AsyncMock(return_value=fake)
     with patch("scribe.mcp.tools.notes.notes_svc.update_note", mock):
         await update_note(note_id=1, tags=None)

tests/test_mcp_tool_planning.py

@@ -2,14 +2,8 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
-from scribe.mcp._context import _user_id_ctx
 
-
-@pytest.fixture(autouse=True)
-def _bind_user():
-    token = _user_id_ctx.set(7)
-    yield
-    _user_id_ctx.reset(token)
+pytestmark = pytest.mark.usefixtures("_bind_user")
 
 
 @pytest.mark.asyncio

tests/test_mcp_tool_processes.py

@@ -2,27 +2,10 @@
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
-
-from scribe.mcp._context import _user_id_ctx
+from tests.helpers import fake_note
 
 
-@pytest.fixture(autouse=True)
-def _bind_user():
-    token = _user_id_ctx.set(7)
-    yield
-    _user_id_ctx.reset(token)
-
-
-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
+pytestmark = pytest.mark.usefixtures("_bind_user")
 
 
 @pytest.mark.asyncio
@@ -36,7 +19,7 @@ async def test_create_process_requires_title_and_body():
 
 @pytest.mark.asyncio
 async def test_create_process_sets_note_type():
-    created = _fake_note()
+    created = fake_note(title="Drift Audit", note_type="process")
     with patch("scribe.mcp.tools.processes.dedup_svc.find_duplicate_note",
                AsyncMock(return_value=None)), \
          patch("scribe.services.notes.create_note",
@@ -72,7 +55,7 @@ async def test_create_process_blocks_a_near_duplicate():
 
 @pytest.mark.asyncio
 async def test_create_process_force_bypasses_the_gate():
-    created = _fake_note()
+    created = fake_note(title="Drift Audit", note_type="process")
     with patch("scribe.mcp.tools.processes.dedup_svc.find_duplicate_note",
                AsyncMock()) as find_mock, \
          patch("scribe.services.notes.create_note",
@@ -84,7 +67,7 @@ async def test_create_process_force_bypasses_the_gate():
 
 @pytest.mark.asyncio
 async def test_get_process_returns_body_and_candidates():
-    note = _fake_note(id=7)
+    note = fake_note(id=7, title="Drift Audit", note_type="process")
     with patch("scribe.services.notes.resolve_process",
                AsyncMock(return_value=(note, [{"id": 9, "title": "Drift Audit Notes"}]))):
         from scribe.mcp.tools.processes import get_process
@@ -101,7 +84,7 @@ 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)
+    note = fake_note(id=7, title="Drift Audit", user_id=9, note_type="process")
     with patch("scribe.services.notes.resolve_process",
                AsyncMock(return_value=(note, []))), \
          patch("scribe.services.access.describe_provenance",
@@ -126,7 +109,7 @@ async def test_get_process_not_found_raises():
 async def test_update_process_rejects_non_process_note():
     # Resolves share-aware now, so the patch target is get_note_for_user, which
     # returns (note, permission).
-    plain = _fake_note(id=3, note_type="note")
+    plain = fake_note(id=3, title="Drift Audit", note_type="note")
     plain.deleted_at = None
     with patch("scribe.services.notes.get_note_for_user",
                AsyncMock(return_value=(plain, "owner"))):
@@ -140,7 +123,7 @@ async def test_update_process_refuses_a_read_only_share_with_the_reason():
     """An editor grant lets the holder edit someone else's process; a viewer
     grant must be refused, and saying "not found" about a process the caller can
     open would just send them looking for a missing id."""
-    theirs = _fake_note(id=5, user_id=9)
+    theirs = fake_note(id=5, title="Drift Audit", user_id=9, note_type="process")
     theirs.deleted_at = None
     with patch("scribe.services.notes.get_note_for_user",
                AsyncMock(return_value=(theirs, "viewer"))), \
@@ -154,7 +137,7 @@ async def test_update_process_refuses_a_read_only_share_with_the_reason():
 
 @pytest.mark.asyncio
 async def test_delete_process_trashes_it_recoverably():
-    proc = _fake_note(id=4)
+    proc = fake_note(id=4, title="Drift Audit", note_type="process")
     proc.deleted_at = None
     with patch("scribe.services.notes.get_note_for_user",
                AsyncMock(return_value=(proc, "owner"))), \
@@ -172,7 +155,7 @@ 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 = fake_note(id=3, title="Drift Audit", note_type="note")
     plain.deleted_at = None
     with patch("scribe.services.notes.get_note_for_user",
                AsyncMock(return_value=(plain, "owner"))), \

tests/test_mcp_tool_projects.py

@@ -3,18 +3,13 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
-from scribe.mcp._context import _user_id_ctx
 from scribe.mcp.tools.projects import (
     list_projects, get_project, create_project,
     update_project, enter_project,
 )
 
 
-@pytest.fixture(autouse=True)
-def _bind_user():
-    token = _user_id_ctx.set(7)
-    yield
-    _user_id_ctx.reset(token)
+pytestmark = pytest.mark.usefixtures("_bind_user")
 
 
 @pytest.fixture(autouse=True)

tests/test_mcp_tool_rulebooks.py

@@ -3,14 +3,8 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
-from scribe.mcp._context import _user_id_ctx
 
-
-@pytest.fixture(autouse=True)
-def _bind_user():
-    token = _user_id_ctx.set(7)
-    yield
-    _user_id_ctx.reset(token)
+pytestmark = pytest.mark.usefixtures("_bind_user")
 
 
 def _fake_rulebook(id=1, title="t"):

tests/test_mcp_tool_search.py

@@ -1,12 +1,13 @@
 """search tool — proves the tool pattern (context + service call + dict shape).
 
 Service call is mocked; no DB needed."""
-from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import AsyncMock, patch
 
 import pytest
 
 from scribe.mcp._context import _user_id_ctx
 from scribe.mcp.tools.search import search
+from tests.helpers import fake_note
 
 
 @pytest.fixture(autouse=True)
@@ -17,22 +18,6 @@ def _reset_user_ctx():
     _user_id_ctx.reset(token)
 
 
-def _fake_note(*, id: int, title: str, body: str = "",
-               tags: list[str] | None = None, is_task: bool = False,
-               user_id: int = 7) -> MagicMock:
-    note = MagicMock()
-    note.id = id
-    note.title = title
-    note.body = body
-    note.tags = tags or []
-    note.is_task = is_task
-    # A real int, matching the bound caller by default: results compare it to
-    # decide whether to attach a shared/owner marker, and an auto-MagicMock would
-    # read as "someone else's" and send the tool looking up a username.
-    note.user_id = user_id
-    return note
-
-
 @pytest.mark.asyncio
 async def test_fable_search_raises_without_context():
     with pytest.raises(RuntimeError, match="no MCP user context"):
@@ -42,8 +27,7 @@ async def test_fable_search_raises_without_context():
 @pytest.mark.asyncio
 async def test_fable_search_returns_repackaged_results():
     _user_id_ctx.set(7)
-    fake = _fake_note(id=1, title="kafka rebalance", body="HPA details",
-                      tags=["ops"], is_task=False)
+    fake = fake_note(id=1, title="kafka rebalance", is_task=False, body="HPA details", tags=["ops"])
     with patch(
         "scribe.mcp.tools.search.semantic_search_notes",
         AsyncMock(return_value=[(0.93, fake)]),
@@ -65,7 +49,7 @@ async def test_fable_search_returns_repackaged_results():
 async def test_fable_search_body_is_truncated_to_240_chars():
     _user_id_ctx.set(7)
     long_body = "x" * 500
-    fake = _fake_note(id=1, title="t", body=long_body)
+    fake = fake_note(id=1, title="t", body=long_body)
     with patch(
         "scribe.mcp.tools.search.semantic_search_notes",
         AsyncMock(return_value=[(0.5, fake)]),

tests/test_mcp_tool_snippets.py

@@ -3,14 +3,8 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
-from scribe.mcp._context import _user_id_ctx
 
-
-@pytest.fixture(autouse=True)
-def _bind_user():
-    token = _user_id_ctx.set(7)
-    yield
-    _user_id_ctx.reset(token)
+pytestmark = pytest.mark.usefixtures("_bind_user")
 
 
 def _fake_snippet(user_id: int = 7):

tests/test_mcp_tool_systems.py

@@ -2,6 +2,7 @@
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
+from tests.helpers import fake_note
 
 
 def _fake_system(sid=1, name="Reader", project_id=5):
@@ -88,12 +89,6 @@ async def test_untagged_hint_names_the_projects_systems():
     assert "create_system" in hint
 
 
-def _fake_note(title):
-    n = MagicMock()
-    n.title = title
-    return n
-
-
 @pytest.mark.asyncio
 async def test_untagged_hint_zero_systems_prompts_first_create_and_fails_open():
     from scribe.mcp.tools.systems import untagged_systems_hint
@@ -120,7 +115,7 @@ async def test_untagged_hint_escalates_in_a_mature_zero_systems_project():
     concrete deliverable, because that is the property separating the nudges
     that convert from the prose that doesn't."""
     from scribe.mcp.tools.systems import untagged_systems_hint
-    recent = [_fake_note("Fix scrape retry backoff"), _fake_note("Worker pool sizing")]
+    recent = [fake_note(title="Fix scrape retry backoff"), fake_note(title="Worker pool sizing")]
     with patch("scribe.mcp.tools.systems.systems_svc") as svc, \
          patch("scribe.mcp.tools.systems.notes_svc") as notes:
         svc.list_systems = AsyncMock(return_value=[])

tests/test_mcp_tool_tags.py

@@ -7,15 +7,10 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
-from scribe.mcp._context import _user_id_ctx
 from scribe.mcp.tools.tags import list_tags, _aggregate_tag_counts
 
 
-@pytest.fixture(autouse=True)
-def _bind_user():
-    token = _user_id_ctx.set(7)
-    yield
-    _user_id_ctx.reset(token)
+pytestmark = pytest.mark.usefixtures("_bind_user")
 
 
 def test_aggregate_tag_counts_basic():

tests/test_mcp_tool_tasks.py

@@ -4,18 +4,13 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
-from scribe.mcp._context import _user_id_ctx
 from scribe.mcp.tools.tasks import (
     list_tasks, get_task, create_task,
     update_task, add_task_log,
 )
 
 
-@pytest.fixture(autouse=True)
-def _bind_user():
-    token = _user_id_ctx.set(7)
-    yield
-    _user_id_ctx.reset(token)
+pytestmark = pytest.mark.usefixtures("_bind_user")
 
 
 def _fake_task(*, parent_id: int | None = None, user_id: int = 7,

tests/test_mcp_tool_tasks_kind.py

@@ -1,27 +1,16 @@
-from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import AsyncMock, patch
 
 import pytest
-
-from scribe.mcp._context import _user_id_ctx
+from tests.helpers import fake_note
 
 
-@pytest.fixture(autouse=True)
-def _bind_user():
-    token = _user_id_ctx.set(7)
-    yield
-    _user_id_ctx.reset(token)
-
-
-def _fake_note(task_kind="work"):
-    n = MagicMock()
-    n.to_dict.return_value = {"id": 1, "title": "T", "task_kind": task_kind}
-    return n
+pytestmark = pytest.mark.usefixtures("_bind_user")
 
 
 @pytest.mark.asyncio
 async def test_create_task_passes_kind():
     # kind=plan is retired (plans are milestones); 'issue' exercises passthrough.
-    mock = AsyncMock(return_value=_fake_note(task_kind="issue"))
+    mock = AsyncMock(return_value=fake_note(task_kind="issue"))
     with patch("scribe.mcp.tools.tasks.notes_svc.create_note", mock):
         from scribe.mcp.tools.tasks import create_task
         await create_task(title="P", kind="issue")

tests/test_mcp_tool_trash.py

@@ -3,14 +3,8 @@ from unittest.mock import AsyncMock, patch
 
 import pytest
 
-from scribe.mcp._context import _user_id_ctx
 
-
-@pytest.fixture(autouse=True)
-def _bind_user():
-    token = _user_id_ctx.set(7)
-    yield
-    _user_id_ctx.reset(token)
+pytestmark = pytest.mark.usefixtures("_bind_user")
 
 
 @pytest.mark.asyncio

tests/test_note_usage.py

@@ -8,22 +8,9 @@ before this it surfaced snippets while leaving no trace anywhere.
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
-import pytest_asyncio
 
 
-@pytest.fixture(autouse=True)
-def _no_supersession():
-    """The auto-inject menu now asks which of its lines are superseded (#278).
-
-    That is a real database call on a path these tests exercise without one.
-    Stubbed to "nothing superseded" — the ordinary state — rather than hidden
-    behind a try/except in the product, which would make the code lie about
-    what it does. The label's own behaviour is covered in
-    tests/test_supersession_ranking.py.
-    """
-    with patch("scribe.services.plugin_context.superseded_ids",
-               AsyncMock(return_value=set())):
-        yield
+pytestmark = pytest.mark.usefixtures("_no_supersession")
 
 
 from scribe.services import note_usage
@@ -33,19 +20,7 @@ from scribe.services.note_usage import (
     record_surfaced,
     usage_for_notes,
 )
-
-
-def _note(nid, title, user_id=1):
-    n = MagicMock()
-    n.id, n.title, n.user_id = nid, title, user_id
-    n.note_type = "snippet"
-    # See the same note in test_write_path_trigger: an auto-created mock on
-    # `.data` is truthy and would leak into a rendered menu line (#2244).
-    n.data = None
-    # And on `.is_task`, which the kind marker reads FIRST — truthy there makes
-    # every one of these snippets read as a task (#2246).
-    n.is_task, n.task_kind = False, "work"
-    return n
+from tests.helpers import fake_note
 
 
 # --- recording ------------------------------------------------------------
@@ -207,7 +182,7 @@ async def test_auto_inject_records_what_survived_the_margin_gate():
     two numbers must not silently mean different things per surface."""
     from scribe.services import plugin_context
 
-    hits = [(0.90, _note(1, "kept")), (0.40, _note(2, "cut by the margin gate"))]
+    hits = [(0.90, fake_note(id=1, title="kept", user_id=1, note_type="snippet")), (0.40, fake_note(id=2, title="cut by the margin gate", user_id=1, note_type="snippet"))]
     with (
         patch.object(
             plugin_context,
@@ -271,13 +246,6 @@ def test_every_getter_that_can_be_surfaced_also_records_a_pull():
 # and split the chain so a failure names its half.
 
 
-@pytest_asyncio.fixture
-async def _dispose_engine():
-    from scribe.models import engine
-    yield
-    await engine.dispose()
-
-
 async def _purge(note_id: int) -> None:
     from sqlalchemy import delete
 

tests/test_pattern_coverage.py

@@ -23,6 +23,7 @@ from scribe.services.coverage import (
     shapes_from_archive,
 )
 from scribe.services.shape_ledger import location_covers
+from tests.helpers import ensure_user
 
 # --- unit: the definition extractor (shared vectors with the hook) -----------
 
@@ -204,32 +205,16 @@ def _selector(tar_bytes: bytes):
     return ForgeSelector((_forge(tar_bytes),))
 
 
-@pytest_asyncio.fixture
-async def _dispose_engine():
-    from scribe.models import engine
-    yield
-    await engine.dispose()
-
-
 @pytest_asyncio.fixture
 async def seeded(_dispose_engine):
     """User + project + binding + two snippets that cover 2 of TREE's 4 shapes."""
-    from sqlalchemy import select
-
     from scribe.models import async_session
     from scribe.models.project import Project
-    from scribe.models.user import User
     from scribe.services import snippets as svc
     from scribe.services.repo_bindings import set_binding
 
     async with async_session() as s:
-        user = (
-            await s.execute(select(User).where(User.username == "coverage_itest"))
-        ).scalar_one_or_none()
-        if user is None:
-            user = User(username="coverage_itest")
-            s.add(user)
-            await s.flush()
+        user = await ensure_user(s, "coverage_itest")
         project = Project(user_id=user.id, title="Widget")
         s.add(project)
         await s.flush()
@@ -396,9 +381,7 @@ async def test_explicit_refresh_names_its_failures(seeded):
     an agent mid-task must learn WHY nothing measured — 'None' is exactly the
     stranding the button-only path caused."""
     from scribe.models import async_session
-    from scribe.models.user import User
     from scribe.services.coverage import refresh_for_caller
-    from sqlalchemy import select
 
     uid, pid = seeded["uid"], seeded["pid"]
     # The owner has no forge connection rows → the error names the fix.
@@ -408,13 +391,7 @@ async def test_explicit_refresh_names_its_failures(seeded):
 
     # A stranger gets not-found/no-write, never a measurement.
     async with async_session() as s:
-        other = (await s.execute(
-            select(User).where(User.username == "coverage_outsider")
-        )).scalar_one_or_none()
-        if other is None:
-            other = User(username="coverage_outsider")
-            s.add(other)
-            await s.flush()
+        other = await ensure_user(s, "coverage_outsider")
         other_id = other.id
         await s.commit()
     with pytest.raises(ValueError) as err:

tests/test_retrieval_scopes.py

@@ -15,38 +15,10 @@ shared record would be findable by wording and invisible by meaning.
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
+from tests.helpers import fake_note
 
 
-@pytest.fixture(autouse=True)
-def _no_supersession():
-    """The auto-inject menu now asks which of its lines are superseded (#278).
-
-    That is a real database call on a path these tests exercise without one.
-    Stubbed to "nothing superseded" — the ordinary state — rather than hidden
-    behind a try/except in the product, which would make the code lie about
-    what it does. The label's own behaviour is covered in
-    tests/test_supersession_ranking.py.
-    """
-    with patch("scribe.services.plugin_context.superseded_ids",
-               AsyncMock(return_value=set())):
-        yield
-
-
-
-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 = []
-    # 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
+pytestmark = pytest.mark.usefixtures("_no_supersession")
 
 
 @pytest.fixture
@@ -148,7 +120,7 @@ async def test_injected_menu_attributes_another_users_note():
     menu line is the only provenance the agent sees, so it has to name them."""
     from scribe.services import plugin_context
 
-    mine, theirs = _note(1, user_id=7, title="Mine"), _note(2, user_id=9, title="Theirs")
+    mine, theirs = fake_note(id=1, title="Mine", user_id=7), fake_note(id=2, title="Theirs", user_id=9)
     with patch.object(plugin_context, "semantic_search_notes",
                       AsyncMock(return_value=[(0.9, theirs), (0.88, mine)])), \
          patch.object(plugin_context, "get_autoinject_config",
@@ -176,11 +148,11 @@ async def test_injected_menu_labels_the_record_kind():
     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")),
+        (0.92, fake_note(id=1, title="debounce — rate-limit a callback", note_type="snippet")),
+        (0.91, fake_note(id=2, title="Release checklist", note_type="process")),
+        (0.90, fake_note(id=3, title="Auth token expiry", is_task=True, task_kind="issue")),
+        (0.89, fake_note(id=4, title="Ship the drafter", is_task=True)),
+        (0.88, fake_note(id=5, title="Why we dropped CalDAV")),
     ]
     with patch.object(plugin_context, "semantic_search_notes",
                       AsyncMock(return_value=hits)), \

tests/test_services_dedup.py

@@ -9,6 +9,7 @@ from scribe.services.dedup import (
     find_duplicate_note,
     find_duplicate_rule,
 )
+from tests.helpers import fake_note
 
 
 def _session_returning(note):
@@ -22,15 +23,9 @@ def _session_returning(note):
     return s
 
 
-def _fake_note(id=1, title="T", note_type="note"):
-    n = MagicMock()
-    n.id, n.title, n.note_type = id, title, note_type
-    return n
-
-
 @pytest.mark.asyncio
 async def test_title_exact_match_returns_title_duplicate():
-    note = _fake_note(id=10, title="Setup CI")
+    note = fake_note(id=10, title="Setup CI")
     with patch("scribe.services.dedup.async_session",
                return_value=_session_returning(note)):
         # whitespace/case differences are normalized away
@@ -54,7 +49,7 @@ async def test_short_body_skips_semantic_check():
 
 @pytest.mark.asyncio
 async def test_semantic_match_when_body_substantial():
-    hit = _fake_note(id=20, title="Existing", note_type="note")
+    hit = fake_note(id=20, title="Existing", note_type="note")
     sem = AsyncMock(return_value=[(0.93, hit)])
     with patch("scribe.services.dedup.async_session",
                return_value=_session_returning(None)), \
@@ -82,7 +77,7 @@ async def test_gate_catches_a_duplicate_hiding_in_a_later_chunk():
     n_chunks = len(chunk_document("Title", body))
     assert n_chunks > 1, "test body must actually chunk"
 
-    hit = _fake_note(id=30, title="The existing decision", note_type="note")
+    hit = fake_note(id=30, title="The existing decision", note_type="note")
     # Every chunk misses except the LAST one the gate will ask about.
     sem = AsyncMock(side_effect=[[] for _ in range(n_chunks - 1)] + [[(0.94, hit)]])
     with patch("scribe.services.dedup.async_session",
@@ -97,7 +92,7 @@ async def test_gate_catches_a_duplicate_hiding_in_a_later_chunk():
 
 @pytest.mark.asyncio
 async def test_semantic_match_of_other_note_type_is_ignored():
-    other = _fake_note(id=21, title="X", note_type="process")
+    other = fake_note(id=21, title="X", note_type="process")
     sem = AsyncMock(return_value=[(0.97, other)])
     with patch("scribe.services.dedup.async_session",
                return_value=_session_returning(None)), \
@@ -108,7 +103,7 @@ async def test_semantic_match_of_other_note_type_is_ignored():
 
 @pytest.mark.asyncio
 async def test_rule_title_match_in_topic():
-    rule = _fake_note(id=47, title="Honor the multi-user sharing ACL")
+    rule = fake_note(id=47, title="Honor the multi-user sharing ACL")
     with patch("scribe.services.dedup.async_session",
                return_value=_session_returning(rule)):
         dup = await find_duplicate_rule(
@@ -171,8 +166,7 @@ def _session_sequence(results):
 async def test_same_location_is_a_duplicate_however_it_is_described():
     """The measured false NEGATIVE: identical code at an identical
     repo·path·symbol was created because the prose around it differed."""
-    existing = _fake_note(id=30, title=".btn-primary — a page's main action",
-                          note_type="snippet")
+    existing = fake_note(id=30, title=".btn-primary — a page's main action", note_type="snippet")
     sem = AsyncMock()
     with patch("scribe.services.dedup.async_session",
                return_value=_session_sequence([None, existing])), \
@@ -193,7 +187,7 @@ async def test_same_location_is_a_duplicate_however_it_is_described():
 
 @pytest.mark.asyncio
 async def test_identical_code_is_a_duplicate_at_a_different_location():
-    existing = _fake_note(id=31, title="group_pairs", note_type="snippet")
+    existing = fake_note(id=31, title="group_pairs", note_type="snippet")
     with patch("scribe.services.dedup.async_session",
                return_value=_session_sequence([None, None, existing])), \
          patch("scribe.services.dedup.embeddings_svc.semantic_search_notes",

tests/test_services_design_systems.py

@@ -8,16 +8,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 
 from scribe.services.design_systems import DesignSystemCycle
-
-
-def _make_mock_session():
-    s = AsyncMock()
-    s.__aenter__ = AsyncMock(return_value=s)
-    s.__aexit__ = AsyncMock(return_value=False)
-    s.add = MagicMock()
-    s.commit = AsyncMock()
-    s.refresh = AsyncMock()
-    return s
+from tests.helpers import make_mock_session
 
 
 # --- creating ---------------------------------------------------------------
@@ -38,7 +29,7 @@ async def test_create_with_a_parent_is_denied_without_write_on_that_parent():
 async def test_create_without_a_parent_never_consults_the_parent_acl():
     """A family system has no parent, and creating one must not be gated on a
     permission check for a system that does not exist."""
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     captured = {}
     mock_session.add = MagicMock(
         side_effect=lambda obj: captured.update(
@@ -63,7 +54,7 @@ async def test_reparenting_into_a_loop_raises_rather_than_returning_none():
     """None already means "not found, or not yours". A caller that conflated the
     two would report "no such design system" for what is really "that parent is
     one of its own descendants", so the cycle gets its own exception type."""
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     mock_session.get = AsyncMock(return_value=MagicMock(deleted_at=None, parent_id=None))
 
     with patch("scribe.services.design_systems.async_session") as mock_cls, \
@@ -85,7 +76,7 @@ async def test_clearing_the_parent_is_allowed_and_is_not_read_as_no_change():
     value rather than "leave alone", which is why it is handled apart from the
     others."""
     system = MagicMock(deleted_at=None, parent_id=5)
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     mock_session.get = AsyncMock(return_value=system)
 
     with patch("scribe.services.design_systems.async_session") as mock_cls, \
@@ -116,7 +107,7 @@ async def test_token_values_default_to_an_empty_map_not_json_null():
     """The column is NOT NULL so that absence has exactly ONE spelling. Passing
     None straight through would store JSON null and hand every reader back the
     second empty state the schema was shaped to remove."""
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     captured = {}
     mock_session.add = MagicMock(
         side_effect=lambda obj: captured.update(value_by_mode=obj.value_by_mode)
@@ -159,7 +150,7 @@ async def test_pointing_a_project_needs_only_READ_on_the_system():
     through another project is a legitimate choice here. Requiring write would
     make a shared family style unusable by the people it was shared with."""
     project = MagicMock(deleted_at=None, design_system_id=None)
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     mock_session.get = AsyncMock(return_value=project)
 
     with patch("scribe.services.design_systems.async_session") as mock_cls, \
@@ -179,7 +170,7 @@ async def test_clearing_a_projects_design_system_skips_the_system_acl():
     """Un-styling a project must not require permission on the system it is
     letting go of — including one that has since been deleted."""
     project = MagicMock(deleted_at=None, design_system_id=3)
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     mock_session.get = AsyncMock(return_value=project)
 
     with patch("scribe.services.design_systems.async_session") as mock_cls, \
@@ -219,7 +210,7 @@ async def test_resolve_scopes_the_hierarchy_to_the_systems_OWNER_not_the_caller(
     """
     owner, caller = 42, 7
     system = MagicMock(deleted_at=None, owner_user_id=owner)
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     mock_session.get = AsyncMock(return_value=system)
     mock_session.execute = AsyncMock(
         return_value=MagicMock(scalars=MagicMock(return_value=MagicMock(all=lambda: [])))
@@ -243,7 +234,7 @@ async def test_resolve_scopes_the_hierarchy_to_the_systems_OWNER_not_the_caller(
 @pytest.mark.asyncio
 async def test_create_token_supersedes_defaults_to_an_empty_list_not_json_null():
     """Same NOT NULL reasoning as value_by_mode: absence gets one spelling."""
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     captured = {}
     mock_session.add = MagicMock(
         side_effect=lambda obj: captured.update(supersedes=obj.supersedes)
@@ -259,7 +250,7 @@ async def test_create_token_supersedes_defaults_to_an_empty_list_not_json_null()
 
 @pytest.mark.asyncio
 async def test_create_token_records_the_literals_it_replaces():
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     captured = {}
     mock_session.add = MagicMock(
         side_effect=lambda obj: captured.update(supersedes=obj.supersedes)
@@ -348,7 +339,7 @@ async def test_design_context_merges_guidance_ANCESTOR_FIRST():
                     description="one app", guidance="Accent on the wordmark.")
     app.title = "App"
 
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     mock_session.get = AsyncMock(return_value=app)
     mock_session.execute = AsyncMock(return_value=MagicMock(
         scalars=MagicMock(return_value=MagicMock(all=lambda: [app, family]))
@@ -395,7 +386,7 @@ async def test_design_context_omits_systems_with_no_guidance():
                     description="", guidance="   ")
     app.title = "App"
 
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     mock_session.get = AsyncMock(return_value=app)
     mock_session.execute = AsyncMock(return_value=MagicMock(
         scalars=MagicMock(return_value=MagicMock(all=lambda: [app, family]))

tests/test_services_knowledge_counts.py

@@ -2,13 +2,7 @@
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
-
-
-def _make_mock_session():
-    s = AsyncMock()
-    s.__aenter__ = AsyncMock(return_value=s)
-    s.__aexit__ = AsyncMock(return_value=False)
-    return s
+from tests.helpers import make_mock_session
 
 
 def _grouped(rows):
@@ -25,7 +19,7 @@ def _scalar(n):
 
 @pytest.mark.asyncio
 async def test_counts_include_process_in_facet_and_total():
-    session = _make_mock_session()
+    session = make_mock_session()
     # 1) grouped non-task counts, 2) task count, 3) plan count
     session.execute = AsyncMock(side_effect=[
         _grouped([("note", 3), ("process", 2)]),

tests/test_services_notes_process.py

@@ -5,13 +5,7 @@ Mocks async_session — no real DB, matching the other notes-service tests.
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
-
-
-def _make_mock_session():
-    s = AsyncMock()
-    s.__aenter__ = AsyncMock(return_value=s)
-    s.__aexit__ = AsyncMock(return_value=False)
-    return s
+from tests.helpers import fake_note, make_mock_session
 
 
 def _result(first=None, all_=None):
@@ -22,17 +16,10 @@ def _result(first=None, all_=None):
     return r
 
 
-def _note(id, title):
-    n = MagicMock()
-    n.id = id
-    n.title = title
-    return n
-
-
 @pytest.mark.asyncio
 async def test_resolve_process_by_numeric_id():
-    note = _note(5, "Drift Audit")
-    session = _make_mock_session()
+    note = fake_note(id=5, title="Drift Audit")
+    session = make_mock_session()
     # numeric id → first execute (id lookup) hits
     session.execute = AsyncMock(side_effect=[_result(first=note)])
     with patch("scribe.services.notes.async_session") as cls:
@@ -46,8 +33,8 @@ async def test_resolve_process_by_numeric_id():
 
 @pytest.mark.asyncio
 async def test_resolve_process_exact_title_beats_substring():
-    note = _note(7, "Drift Audit")
-    session = _make_mock_session()
+    note = fake_note(id=7, title="Drift Audit")
+    session = make_mock_session()
     # non-digit → exact-title query (first execute) hits; substring never runs
     session.execute = AsyncMock(side_effect=[_result(first=note)])
     with patch("scribe.services.notes.async_session") as cls:
@@ -61,9 +48,9 @@ async def test_resolve_process_exact_title_beats_substring():
 
 @pytest.mark.asyncio
 async def test_resolve_process_substring_returns_candidates():
-    n1 = _note(7, "Drift Audit Remediation")
-    n2 = _note(9, "Drift Audit Notes")
-    session = _make_mock_session()
+    n1 = fake_note(id=7, title="Drift Audit Remediation")
+    n2 = fake_note(id=9, title="Drift Audit Notes")
+    session = make_mock_session()
     # exact miss, then substring returns two (most-recent first)
     session.execute = AsyncMock(side_effect=[_result(first=None), _result(all_=[n1, n2])])
     with patch("scribe.services.notes.async_session") as cls:
@@ -77,7 +64,7 @@ async def test_resolve_process_substring_returns_candidates():
 
 @pytest.mark.asyncio
 async def test_resolve_process_no_match():
-    session = _make_mock_session()
+    session = make_mock_session()
     session.execute = AsyncMock(side_effect=[_result(first=None), _result(all_=[])])
     with patch("scribe.services.notes.async_session") as cls:
         cls.return_value = session

tests/test_services_notes_task_kind.py

@@ -2,21 +2,12 @@
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
-
-
-def _make_mock_session():
-    s = AsyncMock()
-    s.__aenter__ = AsyncMock(return_value=s)
-    s.__aexit__ = AsyncMock(return_value=False)
-    s.add = MagicMock()
-    s.commit = AsyncMock()
-    s.refresh = AsyncMock()
-    return s
+from tests.helpers import make_mock_session
 
 
 @pytest.mark.asyncio
 async def test_create_note_passes_task_kind_to_model():
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     captured = {}
 
     def _capture_add(obj):
@@ -33,7 +24,7 @@ async def test_create_note_passes_task_kind_to_model():
 
 @pytest.mark.asyncio
 async def test_list_notes_filters_by_task_kind():
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     mock_session.scalar = AsyncMock(return_value=0)
     exec_result = MagicMock()
     exec_result.scalars.return_value.all.return_value = []

tests/test_services_plugin_context.py

@@ -1,22 +1,10 @@
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
+from tests.helpers import fake_note
 
 
-@pytest.fixture(autouse=True)
-def _no_supersession():
-    """The auto-inject menu now asks which of its lines are superseded (#278).
-
-    That is a real database call on a path these tests exercise without one.
-    Stubbed to "nothing superseded" — the ordinary state — rather than hidden
-    behind a try/except in the product, which would make the code lie about
-    what it does. The label's own behaviour is covered in
-    tests/test_supersession_ranking.py.
-    """
-    with patch("scribe.services.plugin_context.superseded_ids",
-               AsyncMock(return_value=set())):
-        yield
-
+pytestmark = pytest.mark.usefixtures("_no_supersession")
 
 
 def _rule(rid, title, topic_id):
@@ -26,21 +14,6 @@ def _rule(rid, title, topic_id):
     return r
 
 
-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 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
-    # The write-path menu reads note.data for a language tag; an auto-mock there
-    # is truthy and renders its repr into the marker.
-    n.data = None
-    return n
-
-
 # ─── knowledge auto-inject (Path A) ──────────────────────────────────────────
 
 
@@ -88,9 +61,9 @@ async def test_build_autoinject_hint_disabled_returns_empty_and_skips_search():
 async def test_build_autoinject_hint_titles_only_with_margin_gate():
     from scribe.services import plugin_context as pc
     # top=0.80; 0.74 within band (0.10), 0.61 outside → dropped.
-    hits = [(0.80, _note(11, "Pool sizing decision")),
-            (0.74, _note(22, "run_maintenance thresholds")),
-            (0.61, _note(33, "unrelated-ish"))]
+    hits = [(0.80, fake_note(id=11, title="Pool sizing decision", user_id=1)),
+            (0.74, fake_note(id=22, title="run_maintenance thresholds", user_id=1)),
+            (0.61, fake_note(id=33, title="unrelated-ish", user_id=1))]
     rec = MagicMock()
     with patch.object(pc, "get_autoinject_config",
                       AsyncMock(return_value={"enabled": True, "threshold": 0.55, "top_k": 3})), \
@@ -352,10 +325,10 @@ async def test_a_snippet_takes_the_last_slot_when_none_won_on_score():
     records ABOUT building the retrieval system, and zero snippets. Scribe's own
     records are about software work, so they share vocabulary with any coding
     prompt while answering none of them."""
-    main = [(0.66, _note(1, "Step 3: title-first auto-inject")),
-            (0.65, _note(2, "Task-reminder dedup query crashes", is_task=True)),
-            (0.64, _note(3, "Drafter hardening · write-path trigger", is_task=True))]
-    reuse = [(0.58, _note(9, "debounce — collapse rapid calls", note_type="snippet"))]
+    main = [(0.66, fake_note(id=1, title="Step 3: title-first auto-inject", user_id=1)),
+            (0.65, fake_note(id=2, title="Task-reminder dedup query crashes", user_id=1, is_task=True)),
+            (0.64, fake_note(id=3, title="Drafter hardening · write-path trigger", user_id=1, is_task=True))]
+    reuse = [(0.58, fake_note(id=9, title="debounce — collapse rapid calls", user_id=1, note_type="snippet"))]
 
     out, calls = await _autoinject(main, reuse)
 
@@ -370,8 +343,8 @@ async def test_a_snippet_takes_the_last_slot_when_none_won_on_score():
 async def test_the_reserved_query_is_skipped_when_a_snippet_already_won():
     """No second query, and no slot spent twice, when ranking already did the
     right thing — the fix must be invisible in the case it isn't needed."""
-    main = [(0.81, _note(9, "debounce helper", note_type="snippet")),
-            (0.80, _note(1, "some task", is_task=True))]
+    main = [(0.81, fake_note(id=9, title="debounce helper", user_id=1, note_type="snippet")),
+            (0.80, fake_note(id=1, title="some task", user_id=1, is_task=True))]
 
     out, calls = await _autoinject(main, [])
 
@@ -384,7 +357,7 @@ async def test_a_weak_snippet_does_not_buy_the_slot():
     """The reserved hit skips the MARGIN band — that band is what snippets lose
     to — but never the threshold. Silence stays the default; a slot spent on an
     irrelevant snippet is how a menu teaches people to ignore it."""
-    main = [(0.66, _note(1, "a task", is_task=True))]
+    main = [(0.66, fake_note(id=1, title="a task", user_id=1, is_task=True))]
 
     out, calls = await _autoinject(main, [])     # threshold returned nothing
 
@@ -397,8 +370,8 @@ async def test_the_reserved_hit_is_not_held_to_the_margin_band():
     """0.58 is 0.08 below the top hit. Under the band it would survive; the point
     is that it must survive even when it wouldn't — a snippet losing to a
     same-vocabulary project record by a wide margin is the whole bug."""
-    main = [(0.90, _note(1, "a task", is_task=True))]
-    reuse = [(0.58, _note(9, "debounce", note_type="snippet"))]
+    main = [(0.90, fake_note(id=1, title="a task", user_id=1, is_task=True))]
+    reuse = [(0.58, fake_note(id=9, title="debounce", user_id=1, note_type="snippet"))]
 
     out, _calls = await _autoinject(main, reuse)
 
@@ -409,15 +382,15 @@ async def test_the_reserved_hit_is_not_held_to_the_margin_band():
 async def test_a_process_counts_as_reuse_too():
     """A stored process answers 'how do we do X here' the same way a snippet
     answers 'what do we already have' — both lose to the same project records."""
-    main = [(0.70, _note(1, "a task", is_task=True))]
-    reuse = [(0.60, _note(8, "DRY pass process", note_type="process"))]
+    main = [(0.70, fake_note(id=1, title="a task", user_id=1, is_task=True))]
+    reuse = [(0.60, fake_note(id=8, title="DRY pass process", user_id=1, note_type="process"))]
 
     out, _ = await _autoinject(main, reuse)
     assert 8 in out["note_ids"]
 
     # …and one already on the menu suppresses the reserved query.
     out2, calls2 = await _autoinject(
-        [(0.70, _note(8, "DRY pass process", note_type="process"))], [])
+        [(0.70, fake_note(id=8, title="DRY pass process", user_id=1, note_type="process"))], [])
     assert len(calls2) == 1
 
 
@@ -431,9 +404,8 @@ async def test_write_path_semantic_arm_asks_for_experience_not_just_snippets():
     it says what NOT to do."""
     from scribe.services import plugin_context as pc
 
-    hits = [(0.72, _note(9, "debounce helper", note_type="snippet")),
-            (0.70, _note(7, "Debounce dropped the trailing call", is_task=True,
-                         task_kind="issue"))]
+    hits = [(0.72, fake_note(id=9, title="debounce helper", user_id=1, note_type="snippet")),
+            (0.70, fake_note(id=7, title="Debounce dropped the trailing call", user_id=1, is_task=True, task_kind="issue"))]
     search = AsyncMock(return_value=hits)
     rec = MagicMock()
     with patch.object(pc, "get_writepath_config",
@@ -466,9 +438,8 @@ async def test_write_path_labels_a_non_snippet_hit_with_its_kind():
     menu's default and the header's default reading."""
     from scribe.services import plugin_context as pc
 
-    hits = [(0.72, _note(9, "debounce helper", note_type="snippet")),
-            (0.71, _note(7, "Debounce dropped the trailing call", is_task=True,
-                         task_kind="issue"))]
+    hits = [(0.72, fake_note(id=9, title="debounce helper", user_id=1, note_type="snippet")),
+            (0.71, fake_note(id=7, title="Debounce dropped the trailing call", user_id=1, is_task=True, task_kind="issue"))]
     with patch.object(pc, "get_writepath_config",
                       AsyncMock(return_value={"enabled": True, "threshold": 0.6,
                                               "top_k": 3})), \

tests/test_services_retrieval_telemetry.py

@@ -7,7 +7,6 @@ integration test against real Postgres.
 from types import SimpleNamespace
 
 import pytest
-import pytest_asyncio
 
 from scribe.services.retrieval_telemetry import (
     _build_payload,
@@ -72,13 +71,6 @@ def test_record_retrieval_without_event_loop_is_safe():
 # ─── persistence (integration) ───────────────────────────────────────────────
 
 
-@pytest_asyncio.fixture
-async def _dispose_engine():
-    from scribe.models import engine
-    yield
-    await engine.dispose()
-
-
 @pytest.mark.integration
 @pytest.mark.asyncio
 async def test_insert_retrieval_log_roundtrip(_dispose_engine):

tests/test_services_rulebooks.py

@@ -6,16 +6,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 from datetime import datetime, timezone
 
 import pytest
-
-
-def _make_mock_session():
-    s = AsyncMock()
-    s.__aenter__ = AsyncMock(return_value=s)
-    s.__aexit__ = AsyncMock(return_value=False)
-    s.add = MagicMock()
-    s.commit = AsyncMock()
-    s.refresh = AsyncMock()
-    return s
+from tests.helpers import make_mock_session
 
 
 def _fake_rulebook(id=1, owner_user_id=7, title="FabledSword family", description=""):
@@ -35,7 +26,7 @@ def _fake_rulebook(id=1, owner_user_id=7, title="FabledSword family", descriptio
 
 @pytest.mark.asyncio
 async def test_create_rulebook_stores_to_db():
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     with patch("scribe.services.rulebooks.async_session") as mock_cls:
         mock_cls.return_value = mock_session
         from scribe.services.rulebooks import create_rulebook
@@ -49,7 +40,7 @@ async def test_create_rulebook_stores_to_db():
 @pytest.mark.asyncio
 async def test_list_rulebooks_returns_owned_only():
     rb = _fake_rulebook(id=1)
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     mock_result = MagicMock()
     mock_result.scalars.return_value.all.return_value = [rb]
     mock_session.execute = AsyncMock(return_value=mock_result)
@@ -63,7 +54,7 @@ async def test_list_rulebooks_returns_owned_only():
 @pytest.mark.asyncio
 async def test_get_rulebook_returns_none_when_not_owner():
     """get_rulebook scopes by owner_user_id — wrong user gets None."""
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     mock_result = MagicMock()
     mock_result.scalar_one_or_none.return_value = None
     mock_session.execute = AsyncMock(return_value=mock_result)
@@ -77,7 +68,7 @@ async def test_get_rulebook_returns_none_when_not_owner():
 @pytest.mark.asyncio
 async def test_update_rulebook_only_sets_provided_fields():
     rb = _fake_rulebook(id=1, title="old")
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     mock_result = MagicMock()
     mock_result.scalar_one_or_none.return_value = rb
     mock_session.execute = AsyncMock(return_value=mock_result)
@@ -91,7 +82,7 @@ async def test_update_rulebook_only_sets_provided_fields():
 @pytest.mark.asyncio
 async def test_delete_rulebook_calls_delete():
     rb = _fake_rulebook(id=1)
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     mock_result = MagicMock()
     mock_result.scalar_one_or_none.return_value = rb
     mock_session.execute = AsyncMock(return_value=mock_result)
@@ -124,7 +115,7 @@ def _fake_topic(id=1, rulebook_id=1, title="git-workflow", description="", order
 @pytest.mark.asyncio
 async def test_create_topic_requires_owned_rulebook():
     """create_topic raises ValueError if the rulebook isn't owned by user."""
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     mock_result = MagicMock()
     mock_result.scalar_one_or_none.return_value = None
     mock_session.execute = AsyncMock(return_value=mock_result)
@@ -143,7 +134,7 @@ async def test_list_topics_returns_topics_for_owned_rulebook():
     topic = _fake_topic(id=10, rulebook_id=1, title="git-workflow")
 
     # Two execute calls: ownership check, then topic select.
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     rb_result = MagicMock()
     rb_result.scalar_one_or_none.return_value = rb
     topic_result = MagicMock()
@@ -182,7 +173,7 @@ def _fake_rule(id=1, topic_id=10, title="dev is home",
 
 @pytest.mark.asyncio
 async def test_create_rule_requires_owned_topic():
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     mock_result = MagicMock()
     mock_result.scalar_one_or_none.return_value = None  # topic not found
     mock_session.execute = AsyncMock(return_value=mock_result)
@@ -199,7 +190,7 @@ async def test_create_rule_requires_owned_topic():
 async def test_list_rules_filters_by_topic_id():
     """list_rules(topic_id=X) returns rules in that topic, ownership-scoped."""
     rule = _fake_rule(id=1, topic_id=10)
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     mock_result = MagicMock()
     mock_result.scalars.return_value.all.return_value = [rule]
     mock_session.execute = AsyncMock(return_value=mock_result)
@@ -212,7 +203,7 @@ async def test_list_rules_filters_by_topic_id():
 
 @pytest.mark.asyncio
 async def test_get_rule_returns_none_when_not_owner():
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     mock_result = MagicMock()
     mock_result.scalar_one_or_none.return_value = None
     mock_session.execute = AsyncMock(return_value=mock_result)
@@ -228,7 +219,7 @@ async def test_get_rule_returns_none_when_not_owner():
 @pytest.mark.asyncio
 async def test_subscribe_project_requires_owned_rulebook():
     """subscribe_project raises if user doesn't own the rulebook."""
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     mock_result = MagicMock()
     mock_result.scalar_one_or_none.return_value = None
     mock_session.execute = AsyncMock(return_value=mock_result)
@@ -253,7 +244,7 @@ def _empty():
 async def test_get_applicable_rules_returns_shape():
     """get_applicable_rules returns the full projection — including the
     new suppression fields and rulebook/topic IDs on each rule."""
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     sub_result = MagicMock()
     sub_result.all.return_value = [(1, "FabledSword family")]
     rules_result = MagicMock()
@@ -291,7 +282,7 @@ async def test_get_applicable_rules_returns_shape():
 @pytest.mark.asyncio
 async def test_get_applicable_rules_truncates_when_over_limit():
     """When limit+1 rows are returned, truncated=True and only `limit` returned."""
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     sub_result = MagicMock()
     sub_result.all.return_value = []
     rules_result = MagicMock()
@@ -314,7 +305,7 @@ async def test_get_applicable_rules_truncates_when_over_limit():
 @pytest.mark.asyncio
 async def test_get_applicable_rules_includes_project_scoped_rules():
     """Project-scoped rules surface in the project_rules field."""
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     proj_rules_result = MagicMock()
     proj_rules_result.all.return_value = [
         (100, "Use alembic", "Always run migrations via alembic, never raw SQL."),
@@ -338,7 +329,7 @@ async def test_get_applicable_rules_includes_project_scoped_rules():
 async def test_get_applicable_rules_surfaces_suppressed_with_context():
     """Suppressed rules and topics come back with full title + rulebook context
     so the UI can render them without an extra round-trip."""
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     suppressed_rules_result = MagicMock()
     suppressed_rules_result.all.return_value = [
         # (rule_id, title, topic_id, topic_title, rulebook_id, rulebook_title)

tests/test_services_systems.py

@@ -2,16 +2,7 @@
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
-
-
-def _make_mock_session():
-    s = AsyncMock()
-    s.__aenter__ = AsyncMock(return_value=s)
-    s.__aexit__ = AsyncMock(return_value=False)
-    s.add = MagicMock()
-    s.commit = AsyncMock()
-    s.refresh = AsyncMock()
-    return s
+from tests.helpers import make_mock_session
 
 
 @pytest.mark.asyncio
@@ -25,7 +16,7 @@ async def test_create_system_denied_without_project_write():
 
 @pytest.mark.asyncio
 async def test_create_system_sets_fields_when_authorized():
-    mock_session = _make_mock_session()
+    mock_session = make_mock_session()
     captured = {}
 
     def _capture_add(obj):

tests/test_services_trash.py

@@ -2,14 +2,7 @@
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
-
-
-def _make_mock_session():
-    s = AsyncMock()
-    s.__aenter__ = AsyncMock(return_value=s)
-    s.__aexit__ = AsyncMock(return_value=False)
-    s.commit = AsyncMock()
-    return s
+from tests.helpers import make_mock_session
 
 
 def _exists_result(found=True):
@@ -20,7 +13,7 @@ def _exists_result(found=True):
 
 @pytest.mark.asyncio
 async def test_delete_note_returns_batch_and_commits():
-    session = _make_mock_session()
+    session = make_mock_session()
     # exists-check, then the subtree descent: one child-lookup (no children
     # here) + one _set stamping the whole subtree.
     no_children = MagicMock()
@@ -38,7 +31,7 @@ async def test_delete_note_returns_batch_and_commits():
 
 @pytest.mark.asyncio
 async def test_delete_returns_none_when_not_found():
-    session = _make_mock_session()
+    session = make_mock_session()
     session.execute = AsyncMock(return_value=_exists_result(False))
     with patch("scribe.services.trash.async_session") as cls:
         cls.return_value = session
@@ -52,7 +45,7 @@ async def test_delete_returns_none_when_not_found():
 
 @pytest.mark.asyncio
 async def test_delete_project_cascades_to_notes_milestones_project_rules_and_suppressions():
-    session = _make_mock_session()
+    session = make_mock_session()
     # exists-check + 6 cascade ops:
     #   notes (soft) → milestones (soft) → project-scoped rules (soft) →
     #   project_rule_suppressions (hard DELETE) → project_topic_suppressions (hard DELETE) →
@@ -73,7 +66,7 @@ async def test_delete_project_cascades_to_notes_milestones_project_rules_and_sup
 
 @pytest.mark.asyncio
 async def test_delete_rulebook_cascades_topics_and_rules():
-    session = _make_mock_session()
+    session = make_mock_session()
     topic_ids_result = MagicMock()
     topic_ids_result.scalars.return_value.all.return_value = [10, 11]
     # exists-check, topic-id select, then 3 updates (rules, topics, rulebook)
@@ -96,7 +89,7 @@ def _rowcount_result(n):
 
 @pytest.mark.asyncio
 async def test_restore_clears_batch_across_all_models():
-    session = _make_mock_session()
+    session = make_mock_session()
     # 6 soft-deletable models, each returns a rowcount
     session.execute = AsyncMock(side_effect=[_rowcount_result(i) for i in [2, 0, 1, 1, 0, 0]])
     with patch("scribe.services.trash.async_session") as cls:
@@ -110,7 +103,7 @@ async def test_restore_clears_batch_across_all_models():
 
 @pytest.mark.asyncio
 async def test_purge_expired_skips_when_retention_zero():
-    session = _make_mock_session()
+    session = make_mock_session()
     session.execute = AsyncMock()
     with patch("scribe.services.trash.async_session") as cls:
         cls.return_value = session
@@ -122,7 +115,7 @@ async def test_purge_expired_skips_when_retention_zero():
 
 @pytest.mark.asyncio
 async def test_purge_expired_deletes_across_models_when_positive():
-    session = _make_mock_session()
+    session = make_mock_session()
     session.execute = AsyncMock(side_effect=[_rowcount_result(1) for _ in range(6)])
     with patch("scribe.services.trash.async_session") as cls:
         cls.return_value = session
@@ -157,7 +150,7 @@ def test_owner_clause_scopes_every_model():
 
 @pytest.mark.asyncio
 async def test_list_trash_groups_by_batch():
-    session = _make_mock_session()
+    session = make_mock_session()
 
     def _note(id, batch, title):
         from datetime import datetime, timezone

tests/test_snippet_provenance.py

@@ -24,6 +24,8 @@ a DB-touching path with only mocked coverage is a path with no coverage).
 import pytest
 import pytest_asyncio
 
+from tests.helpers import ensure_user
+
 from scribe.services.snippets import (
     VERIFY_CHANGED,
     VERIFY_OK,
@@ -79,34 +81,13 @@ def test_verification_records_and_reads_back_the_checked_commit():
 
 # --- integration: the rules through the real service paths -------------------
 
-@pytest_asyncio.fixture
-async def _dispose_engine():
-    from scribe.models import engine
-    yield
-    await engine.dispose()
-
 
 @pytest_asyncio.fixture
 async def user_id(_dispose_engine):
-    # Get-or-create: the lane's database persists across tests, so a second
-    # test re-creating the same username dies on the unique constraint.
-    from sqlalchemy import select
-
     from scribe.models import async_session
-    from scribe.models.user import User
 
     async with async_session() as s:
-        existing = (
-            await s.execute(
-                select(User).where(User.username == "snippet_prov_itest")
-            )
-        ).scalar_one_or_none()
-        if existing is not None:
-            return existing.id
-        user = User(username="snippet_prov_itest")
-        s.add(user)
-        await s.flush()
-        uid = user.id
+        uid = (await ensure_user(s, "snippet_prov_itest")).id
         await s.commit()
     return uid
 

tests/test_supersession_ranking.py

@@ -12,7 +12,7 @@ satisfied by dropping a record instead checks that it is still present.
 """
 from __future__ import annotations
 
-from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import AsyncMock, patch
 
 import pytest
 
@@ -20,12 +20,7 @@ from scribe.services.embeddings import (
     _SUPERSESSION_PENALTY,
     _apply_supersession_penalty,
 )
-
-
-def _note(note_id: int):
-    n = MagicMock()
-    n.id = note_id
-    return n
+from tests.helpers import fake_note
 
 
 def _stale(*ids):
@@ -37,7 +32,7 @@ def _stale(*ids):
 
 @pytest.mark.asyncio
 async def test_a_superseded_record_falls_behind_an_equal_live_one():
-    scored = [(0.70, _note(1)), (0.69, _note(2))]      # 1 leads on raw score
+    scored = [(0.70, fake_note(id=1)), (0.69, fake_note(id=2))]      # 1 leads on raw score
     with _stale(1):
         out = await _apply_supersession_penalty(scored, limit=5)
     assert [int(n.id) for _s, n in out] == [2, 1]
@@ -51,7 +46,7 @@ async def test_a_strong_superseded_record_still_beats_a_weak_live_one():
     answers a question nothing else answers should still surface, just behind
     anything comparable that is current.
     """
-    scored = [(0.90, _note(1)), (0.50, _note(2))]
+    scored = [(0.90, fake_note(id=1)), (0.50, fake_note(id=2))]
     with _stale(1):
         out = await _apply_supersession_penalty(scored, limit=5)
     assert [int(n.id) for _s, n in out] == [1, 2]
@@ -62,7 +57,7 @@ async def test_a_strong_superseded_record_still_beats_a_weak_live_one():
 async def test_the_superseded_record_is_still_returned():
     """The whole point. A test that only checked ordering would pass just as
     happily against an implementation that dropped it."""
-    scored = [(0.70, _note(1))]
+    scored = [(0.70, fake_note(id=1))]
     with _stale(1):
         out = await _apply_supersession_penalty(scored, limit=5)
     assert [int(n.id) for _s, n in out] == [1]
@@ -73,7 +68,7 @@ async def test_the_returned_score_is_the_adjusted_one():
     """Downstream gates must see the adjusted value — the auto-inject margin
     band in particular, which exists to stop near-ties dragging in neighbours
     and would otherwise re-tie exactly what this just separated."""
-    scored = [(0.70, _note(1)), (0.68, _note(2))]
+    scored = [(0.70, fake_note(id=1)), (0.68, fake_note(id=2))]
     with _stale(1):
         out = await _apply_supersession_penalty(scored, limit=5)
     by_id = {int(n.id): s for s, n in out}
@@ -83,7 +78,7 @@ async def test_the_returned_score_is_the_adjusted_one():
 
 @pytest.mark.asyncio
 async def test_nothing_superseded_leaves_the_order_untouched():
-    scored = [(0.70, _note(1)), (0.69, _note(2)), (0.60, _note(3))]
+    scored = [(0.70, fake_note(id=1)), (0.69, fake_note(id=2)), (0.60, fake_note(id=3))]
     with _stale():
         out = await _apply_supersession_penalty(scored, limit=5)
     assert [int(n.id) for _s, n in out] == [1, 2, 3]
@@ -94,7 +89,7 @@ async def test_ties_keep_their_database_order():
     """Stable sort. Equal scores must not reshuffle per call — a menu that
     reorders between identical queries reads as nondeterminism and sends
     someone hunting for a bug that isn't there."""
-    scored = [(0.70, _note(1)), (0.70, _note(2)), (0.70, _note(3))]
+    scored = [(0.70, fake_note(id=1)), (0.70, fake_note(id=2)), (0.70, fake_note(id=3))]
     with _stale():
         out = await _apply_supersession_penalty(scored, limit=5)
     assert [int(n.id) for _s, n in out] == [1, 2, 3]
@@ -104,7 +99,7 @@ async def test_ties_keep_their_database_order():
 async def test_the_limit_is_applied_after_reordering():
     """Over-fetching is pointless if the cut happens first. Three candidates,
     limit 2, and the demoted leader must be the one that falls out."""
-    scored = [(0.70, _note(1)), (0.69, _note(2)), (0.68, _note(3))]
+    scored = [(0.70, fake_note(id=1)), (0.69, fake_note(id=2)), (0.68, fake_note(id=3))]
     with _stale(1):
         out = await _apply_supersession_penalty(scored, limit=2)
     assert [int(n.id) for _s, n in out] == [2, 3]
@@ -115,7 +110,7 @@ async def test_a_failed_lookup_returns_unpenalised_results_not_none():
     """Fail OPEN, and the direction matters. Ranking without the penalty is the
     behaviour that shipped for months; returning nothing would turn a
     supersession hiccup into a broken search."""
-    scored = [(0.70, _note(1)), (0.69, _note(2))]
+    scored = [(0.70, fake_note(id=1)), (0.69, fake_note(id=2))]
     with patch("scribe.services.supersession.superseded_ids",
                AsyncMock(side_effect=RuntimeError("db gone"))):
         out = await _apply_supersession_penalty(scored, limit=5)

tests/test_write_path_trigger.py

@@ -13,6 +13,7 @@ from pathlib import Path
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
+from tests.helpers import fake_note
 
 PLUGIN = Path(__file__).resolve().parents[1] / "plugin"
 HOOK = PLUGIN / "hooks" / "scribe_prior_art.sh"
@@ -23,20 +24,6 @@ def _snippet_item(nid, title, user_id=1):
     return {"id": nid, "title": title, "user_id": user_id, "note_type": "snippet"}
 
 
-def _note(nid, title, user_id=1, note_type="snippet", is_task=False, task_kind="work"):
-    n = MagicMock()
-    n.id, n.title, n.user_id = nid, title, user_id
-    # Explicitly None, not left to MagicMock's auto-attribute: the semantic arm
-    # reads `note.data` for the snippet's language (#2244), and an auto-created
-    # mock there is truthy, so it would render its repr into the menu line.
-    n.data = None
-    # Same reasoning, second instance: since #2246 this arm returns issues and
-    # dev-logs too, so the line names the kind. An auto-mock `is_task` is truthy,
-    # which would label every snippet here "task".
-    n.note_type, n.is_task, n.task_kind = note_type, is_task, task_kind
-    return n
-
-
 def _cfg(**over):
     base = {"enabled": True, "threshold": 0.68, "top_k": 3}
     base.update(over)
@@ -218,7 +205,7 @@ async def test_a_failing_location_lookup_does_not_sink_the_hint():
     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 — …"))])), \
+                      AsyncMock(return_value=[(0.80, fake_note(id=5, title="throttle — …", user_id=1, note_type="snippet"))])), \
          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=REAL_CODE)
@@ -249,7 +236,7 @@ async def test_semantic_arm_is_snippet_only_and_browse_scoped():
 @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"))]
+    hits = [(0.80, fake_note(id=11, title="near", user_id=1, note_type="snippet")), (0.74, fake_note(id=22, title="alsoNear", user_id=1, note_type="snippet")), (0.61, fake_note(id=33, title="far", user_id=1, note_type="snippet"))]
     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)), \
@@ -288,7 +275,7 @@ async def test_place_beats_meaning_and_the_cap_covers_both_arms():
     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"))])), \
+                      AsyncMock(return_value=[(0.9, fake_note(id=7, title="scored", user_id=1, note_type="snippet"))])), \
          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=REAL_CODE)
@@ -363,7 +350,7 @@ async def test_telemetry_uses_its_own_source():
     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"))])), \
+                      AsyncMock(return_value=[(0.9, fake_note(id=7, title="scored", user_id=1, note_type="snippet"))])), \
          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=REAL_CODE, project_id=4)
@@ -497,7 +484,7 @@ async def test_a_real_helper_clears_the_floor():
     """The floor errs toward keeping recall — anything that plausibly IS a
     reusable helper has to get through, or the feature stops working."""
     from scribe.services import plugin_context as pc
-    search = AsyncMock(return_value=[(0.80, _note(5, "debounce — …"))])
+    search = AsyncMock(return_value=[(0.80, fake_note(id=5, title="debounce — …", user_id=1, note_type="snippet"))])
     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), \
@@ -712,7 +699,7 @@ async def test_a_cross_language_hit_is_labelled_and_explained():
     """The fail state this closes: an unlabelled Python hit offered while writing
     TypeScript is either dismissed as irrelevant or pasted into the .ts file."""
     from scribe.services import plugin_context as pc
-    note = _note(7, "group_pairs — collapse related pairs into groups")
+    note = fake_note(id=7, title="group_pairs — collapse related pairs into groups", user_id=1, note_type="snippet")
     note.data = {"language": "python"}
     with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
          patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
@@ -732,7 +719,7 @@ async def test_a_same_language_hit_is_not_labelled_and_gets_no_preamble():
     """The common case keeps a clean line; the explanation only appears when
     there is something on the menu it explains."""
     from scribe.services import plugin_context as pc
-    note = _note(7, "debounce — rate-limit a callback")
+    note = fake_note(id=7, title="debounce — rate-limit a callback", user_id=1, note_type="snippet")
     note.data = {"language": "python"}
     with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
          patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
@@ -749,7 +736,7 @@ async def test_a_same_language_hit_is_not_labelled_and_gets_no_preamble():
 async def test_an_unknown_target_extension_never_invents_a_mismatch():
     """A wrong "· python" tag is worse than no tag at all."""
     from scribe.services import plugin_context as pc
-    note = _note(7, "helper — does a thing")
+    note = fake_note(id=7, title="helper — does a thing", user_id=1, note_type="snippet")
     note.data = {"language": "python"}
     with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
          patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
@@ -1054,7 +1041,7 @@ async def test_a_pulled_snippet_already_seen_is_evidence_not_menu():
     writes code resembling it. #7 must be scored for this payload — and handed
     to the stamp as resemblance — without being re-listed in the menu."""
     from scribe.services import plugin_context as pc
-    search = AsyncMock(return_value=[(0.91, _note(7, "pulled")), (0.80, _note(8, "fresh"))])
+    search = AsyncMock(return_value=[(0.91, fake_note(id=7, title="pulled", user_id=1, note_type="snippet")), (0.80, fake_note(id=8, title="fresh", user_id=1, note_type="snippet"))])
     stamp = AsyncMock(return_value=[{
         "path": "src/x.py", "symbol": "debounce", "kind": "sym",
         "snippet_id": 7, "reason": "hook: pulled #7; payload resembles it (0.91)",