Jobs: big jobs stuck in queue. Added Cancel, Resubmit

docs/schema_v2.md

@@ -50,7 +50,7 @@ erDiagram
     JOB {
         UUID id PK
         UUID document_id FK
-        VARCHAR status "queued | processing | completed | partial_success | failed"
+        VARCHAR status "queued | processing | transcribed | completed | partial_success | failed"
         INTEGER retry_count
         TEXT provider
         TEXT model
@@ -97,6 +97,7 @@ erDiagram
 ### Page-Level Execution & AI Outputs
 
 * Execution Granularity: Every single image execution by an AI model produces a dedicated record in job_source.
+* Source vs Execution Status: `source` does not carry a `status` column. Per-source execution state is tracked in `job_source.status` (`pending`, `transcribed`, `failed`).
 * Point-in-Time Auditability: job_source.raw_api_response stores the unparsed REST response envelope for that specific image page call. job_source.ai_metadata stores spatial bounding boxes, token usage, and layout details for that specific image page call.
 * Active Output Caching: Upon successful completion of an image call, source.raw_transcription is updated with the latest output string from job_source.raw_transcription for fast UI rendering.
 

src/transcription/services/jobs.py

@@ -11,6 +11,7 @@ from ..errors import AppError
 from ..errors import ErrorCategory
 from ..db.models import Job
 from ..db.models import JobSource
+from ..db.models import JobSourceStatus
 from ..db.models import JobStatus
 from ..db.models import Source
 from .base import ServiceBase
@@ -20,6 +21,14 @@ class JobDeleteBlockedError(AppError):
     """Raised when a job delete operation is blocked by lifecycle policy."""
 
 
+class JobCancelBlockedError(AppError):
+    """Raised when a job cancel operation is blocked by lifecycle policy."""
+
+
+class JobResubmitBlockedError(AppError):
+    """Raised when a job resubmit operation is blocked by lifecycle policy."""
+
+
 class JobService(ServiceBase):
     """Thin service class for managing jobs in the database."""
 
@@ -221,3 +230,87 @@ class JobService(ServiceBase):
 
             await _session.delete(job)
             await self._finalize(session=_session, caller_session=session)
+
+    async def cancel_job(self, *, job_id: UUID, session: AsyncSession | None = None) -> Job:
+        """Cancel a queued/processing job and stop remaining source work."""
+        async with self._session_scope(session) as _session:
+            query = (
+                select(Job)
+                .options(
+                    selectinload(Job.job_sources).selectinload(JobSource.source),  # pyright: ignore[reportArgumentType]
+                )
+                .where(Job.id == job_id)
+                .execution_options(populate_existing=True)
+            )
+            job = (await _session.exec(query)).first()
+            if job is None:
+                raise ValueError(f"Job with id {job_id} not found")
+
+            if job.status in {JobStatus.TRANSCRIBED, JobStatus.COMPLETED}:
+                raise JobCancelBlockedError(
+                    "Job cancel is not allowed for transcribed/completed jobs",
+                    category=ErrorCategory.VALIDATION,
+                    suggestion="Use resubmit for reprocessing needs, or leave the terminal job unchanged.",
+                )
+
+            now = datetime.now(UTC)
+            job.status = JobStatus.FAILED
+            job.date_updated = now
+
+            for job_source in job.job_sources:
+                if job_source.status == JobSourceStatus.TRANSCRIBED:
+                    continue
+                job_source.status = JobSourceStatus.FAILED
+                job_source.raw_transcription = None
+                job_source.error_detail = "Cancelled by user"
+                job_source.executed_at = now
+                if job_source.source is not None:
+                    job_source.source.raw_transcription = None
+
+            await self._finalize(session=_session, caller_session=session, refresh=(job,))
+            return job
+
+    async def resubmit_non_transcribed_sources(self, *, job_id: UUID, session: AsyncSession | None = None) -> int:
+        """Reset non-transcribed source executions and queue the job for reprocessing."""
+        async with self._session_scope(session) as _session:
+            query = (
+                select(Job)
+                .options(
+                    selectinload(Job.job_sources).selectinload(JobSource.source),  # pyright: ignore[reportArgumentType]
+                )
+                .where(Job.id == job_id)
+                .execution_options(populate_existing=True)
+            )
+            job = (await _session.exec(query)).first()
+            if job is None:
+                raise ValueError(f"Job with id {job_id} not found")
+
+            if job.status == JobStatus.PROCESSING:
+                raise JobResubmitBlockedError(
+                    "Job resubmit is blocked while processing is active",
+                    category=ErrorCategory.VALIDATION,
+                    suggestion="Cancel processing first, then resubmit remaining sources.",
+                )
+
+            candidates = [job_source for job_source in job.job_sources if job_source.status != JobSourceStatus.TRANSCRIBED]
+            if not candidates:
+                raise JobResubmitBlockedError(
+                    "Job has no non-transcribed sources to resubmit",
+                    category=ErrorCategory.VALIDATION,
+                    suggestion="Only failed or pending sources can be resubmitted.",
+                )
+
+            now = datetime.now(UTC)
+            for job_source in candidates:
+                job_source.status = JobSourceStatus.PENDING
+                job_source.raw_transcription = None
+                job_source.error_detail = None
+                job_source.executed_at = now
+                if job_source.source is not None:
+                    job_source.source.raw_transcription = None
+
+            job.status = JobStatus.QUEUED
+            job.date_updated = now
+
+            await self._finalize(session=_session, caller_session=session, refresh=(job,))
+            return len(candidates)

src/transcription/services/transcription.py

@@ -117,6 +117,33 @@ class TranscriptionService(ServiceBase):
             await _session.delete(source)
             await self._finalize(session=_session, caller_session=session)
 
+    async def delete_unlinked_source(self, *, source_id: UUID, session: AsyncSession | None = None) -> None:
+        """Delete a source only when no JobSource links exist."""
+        async with self._session_scope(session) as _session:
+            source = await _session.get(
+                Source,
+                source_id,
+                options=(
+                    selectinload(Source.job_sources),  # pyright: ignore[reportArgumentType]
+                ),
+            )
+            if source is None:
+                raise TranscriptionNotFoundError(
+                    f"Source with id {source_id} not found",
+                    category=ErrorCategory.NOT_FOUND,
+                    suggestion="Verify the source id and retry.",
+                )
+
+            if source.job_sources:
+                raise SourceDeleteBlockedError(
+                    "Source delete blocked because it is linked to one or more jobs",
+                    category=ErrorCategory.VALIDATION,
+                    suggestion="Remove JobSource links first, then retry deletion.",
+                )
+
+            await _session.delete(source)
+            await self._finalize(session=_session, caller_session=session)
+
     async def list_sources(
         self,
         *,
@@ -292,7 +319,11 @@ class TranscriptionService(ServiceBase):
         prompt_name: str = DEFAULT_PROMPT_FILE,
         session: AsyncSession | None = None,
     ) -> Job:
-        """Persist original transcription output fields on a job."""
+        """Persist transcription output for the first ordered source in a job's document.
+
+        This compatibility helper keeps legacy single-source workflows working.
+        New multi-source flows should use ``update_job_source_transcription``.
+        """
         async with self._session_scope(session) as _session:
             job = await _session.get(Job, job_id)
             if job is None:
@@ -314,28 +345,86 @@ class TranscriptionService(ServiceBase):
             )
             source_row = source.first()
             if source_row is not None:
