chore: enable logging format lint rule

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

.pre-commit-config.yaml

@@ -1,6 +1,5 @@
-# Quality gate for V4.6 [HIGH-06]. `ruff check` is blocking. `ty check` is advisory
-# during release stabilization: it reports its whole-project baseline without failing
-# the commit. Restore it to blocking once that baseline is clear.
+# Quality gate for V4.6 [HIGH-06]. `ruff check`, `ruff format --check`, and `ty check`
+# are blocking once known `ty` false positives are suppressed inline with rationale.
 #
 # Both tools are uv-managed dev dependencies and are not on PATH, so each entry must
 # go through `uv run`.
@@ -13,9 +12,16 @@ repos:
         language: system
         types_or: [python, pyi]
         require_serial: true
+      - id: ruff-format
+        name: ruff format check
+        entry: uv run ruff format --check .
+        language: system
+        types_or: [python, pyi]
+        pass_filenames: false
+        require_serial: true
       - id: ty
-        name: ty check (advisory)
-        entry: python -c "import subprocess, sys; subprocess.run(['uv', 'run', 'ty', 'check']); sys.exit(0)"
+        name: ty check
+        entry: uv run ty check
         language: system
         types_or: [python, pyi]
         pass_filenames: false

docs/production-runbook.md

@@ -113,3 +113,17 @@ Declared with `>=` floors and moved by explicit `uv lock --upgrade`. Verify with
 `uv run ruff check .`, `uv run ty check`, and `uv run pytest -q -m "not external"`
 before committing a changed lockfile.
 
+## 7. Type-check suppression policy
+
+`uv run ty check` is a blocking pre-commit gate. Suppressions are allowed only for
+proven SQLAlchemy descriptor false positives where runtime behavior is correct and
+the checker cannot represent the descriptor protocol at that call site.
+
+Every suppression must be:
+
+1. **Targeted** to a single rule (for example `# ty: ignore[unresolved-attribute]`).
+2. **Inline** on the expression it suppresses (not file-wide).
+3. Followed by a **one-line rationale** stating it is a SQLAlchemy descriptor false positive.
+
+Do not use broad or rationale-free suppressions. If a diagnostic is not a known
+false positive, fix the code instead of suppressing it.

docs/reviews/2026-08-23-code-review.md

