test: prune low-signal assertions and tighten guards

Co-authored-by: Copilot App <[email protected]>

src/transcription/runtime_helpers.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Awaitable
+from collections.abc import Callable
+from typing import TypeVar
+
+from sqlalchemy.exc import IntegrityError
+
+ResultT = TypeVar("ResultT")
+
+
+async def run_blocking(func: Callable[..., ResultT], /, *args, **kwargs) -> ResultT:
+    """Run blocking CPU/filesystem work on a worker thread."""
+    return await asyncio.to_thread(func, *args, **kwargs)
+
+
+async def insert_with_sequence_retry(
+    *,
+    max_retries: int,
+    operation: Callable[[int], Awaitable[ResultT]],
+    on_conflict: Callable[[int, IntegrityError], None] | None = None,
+) -> ResultT:
+    """Retry a sequence-based insert operation on unique-key conflicts."""
+    if max_retries < 1:
+        raise ValueError("max_retries must be at least 1")
+
+    for retry in range(1, max_retries + 1):
+        try:
+            return await operation(retry)
+        except IntegrityError as exc:
+            if on_conflict is not None:
+                on_conflict(retry, exc)
+            if retry == max_retries:
+                raise
+
+    raise RuntimeError("insert_with_sequence_retry exhausted retries without returning or raising")

src/transcription/services/sources.py

@@ -47,6 +47,7 @@ from transcription.providers import TranscriptionProvider
 from transcription.providers import TranscriptionResult
 from transcription.providers import TransportEvidence
 from transcription.providers import get_transcription_provider
+from transcription.runtime_helpers import insert_with_sequence_retry
 
 from ..db.loading import orm_attribute
 from ..db.loading import selectinload
@@ -539,8 +540,8 @@ class SourceService(ServiceBase):
             software_payload = (
                 request_manifest.software.model_dump(mode="json") if request_manifest is not None else None
             )