-                existing_job_source = await _session.exec(
+                await self.update_job_source_transcription(
-                    select(JobSource).where(JobSource.job_id == job.id).where(JobSource.source_id == source_row.id)
+                    job_id=job.id,
+                    source_id=source_row.id,
+                    text=text,
+                    error_detail=error_detail,
+                    provider=provider,
+                    model=model,
+                    prompt_name=prompt_name,
+                    session=_session,
                 )
-                job_source = existing_job_source.first()
-                if job_source is None:
-                    job_source = JobSource(
-                        job_id=job.id,
-                        source_id=source_row.id,
-                        status=JobSourceStatus.TRANSCRIBED if text is not None else JobSourceStatus.FAILED,
-                        raw_transcription=text,
-                        error_detail=error_detail,
-                    )
-                    _session.add(job_source)
-                else:
-                    job_source.raw_transcription = text
-                    job_source.error_detail = error_detail
-                    job_source.status = JobSourceStatus.TRANSCRIBED if text is not None else JobSourceStatus.FAILED
-                    job_source.executed_at = datetime.now(UTC)
 
             await self._finalize(session=_session, caller_session=session, refresh=(job,))
             return job
 
+    async def update_job_source_transcription(
+        self,
+        *,
+        job_id: UUID,
+        source_id: UUID,
+        text: str | None,
+        error_detail: str | None = None,
+        provider: str | None = None,
+        model: str | None = None,
+        prompt_name: str = DEFAULT_PROMPT_FILE,
+        session: AsyncSession | None = None,
+    ) -> JobSource:
+        """Persist transcription fields for one source within a specific job."""
+        async with self._session_scope(session) as _session:
+            job = await _session.get(Job, job_id)
+            if job is None:
+                raise TranscriptionNotFoundError(
+                    f"Job with id {job_id} not found",
+                    category=ErrorCategory.NOT_FOUND,
+                    suggestion="Verify the job id and retry.",
+                )
+
+            source = await _session.get(Source, source_id)
+            if source is None:
+                raise TranscriptionNotFoundError(
+                    f"Source with id {source_id} not found",
+                    category=ErrorCategory.NOT_FOUND,
+                    suggestion="Verify the source id and retry.",
+                )
+
+            if source.document_id != job.document_id:
+                raise TranscriptionError(
+                    f"Source {source_id} does not belong to job {job_id}",
+                    category=ErrorCategory.VALIDATION,
+                    suggestion="Link the source to the same document as the job and retry.",
+                )
+
+            job.provider = provider or job.provider or self.settings.provider.value
+            job.model = model or job.model or _resolve_transcript_model(provider=self.provider, settings=self.settings)
+            job.prompt_name = prompt_name or job.prompt_name or DEFAULT_PROMPT_FILE
+            job.date_updated = datetime.now(UTC)
+
+            source.raw_transcription = text
+
+            existing_job_source = await _session.exec(
+                select(JobSource).where(JobSource.job_id == job_id).where(JobSource.source_id == source_id)
+            )
+            job_source = existing_job_source.first()
+            if job_source is None:
+                job_source = JobSource(
+                    job_id=job_id,
+                    source_id=source_id,
+                    status=JobSourceStatus.TRANSCRIBED if text is not None else JobSourceStatus.FAILED,
+                    raw_transcription=text,
+                    error_detail=error_detail,
+                )
+                _session.add(job_source)
+            else:
+                job_source.raw_transcription = text
+                job_source.error_detail = error_detail
+                job_source.status = JobSourceStatus.TRANSCRIBED if text is not None else JobSourceStatus.FAILED
+                job_source.executed_at = datetime.now(UTC)
+
+            await self._finalize(session=_session, caller_session=session, refresh=(job, source, job_source))
+            return job_source
+
     async def upsert_revision_for_source(
         self,
         *,

src/transcription/services/workflows.py

@@ -6,6 +6,7 @@ from sqlmodel.ext.asyncio.session import AsyncSession
 from ..config import Settings
 from ..config import get_settings
 from ..db.models import Job
+from ..db.models import JobSourceStatus
 from ..db.models import JobStatus
 from ..db.models import Source
 from ..errors import AppError
@@ -69,88 +70,114 @@ async def process_queued_job(
         await session.commit()
 
     source_job = await services.jobs.read_job(job_id=job.id, session=session)
-    source = _resolve_primary_source(source_job)
+    sources = _resolve_job_sources(source_job)
-    if source is None:
+    if not sources and not source_job.job_sources:
         candidate_sources = await services.transcriptions.list_sources(document_id=job.document_id, session=session)
-        source = next(iter(sorted(candidate_sources, key=lambda item: item.page_number)), None)
+        sources = list(sorted(candidate_sources, key=lambda item: (item.page_number, item.upload_name.casefold())))
 
-    if source is None:
+    if not sources:
-        error = AppError(
+        return await services.jobs.mark_job_status(job.id, JobStatus.TRANSCRIBED, session=session)
-            f"Job {job.id} has no associated source record.",
-            category=ErrorCategory.VALIDATION,
-            suggestion="Attach at least one source to the job and retry.",
-        )
-        return await _finalize_failed(job=job, services=services, error=error, session=session)
-    started_at = asyncio.get_running_loop().time()
 
-    try:
+    successful_pages: list[tuple[Source, TranscriptionResult]] = []
-        result = await asyncio.wait_for(
+    failed_pages: list[tuple[Source, AppError]] = []
-            transcribe_document_image(source.file_path),
+    externally_stopped = False
-            timeout=runtime_settings.worker_provider_timeout_seconds,
-        )
-        elapsed_seconds = asyncio.get_running_loop().time() - started_at
-        logger.info(
-            "Provider response diagnostics operation=worker.provider_response "
-            "job_id=%s document_id=%s source_id=%s provider=%s model=%s "
-            "finish_reason=%s usage_input_tokens=%s usage_output_tokens=%s usage_total_tokens=%s "
-            "latency_seconds=%.3f text_chars=%s text_lines=%s",
-            job.id,
-            job.document_id,
-            source.id,
-            result.provider,
-            result.model,
-            result.finish_reason or "unknown",
-            result.usage_input_tokens,
-            result.usage_output_tokens,
-            result.usage_total_tokens,
-            elapsed_seconds,
-            len(result.text),
-            _line_count(result.text),
-        )
 
-        _validate_transcription_quality(result=result, settings=runtime_settings)
+    for source in sources:
+        if await _job_no_longer_processing(job_id=job.id, services=services, session=session):
+            externally_stopped = True
+            break
 
-        job = await _finalize_transcribed(job=job, services=services, result=result, session=session)
+        started_at = asyncio.get_running_loop().time()
-        logger.info(
+        try:
-            "Job transcribed operation=worker.process_job job_id=%s document_id=%s source_id=%s provider=%s",
+            result = await asyncio.wait_for(
-            job.id,
+                transcribe_document_image(source.file_path),
-            job.document_id,
+                timeout=runtime_settings.worker_provider_timeout_seconds,
-            source.id,
+            )
-            result.provider,
+            elapsed_seconds = asyncio.get_running_loop().time() - started_at
-        )
+            logger.info(
-    except TimeoutError:
+                "Provider response diagnostics operation=worker.provider_response "
-        error = AppError(
+                "job_id=%s document_id=%s source_id=%s provider=%s model=%s "
-            f"Provider call timed out after {runtime_settings.worker_provider_timeout_seconds:.1f}s",
+                "finish_reason=%s usage_input_tokens=%s usage_output_tokens=%s usage_total_tokens=%s "
-            category=ErrorCategory.EXTERNAL_PROVIDER,
+                "latency_seconds=%.3f text_chars=%s text_lines=%s",
-            suggestion="Retry the job. If this repeats, verify provider latency and request payload size.",
+                job.id,
-            retriable=True,
+                job.document_id,
-        )
+                source.id,
-        job = await _finalize_failed(job=job, services=services, error=error, session=session)
+                result.provider,
-        logger.error(
+                result.model,
-            "Job failed operation=worker.process_job job_id=%s document_id=%s source_id=%s error_id=%s category=%s",
+                result.finish_reason or "unknown",
-            job.id,
+                result.usage_input_tokens,
-            job.document_id,
+                result.usage_output_tokens,
-            source.id,
+                result.usage_total_tokens,
-            error.error_id,
+                elapsed_seconds,
-            error.category.value,
+                len(result.text),
-        )
+                _line_count(result.text),
-    except Exception as exc:  # noqa: BLE001
+            )
-        match exc:
-            case AppError() as error:
-                pass
-            case _:
-                error = classify_unexpected_error(exc, operation="worker.process_job")
 
-        job = await _finalize_failed(job=job, services=services, error=error, session=session)
+            _validate_transcription_quality(result=result, settings=runtime_settings)
-        logger.error(
+            successful_pages.append((source, result))
-            "Job failed operation=worker.process_job job_id=%s document_id=%s source_id=%s error_id=%s category=%s",
+        except TimeoutError:
-            job.id,
+            error = AppError(
-            job.document_id,
+                f"Provider call timed out after {runtime_settings.worker_provider_timeout_seconds:.1f}s",
-            source.id,
+                category=ErrorCategory.EXTERNAL_PROVIDER,
-            error.error_id,
+                suggestion="Retry the job. If this repeats, verify provider latency and request payload size.",
-            error.category.value,
+                retriable=True,
-        )
+            )
-    return job
+            failed_pages.append((source, error))
+            logger.error(
+                "Source failed operation=worker.process_job job_id=%s document_id=%s source_id=%s error_id=%s category=%s",
+                job.id,
+                job.document_id,
+                source.id,
+                error.error_id,
+                error.category.value,
+            )
+        except Exception as exc:  # noqa: BLE001
+            match exc:
+                case AppError() as error:
+                    pass
+                case _:
+                    error = classify_unexpected_error(exc, operation="worker.process_job")
+
+            failed_pages.append((source, error))
+            logger.error(
+                "Source failed operation=worker.process_job job_id=%s document_id=%s source_id=%s error_id=%s category=%s",
+                job.id,
+                job.document_id,
+                source.id,
+                error.error_id,
+                error.category.value,
+            )
+
+        if await _job_no_longer_processing(job_id=job.id, services=services, session=session):
+            externally_stopped = True
+            break
+
+    terminal_status = JobStatus.TRANSCRIBED
+    if externally_stopped:
+        terminal_status = JobStatus.FAILED
+    elif failed_pages and successful_pages:
+        terminal_status = JobStatus.PARTIAL_SUCCESS
+    elif failed_pages and not successful_pages:
+        terminal_status = JobStatus.FAILED
+
+    updated_job = await _finalize_batch_outcome(
+        job=job,
+        services=services,
+        successful_pages=successful_pages,
+        failed_pages=failed_pages,
+        status=terminal_status,
+        session=session,
+    )
+
+    logger.info(
+        "Job finished operation=worker.process_job job_id=%s document_id=%s status=%s success_pages=%s failed_pages=%s",
+        updated_job.id,
+        updated_job.document_id,
+        updated_job.status.value,
+        len(successful_pages),
+        len(failed_pages),
+    )
+    return updated_job
 
 
 async def process_next_queued_job(
@@ -307,6 +334,95 @@ def _resolve_primary_source(job: Job) -> Source | None:
     return next((job_source.source for job_source in job.job_sources if job_source.source is not None), None)
 
 
+def _resolve_job_sources(job: Job) -> list[Source]:
+    """Resolve non-transcribed linked sources for a job in deterministic page order."""
+    if not job.job_sources:
+        return []
+
+    sources = [
+        job_source.source
+        for job_source in job.job_sources
+        if job_source.source is not None and job_source.status != JobSourceStatus.TRANSCRIBED
+    ]
+    return list(sorted(sources, key=lambda item: (item.page_number, item.upload_name.casefold())))
+
+
+async def _job_no_longer_processing(
+    *,
+    job_id,
+    services: ServiceBundle,
+    session: AsyncSession | None = None,
+) -> bool:
+    """Return True when job status changed externally from PROCESSING."""
+    latest_job = await services.jobs.read_job(job_id=job_id, session=session)
+    return latest_job.status != JobStatus.PROCESSING
+
+
+async def _finalize_batch_outcome(
+    *,
+    job: Job,
+    services: ServiceBundle,
+    successful_pages: list[tuple[Source, TranscriptionResult]],
+    failed_pages: list[tuple[Source, AppError]],
+    status: JobStatus,
+    session: AsyncSession | None = None,
+) -> Job:
+    """Transaction B: write per-source outcomes and terminal job status atomically."""
+    if session is None:
+        async with services.jobs._session_scope() as local_session:
+            for source, result in successful_pages:
+                await services.transcriptions.update_job_source_transcription(
+                    job_id=job.id,
+                    source_id=source.id,
+                    text=result.text,
+                    error_detail=None,
+                    provider=result.provider,
+                    model=result.model,
+                    prompt_name=result.prompt_name,
+                    session=local_session,
+                )
+
+            for source, error in failed_pages:
+                await services.transcriptions.update_job_source_transcription(
+                    job_id=job.id,
+                    source_id=source.id,
+                    text=None,
+                    error_detail=format_error_detail(error),
+                    prompt_name=DEFAULT_PROMPT_FILE,
+                    session=local_session,
+                )
+
+            updated_job = await services.jobs.mark_job_status(job.id, status, session=local_session)
+            await local_session.commit()
+            return updated_job
+
+    for source, result in successful_pages:
+        await services.transcriptions.update_job_source_transcription(
+            job_id=job.id,
+            source_id=source.id,
+            text=result.text,
+            error_detail=None,
+            provider=result.provider,
+            model=result.model,
+            prompt_name=result.prompt_name,
+            session=session,
+        )
+
+    for source, error in failed_pages:
+        await services.transcriptions.update_job_source_transcription(
+            job_id=job.id,
+            source_id=source.id,
+            text=None,
+            error_detail=format_error_detail(error),
+            prompt_name=DEFAULT_PROMPT_FILE,
+            session=session,
+        )
+
+    updated_job = await services.jobs.mark_job_status(job.id, status, session=session)
+    await session.commit()
+    return updated_job
+
+
 def _validate_transcription_quality(*, result: TranscriptionResult, settings: Settings) -> None:
     text_chars = len(result.text)
     text_lines = _line_count(result.text)

src/transcription/ui/pages/jobs_page.py

@@ -8,10 +8,13 @@ from uuid import UUID
 from fastapi import Request
 from nicegui import ui
 
+from transcription.db.models import JobSourceStatus
 from transcription.db.models import JobStatus
 from transcription.db.session import session_scope
 from transcription.services.documents import DocumentService
 from transcription.services.jobs import JobDeleteBlockedError
+from transcription.services.jobs import JobCancelBlockedError
+from transcription.services.jobs import JobResubmitBlockedError
 from transcription.services.jobs import JobService
 from transcription.services.store import create_job_for_document
 from transcription.ui.components.app_shell import render_navigation_header
@@ -203,7 +206,7 @@ def register_page() -> None:  # noqa: PLR0915
                 ui.button("Back to Jobs", on_click=lambda: ui.navigate.to("/jobs"), icon="arrow_back").props("flat")
 
     @ui.page("/jobs/{job_id}")
-    async def job_detail_page(job_id: str, session_factory: SessionFactoryDep) -> None:
+    async def job_detail_page(job_id: str, request: Request, session_factory: SessionFactoryDep) -> None:
         apply_archival_theme()
         jobs_service = JobService(session_factory=session_factory)
         render_navigation_header(current_path="/jobs")
@@ -225,6 +228,20 @@ def register_page() -> None:  # noqa: PLR0915
                 page_header(f"Job Record: {job.id}")
                 with ui.row().classes("items-center gap-2"):
                     archival_badge(job.status.value.upper())
+
+                    if job.status in {JobStatus.QUEUED, JobStatus.PROCESSING}:
+                        destructive_button(
+                            "Cancel",
+                            on_click=lambda: ui.navigate.to(f"/jobs/{job.id}/cancel"),
+                            icon="stop_circle",
+                            extra_classes="text-xs",
+                        )
+
+                    if job.status != JobStatus.TRANSCRIBED:
+                        ui.button("Resubmit", on_click=lambda: ui.navigate.to(f"/jobs/{job.id}/resubmit"), icon="replay").props(
+                            "outlined"
+                        ).classes("text-xs")
+
                     destructive_button(
                         "Delete Job",
                         on_click=lambda: ui.navigate.to(f"/jobs/{job.id}/delete"),
@@ -254,6 +271,114 @@ def register_page() -> None:  # noqa: PLR0915
                             icon="description",
                         ).props("flat text-xs").classes("ui-link-primary w-full")
 
+    @ui.page("/jobs/{job_id}/cancel")
+    async def job_cancel_page(job_id: str, request: Request, session_factory: SessionFactoryDep) -> None:
+        apply_archival_theme()
+        jobs_service = JobService(session_factory=session_factory)
+        render_navigation_header(current_path="/jobs")
+
+        try:
+            parsed_job_id = UUID(job_id)
+        except ValueError:
+            ui.label("Invalid job id").classes("text-h6 text-red-800 p-4")
+            return
+
+        try:
+            job = await jobs_service.read_job(job_id=parsed_job_id)
+        except ValueError:
+            ui.label("Job not found").classes("text-h6 text-red-800 p-4")
+            return
+
+        with ui.column().classes("w-full max-w-xl mx-auto p-4 gap-4"):
+            page_header("Cancel Processing Job")
+
+            with archival_card(extra_classes="gap-2"):
+                ui.label(f"Job ID: {job.id}").classes("text-sm font-semibold font-mono ui-text-primary")
+                metadata_row("Current Status:", job.status.value)
+                ui.label("Cancel stops processing and marks remaining non-transcribed sources as failed.").classes(
+                    "text-xs ui-text-muted"
+                )
+
+            async def submit_cancel() -> None:
+                try:
+                    await jobs_service.cancel_job(job_id=job.id)
+                except JobCancelBlockedError as exc:
+                    ui.notify(exc.message, type="warning")
+                    return
+                except ValueError:
+                    ui.notify("Job not found.", type="warning")
+                    ui.navigate.to("/jobs")
+                    return
+                except Exception as exc:  # noqa: BLE001
+                    show_error(exc, title="Cancel job failed", operation="jobs.cancel")
+                    return
+
+                resolve_worker_notifier(request.app.state).notify()
+                ui.notify("Job cancelled", type="positive")
+                ui.navigate.to(f"/jobs/{job.id}")
+
+            with ui.row().classes("w-full items-center gap-2 mt-2"):
+                destructive_button(
+                    "Cancel job",
+                    on_click=submit_cancel,
+                    icon="stop_circle",
+                    variant="solid",
+                )
+                ui.button("Back to Job", on_click=lambda: ui.navigate.to(f"/jobs/{job.id}"), icon="arrow_back").props("flat")
+
+    @ui.page("/jobs/{job_id}/resubmit")
+    async def job_resubmit_page(job_id: str, request: Request, session_factory: SessionFactoryDep) -> None:
+        apply_archival_theme()
+        jobs_service = JobService(session_factory=session_factory)
+        render_navigation_header(current_path="/jobs")
+
+        try:
+            parsed_job_id = UUID(job_id)
+        except ValueError:
+            ui.label("Invalid job id").classes("text-h6 text-red-800 p-4")
+            return
+
+        try:
+            job = await jobs_service.read_job(job_id=parsed_job_id)
+        except ValueError:
+            ui.label("Job not found").classes("text-h6 text-red-800 p-4")
+            return
+
+        non_transcribed_count = sum(1 for job_source in job.job_sources if job_source.status != JobSourceStatus.TRANSCRIBED)
+
+        with ui.column().classes("w-full max-w-xl mx-auto p-4 gap-4"):
+            page_header("Resubmit Job")
+
+            with archival_card(extra_classes="gap-2"):
+                ui.label(f"Job ID: {job.id}").classes("text-sm font-semibold font-mono ui-text-primary")
+                metadata_row("Current Status:", job.status.value)
+                metadata_row("Non-Transcribed Sources:", str(non_transcribed_count))
+                ui.label("Resubmit queues all non-transcribed linked sources. New results overwrite prior page-level results.").classes(
+                    "text-xs ui-text-muted"
+                )
+
+            async def submit_resubmit() -> None:
+                try:
+                    resubmitted_count = await jobs_service.resubmit_non_transcribed_sources(job_id=job.id)
+                except JobResubmitBlockedError as exc:
+                    ui.notify(exc.message, type="warning")
+                    return
+                except ValueError:
+                    ui.notify("Job not found.", type="warning")
+                    ui.navigate.to("/jobs")
+                    return
+                except Exception as exc:  # noqa: BLE001
+                    show_error(exc, title="Resubmit failed", operation="jobs.resubmit")
+                    return
+
+                resolve_worker_notifier(request.app.state).notify()
+                ui.notify(f"Resubmitted {resubmitted_count} source(s)", type="positive")
+                ui.navigate.to(f"/jobs/{job.id}")
+
+            with ui.row().classes("w-full items-center gap-2 mt-2"):
+                ui.button("Resubmit now", on_click=submit_resubmit, icon="replay").classes("ui-btn-primary")
+                ui.button("Back to Job", on_click=lambda: ui.navigate.to(f"/jobs/{job.id}"), icon="arrow_back").props("flat")
+
     @ui.page("/jobs/{job_id}/delete")
     async def job_delete_page(job_id: str, session_factory: SessionFactoryDep) -> None:
         apply_archival_theme()

src/transcription/ui/pages/sources_page.py

@@ -13,6 +13,7 @@ from transcription.db.models import JobSource, Source
 from transcription.services.documents import DocumentError, DocumentService
 from transcription.services.jobs import JobService
 from transcription.services.transcription import (
+    SourceDeleteBlockedError,
     TranscriptionNotFoundError,
     TranscriptionService,
 )
@@ -21,6 +22,7 @@ from transcription.ui.components.cards import archival_card
 from transcription.ui.components.data_display import metadata_row
 from transcription.ui.components.document_panzoom import render_document_panzoom
 from transcription.ui.components.error_presenter import show_error
+from transcription.ui.components.primitives import destructive_button
 from transcription.ui.components.primitives import section_header_row
 from transcription.ui.components.table.sources import SourceTableRow, render_sources_table
 from transcription.ui.theme import apply_archival_theme
@@ -133,20 +135,28 @@ def register_page() -> None:
             with section_header_row():
                 page_header(f"Source Page {source.page_number}: {source.upload_name}", subtitle=f"Source ID: {source.id}")
 
-                if back_path is not None:
+                with ui.row().classes("items-center gap-2"):
-                    back_label = (
+                    if back_path is not None:
-                        "Back to Document"
+                        back_label = (
-                        if "document_id" in request.query_params
+                            "Back to Document"
-                        else "Back to Job"
+                            if "document_id" in request.query_params
-                        if "job_id" in request.query_params
+                            else "Back to Job"
-                        else "Back to Sources"
+                            if "job_id" in request.query_params
-                    )
+                            else "Back to Sources"
-                    ui.button(back_label, on_click=lambda route=back_path: ui.navigate.to(route), icon="arrow_back").classes(
+                        )
-                        "ui-btn-primary text-xs"
+                        ui.button(back_label, on_click=lambda route=back_path: ui.navigate.to(route), icon="arrow_back").classes(
-                    )
+                            "ui-btn-primary text-xs"
-                else:
+                        )
-                    ui.button("Back to Sources", on_click=lambda: ui.navigate.to("/sources"), icon="arrow_back").props(
+                    else:
-                        "flat text-xs"
+                        ui.button("Back to Sources", on_click=lambda: ui.navigate.to("/sources"), icon="arrow_back").props(
+                            "flat text-xs"
+                        )
+
+                    destructive_button(
+                        "Delete Source",
+                        on_click=lambda: ui.navigate.to(f"/sources/{source.id}/delete{_back_query(request.query_params)}"),
+                        icon="delete",
+                        extra_classes="text-xs",
                     )
 
             with ui.grid().classes("w-full grid-cols-12 gap-4"):
@@ -196,6 +206,73 @@ def register_page() -> None:
                         with ui.row().classes("w-full items-center gap-2 mt-2"):
                             ui.button("Save Revision", on_click=save_revision, icon="save").classes("ui-btn-primary text-xs")
 
+    @ui.page("/sources/{source_id}/delete")
+    async def source_delete_page(source_id: str, request: Request, session_factory: SessionFactoryDep) -> None:
+        apply_archival_theme()
+        sources_service = TranscriptionService(session_factory=session_factory)
+        render_navigation_header(current_path="/sources")
+
+        try:
+            parsed_source_id = UUID(source_id)
+        except ValueError:
+            ui.label("Invalid source id").classes("text-h6 text-red-800 p-4")
+            return
+
+        try:
+            source = await sources_service.read_source_detail(source_id=parsed_source_id)
+        except TranscriptionNotFoundError:
+            ui.label("Source not found").classes("text-h6 text-red-800 p-4")
+            return
+        except Exception as exc:  # noqa: BLE001
+            show_error(exc, title="Load failed", operation="sources.delete.load")
+            return
+
+        back_path = _back_path_from_query(request.query_params) or "/sources"
+        next_sources_path = f"/sources{_back_query(request.query_params)}"
+        linked_count = len(source.job_sources)
+
+        with ui.column().classes("w-full max-w-xl mx-auto p-4 gap-4"):
+            page_header("Delete Source Record")
+
+            with archival_card(extra_classes="gap-2"):
+                metadata_row("Source ID:", str(source.id))
+                metadata_row("Upload Name:", source.upload_name)
+                metadata_row("Linked Jobs:", str(linked_count))
+
+                if linked_count > 0:
+                    ui.label("Delete is only available for unlinked sources.").classes("text-xs text-red-800 font-bold mt-2")
+                    ui.label("This source is linked to one or more jobs and cannot be deleted from this view.").classes(
+                        "text-xs ui-text-muted italic"
+                    )
+                else:
+                    ui.label("This action permanently deletes the source record.").classes("text-xs text-red-800 font-medium")
+
+            async def submit_delete() -> None:
+                try:
+                    await sources_service.delete_unlinked_source(source_id=source.id)
+                except SourceDeleteBlockedError as exc:
+                    ui.notify(exc.message, type="warning")
+                    return
+                except TranscriptionNotFoundError:
+                    ui.notify("Source not found.", type="warning")
+                    ui.navigate.to(next_sources_path)
+                    return
+                except Exception as exc:  # noqa: BLE001
+                    show_error(exc, title="Delete source failed", operation="sources.delete")
+                    return
+
+                ui.notify("Source deleted", type="positive")
+                ui.navigate.to(next_sources_path)
+
+            with ui.row().classes("w-full items-center gap-2 mt-2"):
+                destructive_button(
+                    "Delete source permanently",
+                    on_click=submit_delete,
+                    icon="delete_forever",
+                    variant="solid",
+                )
+                ui.button("Cancel", on_click=lambda route=back_path: ui.navigate.to(route), icon="arrow_back").props("flat")
+
     @ui.page("/documents/{document_id}/sources")
     async def document_sources_page(document_id: str) -> RedirectResponse:
         return RedirectResponse(url=f"/ui/sources?document_id={document_id}")

tests/integration/test_pipeline_flow.py

@@ -1,25 +1,49 @@
 """Integration tests for end-to-end upload and worker pipeline behavior."""
 
 from pathlib import Path
+from uuid import uuid4
 
 import pytest
 
 from transcription.config import Settings
+from transcription.db.models import Document
 from transcription.db.models import Job
+from transcription.db.models import JobSourceStatus
 from transcription.db.models import JobStatus
 from transcription.providers.base import TranscriptionResult
 from transcription.services import ServiceBundle
+from transcription.services.store import create_job_for_document
 from transcription.services.store import create_upload_job
 from transcription.services.workflows import advance_job
 
 
+def _build_services(default_session_factory) -> ServiceBundle:
+    services = ServiceBundle()
+    object.__setattr__(
+        services,
+        "documents",
+        services.documents.__class__(session_factory=default_session_factory),
+    )
+    object.__setattr__(
+        services,
+        "jobs",
+        services.jobs.__class__(session_factory=default_session_factory),
+    )
+    object.__setattr__(
+        services,
+        "transcriptions",
+        services.transcriptions.__class__(session_factory=default_session_factory),
+    )
+    return services
+
+
 @pytest.mark.integration
 class TestPipelineSuccessFlow:
     """Verify end-to-end success lifecycle behavior."""
 
     @pytest.mark.asyncio
     async def test_upload_then_worker_persists_transcribed_terminal_state(
-        self, async_session, tmp_path: Path, monkeypatch
+        self, async_session, default_session_factory, tmp_path: Path, monkeypatch
     ):
         """Upload followed by worker processing persists job transcription and transcribed status."""
         settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
@@ -59,12 +83,12 @@ class TestPipelineSuccessFlow:
             _fake_transcribe_document_image,
         )
 
-        services = ServiceBundle()
+        services = _build_services(default_session_factory)
-        queued_job = await services.jobs.read_next_queued_job(session=async_session)
+        queued_job = await services.jobs.read_job(job_id=upload_result.job_id, session=async_session)
         processed = queued_job is not None
         if queued_job is not None:
             await advance_job(job=queued_job, services=services, session=async_session)
-        job = await async_session.get(Job, upload_result.job_id)
+        job = await services.jobs.read_job(job_id=upload_result.job_id, session=async_session)
 
         assert processed is True
         assert job is not None
@@ -72,13 +96,212 @@ class TestPipelineSuccessFlow:
         assert any(job_source.raw_transcription == "Pipeline transcript" for job_source in job.job_sources)
         assert all(job_source.error_detail is None for job_source in job.job_sources)
 
+    @pytest.mark.asyncio
+    async def test_worker_transcribes_all_sources_for_multi_page_job(
+        self,
+        async_session,
+        default_session_factory,
+        tmp_path: Path,
+        monkeypatch,
+    ):
+        """Worker stores transcription output for every source linked to the queued job."""
+        settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
+        document = Document(id=uuid4(), name="multi-page-document")
+        async_session.add(document)
+        await async_session.commit()
+
+        create_result = await create_job_for_document(
+            document_id=document.id,
+            uploads=[
+                ("page-01.jpg", b"one"),
+                ("page-02.jpg", b"two"),
+                ("page-03.jpg", b"three"),
+            ],
+            session=async_session,
+            settings=settings,
+        )
+
+        async def _fake_transcribe_document_image(
+            image_path,
+            *,
+            prompt_name="transcribe_document.md",
+            settings=None,
+            provider=None,
+        ) -> TranscriptionResult:
+            page_name = Path(image_path).name
+            _ = (prompt_name, settings, provider)
+            return TranscriptionResult(
+                text=f"Transcript for {page_name}",
+                provider="openrouter",
+                model="test-model",
+                prompt_name="transcribe_document.md",
+            )
+
+        monkeypatch.setattr(
+            "transcription.services.workflows.transcribe_document_image",
+            _fake_transcribe_document_image,
+        )
+
+        services = _build_services(default_session_factory)
+        queued_job = await services.jobs.read_job(job_id=create_result.job_id, session=async_session)
+        assert queued_job is not None
+
+        await advance_job(job=queued_job, services=services, session=async_session)
+        job = await services.jobs.read_job(job_id=create_result.job_id, session=async_session)
+
+        assert job.status == JobStatus.TRANSCRIBED
+        assert len(job.job_sources) == 3
+        assert all(job_source.status == JobSourceStatus.TRANSCRIBED for job_source in job.job_sources)
+        assert all(job_source.raw_transcription for job_source in job.job_sources)
+        assert all(job_source.source is not None and job_source.source.raw_transcription for job_source in job.job_sources)
+
+    @pytest.mark.asyncio
+    async def test_worker_marks_partial_success_when_some_sources_fail(
+        self,
+        async_session,
+        default_session_factory,
+        tmp_path: Path,
+        monkeypatch,
+    ):
+        """Mixed page outcomes produce PARTIAL_SUCCESS and preserve per-source status."""
+        settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
+        document = Document(id=uuid4(), name="partial-page-document")
+        async_session.add(document)
+        await async_session.commit()
+
+        create_result = await create_job_for_document(
+            document_id=document.id,
+            uploads=[
+                ("page-01.jpg", b"one"),
+                ("page-02.jpg", b"two"),
+            ],
+            session=async_session,
+            settings=settings,
+        )
+
+        call_count = 0
+
+        async def _fake_transcribe_document_image(
+            image_path,
+            *,
+            prompt_name="transcribe_document.md",
+            settings=None,
+            provider=None,
+        ) -> TranscriptionResult:
+            nonlocal call_count
+            call_count += 1
+            _ = (prompt_name, settings, provider)
+            if call_count == 2:
+                raise RuntimeError("simulated page failure")
+            return TranscriptionResult(
+                text="Transcript for first page",
+                provider="openrouter",
+                model="test-model",
+                prompt_name="transcribe_document.md",
+            )
+
+        monkeypatch.setattr(
+            "transcription.services.workflows.transcribe_document_image",
+            _fake_transcribe_document_image,
+        )
+
+        services = _build_services(default_session_factory)
+        queued_job = await services.jobs.read_job(job_id=create_result.job_id, session=async_session)
+        assert queued_job is not None
+
+        await advance_job(job=queued_job, services=services, session=async_session)
+        job = await services.jobs.read_job(job_id=create_result.job_id, session=async_session)
+
+        assert job.status == JobStatus.PARTIAL_SUCCESS
+        assert len(job.job_sources) == 2
+        statuses = {job_source.status for job_source in job.job_sources}
+        assert statuses == {JobSourceStatus.TRANSCRIBED, JobSourceStatus.FAILED}
+        assert any(job_source.error_detail is not None for job_source in job.job_sources)
+
+    @pytest.mark.asyncio
+    async def test_worker_skips_already_transcribed_sources_on_resubmit(
+        self,
+        async_session,
+        default_session_factory,
+        tmp_path: Path,
+        monkeypatch,
+    ):
+        """Queued jobs only process non-transcribed JobSource records."""
+        settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
+        document = Document(id=uuid4(), name="resubmit-filter-document")
+        async_session.add(document)
+        await async_session.commit()
+
+        create_result = await create_job_for_document(
+            document_id=document.id,
+            uploads=[
+                ("page-01.jpg", b"one"),
+                ("page-02.jpg", b"two"),
+            ],
+            session=async_session,
+            settings=settings,
+        )
+
+        services = _build_services(default_session_factory)
+        job = await services.jobs.read_job(job_id=create_result.job_id, session=async_session)
+        page_one = next(js for js in job.job_sources if js.source is not None and js.source.page_number == 1)
+        page_two = next(js for js in job.job_sources if js.source is not None and js.source.page_number == 2)
+
+        page_one.status = JobSourceStatus.TRANSCRIBED
+        page_one.raw_transcription = "existing transcript"
+        page_two.status = JobSourceStatus.PENDING
+        page_two.raw_transcription = None
+        await services.transcriptions.update_job_source(job_source=page_one, session=async_session)
+        await services.transcriptions.update_job_source(job_source=page_two, session=async_session)
+        await services.jobs.update_job_state(job_id=job.id, status=JobStatus.QUEUED, session=async_session)
+        await async_session.commit()
+
+        call_count = 0
+
+        async def _fake_transcribe_document_image(
+            image_path,
+            *,
+            prompt_name="transcribe_document.md",
+            settings=None,
+            provider=None,
+        ) -> TranscriptionResult:
+            nonlocal call_count
+            _ = (image_path, prompt_name, settings, provider)
+            call_count += 1
+            return TranscriptionResult(
+                text="new transcript",
+                provider="openrouter",
+                model="test-model",
+                prompt_name="transcribe_document.md",
+            )
+
+        monkeypatch.setattr(
+            "transcription.services.workflows.transcribe_document_image",
+            _fake_transcribe_document_image,
+        )
+
+        queued_job = await services.jobs.read_job(job_id=create_result.job_id, session=async_session)
+        assert queued_job is not None
+        await advance_job(job=queued_job, services=services, session=async_session)
+
+        refreshed = await services.jobs.read_job(job_id=create_result.job_id, session=async_session)
+        assert call_count == 1
+        statuses = {js.status for js in refreshed.job_sources}
+        assert statuses == {JobSourceStatus.TRANSCRIBED}
+
 
 @pytest.mark.integration
 class TestPipelineFailureFlow:
     """Verify end-to-end failure lifecycle behavior."""
 
     @pytest.mark.asyncio
-    async def test_upload_then_worker_persists_failed_terminal_state(self, async_session, tmp_path: Path, monkeypatch):
+    async def test_upload_then_worker_persists_failed_terminal_state(
+        self,
+        async_session,
+        default_session_factory,
+        tmp_path: Path,
+        monkeypatch,
+    ):
         """Upload followed by worker processing persists error detail and failed status on the job."""
         settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
         upload_result = await create_upload_job(
@@ -103,12 +326,12 @@ class TestPipelineFailureFlow:
             _fake_transcribe_document_image,
         )
 
-        services = ServiceBundle()
+        services = _build_services(default_session_factory)
-        queued_job = await services.jobs.read_next_queued_job(session=async_session)
+        queued_job = await services.jobs.read_job(job_id=upload_result.job_id, session=async_session)
         processed = queued_job is not None
         if queued_job is not None:
             await advance_job(job=queued_job, services=services, session=async_session)
-        job = await async_session.get(Job, upload_result.job_id)
+        job = await services.jobs.read_job(job_id=upload_result.job_id, session=async_session)
 
         assert processed is True
         assert job is not None

tests/services/test_job_service.py

@@ -13,6 +13,8 @@ from transcription.db.models import JobStatus
 from transcription.db.models import Source
 from transcription.services.documents import DocumentService
 from transcription.services.jobs import JobDeleteBlockedError
+from transcription.services.jobs import JobCancelBlockedError
+from transcription.services.jobs import JobResubmitBlockedError
 from transcription.services.jobs import JobService
 
 
@@ -224,3 +226,157 @@ class TestJobService:
 
         with pytest.raises(ValueError):
             await job_service.read_job(job_id=job.id)
+
+    @pytest.mark.asyncio
+    async def test_cancel_job_marks_non_transcribed_sources_failed(
+        self,
+        job_service: JobService,
+        document_service: DocumentService,
+    ):
+        document = Document(id=uuid4(), name="cancel-job-doc")
+        await document_service.create_document(document=document)
+
+        job = Job(document_id=document.id, status=JobStatus.QUEUED)
+        await job_service.create_job(job=job)
+
+        async with job_service._session_scope() as session:
+            source_one = Source(
+                document_id=document.id,
+                page_number=1,
+                upload_name="cancel-1.jpg",
+                filename="stored-cancel-1.jpg",
+                file_path="/uploads/stored-cancel-1.jpg",
+            )
+            source_two = Source(
+                document_id=document.id,
+                page_number=2,
+                upload_name="cancel-2.jpg",
+                filename="stored-cancel-2.jpg",
+                file_path="/uploads/stored-cancel-2.jpg",
+            )
+            session.add(source_one)
+            session.add(source_two)
+            await session.flush()
+
+            session.add(
+                JobSource(
+                    job_id=job.id,
+                    source_id=source_one.id,
+                    status=JobSourceStatus.TRANSCRIBED,
+                    raw_transcription="done",
+                )
+            )
+            session.add(
+                JobSource(
+                    job_id=job.id,
+                    source_id=source_two.id,
+                    status=JobSourceStatus.PENDING,
+                )
+            )
+            await session.commit()
+
+        cancelled = await job_service.cancel_job(job_id=job.id)
+        assert cancelled.status == JobStatus.FAILED
+
+        refreshed = await job_service.read_job(job_id=job.id)
+        statuses = {item.status for item in refreshed.job_sources}
+        assert JobSourceStatus.TRANSCRIBED in statuses
+        assert JobSourceStatus.FAILED in statuses
+        pending_entry = next(item for item in refreshed.job_sources if item.status == JobSourceStatus.FAILED)
+        assert pending_entry.error_detail == "Cancelled by user"
+
+    @pytest.mark.asyncio
+    async def test_resubmit_non_transcribed_sources_resets_only_non_transcribed(
+        self,
+        job_service: JobService,
+        document_service: DocumentService,
+    ):
+        document = Document(id=uuid4(), name="resubmit-job-doc")
+        await document_service.create_document(document=document)
+
+        job = Job(document_id=document.id, status=JobStatus.FAILED)
+        await job_service.create_job(job=job)
+
+        async with job_service._session_scope() as session:
+            source_one = Source(
+                document_id=document.id,
+                page_number=1,
+                upload_name="resubmit-1.jpg",
+                filename="stored-resubmit-1.jpg",
+                file_path="/uploads/stored-resubmit-1.jpg",
+                raw_transcription="existing text",
+            )
+            source_two = Source(
+                document_id=document.id,
+                page_number=2,
+                upload_name="resubmit-2.jpg",
+                filename="stored-resubmit-2.jpg",
+                file_path="/uploads/stored-resubmit-2.jpg",
+                raw_transcription="done text",
+            )
+            session.add(source_one)
+            session.add(source_two)
+            await session.flush()
+
+            session.add(
+                JobSource(
+                    job_id=job.id,
+                    source_id=source_one.id,
+                    status=JobSourceStatus.FAILED,
+                    raw_transcription=None,
+                    error_detail="prior error",
+                )
+            )
+            session.add(
+                JobSource(
+                    job_id=job.id,
+                    source_id=source_two.id,
+                    status=JobSourceStatus.TRANSCRIBED,
+                    raw_transcription="done text",
+                )
+            )
+            await session.commit()
+
+        count = await job_service.resubmit_non_transcribed_sources(job_id=job.id)
+        assert count == 1
+
+        refreshed = await job_service.read_job(job_id=job.id)
+        assert refreshed.status == JobStatus.QUEUED
+
+        failed_entry = next(item for item in refreshed.job_sources if item.source is not None and item.source.page_number == 1)
+        transcribed_entry = next(item for item in refreshed.job_sources if item.source is not None and item.source.page_number == 2)
+        assert failed_entry.status == JobSourceStatus.PENDING
+        assert failed_entry.error_detail is None
+        assert failed_entry.source is not None
+        assert failed_entry.source.raw_transcription is None
+        assert transcribed_entry.status == JobSourceStatus.TRANSCRIBED
+
+    @pytest.mark.asyncio
+    async def test_resubmit_non_transcribed_sources_blocks_when_processing(
+        self,
+        job_service: JobService,
+        document_service: DocumentService,
+    ):
+        document = Document(id=uuid4(), name="resubmit-blocked-doc")
+        await document_service.create_document(document=document)
+
+        job = Job(document_id=document.id, status=JobStatus.PROCESSING)
+        await job_service.create_job(job=job)
+
+        with pytest.raises(JobResubmitBlockedError):
+            await job_service.resubmit_non_transcribed_sources(job_id=job.id)
+
+    @pytest.mark.asyncio
+    async def test_cancel_job_blocks_transcribed_terminal_jobs(
+        self,
+        job_service: JobService,
+        document_service: DocumentService,
+    ):
+        document = Document(id=uuid4(), name="cancel-blocked-doc")
+        await document_service.create_document(document=document)
+
+        job = Job(document_id=document.id, status=JobStatus.TRANSCRIBED)
+        await job_service.create_job(job=job)
+
+        with pytest.raises(JobCancelBlockedError):
+            await job_service.cancel_job(job_id=job.id)

tests/services/test_transcription_service.py

@@ -154,3 +154,54 @@ class TestTranscriptionServiceRevisionUpsert:
 
         with pytest.raises(SourceDeleteBlockedError):
             await transcriptions.delete_source_from_job_context(job_id=job_one.id, source_id=source.id)
+
+    @pytest.mark.asyncio
+    async def test_delete_unlinked_source_succeeds(self, default_session_factory):
+        documents = DocumentService(session_factory=default_session_factory)
+        transcriptions = TranscriptionService(session_factory=default_session_factory)
+
+        document = Document(id=uuid4(), name="delete-unlinked-source")
+        await documents.create_document(document=document)
+
+        source = Source(
+            document_id=document.id,
+            page_number=1,
+            upload_name="orphan.jpg",
+            filename="orphan.jpg",
+            file_path="uploads/orphan.jpg",
+        )
+        await transcriptions.create_source(source=source)
+
+        await transcriptions.delete_unlinked_source(source_id=source.id)
+
+        with pytest.raises(TranscriptionNotFoundError):
+            await transcriptions.read_source(source.id)
+
+    @pytest.mark.asyncio
+    async def test_delete_unlinked_source_blocks_when_linked(self, default_session_factory):
+        documents = DocumentService(session_factory=default_session_factory)
+        jobs = JobService(session_factory=default_session_factory)
+        transcriptions = TranscriptionService(session_factory=default_session_factory)
+
+        document = Document(id=uuid4(), name="delete-unlinked-blocked")
+        await documents.create_document(document=document)
+
+        job = Job(document_id=document.id, status=JobStatus.QUEUED)
+        await jobs.create_job(job=job)
+
+        source = Source(
+            document_id=document.id,
+            page_number=1,
+            upload_name="linked.jpg",
+            filename="linked.jpg",
+            file_path="uploads/linked.jpg",
+        )
+        async with transcriptions._session_scope() as session:
+            session.add(source)
+            await session.flush()
+            session.add(JobSource(job_id=job.id, source_id=source.id, status=JobSourceStatus.PENDING))
+            await session.commit()
+            await session.refresh(source)
+
+        with pytest.raises(SourceDeleteBlockedError):
+            await transcriptions.delete_unlinked_source(source_id=source.id)

tests/ui/test_jobs_page.py

@@ -86,7 +86,7 @@ class TestPageRendering:
         assert "document links" in response.text.lower()
         assert "Sources" in response.text
         assert "Jobs" in response.text
-        assert "Delete job" not in response.text
+        assert "Delete Job" in response.text
 
     def test_job_detail_page_rejects_invalid_id(self, app_client):
         """GET /ui/jobs/{job_id} shows validation feedback for malformed IDs."""
@@ -105,19 +105,41 @@ class TestPageRendering:
         assert response.status_code == 200
         assert "Job not found" in response.text
 
-    def test_job_detail_page_hides_delete_action(self, app_client, seed_job):
+    def test_job_detail_page_shows_cancel_and_resubmit_when_queued(self, app_client, seed_job):
-        """GET /ui/jobs/{job_id} does not expose job deletion controls in this revision."""
+        """GET /ui/jobs/{job_id} exposes cancel/resubmit controls for queued jobs."""
         _, client = app_client
         job_id = seed_job(
             filename="no-revision.pdf",
-            status=JobStatus.TRANSCRIBED,
+            status=JobStatus.QUEUED,
-            transcription_text="original text",
+            transcription_text=None,
         )
 
         response = client.get(f"/ui/jobs/{job_id}")
 
         assert response.status_code == 200
-        assert "Delete job" not in response.text
+        assert "Cancel" in response.text
+        assert "Resubmit" in response.text
+        assert "Delete Job" in response.text
+
+    def test_job_cancel_page_renders_confirmation(self, app_client, seed_job):
+        _, client = app_client
+        job_id = seed_job(filename="cancel-ready.pdf", status=JobStatus.PROCESSING)
+
+        response = client.get(f"/ui/jobs/{job_id}/cancel")
+
+        assert response.status_code == 200
+        assert "Cancel Processing Job" in response.text
+        assert "Cancel job" in response.text
+
+    def test_job_resubmit_page_renders_confirmation(self, app_client, seed_job):
+        _, client = app_client
+        job_id = seed_job(filename="resubmit-ready.pdf", status=JobStatus.FAILED, transcription_text=None)
+
+        response = client.get(f"/ui/jobs/{job_id}/resubmit")
+
+        assert response.status_code == 200
+        assert "Resubmit Job" in response.text
+        assert "Resubmit now" in response.text
 
     def test_job_delete_page_shows_confirmation_when_not_processing(self, app_client, seed_job):
         _, client = app_client

tests/ui/test_sources_page.py

@@ -138,3 +138,52 @@ class TestSourcesPageRendering:
         assert "human revision text" in response.text
         assert "Page Number:" in response.text
         assert "Stored Filename:" in response.text
+        assert "Delete Source" in response.text
+
+    def test_source_delete_page_blocks_when_source_is_job_linked(self, app_client, seed_job):
+        _, client = app_client
+        job_id = seed_job(filename="linked-source.png", transcription_text="linked text")
+
+        async def _get_source_id() -> str:
+            async with session_scope() as session:
+                job = await session.get(Job, job_id)
+                assert job is not None
+                source = (
+                    await session.exec(select(Source).where(Source.document_id == job.document_id))
+                ).first()
+                assert source is not None
+                return str(source.id)
+
+        source_id = asyncio.run(_get_source_id())
+        response = client.get(f"/ui/sources/{source_id}/delete")
+
+        assert response.status_code == 200
+        assert "Delete Source Record" in response.text
+        assert "Delete is only available for unlinked sources." in response.text
+
+    def test_source_delete_page_allows_unlinked_source(self, app_client):
+        _, client = app_client
+
+        async def _seed_unlinked_source() -> str:
+            async with session_scope() as session:
+                document = Document(name="Unlinked Source Doc", document_type="memo")
+                session.add(document)
+                await session.flush()
+                source = Source(
+                    document_id=document.id,
+                    page_number=1,
+                    upload_name="orphan-source.png",
+                    filename="orphan-source.png",
+                    file_path="/tmp/orphan-source.png",
+                )
+                session.add(source)
+                await session.commit()
+                return str(source.id)
+
+        source_id = asyncio.run(_seed_unlinked_source())
+        response = client.get(f"/ui/sources/{source_id}/delete")
+
+        assert response.status_code == 200
+        assert "Delete Source Record" in response.text
+        assert "Delete source permanently" in response.text
+        assert "Delete is only available for unlinked sources." not in response.text