@@ -384,6 +384,7 @@ The critical gap is transaction atomicity (HIGH-04) — the audit verdict is **"
 ```python
 # src/transcription/services/base.py
 
+
 @asynccontextmanager
 async def unit_of_work(
     services: ServiceBundle,
@@ -396,10 +397,12 @@ async def unit_of_work(
     HIGH-01 structurally hard to reintroduce.
     """
 
+
 async def run_blocking[T](fn: Callable[[], T]) -> T:
     """Run a CPU- or disk-bound callable off the event loop."""
     return await asyncio.to_thread(fn)
 
+
 async def insert_with_sequence_retry(
     session: AsyncSession,
     *,

ruff.toml

@@ -26,6 +26,7 @@ extend-select = [
 	"E", "W",   # https://docs.astral.sh/ruff/rules/#pycodestyle-e-w
 	"F",        # https://docs.astral.sh/ruff/rules/#pyflakes-f
     "FURB",     # https://docs.astral.sh/ruff/rules/#refurb-furb
+    "G",        # https://docs.astral.sh/ruff/rules/#flake8-logging-format-g
 	"I",        # https://docs.astral.sh/ruff/rules/#isort-i
     "N",        # https://docs.astral.sh/ruff/rules/#pep8-naming-n
     "PD",       # https://docs.astral.sh/ruff/rules/#pandas-vet-pd

src/transcription/config.py

@@ -207,7 +207,7 @@ LOGGING_CONFIG: dict[str, Any] = {
             "maxBytes": 10 * 1024 * 1024,
             "backupCount": 5,
             "encoding": "utf-8",
-        }
+        },
     },
     "root": {
         "level": "INFO",

src/transcription/db/migration.py

@@ -93,11 +93,15 @@ def export_bundle(*, source_db_url: str, source_upload_dir: Path, bundle_dir: Pa
                         if legacy_column not in export_columns:
                             export_columns.append(legacy_column)
                 if table_name == "person" and "portrait_path" in source_table.columns:
-                    legacy_portrait_rows = connection.execute(
+                    legacy_portrait_rows = (
+                        connection.execute(
                             select(source_table.c["id"], source_table.c["portrait_path"]).where(
                                 source_table.c["portrait_path"].is_not(None)
                             )
-                    ).mappings().all()
+                        )
+                        .mappings()
+                        .all()
+                    )
                 rows = connection.execute(select(*(source_table.c[name] for name in export_columns))).mappings().all()
                 payload["tables"][table_name] = [
                     _serialize_row(row, table_name=table_name, source_upload_dir=source_upload_dir) for row in rows
@@ -344,15 +348,9 @@ def _prepare_photo_payload_and_uploads(  # noqa: PLR0915
     photo_rows[:] = retained_rows
 
     existing_homepage_rows = [row for row in photo_rows if row.get("person_id") is None]
-    existing_person_ids = {
-        str(row["person_id"])
-        for row in photo_rows
-        if row.get("person_id") is not None
-    }
+    existing_person_ids = {str(row["person_id"]) for row in photo_rows if row.get("person_id") is not None}
     existing_primary_person_ids = {
-        str(row["person_id"])
-        for row in photo_rows
-        if row.get("person_id") is not None and bool(row.get("is_primary"))
+        str(row["person_id"]) for row in photo_rows if row.get("person_id") is not None and bool(row.get("is_primary"))
     }
     has_homepage_primary = any(bool(row.get("is_primary")) for row in existing_homepage_rows)
 

src/transcription/db/models.py

@@ -435,9 +435,7 @@ class Source(SQLModel, table=True):
         """
         job_sources = _loaded_attribute(self, "job_sources") or ()
         dated = [
-            (job, job_source)
-            for job_source in job_sources
-            if (job := _loaded_attribute(job_source, "job")) is not None
+            (job, job_source) for job_source in job_sources if (job := _loaded_attribute(job_source, "job")) is not None
         ]
         if dated:
             return max(dated, key=lambda pair: pair[0].date_created)[1]

src/transcription/db/runtime.py

@@ -50,8 +50,7 @@ def initialize_database_runtime(*, settings: Settings | None = None) -> Database
         runtime_url = runtime.engine.url.render_as_string(hide_password=False)
         if runtime_url != database_url:
             raise RuntimeError(
-                "Database runtime is already initialized for a different database: "
-                f"{runtime_url!r} != {database_url!r}"
+                f"Database runtime is already initialized for a different database: {runtime_url!r} != {database_url!r}"
             )
         return runtime
 

src/transcription/services/documents.py

@@ -582,8 +582,7 @@ class DocumentService(ServiceBase):
                 )
 
             existing_tags = (
-                (await _session.exec(select(Tag).where(col(Tag.normalized_label).in_(label_keys))))
-                .all()
+                (await _session.exec(select(Tag).where(col(Tag.normalized_label).in_(label_keys)))).all()
                 if label_keys
                 else []
             )
@@ -598,9 +597,7 @@ class DocumentService(ServiceBase):
                     tags_by_key[key] = tag
                 selected_tag_ids.add(tag.id)
 
-            links = (
-                await _session.exec(select(DocumentTag).where(DocumentTag.document_id == document_id))
-            ).all()
+            links = (await _session.exec(select(DocumentTag).where(DocumentTag.document_id == document_id))).all()
             existing_ids = {link.tag_id for link in links}
 
             for link in links:

src/transcription/services/jobs.py

@@ -278,9 +278,7 @@ class JobService(ServiceBase):
                 )
             attempt_count = (
                 await _session.exec(
-                    select(func.count())
-                    .select_from(ExecutionAttempt)
-                    .where(ExecutionAttempt.job_id == job_id)
+                    select(func.count()).select_from(ExecutionAttempt).where(ExecutionAttempt.job_id == job_id)
                 )
             ).one()
             if attempt_count:
@@ -320,11 +318,7 @@ class JobService(ServiceBase):
                 )
 
             attempts = list(
-                (
-                    await session.exec(
-                        select(ExecutionAttempt).where(ExecutionAttempt.job_id == job_id)
-                    )
-                ).all()
+                (await session.exec(select(ExecutionAttempt).where(ExecutionAttempt.job_id == job_id))).all()
             )
             for attempt in attempts:
                 await session.delete(attempt)

src/transcription/services/media_storage.py

@@ -67,10 +67,14 @@ async def persist_named_media(
 ) -> Path:
     """Resolve a target directory/name and persist media bytes safely."""
     target_dir = root if namespace is None else root / Path(namespace)
-    stored_name = Path(filename).name if preserve_original_name else build_stored_filename(
+    stored_name = (
+        Path(filename).name
+        if preserve_original_name
+        else build_stored_filename(
             filename=filename,
             filename_stem=filename_stem,
         )
+    )
     return await write_media_bytes(
         target_dir=target_dir,
         stored_name=stored_name,

src/transcription/services/people.py

@@ -277,8 +277,7 @@ class PeopleService(ServiceBase):
                 )
 
             existing_tags = (
-                (await _session.exec(select(Tag).where(col(Tag.normalized_label).in_(label_keys))))
-                .all()
+                (await _session.exec(select(Tag).where(col(Tag.normalized_label).in_(label_keys)))).all()
                 if label_keys
                 else []
             )

src/transcription/services/photos.py

@@ -178,25 +178,33 @@ class PhotosService(ServiceBase):
     async def _list_owner_photos(self, *, session: AsyncSession, person_id: UUID | None) -> list[Photo]:
         query = select(Photo)
         if person_id is None:
-            query = query.where(Photo.person_id.is_(None))
+            query = query.where(Photo.person_id.is_(None))  # ty: ignore[unresolved-attribute] - SQLAlchemy descriptor false positive.
         else:
             query = query.where(Photo.person_id == person_id)
-        query = query.order_by(Photo.created_at.asc(), Photo.id.asc())
+        query = query.order_by(
+            Photo.created_at.asc(),  # ty: ignore[unresolved-attribute] - SQLAlchemy descriptor false positive.
+            Photo.id.asc(),  # ty: ignore[unresolved-attribute] - SQLAlchemy descriptor false positive.
+        )
         return list((await session.exec(query)).all())
 
     async def _owner_oldest_photo(self, *, session: AsyncSession, person_id: UUID | None) -> Photo | None:
         query = select(Photo)
         if person_id is None:
-            query = query.where(Photo.person_id.is_(None))
+            query = query.where(Photo.person_id.is_(None))  # ty: ignore[unresolved-attribute] - SQLAlchemy descriptor false positive.
         else:
             query = query.where(Photo.person_id == person_id)
-        query = query.order_by(Photo.created_at.asc(), Photo.id.asc()).limit(1)
+        query = query.order_by(
+            Photo.created_at.asc(),  # ty: ignore[unresolved-attribute] - SQLAlchemy descriptor false positive.
+            Photo.id.asc(),  # ty: ignore[unresolved-attribute] - SQLAlchemy descriptor false positive.
+        ).limit(1)
         return (await session.exec(query)).first()
 
     async def _clear_owner_primary(self, *, session: AsyncSession, person_id: UUID | None) -> None:
-        query = select(Photo).where(Photo.is_primary.is_(True))
+        query = select(Photo).where(
+            Photo.is_primary.is_(True)  # ty: ignore[unresolved-attribute] - SQLAlchemy descriptor false positive.
+        )
         if person_id is None:
-            query = query.where(Photo.person_id.is_(None))
+            query = query.where(Photo.person_id.is_(None))  # ty: ignore[unresolved-attribute] - SQLAlchemy descriptor false positive.
         else:
             query = query.where(Photo.person_id == person_id)
         for current in (await session.exec(query)).all():

src/transcription/services/quality.py

@@ -65,9 +65,7 @@ def analyze_transcription_quality(text: str) -> tuple[QualityWarning, ...]:
             warnings.append(
                 QualityWarning(
                     code=QualityWarningCode.REDUNDANT_HANDWRITING_WRAPPERS,
-                    detail=(
-                        "A wholly handwritten document also uses repeated whole-line handwriting wrappers."
-                    ),
+                    detail=("A wholly handwritten document also uses repeated whole-line handwriting wrappers."),
                 )
             )
 

src/transcription/services/registry.py

@@ -249,8 +249,7 @@ class RegistryService[ModelT: RegistryEntry](ServiceBase):
                     f"Built-in {self.noun} {entry.label!r} cannot be deleted",
                     category=ErrorCategory.CONFLICT,
                     suggestion=(
-                        f"Deactivate the {self.short_noun} instead; "
-                        "its built-in meaning must remain available."
+                        f"Deactivate the {self.short_noun} instead; its built-in meaning must remain available."
                     ),
                 )
             if await self._is_referenced(session=_session, entry=entry):
@@ -258,8 +257,7 @@ class RegistryService[ModelT: RegistryEntry](ServiceBase):
                     f"{self.noun} {entry.label!r} is referenced and cannot be deleted",
                     category=ErrorCategory.CONFLICT,
                     suggestion=(
-                        f"Deactivate the {self.short_noun} instead; "
-                        f"{self.referenced_retainer} will retain it."
+                        f"Deactivate the {self.short_noun} instead; {self.referenced_retainer} will retain it."
                     ),
                 )
             await _session.delete(entry)

src/transcription/services/sources.py

@@ -351,9 +351,7 @@ class SourceService(ServiceBase):
         async with self._session_scope(session) as _session:
             job_source = (
                 await _session.exec(
-                    select(JobSource)
-                    .where(JobSource.job_id == job_id)
-                    .where(JobSource.source_id == source_id)
+                    select(JobSource).where(JobSource.job_id == job_id).where(JobSource.source_id == source_id)
                 )
             ).first()
         if job_source is None:
@@ -400,9 +398,7 @@ class SourceService(ServiceBase):
             linked_job_sources = list(source.job_sources)
             attempt_count = (
                 await _session.exec(
-                    select(func.count())
-                    .select_from(ExecutionAttempt)
-                    .where(ExecutionAttempt.source_id == source_id)
+                    select(func.count()).select_from(ExecutionAttempt).where(ExecutionAttempt.source_id == source_id)
                 )
             ).one()
             if attempt_count:
@@ -586,11 +582,7 @@ class SourceService(ServiceBase):
             _session.add(attempt)
             await _session.flush()
 
-            if (
-                text is not None
-                and source.raw_transcription is None
-                and source.preferred_execution_attempt_id is None
-            ):
+            if text is not None and source.raw_transcription is None and source.preferred_execution_attempt_id is None:
                 source.raw_transcription = text
                 source.preferred_execution_attempt_id = attempt.id
 

src/transcription/services/workflows.py

@@ -190,7 +190,7 @@ async def advance_job(
                     session=session,
                 )
             else:
-                logger.error(f"Job {job.id} has failed and reached max retries.")
+                logger.error("Job %s has failed and reached max retries.", job.id)
                 return
         case _:
             return
@@ -207,7 +207,7 @@ async def process_queued_job(  # noqa: PLR0915
     runtime_settings = settings or get_settings()
     current_status = _coerce_job_status(job.status)
     if current_status not in {JobStatus.QUEUED, JobStatus.PROCESSING}:
-        logger.warning(f"Job {job.id} is not queued. Current status: {job.status}")
+        logger.warning("Job %s is not queued. Current status: %s", job.id, job.status)
         return
 
     # Transaction A: claim job for processing. Reached only when a caller hands us a

src/transcription/ui/components/upload_panel.py

@@ -43,8 +43,4 @@ def render_upload_picker(
         props.append("webkitdirectory directory")
     if multiple:
         props.append("multiple")
-    return (
-        ui.upload(on_upload=on_upload, auto_upload=True, label=label)
-        .props(" ".join(props))
-        .classes("w-full")
-    )
+    return ui.upload(on_upload=on_upload, auto_upload=True, label=label).props(" ".join(props)).classes("w-full")

src/transcription/ui/pages/documents_page.py

@@ -377,9 +377,7 @@ def register_page() -> None:  # noqa: PLR0915
                 if document.sources or document.jobs:
                     render_delete_blocked_notice(
                         reason="Delete is blocked because related records exist.",
-                        detail=dependency_summary(
-                            [("Sources", bool(document.sources)), ("Jobs", bool(document.jobs))]
-                        ),
+                        detail=dependency_summary([("Sources", bool(document.sources)), ("Jobs", bool(document.jobs))]),
                         guidance="Remove related records first, then retry deletion.",
                         back_label="Back to Document",
                         back_target=f"/documents/{document.id}",
@@ -497,14 +495,18 @@ def _render_document_form_fields(
             if document is not None
             else []
         )
-        tags_input = ui.select(
+        tags_input = (
+            ui.select(
                 sorted(tag_options, key=str.casefold),
                 label="Tags",
                 value=selected_tags,
                 multiple=True,
                 with_input=True,
                 new_value_mode="add-unique",
-        ).props("outlined use-chips").classes("w-full ui-form-surface")
+            )
+            .props("outlined use-chips")
+            .classes("w-full ui-form-surface")
+        )
 
         linked_people.render()
 

src/transcription/ui/pages/home_page.py

@@ -88,6 +88,7 @@ def _render_homepage_gallery(
         ui.label(f"{active_index[0] + 1} of {len(photos)}").classes("text-xs ui-text-muted")
 
     if enable_rotation and rotate_enabled is not None:
+
         def set_rotation(enabled: bool) -> None:
             rotate_enabled[0] = enabled
             if on_change is not None:
@@ -128,10 +129,14 @@ def _render_homepage_editor(*, render_image_panel, markdown_input, on_upload) ->
             render_image_panel()
 
         with ui.column().classes("col-span-12 lg:col-span-5 gap-4"), archival_card(title="Home Text"):
-            markdown_input[0] = ui.textarea(
+            markdown_input[0] = (
+                ui.textarea(
                     label="Homepage markdown",
                     value=read_homepage_markdown(),
-            ).props("outlined autogrow").classes("w-full")
+                )
+                .props("outlined autogrow")
+                .classes("w-full")
+            )
 
         with ui.column().classes("col-span-12 lg:col-span-3"):
             ui.element("div")
@@ -198,10 +203,14 @@ def register_page() -> None:  # noqa: PLR0915
 
                 if photos:
                     current_photo = photos[active_index[0]]
-                    description_input = ui.input(
+                    description_input = (
+                        ui.input(
                             label="Image description",
                             value=current_photo.description or "",
-                    ).props("outlined dense").classes("w-full")
+                        )
+                        .props("outlined dense")
+                        .classes("w-full")
+                    )
 
                     async def save_description() -> None:
                         try:

src/transcription/ui/pages/jobs_page.py

@@ -361,9 +361,7 @@ def register_page() -> None:  # noqa: PLR0915
             return
 
         resubmittable_count = sum(
-            1
-            for js in job.job_sources
-            if js.status in {JobSourceStatus.FAILED, JobSourceStatus.CANCELLED}
+            1 for js in job.job_sources if js.status in {JobSourceStatus.FAILED, JobSourceStatus.CANCELLED}
         )
 
         with ui.column().classes("w-full max-w-xl mx-auto p-4 gap-4"):
@@ -374,8 +372,7 @@ def register_page() -> None:  # noqa: PLR0915
                 metadata_row("Current Status:", job.status.value.upper())
                 metadata_row("Resubmittable Sources:", str(resubmittable_count))
                 ui.label(
-                    "Resubmit queues failed and cancelled linked sources. "
-                    "Prior execution evidence remains preserved."
+                    "Resubmit queues failed and cancelled linked sources. Prior execution evidence remains preserved."
                 ).classes("text-xs ui-text-muted")
 
             async def submit_resubmit() -> None:
@@ -440,9 +437,7 @@ def register_page() -> None:  # noqa: PLR0915
                     ui.label(
                         "Related JobSource links, execution attempts, transport responses, and attempt artifacts "
                         "will be removed. Source records and files remain until deleted separately."
-                    ).classes(
-                        "text-xs ui-text-muted"
-                    )
+                    ).classes("text-xs ui-text-muted")
 
             async def submit_delete() -> None:
                 try:

src/transcription/ui/pages/people_page.py

@@ -368,9 +368,9 @@ def register_page() -> None:  # noqa: PLR0915
                         label="",
                         on_upload=on_photo_selected,
                         auto_upload=True,
-                    ).props(
-                        f'multiple accept="{",".join(sorted(IMAGE_UPLOAD_EXTENSIONS))}"'
-                    ).classes("hidden person-photo-upload")
+                    ).props(f'multiple accept="{",".join(sorted(IMAGE_UPLOAD_EXTENSIONS))}"').classes(
+                        "hidden person-photo-upload"
+                    )
                     ui.button(
                         "Upload Photo(s)",
                         on_click=lambda: ui.run_javascript(
@@ -606,14 +606,18 @@ def _render_person_form_fields(
             if person is not None
             else []
         )
-        tags_input = ui.select(
+        tags_input = (
+            ui.select(
                 sorted(tag_options, key=str.casefold),
                 label="Tags",
                 value=selected_tags,
                 multiple=True,
                 with_input=True,
                 new_value_mode="add-unique",
-        ).props("outlined use-chips").classes("w-full ui-form-surface")
+            )
+            .props("outlined use-chips")
+            .classes("w-full ui-form-surface")
+        )
 
     return PersonFormFields(
         last_name=last_name_input,

src/transcription/ui/pages/sources_page.py

@@ -651,19 +651,11 @@ def _render_machine_candidates(
     evidence_service: EvidenceService,
 ) -> None:
     successful = [
-        attempt
-        for attempt in attempts
-        if attempt.status == JobSourceStatus.TRANSCRIBED and attempt.raw_transcription
-    ]
-    candidates = [
-        attempt for attempt in successful if attempt.id != source.preferred_execution_attempt_id
+        attempt for attempt in attempts if attempt.status == JobSourceStatus.TRANSCRIBED and attempt.raw_transcription
     ]
+    candidates = [attempt for attempt in successful if attempt.id != source.preferred_execution_attempt_id]
     preferred_attempt = next(
-        (
-            attempt
-            for attempt in successful
-            if attempt.id == source.preferred_execution_attempt_id
-        ),
+        (attempt for attempt in successful if attempt.id == source.preferred_execution_attempt_id),
         None,
     )
 
@@ -697,9 +689,7 @@ def _render_machine_candidates(
                     with ui.grid().classes("w-full grid-cols-1 md:grid-cols-2 gap-3"):
                         with ui.column().classes("gap-1"):
                             ui.label("Preferred machine transcription").classes("text-xs font-semibold")
-                            ui.label(source.raw_transcription).classes(
-                                "p-2 ui-note-box text-xs whitespace-pre-wrap"
-                            )
+                            ui.label(source.raw_transcription).classes("p-2 ui-note-box text-xs whitespace-pre-wrap")
                         with ui.column().classes("gap-1"):
                             ui.label("Candidate transcription").classes("text-xs font-semibold")
                             ui.label(attempt.raw_transcription or "").classes(

tests/integration/test_pipeline_flow.py

@@ -26,9 +26,7 @@ from transcription.services.workflows import advance_job
 async def _attempts_for_job(session, job) -> list[ExecutionAttempt]:
     """Load execution attempts for a job; V4.7 moved evidence off JobSource."""
     job_source_ids = [job_source.id for job_source in job.job_sources]
-    result = await session.exec(
-        select(ExecutionAttempt).where(col(ExecutionAttempt.job_source_id).in_(job_source_ids))
-    )
+    result = await session.exec(select(ExecutionAttempt).where(col(ExecutionAttempt.job_source_id).in_(job_source_ids)))
     return list(result.all())
 
 

tests/services/test_job_service.py

@@ -154,11 +154,7 @@ class TestJobService:
             finally:
                 event.remove(bind, "before_cursor_execute", capture)
 
-        claim_sql = [
-            item
-            for item in statements
-            if item.lstrip().upper().startswith(("SELECT", "UPDATE"))
-        ]
+        claim_sql = [item for item in statements if item.lstrip().upper().startswith(("SELECT", "UPDATE"))]
         assert len(claim_sql) == 1, claim_sql
         assert "LIMIT" in claim_sql[0].upper()
         assert "JOIN" not in claim_sql[0].upper()

tests/services/test_photo_service.py

@@ -7,10 +7,13 @@ from transcription.db.models import Person
 from transcription.services.people import PeopleService
 from transcription.services.photos import PhotosService
 
-PNG_BYTES = bytes.fromhex(
+PNG_BYTES = (
+    bytes.fromhex(
         "89504e470d0a1a0a0000000d49484452000000010000000108060000001f15c4890000000a49444154789c6360000002000100"
         "05fe02fea7b1b8000000004945"
-) + b"NDAE\xae\x42\x60\x82"
+    )
+    + b"NDAE\xae\x42\x60\x82"
+)
 
 
 @pytest.mark.asyncio

tests/services/test_quality.py

@@ -35,9 +35,7 @@ def test_quality_analysis_accepts_clean_transcription():
 
 @pytest.mark.unit
 def test_quality_warning_payload_is_versioned():
-    payload = quality_warning_payload(
-        analyze_transcription_quality("[document body typeset]\nBroken \ufffd")
-    )
+    payload = quality_warning_payload(analyze_transcription_quality("[document body typeset]\nBroken \ufffd"))
 
     assert payload["schema_name"] == "transcription.quality-warnings"
     assert payload["schema_version"] == "1"

tests/services/test_store.py

@@ -73,10 +73,7 @@ async def test_create_job_for_document_sorts_sources_and_creates_links(async_ses
     assert all(source.filename.endswith(".pdf") for source in sources)
     assert all("A_page" not in source.filename and "b_page" not in source.filename for source in sources)
     assert all(Path(source.filename).stem == str(source.id) for source in sources)
-    assert all(
-        source.file_path == f"documents/{document.id}/{source.filename}"
-        for source in sources
-    )
+    assert all(source.file_path == f"documents/{document.id}/{source.filename}" for source in sources)
     assert [source.file_hash for source in sources] == [
         "ca978112ca1bbdcafac231b39a23dc4da786eff8147c4e72b9807785afee48bb",
         "3e23e8160039594a33894f6564e1b1348bbd7a0088d42c4acb73eeaed59c009d",

tests/services/test_transcription_external.py

@@ -43,13 +43,7 @@ pytestmark = [
 def _real_image_paths() -> list[Path]:
     if not REAL_IMAGES_DIR.exists():
         return []
-    return sorted(
-        [
-            p
-            for p in REAL_IMAGES_DIR.iterdir()
-            if p.is_file() and p.suffix.lower() in SUPPORTED_EXTENSIONS
-        ]
-    )
+    return sorted([p for p in REAL_IMAGES_DIR.iterdir() if p.is_file() and p.suffix.lower() in SUPPORTED_EXTENSIONS])
 
 
 def _artifact_filename(image_path: Path) -> str:
@@ -78,11 +72,7 @@ class TestRealImageExternalTranscription:
         ARTIFACTS_DIR.mkdir(parents=True, exist_ok=True)
         artifact_path = ARTIFACTS_DIR / _artifact_filename(image_path)
         artifact_text = (
-            f"source: {image_path.name}\n"
-            f"provider: {result.provider}\n"
-            f"model: {result.model}\n"
-            "---\n"
-            f"{result.text}\n"
+            f"source: {image_path.name}\nprovider: {result.provider}\nmodel: {result.model}\n---\n{result.text}\n"
         )
         artifact_path.write_text(artifact_text, encoding="utf-8")
         assert artifact_path.exists()

tests/services/test_workflows_reliability.py

@@ -107,15 +107,12 @@ class TestWorkflowReliability:
 
         async with services.jobs._session_scope() as session:
             attempts = (
-                (
                 await session.exec(
                     select(ExecutionAttempt).where(
                         col(ExecutionAttempt.job_source_id).in_([js.id for js in result.job_sources])
                     )
                 )
-                )
-                .all()
-            )
+            ).all()
         error_detail = next(attempt.error_detail for attempt in attempts if attempt.error_detail is not None)
         assert "timed out" in error_detail.lower()
         assert "20.0s" in error_detail
@@ -179,15 +176,12 @@ class TestWorkflowReliability:
 
         async with services.jobs._session_scope() as session:
             attempts = (
-                (
                 await session.exec(
                     select(ExecutionAttempt).where(
                         col(ExecutionAttempt.job_source_id).in_([js.id for js in result.job_sources])
                     )
                 )
-                )
-                .all()
-            )
+            ).all()
         assert len(attempts) == 1
         duration_ms = attempts[0].duration_ms
 
@@ -242,15 +236,12 @@ class TestWorkflowReliability:
 
         async with services.jobs._session_scope() as session:
             attempts = (
-                (
                 await session.exec(
                     select(ExecutionAttempt).where(
                         col(ExecutionAttempt.job_source_id).in_([js.id for js in result.job_sources])
                     )
                 )
-                )
-                .all()
-            )
+            ).all()
         assert len(attempts) == 1
         attempt = attempts[0]
         timing = (attempt.normalized_metadata or {}).get("processing_timing")

tests/test_db.py

@@ -163,16 +163,13 @@ async def test_create_all_declares_hot_path_indexes(tmp_path):
                     )
                 }
                 job_source_unique = [
-                    constraint["column_names"]
-                    for constraint in database.get_unique_constraints("job_source")
+                    constraint["column_names"] for constraint in database.get_unique_constraints("job_source")
                 ]
                 document_tag_unique = [
-                    constraint["column_names"]
-                    for constraint in database.get_unique_constraints("document_tag")
+                    constraint["column_names"] for constraint in database.get_unique_constraints("document_tag")
                 ]
                 person_tag_unique = [
-                    constraint["column_names"]
-                    for constraint in database.get_unique_constraints("person_tag")
+                    constraint["column_names"] for constraint in database.get_unique_constraints("person_tag")
                 ]
                 return indexes, job_source_unique, document_tag_unique, person_tag_unique
 
@@ -252,11 +249,11 @@ async def test_reconcile_legacy_job_source_columns_drops_executed_at(tmp_path):
             await connection.execute(
                 text(
                     'create table "job_source" ('
-                    'id char(32) not null primary key, '
-                    'job_id char(32) not null, '
-                    'source_id char(32) not null, '
-                    'status varchar(11) not null, '
-                    'executed_at datetime not null, '
+                    "id char(32) not null primary key, "
+                    "job_id char(32) not null, "
+                    "source_id char(32) not null, "
+                    "status varchar(11) not null, "
+                    "executed_at datetime not null, "
                     'constraint "uq_job_source_job_source" unique ("job_id", "source_id"), '
                     'foreign key("job_id") references "job" ("id"), '
                     'foreign key("source_id") references "source" ("id")'
@@ -291,14 +288,14 @@ async def test_reconcile_canonical_media_paths_normalizes_source_and_photo_paths
             await connection.execute(
                 text(
                     'insert into "person" (id, given_names, last_name, created_at, updated_at) '
-                    'values (:id, :given_names, :last_name, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)'
+                    "values (:id, :given_names, :last_name, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)"
                 ),
                 {"id": "11" * 16, "given_names": "Portrait", "last_name": "Person"},
             )
             await connection.execute(
                 text(
                     'insert into "photo" (id, person_id, path, is_primary, created_at, updated_at) '
-                    'values (:id, :person_id, :path, :is_primary, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)'
+                    "values (:id, :person_id, :path, :is_primary, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)"
                 ),
                 {
                     "id": "44" * 16,
@@ -310,7 +307,7 @@ async def test_reconcile_canonical_media_paths_normalizes_source_and_photo_paths
             await connection.execute(
                 text(
                     'insert into "document" (id, name, created_at, updated_at) '
-                    'values (:id, :name, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)'
+                    "values (:id, :name, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)"
                 ),
                 {"id": "22" * 16, "name": "Doc"},
             )

tests/test_meta_contract_guards.py

@@ -109,10 +109,7 @@ def _declared_env_example_keys(*, include_commented: bool) -> set[str]:
 
 def _active_env_example_values() -> dict[str, str]:
     text = _read(".env.example")
-    return {
-        key: value.strip()
-        for key, value in re.findall(r"^\s*([A-Z0-9_]+)\s*=\s*(.*)$", text, flags=re.MULTILINE)
-    }
+    return {key: value.strip() for key, value in re.findall(r"^\s*([A-Z0-9_]+)\s*=\s*(.*)$", text, flags=re.MULTILINE)}
 
 
 def _normalize_env_path_value(value: str | None) -> str | None:

tests/test_orphan_sweep.py

@@ -57,9 +57,7 @@ def _source_files() -> list[Path]:
 
 
 def _is_registered_with_framework(node: ast.FunctionDef | ast.AsyncFunctionDef | ast.ClassDef) -> bool:
-    return any(
-        ast.unparse(decorator).startswith(REGISTRATION_DECORATOR_PREFIXES) for decorator in node.decorator_list
-    )
+    return any(ast.unparse(decorator).startswith(REGISTRATION_DECORATOR_PREFIXES) for decorator in node.decorator_list)
 
 
 def _public_definitions() -> dict[str, str]:

tests/test_storage_reconciliation.py

@@ -36,7 +36,12 @@ async def _document_ids(settings: Settings) -> set[str]:
 async def _source_counts_by_document(settings: Settings) -> dict[str, int]:
     async with session_scope(settings=settings) as session:
         rows = await session.exec(
-            select(Source.document_id, func.count(Source.id)).group_by(Source.document_id)
+            select(
+                Source.document_id,
+                func.count(Source.id),  # ty: ignore[invalid-argument-type] - SQLAlchemy descriptor false positive.
+            ).group_by(
+                Source.document_id  # ty: ignore[invalid-argument-type] - SQLAlchemy descriptor false positive.
+            )
         )
         return {str(document_id): int(count) for document_id, count in rows}
 
@@ -93,9 +98,7 @@ async def assert_storage_reconciliation(*, upload_dir: Path, settings: Settings)
     if mismatches:
         report = "\n".join(f"- {item}" for item in mismatches)
         raise AssertionError(
-            "Storage reconciliation mismatch(es) detected.\n"
-            f"Reconciling item count: {len(mismatches)}\n"
-            f"{report}"
+            f"Storage reconciliation mismatch(es) detected.\nReconciling item count: {len(mismatches)}\n{report}"
         )
 
 

tests/test_ui_boundaries.py

@@ -129,11 +129,7 @@ def _notifies_negative(tree: ast.Module) -> bool:
         ):
             continue
         for keyword in node.keywords:
-            if (
-                keyword.arg == "type"
-                and isinstance(keyword.value, ast.Constant)
-                and keyword.value.value == "negative"
-            ):
+            if keyword.arg == "type" and isinstance(keyword.value, ast.Constant) and keyword.value.value == "negative":
                 return True
     return False
 
@@ -149,6 +145,4 @@ def test_no_page_hand_rolls_error_notifications():
     offenders = sorted(
         path.stem for path in _page_paths() if _notifies_negative(ast.parse(path.read_text(encoding="utf-8")))
     )
-    assert offenders == [], (
-        f"Pages must render errors via error_presenter.show_error, not ui.notify: {offenders}"
-    )
+    assert offenders == [], f"Pages must render errors via error_presenter.show_error, not ui.notify: {offenders}"

tests/tools/test_export_import_migration.py

@@ -153,7 +153,7 @@ def test_export_import_migration_backfills_legacy_portraits_and_homepage_images(
             connection.execute(
                 text(
                     'insert into "person" (id, full_name, portrait_path, created_at, updated_at) '
-                    'values (:id, :full_name, :portrait_path, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)'
+                    "values (:id, :full_name, :portrait_path, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)"
                 ),
                 {"id": person_id, "full_name": "Legacy Portrait", "portrait_path": "persons/legacy/portrait.png"},
             )
@@ -264,7 +264,7 @@ def test_migration_backfills_legacy_media_when_photo_table_contains_stale_rows(t
             connection.execute(
                 text(
                     'insert into "person" (id, full_name, portrait_path, created_at, updated_at) '
-                    'values (:id, :full_name, :portrait_path, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)'
+                    "values (:id, :full_name, :portrait_path, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)"
                 ),
                 {"id": person_id, "full_name": "Legacy Portrait", "portrait_path": "persons/legacy/portrait.png"},
             )

tests/ui/conftest.py

@@ -56,8 +56,7 @@ def app_client(tmp_path_factory: pytest.TempPathFactory) -> Generator[tuple[Fast
         expected_path = Path(database.path).resolve()
         if runtime_path != expected_path:
             raise RuntimeError(
-                "Refusing to initialize destructive UI fixtures against "
-                f"{runtime_path}; expected {expected_path}"
+                f"Refusing to initialize destructive UI fixtures against {runtime_path}; expected {expected_path}"
             )
         yield app, client
 

tests/ui/test_formatters.py

@@ -30,12 +30,8 @@ def test_person_selector_label_disambiguates_without_changing_identity():
 
 
 def test_family_search_url_uses_fixed_person_details_route():
-    assert family_search_url("G8T4-MDQ") == (
-        "https://www.familysearch.org/tree/person/details/G8T4-MDQ"
-    )
+    assert family_search_url("G8T4-MDQ") == ("https://www.familysearch.org/tree/person/details/G8T4-MDQ")
 
 
 def test_google_maps_search_url_encodes_place_query():
-    assert google_maps_search_url("New York, NY") == (
-        "https://www.google.com/maps/search/?api=1&query=New+York%2C+NY"
-    )
+    assert google_maps_search_url("New York, NY") == ("https://www.google.com/maps/search/?api=1&query=New+York%2C+NY")

tests/ui/test_jobs_page.py

@@ -184,9 +184,7 @@ class TestJobsPageRendering:
         assert "Delete is blocked while the job is processing." in response.text
 
     @pytest.mark.asyncio
-    async def test_job_delete_page_allows_deletion_for_queued_job(
-        self, app_client, seed_document_with_unlinked_job
-    ):
+    async def test_job_delete_page_allows_deletion_for_queued_job(self, app_client, seed_document_with_unlinked_job):
         _, client = app_client
         _, job_id = seed_document_with_unlinked_job
 

tests/ui/test_upload_page.py

@@ -31,7 +31,7 @@ class TestPageRendering:
         assert response.status_code == 200
         assert "Home" in response.text
         assert "Edit Home Page" in response.text
-        assert '/homepage' in response.text
+        assert "/homepage" in response.text
 
     def test_homepage_edit_page_renders(self, app_client):
         """GET /ui/homepage/edit renders the edit page."""

tools/migrate_v46_to_v47.py

@@ -132,9 +132,7 @@ def _preflight(connection: Connection, *, strict: bool) -> None:
 def rotate_stored_images(connection: Connection, *, dry_run: bool) -> int:
     """Step 1: rewrite every mis-oriented stored image and its recorded digest."""
     source = SQLModel.metadata.tables["source"]
-    rows = connection.execute(
-        select(source.c.id, source.c.file_path, source.c.filename)
-    ).all()
+    rows = connection.execute(select(source.c.id, source.c.file_path, source.c.filename)).all()
 
     rotated = 0
     missing = 0

tools/migrate_v47_to_v48.py

@@ -55,10 +55,7 @@ def _uniqueness_already_enforced(connection: Connection) -> bool:
     if any(constraint.get("name") == UNIQUE_NAME for constraint in unique_constraints):
         return True
     indexes = inspector.get_indexes("job_source")
-    return any(
-        index.get("name") == UNIQUE_NAME and index.get("unique") is True
-        for index in indexes
-    )
+    return any(index.get("name") == UNIQUE_NAME and index.get("unique") is True for index in indexes)
 
 
 def _choose_keeper(
@@ -100,11 +97,7 @@ def deduplicate_job_source_membership(connection: Connection, *, dry_run: bool)
                 job_source.c.source_id,
                 func.max(execution_attempt.c.created_at).label("latest_attempt_at"),
             )
-            .select_from(
-                job_source.outerjoin(
-                    execution_attempt, execution_attempt.c.job_source_id == job_source.c.id
-                )
-            )
+            .select_from(job_source.outerjoin(execution_attempt, execution_attempt.c.job_source_id == job_source.c.id))
             .group_by(job_source.c.id, job_source.c.job_id, job_source.c.source_id)
         )
         .mappings()
@@ -150,13 +143,9 @@ def add_job_source_uniqueness(connection: Connection, *, dry_run: bool) -> None:
 
     dialect = connection.dialect.name
     if dialect == "postgresql":
-        statement = text(
-            f'alter table "job_source" add constraint "{UNIQUE_NAME}" unique ("job_id", "source_id")'
-        )
+        statement = text(f'alter table "job_source" add constraint "{UNIQUE_NAME}" unique ("job_id", "source_id")')
     else:
-        statement = text(
-            f'create unique index "{UNIQUE_NAME}" on "job_source" ("job_id", "source_id")'
-        )
+        statement = text(f'create unique index "{UNIQUE_NAME}" on "job_source" ("job_id", "source_id")')
     print(f"  applying {UNIQUE_NAME}")
     if not dry_run:
         connection.execute(statement)