diff --git a/src/transcription/db/models.py b/src/transcription/db/models.py index 5040ff5..77dab18 100644 --- a/src/transcription/db/models.py +++ b/src/transcription/db/models.py @@ -70,6 +70,7 @@ class JobSourceStatus(StrEnum): PENDING = "pending" TRANSCRIBED = "transcribed" FAILED = "failed" + CANCELLED = "cancelled" class JobPurpose(StrEnum): @@ -269,15 +270,6 @@ class Job(SQLModel, table=True): return "unknown" - @property - def error_detail(self) -> str | None: - """Return the first available source-level error detail for the job.""" - for job_source in _loaded_attribute(self, "job_sources") or (): - if job_source.error_detail: - return job_source.error_detail - - return None - class Source(SQLModel, table=True): """A document source image or PDF page.""" @@ -321,10 +313,21 @@ class Source(SQLModel, table=True): @property def latest_job_source(self) -> Optional["JobSource"]: - """Return the most recent job execution record for this source.""" - if not self.job_sources: - return None - return max(self.job_sources, key=lambda js: js.executed_at) + """Return the most recent job execution record for this source. + + ``JobSource`` carries no timestamp of its own, so recency is the parent + job's creation time. ``(job_id, source_id)`` is unique per source, so + this is exactly "the most recent job that included this page". + """ + job_sources = _loaded_attribute(self, "job_sources") or () + dated = [ + (job, job_source) + for job_source in job_sources + if (job := _loaded_attribute(job_source, "job")) is not None + ] + if dated: + return max(dated, key=lambda pair: pair[0].date_created)[1] + return job_sources[0] if job_sources else None @property def latest_status(self) -> JobSourceStatus | None: @@ -334,9 +337,19 @@ class Source(SQLModel, table=True): @property def latest_error_detail(self) -> str | None: - """Return the error detail from the latest job run, if present.""" + """Return the error detail of the latest attempt on the latest job run. + + Failure detail lives on ``ExecutionAttempt``; ``JobSource`` records only + which page a job is working on and how far it got. + """ latest = self.latest_job_source - return latest.error_detail if latest else None + if latest is None: + return None + attempts = _loaded_attribute(latest, "execution_attempts") or () + for attempt in sorted(attempts, key=lambda item: item.attempt_number, reverse=True): + if attempt.error_detail: + return attempt.error_detail + return None @property def document_name(self) -> str | None: @@ -363,11 +376,6 @@ class JobSource(SQLModel, table=True): nullable=False, ), ) - raw_transcription: str | None = None - ai_metadata: dict[str, JsonValue] | None = Field(default=None, sa_column=Column(JSONBCompat(), nullable=True)) - raw_api_response: dict[str, JsonValue] | None = Field(default=None, sa_column=Column(JSONBCompat(), nullable=True)) - error_detail: str | None = None - executed_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) job: Optional["Job"] = Relationship(back_populates="job_sources", sa_relationship_kwargs={"lazy": "raise"}) source: Optional["Source"] = Relationship(back_populates="job_sources", sa_relationship_kwargs={"lazy": "raise"}) @@ -388,7 +396,19 @@ class ExecutionAttempt(SQLModel, table=True): job_id: UUID = Field(foreign_key="job.id", index=True) source_id: UUID = Field(foreign_key="source.id", index=True) attempt_number: int = Field(ge=1) - status: JobSourceStatus + status: JobSourceStatus = Field( + sa_column=Column( + # Declared identically to job_source.status. Without values_callable + # SQLAlchemy persists enum *names*, which is defect [45]: the two + # columns spelled the same status differently and never compared equal. + SAEnum( + JobSourceStatus, + values_callable=lambda enum_cls: [item.value for item in enum_cls], + native_enum=False, + ), + nullable=False, + ) + ) provider: str model: str | None = None request_manifest: dict[str, JsonValue] | None = Field(default=None, sa_column=Column(JSONBCompat(), nullable=True)) diff --git a/src/transcription/services/jobs.py b/src/transcription/services/jobs.py index e306574..425d94b 100644 --- a/src/transcription/services/jobs.py +++ b/src/transcription/services/jobs.py @@ -338,10 +338,7 @@ class JobService(ServiceBase): 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 + job_source.status = JobSourceStatus.CANCELLED await self._finalize(session=_session, caller_session=session, refresh=(job,)) return job @@ -368,20 +365,21 @@ class JobService(ServiceBase): suggestion="Cancel processing first, then resubmit remaining sources.", ) - candidates = [job_source for job_source in job.job_sources if job_source.status == JobSourceStatus.FAILED] + # Cancelled pages are re-attemptable: before V4.7 cancel wrote FAILED, + # so resubmit already reset them. Excluding CANCELLED here would make + # cancelled work permanently unrecoverable. + resubmittable = {JobSourceStatus.FAILED, JobSourceStatus.CANCELLED} + candidates = [job_source for job_source in job.job_sources if job_source.status in resubmittable] if not candidates: raise JobResubmitBlockedError( - "Job has no failed sources to resubmit", + "Job has no failed or cancelled sources to resubmit", category=ErrorCategory.VALIDATION, - suggestion="Only failed sources can be resubmitted.", + suggestion="Only failed or cancelled 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 job.status = JobStatus.QUEUED job.date_updated = now diff --git a/src/transcription/services/sources.py b/src/transcription/services/sources.py index cf0873b..f1c50c0 100644 --- a/src/transcription/services/sources.py +++ b/src/transcription/services/sources.py @@ -350,7 +350,11 @@ class SourceService(ServiceBase): async with self._session_scope(session) as _session: query = select(Source).options( selectinload(Source.document), - selectinload(Source.job_sources), + # Both are needed by Source.latest_job_source and + # latest_error_detail: recency comes from the parent job, and + # failure detail lives on the attempt, not the junction row. + selectinload(Source.job_sources).selectinload(orm_attribute(JobSource.job)), + selectinload(Source.job_sources).selectinload(orm_attribute(JobSource.execution_attempts)), ) if document_id is not None: query = query.where(Source.document_id == document_id) @@ -567,24 +571,12 @@ class SourceService(ServiceBase): select(JobSource).where(JobSource.job_id == job_id).where(JobSource.source_id == source_id) ) job_source = existing_job_source.first() + outcome = JobSourceStatus.TRANSCRIBED if text is not None else JobSourceStatus.FAILED 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, - ai_metadata=metadata_payload, - raw_api_response=raw_response_payload, - error_detail=error_detail, - ) + job_source = JobSource(job_id=job_id, source_id=source_id, status=outcome) _session.add(job_source) else: - job_source.raw_transcription = text - job_source.ai_metadata = metadata_payload - job_source.raw_api_response = raw_response_payload - 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) + job_source.status = outcome finish_time = finished_at or datetime.now(UTC) start_time = started_at or finish_time @@ -605,7 +597,7 @@ class SourceService(ServiceBase): job_id=job_id, source_id=source_id, attempt_number=(attempt_number or 0) + 1, - status=JobSourceStatus.TRANSCRIBED if text is not None else JobSourceStatus.FAILED, + status=outcome, provider=provider or job.provider or self.settings.provider.value, model=model or job.model, request_manifest=manifest_payload, diff --git a/src/transcription/services/workflows.py b/src/transcription/services/workflows.py index 94848af..98d2ada 100644 --- a/src/transcription/services/workflows.py +++ b/src/transcription/services/workflows.py @@ -411,7 +411,13 @@ async def process_next_queued_job( def _resolve_job_sources(job: Job) -> list[Source]: - """Resolve non-transcribed linked sources for a job in deterministic page order.""" + """Resolve pending linked sources for a job in deterministic page order. + + A page is work if it has not already succeeded. CANCELLED is included + deliberately: resubmit resets cancelled pages to PENDING, so they are + re-attemptable, and a cancelled page that somehow reaches a running job is + unfinished work rather than a terminal outcome. + """ if not job.job_sources: return [] diff --git a/src/transcription/ui/pages/jobs_page.py b/src/transcription/ui/pages/jobs_page.py index 8fe7347..450f67b 100644 --- a/src/transcription/ui/pages/jobs_page.py +++ b/src/transcription/ui/pages/jobs_page.py @@ -284,9 +284,10 @@ def register_page() -> None: # noqa: PLR0915 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" - ) + ui.label( + "Cancel stops processing and marks remaining non-transcribed sources as cancelled. " + "Cancelled sources can be resubmitted." + ).classes("text-xs ui-text-muted") async def submit_cancel() -> None: try: @@ -327,7 +328,11 @@ def register_page() -> None: # noqa: PLR0915 render_record_not_found("Job") return - failed_count = sum(1 for js in job.job_sources if js.status == JobSourceStatus.FAILED) + resubmittable_count = sum( + 1 + for js in job.job_sources + if js.status in {JobSourceStatus.FAILED, JobSourceStatus.CANCELLED} + ) with ui.column().classes("w-full max-w-xl mx-auto p-4 gap-4"): page_header("Resubmit Job") @@ -335,9 +340,10 @@ def register_page() -> None: # noqa: PLR0915 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("Failed Sources:", str(failed_count)) + metadata_row("Resubmittable Sources:", str(resubmittable_count)) ui.label( - "Resubmit queues only failed linked sources. Prior execution evidence remains preserved." + "Resubmit queues failed and cancelled linked sources. " + "Prior execution evidence remains preserved." ).classes("text-xs ui-text-muted") async def submit_resubmit() -> None: diff --git a/src/transcription/ui/pages/sources_page.py b/src/transcription/ui/pages/sources_page.py index 6393df8..fb7b983 100644 --- a/src/transcription/ui/pages/sources_page.py +++ b/src/transcription/ui/pages/sources_page.py @@ -12,6 +12,7 @@ from nicegui import ui from transcription.config import Settings from transcription.db.models import ExecutionAttempt from transcription.db.models import JobSource +from transcription.db.models import JobSourceStatus from transcription.db.models import Source from transcription.services.sources import LatestExecutionAttempt from transcription.services.sources import SourceDeleteBlockedError @@ -120,7 +121,7 @@ def register_page() -> None: # noqa: PLR0915 try: source = await sources_service.read_source_detail(parsed_source_id) navigation = await sources_service.read_source_navigation(parsed_source_id) - latest_job_source = _latest_job_source(source) + latest_job_source = source.latest_job_source latest_attempt = ( await sources_service.read_latest_execution_attempt(job_source_id=latest_job_source.id) if latest_job_source is not None @@ -134,7 +135,7 @@ def register_page() -> None: # noqa: PLR0915 show_error(exc, title="Load failed", operation="sources.read") return - original_transcription = _resolve_original_transcription(source=source, latest_job_source=latest_job_source) + original_transcription = _resolve_original_transcription(source=source, latest_attempt=latest_attempt) with ui.column().classes("w-full max-w-[1800px] mx-auto p-4 gap-4"): with section_header_row(): @@ -336,7 +337,10 @@ def _render_source_job_metadata_zone( archival_badge(status) metadata_row("Job ID:", str(latest_job_source.job_id)) - metadata_row("Executed:", latest_job_source.executed_at.isoformat()) + metadata_row( + "Executed:", + latest_attempt.attempt.finished_at.isoformat() if latest_attempt is not None else "not yet executed", + ) metadata_row( "Provider:", latest_job_source.job.provider if latest_job_source.job and latest_job_source.job.provider else "unknown", @@ -352,30 +356,18 @@ def _render_source_job_metadata_zone( else "unknown", ) - if latest_job_source.error_detail: + if latest_attempt is not None and latest_attempt.attempt.error_detail: with ui.column().classes("w-full mt-2"): ui.label("Failure Detail:").classes("ui-text-muted text-xs mb-1") - ui.label(latest_job_source.error_detail).classes("p-2 ui-note-box text-xs") + ui.label(latest_attempt.attempt.error_detail).classes("p-2 ui-note-box text-xs") - _render_provider_evidence( - latest_job_source, - latest_attempt=latest_attempt, - ) + _render_provider_evidence(latest_attempt=latest_attempt) -def _render_provider_evidence( - job_source: JobSource, - *, - latest_attempt: LatestExecutionAttempt | None, -) -> None: +def _render_provider_evidence(*, latest_attempt: LatestExecutionAttempt | None) -> None: ui.label("Provider Evidence").classes("text-xs font-semibold ui-text-primary mt-3") if latest_attempt is None: render_empty_state("Exact transport evidence was not captured for this historical execution.", italic=True) - _render_json_evidence("Normalized Metadata (AI Metadata)", job_source.ai_metadata) - _render_json_evidence( - "OpenRouter SDK Response Snapshot (Raw API Response compatibility field)", - job_source.raw_api_response, - ) return attempt = latest_attempt.attempt @@ -516,10 +508,13 @@ def _render_source_transcription_zone( icon="refresh", ).props("flat") - if latest_job_source is not None and latest_job_source.status.value == "failed": - ui.label("Source has a failed job execution. Save a human revision to preserve corrected text.").classes( - "text-xs ui-text-muted italic" - ) + if latest_job_source is not None and latest_job_source.status in { + JobSourceStatus.FAILED, + JobSourceStatus.CANCELLED, + }: + ui.label( + "Source has an unfinished job execution. Save a human revision to preserve corrected text." + ).classes("text-xs ui-text-muted italic") def _render_machine_candidates( @@ -651,13 +646,7 @@ def _reset_revision_text(revision_input: ui.textarea, source: Source, original_t revision_input.value = fallback_text -def _latest_job_source(source: Source) -> JobSource | None: - if not source.job_sources: - return None - return max(source.job_sources, key=lambda item: item.executed_at) - - -def _resolve_original_transcription(*, source: Source, latest_job_source: JobSource | None) -> str | None: - if source.raw_transcription is None and latest_job_source is not None: - return latest_job_source.raw_transcription +def _resolve_original_transcription(*, source: Source, latest_attempt: LatestExecutionAttempt | None) -> str | None: + if source.raw_transcription is None and latest_attempt is not None: + return latest_attempt.attempt.raw_transcription return source.raw_transcription diff --git a/tests/integration/test_pipeline_flow.py b/tests/integration/test_pipeline_flow.py index 7f6bbc1..6d8ef7d 100644 --- a/tests/integration/test_pipeline_flow.py +++ b/tests/integration/test_pipeline_flow.py @@ -6,9 +6,12 @@ from uuid import uuid4 import pytest from PIL import Image +from sqlmodel import col +from sqlmodel import select from transcription.config import Settings from transcription.db.models import Document +from transcription.db.models import ExecutionAttempt from transcription.db.models import JobSourceStatus from transcription.db.models import JobStatus from transcription.providers.base import ProviderUsage @@ -20,6 +23,15 @@ from transcription.services.store import create_job_for_document from transcription.services.workflows import advance_job +async def _attempts_for_job(session, job) -> list[ExecutionAttempt]: + """Load execution attempts for a job; V4.7 moved evidence off JobSource.""" + job_source_ids = [job_source.id for job_source in job.job_sources] + result = await session.exec( + select(ExecutionAttempt).where(col(ExecutionAttempt.job_source_id).in_(job_source_ids)) + ) + return list(result.all()) + + def _jpeg_bytes(color: str = "white") -> bytes: output = io.BytesIO() Image.new("RGB", (2, 2), color=color).save(output, format="JPEG") @@ -116,21 +128,24 @@ class TestPipelineSuccessFlow: assert processed is True assert job is not None assert job.status == JobStatus.TRANSCRIBED - assert any(job_source.raw_transcription == "Pipeline transcript" for job_source in job.job_sources) + attempts = await _attempts_for_job(async_session, job) + assert any(attempt.raw_transcription == "Pipeline transcript" for attempt in attempts) assert job.prompt_name == "transcribe_document.md" assert job.user_prompt is not None assert job.temperature == 0.2 assert job.top_p == 0.85 assert any( - job_source.ai_metadata == {"finish_reason": "stop", "usage": {"total_tokens": 42}} - for job_source in job.job_sources + attempt.normalized_metadata is not None + and attempt.normalized_metadata["finish_reason"] == "stop" + and attempt.normalized_metadata["usage"] == {"total_tokens": 42} + for attempt in attempts ) assert any( - job_source.raw_api_response + attempt.sdk_response_snapshot == {"id": "resp_123", "choices": [{"message": {"content": "Pipeline transcript"}}]} - for job_source in job.job_sources + for attempt in attempts ) - assert all(job_source.error_detail is None for job_source in job.job_sources) + assert all(attempt.error_detail is None for attempt in attempts) @pytest.mark.asyncio async def test_worker_transcribes_all_sources_for_multi_page_job( @@ -191,7 +206,8 @@ class TestPipelineSuccessFlow: 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) + attempts = await _attempts_for_job(async_session, job) + assert all(attempt.raw_transcription for attempt in attempts) assert all( job_source.source is not None and job_source.source.raw_transcription for job_source in job.job_sources ) @@ -261,7 +277,8 @@ class TestPipelineSuccessFlow: 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) + attempts = await _attempts_for_job(async_session, job) + assert any(attempt.error_detail is not None for attempt in attempts) @pytest.mark.asyncio async def test_worker_skips_already_transcribed_sources_on_resubmit( @@ -293,9 +310,7 @@ class TestPipelineSuccessFlow: 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.sources.update_job_source(job_source=page_one, session=async_session) await services.sources.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) @@ -387,11 +402,10 @@ class TestPipelineFailureFlow: assert processed is True assert job is not None assert job.status == JobStatus.FAILED - assert all(job_source.raw_transcription is None for job_source in job.job_sources) - assert any(job_source.error_detail is not None for job_source in job.job_sources) - error_detail = next( - job_source.error_detail for job_source in job.job_sources if job_source.error_detail is not None - ) + attempts = await _attempts_for_job(async_session, job) + assert all(attempt.raw_transcription is None for attempt in attempts) + assert any(attempt.error_detail is not None for attempt in attempts) + error_detail = next(attempt.error_detail for attempt in attempts if attempt.error_detail is not None) assert "pipeline provider failure" in error_detail assert "[internal_unexpected_error]" in error_detail assert "error_id=" in error_detail diff --git a/tests/services/test_job_service.py b/tests/services/test_job_service.py index 648bbcf..a0dddf0 100644 --- a/tests/services/test_job_service.py +++ b/tests/services/test_job_service.py @@ -342,7 +342,6 @@ class TestJobService: job_id=job.id, source_id=source_one.id, status=JobSourceStatus.TRANSCRIBED, - raw_transcription="done", ) ) session.add( @@ -360,9 +359,8 @@ class TestJobService: 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" + assert JobSourceStatus.CANCELLED in statuses + assert JobSourceStatus.FAILED not in statuses @pytest.mark.asyncio async def test_resubmit_failed_sources_resets_only_failed( @@ -406,8 +404,6 @@ class TestJobService: job_id=job.id, source_id=source_one.id, status=JobSourceStatus.FAILED, - raw_transcription=None, - error_detail="prior error", ) ) session.add( @@ -415,7 +411,6 @@ class TestJobService: job_id=job.id, source_id=source_two.id, status=JobSourceStatus.TRANSCRIBED, - raw_transcription="done text", ) ) await session.commit() @@ -433,7 +428,6 @@ class TestJobService: 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 == "existing text" assert transcribed_entry.status == JobSourceStatus.TRANSCRIBED @@ -486,7 +480,6 @@ class TestJobService: job_id=job.id, source_id=source_two.id, status=JobSourceStatus.TRANSCRIBED, - raw_transcription="done text", ) ) await session.commit() @@ -494,6 +487,46 @@ class TestJobService: with pytest.raises(JobResubmitBlockedError): await job_service.resubmit_failed_sources(job_id=job.id) + @pytest.mark.asyncio + async def test_resubmit_failed_sources_includes_cancelled( + self, + job_service: JobService, + document_service: DocumentService, + ): + """Cancel is recoverable: cancelled pages are re-attempted on resubmit.""" + document = Document(id=uuid4(), name="resubmit-cancelled-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 = Source( + document_id=document.id, + page_number=1, + upload_name="resubmit-cancelled.jpg", + filename="stored-resubmit-cancelled.jpg", + file_path="/uploads/stored-resubmit-cancelled.jpg", + file_hash="3" * 64, + file_size_bytes=1, + ) + session.add(source) + await session.flush() + session.add( + JobSource( + job_id=job.id, + source_id=source.id, + status=JobSourceStatus.CANCELLED, + ) + ) + await session.commit() + + assert await job_service.resubmit_failed_sources(job_id=job.id) == 1 + + refreshed = await job_service.read_job(job_id=job.id) + assert refreshed.status == JobStatus.QUEUED + assert refreshed.job_sources[0].status == JobSourceStatus.PENDING + @pytest.mark.asyncio async def test_resubmit_failed_sources_blocks_when_processing( self, diff --git a/tests/services/test_v2_crud.py b/tests/services/test_v2_crud.py index 21c0024..90101aa 100644 --- a/tests/services/test_v2_crud.py +++ b/tests/services/test_v2_crud.py @@ -322,6 +322,10 @@ async def test_update_job_source_transcription_persists_provider_json_payloads(d stored_rows = await transcriptions.list_job_sources(job_id=job.id) assert len(stored_rows) == 1 - assert stored_rows[0].raw_transcription == "provider transcript" - assert stored_rows[0].ai_metadata == metadata - assert stored_rows[0].raw_api_response == raw_payload + assert stored_rows[0].status == JobSourceStatus.TRANSCRIBED + + attempt = await transcriptions.read_latest_execution_attempt(job_source_id=stored_rows[0].id) + assert attempt is not None + assert attempt.attempt.raw_transcription == "provider transcript" + assert attempt.attempt.normalized_metadata == metadata + assert attempt.attempt.sdk_response_snapshot == raw_payload diff --git a/tests/services/test_workflows_reliability.py b/tests/services/test_workflows_reliability.py index cbdffe2..9efdc44 100644 --- a/tests/services/test_workflows_reliability.py +++ b/tests/services/test_workflows_reliability.py @@ -5,9 +5,12 @@ from pathlib import Path from uuid import uuid4 import pytest +from sqlmodel import col +from sqlmodel import select from transcription.config import Settings from transcription.db.models import Document +from transcription.db.models import ExecutionAttempt from transcription.db.models import Job from transcription.db.models import JobSource from transcription.db.models import JobSourceStatus @@ -87,9 +90,18 @@ class TestWorkflowReliability: assert result is not None assert result.status == JobStatus.FAILED - assert result.error_detail is not None - assert "timed out" in result.error_detail.lower() - assert "20.0s" in result.error_detail + + async with services.jobs._session_scope() as session: + attempts = ( + await session.execute( + select(ExecutionAttempt).where( + col(ExecutionAttempt.job_source_id).in_([js.id for js in result.job_sources]) + ) + ) + ).scalars().all() + error_detail = next(attempt.error_detail for attempt in attempts if attempt.error_detail is not None) + assert "timed out" in error_detail.lower() + assert "20.0s" in error_detail @pytest.mark.asyncio async def test_completed_page_is_committed_before_next_provider_call_finishes( diff --git a/tests/test_models.py b/tests/test_models.py index 9a359ad..544f355 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -211,24 +211,17 @@ class TestPersonAndDocumentPersonModel: class TestJobSourceModel: - def test_job_source_persists_json_payloads(self, session): + def test_job_source_is_a_queue_row_not_an_evidence_row(self, session): + """V4.7: job_source carries only queue state; evidence lives on execution_attempt.""" document = _persist_document(session) job = _persist_job(session, document) source = _persist_source(session, document) - job_source = _persist_job_source( - session, - job, - source, - raw_transcription="Page transcript", - ai_metadata={"confidence": 0.91, "boxes": [{"x": 1, "y": 2}]}, - raw_api_response={"provider": "test"}, - ) + job_source = _persist_job_source(session, job, source) fetched = session.get(JobSource, job_source.id) assert fetched is not None assert fetched.status == JobSourceStatus.PENDING - assert fetched.ai_metadata == {"confidence": 0.91, "boxes": [{"x": 1, "y": 2}]} - assert fetched.raw_api_response == {"provider": "test"} + assert set(JobSource.model_fields) == {"id", "job_id", "source_id", "status"} class TestRelationships: diff --git a/tests/ui/conftest.py b/tests/ui/conftest.py index ef88bb2..0b54e26 100644 --- a/tests/ui/conftest.py +++ b/tests/ui/conftest.py @@ -23,6 +23,7 @@ from transcription.db import session as db_session_module from transcription.db import session_scope from transcription.db.models import Document from transcription.db.models import DocumentPerson +from transcription.db.models import ExecutionAttempt from transcription.db.models import Job from transcription.db.models import JobSource from transcription.db.models import JobSourceStatus @@ -131,17 +132,30 @@ async def seed_job(app_client: tuple[FastAPI, TestClient]) -> Callable[..., Awai await session.flush() if transcription_text is not None or error_detail is not None: + outcome = JobSourceStatus.TRANSCRIBED if transcription_text is not None else JobSourceStatus.FAILED + job_source = JobSource(job_id=job.id, source_id=source.id, status=outcome) + session.add(job_source) + await session.flush() + + # V4.7: evidence lives on execution_attempt, not job_source. + executed_at = datetime.now(UTC) session.add( - JobSource( + ExecutionAttempt( + job_source_id=job_source.id, job_id=job.id, source_id=source.id, - status=( - JobSourceStatus.TRANSCRIBED if transcription_text is not None else JobSourceStatus.FAILED - ), + attempt_number=1, + status=outcome, + provider="openrouter", + model="google/gemini-2.5-flash", + response_received=transcription_text is not None, + sdk_response_snapshot=raw_api_response, + normalized_metadata=ai_metadata, raw_transcription=transcription_text, error_detail=error_detail, - ai_metadata=ai_metadata, - raw_api_response=raw_api_response, + started_at=executed_at, + finished_at=executed_at, + duration_ms=0, ) ) diff --git a/tests/ui/test_jobs_page.py b/tests/ui/test_jobs_page.py index fa6dfd3..77f2c92 100644 --- a/tests/ui/test_jobs_page.py +++ b/tests/ui/test_jobs_page.py @@ -125,7 +125,7 @@ class TestJobsPageRendering: assert response.status_code == 200 assert "Resubmit Job" in response.text - assert "Failed Sources:" in response.text + assert "Resubmittable Sources:" in response.text assert "Resubmit now" in response.text @pytest.mark.asyncio diff --git a/tests/ui/test_sources_page.py b/tests/ui/test_sources_page.py index c163bde..461f106 100644 --- a/tests/ui/test_sources_page.py +++ b/tests/ui/test_sources_page.py @@ -7,9 +7,11 @@ import pytest from sqlmodel import select from transcription.db import session_scope +from transcription.db.loading import orm_attribute from transcription.db.loading import selectinload from transcription.db.models import Document from transcription.db.models import Job +from transcription.db.models import JobSource from transcription.db.models import JobSourceStatus from transcription.db.models import JobStatus from transcription.db.models import Source @@ -56,7 +58,7 @@ class TestSourceModelProperties: select(Source) .options( selectinload(Source.document), - selectinload(Source.job_sources), + selectinload(Source.job_sources).selectinload(orm_attribute(JobSource.execution_attempts)), ) .where(Source.document_id == job.document_id) ) @@ -218,8 +220,8 @@ class TestSourcesPageRendering: assert "Save revision" in response.text assert "Previous Page" in response.text assert "Next Page" in response.text - assert "AI Metadata" in response.text - assert "Raw API Response" in response.text + assert "Normalized Metadata" in response.text + assert "OpenRouter SDK Response Snapshot" in response.text assert "finish_reason" in response.text assert "response-123" in response.text diff --git a/tools/migrate_v46_to_v47.py b/tools/migrate_v46_to_v47.py index 95ec334..ad82a07 100644 --- a/tools/migrate_v46_to_v47.py +++ b/tools/migrate_v46_to_v47.py @@ -11,6 +11,10 @@ Steps, in execution order: orientation, in place, and update ``source.file_hash`` and ``source.file_size_bytes`` to describe the rewritten file. 2. Drop the ``processing_artifact`` table and delete its external files. +3. Rewrite ``execution_attempt.status`` from enum *names* to enum *values*, so + it compares equal to ``job_source.status`` (defect [45]). +4. Drop the five evidence columns from ``job_source``, leaving it a pure work + queue of ``id``, ``job_id``, ``source_id`` and ``status``. Design notes: @@ -50,6 +54,7 @@ import sys from collections.abc import Sequence from pathlib import Path +from sqlalchemy import bindparam from sqlalchemy import create_engine from sqlalchemy import inspect as sqlalchemy_inspect from sqlalchemy import select @@ -62,6 +67,7 @@ from transcription.config import Settings from transcription.config import get_settings from transcription.db import models as _models # noqa: F401 (registers every table) from transcription.db.engine import get_database_url +from transcription.db.models import JobSourceStatus from transcription.services.normalization import normalize_orientation from transcription.services.sources import source_mime_type @@ -81,6 +87,17 @@ EXPECTED_ROW_COUNTS = { ARTIFACT_TABLE = "processing_artifact" +#: Evidence columns removed from ``job_source`` in V4.7. Every one of them is +#: duplicated byte-for-byte by ``execution_attempt`` across all 77 rows that +#: carry evidence, so no information is lost by dropping them. +JOB_SOURCE_DROPPED_COLUMNS = ( + "raw_transcription", + "ai_metadata", + "raw_api_response", + "error_detail", + "executed_at", +) + #: The V4.6 default for the deleted ``Settings.artifact_dir``. The setting no #: longer exists, so the historical location is recorded here instead. DEFAULT_ARTIFACT_DIR = Path("data/artifacts") @@ -195,6 +212,74 @@ def drop_processing_artifacts(connection: Connection, artifact_dir: Path, *, dry return count +def normalize_attempt_status(connection: Connection, *, dry_run: bool) -> int: + """Step 3: rewrite ``execution_attempt.status`` from enum names to values. + + Defect [45]: ``execution_attempt.status`` was declared without + ``values_callable``, so SQLAlchemy persisted enum *names* ('TRANSCRIBED') + while ``job_source.status`` persisted *values* ('transcribed'). The two + columns never compared equal on a single one of the 79 rows. The model + declaration is fixed in V4.7; the stored rows are fixed here. + """ + name_to_value = {member.name: member.value for member in JobSourceStatus} + recognised = sorted(set(name_to_value) | set(name_to_value.values())) + unknown = ( + connection.execute( + text("select distinct status from execution_attempt where status not in :values").bindparams( + bindparam("values", recognised, expanding=True) + ) + ) + .scalars() + .all() + ) + if unknown: + raise RuntimeError(f"execution_attempt.status carries unrecognised spellings: {sorted(unknown)}") + + rewritten = 0 + for name, value in sorted(name_to_value.items()): + if name == value: + continue + count = connection.execute( + text("select count(*) from execution_attempt where status = :name"), + {"name": name}, + ).scalar_one() + if not count: + continue + print(f" {name} -> {value}: {count} row(s)") + rewritten += count + if dry_run: + continue + connection.execute( + text("update execution_attempt set status = :value where status = :name"), + {"name": name, "value": value}, + ) + + print(f" rewritten={rewritten}") + return rewritten + + +def strip_job_source_columns(connection: Connection, *, dry_run: bool) -> int: + """Step 4: drop the evidence columns from ``job_source``. + + Uses ``ALTER TABLE ... DROP COLUMN``, supported by SQLite 3.35+ and by + PostgreSQL. Idempotent: a column that is already gone is skipped. + """ + inspector = sqlalchemy_inspect(connection) + present = {column["name"] for column in inspector.get_columns("job_source")} + targets = [name for name in JOB_SOURCE_DROPPED_COLUMNS if name in present] + if not targets: + print(" all evidence columns already dropped") + return 0 + + print(f" dropping {len(targets)} column(s): {', '.join(targets)}") + if dry_run: + return len(targets) + + for name in targets: + connection.execute(text(f'alter table "job_source" drop column "{name}"')) + return len(targets) + + def migrate(*, settings: Settings, artifact_dir: Path, dry_run: bool, strict_counts: bool) -> None: """Apply every V4.7 migration step in order.""" engine = create_engine(_sync_url(settings)) @@ -207,6 +292,12 @@ def migrate(*, settings: Settings, artifact_dir: Path, dry_run: bool, strict_cou print(f"\nStep 2: drop {ARTIFACT_TABLE}") drop_processing_artifacts(connection, artifact_dir, dry_run=dry_run) + + print("\nStep 3: normalize execution_attempt.status spelling") + normalize_attempt_status(connection, dry_run=dry_run) + + print("\nStep 4: strip evidence columns from job_source") + strip_job_source_columns(connection, dry_run=dry_run) finally: engine.dispose()