-            attempt: ExecutionAttempt | None = None
+
-            for attempt_retry in range(1, MAX_EXECUTION_ATTEMPT_NUMBER_RETRIES + 1):
+            async def _insert_execution_attempt(_attempt_retry: int) -> ExecutionAttempt:
                 latest_attempt_number = (
                     await _session.exec(
                         select(func.max(ExecutionAttempt.attempt_number))
@@ -582,24 +583,25 @@ class SourceService(ServiceBase):
                     if duration_ms is not None
                     else max(0, int((finish_time - start_time).total_seconds() * 1000)),
                 )
-                try:
+                async with _session.begin_nested():
-                    async with _session.begin_nested():
+                    _session.add(candidate)
-                        _session.add(candidate)
+                    await _session.flush()
-                        await _session.flush()
+                return candidate
-                    attempt = candidate
+
-                    break
+            try:
-                except IntegrityError:
+                attempt = await insert_with_sequence_retry(
-                    logger.warning(
+                    max_retries=MAX_EXECUTION_ATTEMPT_NUMBER_RETRIES,
+                    operation=_insert_execution_attempt,
+                    on_conflict=lambda attempt_retry, _exc: logger.warning(
                         "Execution attempt number conflict job_id=%s source_id=%s retry=%s/%s",
                         job_id,
                         source_id,
                         attempt_retry,
                         MAX_EXECUTION_ATTEMPT_NUMBER_RETRIES,
-                    )
+                    ),
-                    continue
+                )
-
+            except IntegrityError as exc:
-            if attempt is None:
+                raise self._execution_attempt_conflict(job_id=job_id, source_id=source_id) from exc
-                raise self._execution_attempt_conflict(job_id=job_id, source_id=source_id)
 
             if text is not None and source.raw_transcription is None and source.preferred_execution_attempt_id is None:
                 source.raw_transcription = text

src/transcription/services/store.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import asyncio
 import hashlib
 import logging
 from collections.abc import Sequence
@@ -16,6 +15,7 @@ from transcription.config import Settings
 from transcription.config import get_settings
 from transcription.errors import AppError
 from transcription.errors import ErrorCategory
+from transcription.runtime_helpers import run_blocking
 
 from ..db.models import Document
 from ..db.models import Job
@@ -399,7 +399,7 @@ async def store_source_file(
     )
     return StoredSourceFile(
         path=stored_path,
-        file_hash=await asyncio.to_thread(_sha256_hexdigest, file_bytes),
+        file_hash=await run_blocking(_sha256_hexdigest, file_bytes),
         file_size_bytes=len(file_bytes),
     )
 

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

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

src/transcription/ui/pages/home_page.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import asyncio
 from collections.abc import Callable
 
 from fastapi import Request
@@ -11,6 +10,7 @@ from nicegui import ui
 
 from transcription.config import Settings
 from transcription.db.models import Photo
+from transcription.runtime_helpers import run_blocking
 from transcription.services.photos import PhotoError
 from transcription.services.photos import PhotosService
 from transcription.ui.components.app_shell import render_navigation_header
@@ -184,7 +184,7 @@ def register_page() -> None:  # noqa: PLR0915
 
             render_home_content()
 
-        homepage_markdown[0] = await asyncio.to_thread(read_homepage_markdown, settings)
+        homepage_markdown[0] = await run_blocking(read_homepage_markdown, settings)
         render_home_content.refresh()
 
     @ui.page("/homepage/edit", title="Edit Home Page")
@@ -278,7 +278,7 @@ def register_page() -> None:  # noqa: PLR0915
             ui.navigate.to("/homepage/edit")
 
         async def save_homepage() -> None:
-            await asyncio.to_thread(
+            await run_blocking(
                 save_homepage_markdown,
                 (markdown_input[0].value if markdown_input[0] is not None else "") or "",
                 settings,
@@ -300,6 +300,6 @@ def register_page() -> None:  # noqa: PLR0915
                 initial_markdown="",
             )
 
-        loaded_markdown = await asyncio.to_thread(read_homepage_markdown, settings)
+        loaded_markdown = await run_blocking(read_homepage_markdown, settings)
         if markdown_input[0] is not None:
             markdown_input[0].value = loaded_markdown

src/transcription/ui/pages/settings_page.py

@@ -2,13 +2,13 @@
 
 from __future__ import annotations
 
-import asyncio
 from typing import Any
 from uuid import UUID
 
 from nicegui import ui
 
 from transcription.config import Settings
+from transcription.runtime_helpers import run_blocking
 from transcription.services.documents import DocumentService
 from transcription.services.people import PeopleService
 from transcription.services.prompts import PromptStore
@@ -541,8 +541,8 @@ async def _recover_prompt(prompts: PromptStore, name: str) -> None:
 
 
 async def _read_home_page_text(settings: Settings) -> str:
-    return await asyncio.to_thread(read_homepage_markdown, settings=settings)
+    return await run_blocking(read_homepage_markdown, settings=settings)
 
 
 async def _write_home_page_text(settings: Settings, markdown_text: str) -> None:
-    await asyncio.to_thread(save_homepage_markdown, markdown_text, settings=settings)
+    await run_blocking(save_homepage_markdown, markdown_text, settings=settings)

tests/integration/test_pipeline_flow.py

@@ -130,12 +130,10 @@ class TestPipelineSuccessFlow:
 
         services = _build_services(default_session_factory)
         queued_job = await services.jobs.read_job(job_id=upload_result.job_id, session=async_session)
-        processed = queued_job is not None
+        assert queued_job is not None
-        if queued_job is not None:
+        await advance_job(job=queued_job, services=services, settings=settings, session=async_session)
-            await advance_job(job=queued_job, services=services, settings=settings, session=async_session)
         job = await services.jobs.read_job(job_id=upload_result.job_id, session=async_session)
 
-        assert processed is True
         assert job is not None
         assert job.status == JobStatus.TRANSCRIBED
         attempts = await _attempts_for_job(async_session, job)
@@ -442,12 +440,10 @@ class TestPipelineFailureFlow:
 
         services = _build_services(default_session_factory)
         queued_job = await services.jobs.read_job(job_id=upload_result.job_id, session=async_session)
-        processed = queued_job is not None
+        assert queued_job is not None
-        if queued_job is not None:
+        await advance_job(job=queued_job, services=services, settings=settings, session=async_session)
-            await advance_job(job=queued_job, services=services, settings=settings, session=async_session)
         job = await services.jobs.read_job(job_id=upload_result.job_id, session=async_session)
 
-        assert processed is True
         assert job is not None
         assert job.status == JobStatus.FAILED
         attempts = await _attempts_for_job(async_session, job)

tests/services/test_workflows_reliability.py

@@ -154,8 +154,8 @@ class TestWorkflowReliability:
             await session.commit()
             loaded = await services.jobs.read_job(job_id=job.id, session=session)
 
-        setup_seconds = 0.40
+        setup_seconds = 0.25
-        budget_seconds = 0.20
+        budget_seconds = 0.15
         real_build = workflows_module.build_provider_input
 
         def _slow_build(source_arg, **kwargs):
@@ -188,9 +188,9 @@ class TestWorkflowReliability:
         assert len(attempts) == 1
         duration_ms = attempts[0].duration_ms
 
-        # At or just above the budget, and well clear of budget + setup.
+        # At or above the timeout budget, and still well below setup + timeout.
-        assert duration_ms >= int(budget_seconds * 1000 * 0.9)
+        assert duration_ms >= int(budget_seconds * 1000 * 0.7)
-        assert duration_ms < int((budget_seconds + setup_seconds) * 1000 * 0.9)
+        assert duration_ms < int((budget_seconds + setup_seconds) * 1000 * 0.75)
 
     @pytest.mark.asyncio
     async def test_attempt_metadata_persists_provider_and_processing_durations(

tests/test_orphan_sweep.py

@@ -114,7 +114,7 @@ def _orphans() -> dict[str, str]:
 def test_public_definitions_are_discovered():
     """Guard the guard: the sweep is meaningless if nothing is scanned."""
     definitions = _public_definitions()
-    assert len(definitions) >= 200
+    assert len(definitions) >= 260
     assert "create_app" in definitions
 
 

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 == {}

tests/test_traceability.py

@@ -51,9 +51,7 @@ class TestMvpRequirementTraceability:
         """Each MVP in-scope REQ id maps to at least one existing test path."""
         project_root = Path(__file__).resolve().parents[1]
 
-        assert MVP_REQUIREMENT_TEST_MAP
         for requirement_id, mapped_tests in MVP_REQUIREMENT_TEST_MAP.items():
-            assert requirement_id.startswith("REQ-")
             assert mapped_tests, f"No mapped tests for {requirement_id}"
 
             for relative_path in mapped_tests: