From 6c6589d8ff166fee6c01cf8d9d557d3c564505be Mon Sep 17 00:00:00 2001 From: Jim Lancaster <40281233+zoltan57@users.noreply.github.com> Date: Tue, 4 Aug 2026 09:03:51 -0500 Subject: [PATCH] Jobs: big jobs stuck in queue. Added Cancel, Resubmit --- docs/schema_v2.md | 3 +- src/transcription/services/jobs.py | 93 +++++++ src/transcription/services/transcription.py | 125 +++++++-- src/transcription/services/workflows.py | 268 +++++++++++++------ src/transcription/ui/pages/jobs_page.py | 127 ++++++++- src/transcription/ui/pages/sources_page.py | 105 +++++++- tests/integration/test_pipeline_flow.py | 239 ++++++++++++++++- tests/services/test_job_service.py | 156 +++++++++++ tests/services/test_transcription_service.py | 51 ++++ tests/ui/test_jobs_page.py | 34 ++- tests/ui/test_sources_page.py | 49 ++++ 11 files changed, 1126 insertions(+), 124 deletions(-) diff --git a/docs/schema_v2.md b/docs/schema_v2.md index 0fb7991..6a592fb 100644 --- a/docs/schema_v2.md +++ b/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. diff --git a/src/transcription/services/jobs.py b/src/transcription/services/jobs.py index f071182..ccdb07f 100644 --- a/src/transcription/services/jobs.py +++ b/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) diff --git a/src/transcription/services/transcription.py b/src/transcription/services/transcription.py index ad8b10b..5da5077 100644 --- a/src/transcription/services/transcription.py +++ b/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( - select(JobSource).where(JobSource.job_id == job.id).where(JobSource.source_id == source_row.id) + await self.update_job_source_transcription( + 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, *, diff --git a/src/transcription/services/workflows.py b/src/transcription/services/workflows.py index 55f061a..9dc9214 100644 --- a/src/transcription/services/workflows.py +++ b/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) - if source is None: + sources = _resolve_job_sources(source_job) + 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: - error = AppError( - 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() + if not sources: + return await services.jobs.mark_job_status(job.id, JobStatus.TRANSCRIBED, session=session) - try: - result = await asyncio.wait_for( - transcribe_document_image(source.file_path), - 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), - ) + successful_pages: list[tuple[Source, TranscriptionResult]] = [] + failed_pages: list[tuple[Source, AppError]] = [] + externally_stopped = False - _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) - logger.info( - "Job transcribed operation=worker.process_job job_id=%s document_id=%s source_id=%s provider=%s", - job.id, - job.document_id, - source.id, - result.provider, - ) - except TimeoutError: - error = AppError( - f"Provider call timed out after {runtime_settings.worker_provider_timeout_seconds:.1f}s", - category=ErrorCategory.EXTERNAL_PROVIDER, - suggestion="Retry the job. If this repeats, verify provider latency and request payload size.", - retriable=True, - ) - job = await _finalize_failed(job=job, services=services, error=error, session=session) - logger.error( - "Job 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") + started_at = asyncio.get_running_loop().time() + try: + result = await asyncio.wait_for( + transcribe_document_image(source.file_path), + 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), + ) - job = await _finalize_failed(job=job, services=services, error=error, session=session) - logger.error( - "Job 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, - ) - return job + _validate_transcription_quality(result=result, settings=runtime_settings) + successful_pages.append((source, result)) + except TimeoutError: + error = AppError( + f"Provider call timed out after {runtime_settings.worker_provider_timeout_seconds:.1f}s", + category=ErrorCategory.EXTERNAL_PROVIDER, + suggestion="Retry the job. If this repeats, verify provider latency and request payload size.", + retriable=True, + ) + 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) diff --git a/src/transcription/ui/pages/jobs_page.py b/src/transcription/ui/pages/jobs_page.py index 242f248..16ec9e4 100644 --- a/src/transcription/ui/pages/jobs_page.py +++ b/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() diff --git a/src/transcription/ui/pages/sources_page.py b/src/transcription/ui/pages/sources_page.py index a9b1141..a350905 100644 --- a/src/transcription/ui/pages/sources_page.py +++ b/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: - back_label = ( - "Back to Document" - if "document_id" in request.query_params - else "Back to Job" - 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" - ) - else: - ui.button("Back to Sources", on_click=lambda: ui.navigate.to("/sources"), icon="arrow_back").props( - "flat text-xs" + with ui.row().classes("items-center gap-2"): + if back_path is not None: + back_label = ( + "Back to Document" + if "document_id" in request.query_params + else "Back to Job" + 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" + ) + else: + 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}") diff --git a/tests/integration/test_pipeline_flow.py b/tests/integration/test_pipeline_flow.py index 32f4fbf..1726f1b 100644 --- a/tests/integration/test_pipeline_flow.py +++ b/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() - queued_job = await services.jobs.read_next_queued_job(session=async_session) + services = _build_services(default_session_factory) + queued_job = await services.jobs.read_job(job_id=upload_result.job_id, session=async_session) processed = queued_job is not None 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() - queued_job = await services.jobs.read_next_queued_job(session=async_session) + services = _build_services(default_session_factory) + queued_job = await services.jobs.read_job(job_id=upload_result.job_id, session=async_session) processed = queued_job is not None 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 diff --git a/tests/services/test_job_service.py b/tests/services/test_job_service.py index 4dbc0cc..346f501 100644 --- a/tests/services/test_job_service.py +++ b/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) diff --git a/tests/services/test_transcription_service.py b/tests/services/test_transcription_service.py index 7c5efde..b6cf929 100644 --- a/tests/services/test_transcription_service.py +++ b/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) diff --git a/tests/ui/test_jobs_page.py b/tests/ui/test_jobs_page.py index 8d1dbf6..a064050 100644 --- a/tests/ui/test_jobs_page.py +++ b/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): - """GET /ui/jobs/{job_id} does not expose job deletion controls in this revision.""" + def test_job_detail_page_shows_cancel_and_resubmit_when_queued(self, app_client, seed_job): + """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, - transcription_text="original text", + status=JobStatus.QUEUED, + 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 diff --git a/tests/ui/test_sources_page.py b/tests/ui/test_sources_page.py index 6b27dcb..66ca521 100644 --- a/tests/ui/test_sources_page.py +++ b/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