diff --git a/src/transcription/services/unit_of_work.py b/src/transcription/services/unit_of_work.py
new file mode 100644
index 0000000..feb04d8
--- /dev/null
+++ b/src/transcription/services/unit_of_work.py
@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+from ..db.session import transaction_scope
+from . import ServiceBundle
+
+
+@asynccontextmanager
+async def unit_of_work(
+    *,
+    services: ServiceBundle,
+    session: AsyncSession | None = None,
+) -> AsyncIterator[AsyncSession]:
+    """Yield one shared transactional session for orchestration paths."""
+    if session is not None:
+        yield session
+        return
+
+    async with transaction_scope(session_factory=services.jobs.session_factory) as local_session:
+        yield local_session
diff --git a/src/transcription/services/workflows.py b/src/transcription/services/workflows.py
index afd3345..1e68f40 100644
--- a/src/transcription/services/workflows.py
+++ b/src/transcription/services/workflows.py
@@ -37,6 +37,7 @@ from .sources import build_prompt_execution
 from .sources import build_provider_input
 from .sources import hash_prompt_text
 from .sources import transcribe_document_image
+from .unit_of_work import unit_of_work
 
 logger = logging.getLogger(__name__)
 
@@ -627,7 +628,7 @@ async def _finalize_batch_outcome(
     status change (no sources, or the batch stopped before the last page).
     """
     if session is None:
-        async with services.jobs._session_scope() as local_session:
+        async with unit_of_work(services=services, session=session) as local_session:
             if final_page is not None:
                 await _write_page_outcome(job=job, services=services, page=final_page, session=local_session)
             updated_job = await services.jobs.mark_job_status(job.id, status, session=local_session)
@@ -665,7 +666,7 @@ async def _persist_page_outcome(
     session: AsyncSession | None,
 ) -> None:
     if session is None:
-        async with services.sources._session_scope() as local_session:
+        async with unit_of_work(services=services, session=session) as local_session:
             await _write_page_outcome(job=job, services=services, page=page, session=local_session)
             await local_session.commit()
         return
diff --git a/tests/test_service_boundaries.py b/tests/test_service_boundaries.py
index e38615f..c835414 100644
--- a/tests/test_service_boundaries.py
+++ b/tests/test_service_boundaries.py
@@ -64,3 +64,26 @@ def test_no_service_module_imports_another_service_module():
         name: sorted(_imported_sibling_modules(tree) & set(modules) - {name}) for name, tree in modules.items()
     }
     assert {name: found for name, found in violations.items() if found} == {}
+
+
+def _foreign_session_scope_accesses(tree: ast.Module) -> list[int]:
+    lines: list[int] = []
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Attribute) and node.attr == "_session_scope":
+            if isinstance(node.value, ast.Name) and node.value.id == "self":
+                continue
+            lines.append(node.lineno)
+    return sorted(lines)
+
+
+def test_session_scope_is_not_accessed_via_other_services():
+    """P4-1: orchestration must use a shared unit-of-work entry point, not private service scopes."""
+    violations: dict[str, list[int]] = {}
+    for path in _module_paths():
+        if path.stem == "base":
+            continue
+        tree = ast.parse(path.read_text(encoding="utf-8"))
+        found = _foreign_session_scope_accesses(tree)
+        if found:
+            violations[path.name] = found
+    assert violations == {}