V3 Updated V3 core documents. Added data folder backup/restore before/after running destructive tests.

.gitignore

@@ -18,3 +18,6 @@ wheels/
 # Document images
 uploads/*
 data/*
+
+# Local destructive-test backups
+.test-backups/

README.md

@@ -142,3 +142,47 @@ Prompt files are stored in `prompts/` and loaded from `PROMPT_DIR` (default: `./
 
 The canonical MVP prompt is:
 - `prompts/transcribe_document.md`
+
+## Destructive test procedure (with data backup)
+
+Use the cross-platform Python wrapper below whenever a test run might alter local `./data`.
+
+1. Create backup of `./data`.
+2. Run your test command.
+3. On success, prompt whether to restore now.
+4. On failure, keep backup and current state for inspection.
+
+Preflight behavior:
+
+- Backup preflight is warning-only when `data/transcription.db` appears in use.
+- Restore preflight is blocking: the script prompts you to close conflicting applications, then type `retry` to re-check or `cancel` to skip restore.
+
+### Run with confirmation-gated restore (default)
+
+```bash
+uv run python tools/run_destructive_tests.py -- pytest tests/services/test_job_service.py tests/ui/test_jobs_page.py
+```
+
+After tests pass, the script asks whether to restore backup immediately.
+
+### Run with automatic restore (non-interactive)
+
+```bash
+uv run python tools/run_destructive_tests.py --auto-restore -- pytest
+```
+
+### Run without terminal prompt (decide restore later)
+
+```bash
+uv run python tools/run_destructive_tests.py --skip-restore-prompt -- pytest
+```
+
+This keeps both the current post-test state and the backup, so restore can be decided explicitly later.
+
+### Restore later from a saved backup
+
+```bash
+uv run python tools/run_destructive_tests.py --restore-from data-backup-YYYYMMDD-HHMMSS
+```
+
+Backups are stored in `.test-backups/` and ignored by git.

docs/architecture_v3.md

@@ -6,9 +6,10 @@ This document describes the V3 production architecture of the personal historica
 
 * Preserve source material as immutable transcribed text alongside page-level spatial AI metadata and complete provider API envelopes.
 * Support batching multi-image and folder uploads cleanly into sequential pages (`page_number`).
-* Capture complete input prompt provenance (`system_prompt`, `user_prompt`, `prompt_hash`) and execution parameters (`temperature`, `top_p`) at the page execution level (`JobSource`).
+* Capture complete input prompt provenance (`system_prompt`, `user_prompt`, `prompt_hash`) and execution parameters (`temperature`, `top_p`) at submission time on `Job`.
 * Leverage asynchronous worker pools (`asyncio`) for parallel single-image API execution bounded by rate limiters (`asyncio.Semaphore`).
-* Maintain relational database portability across engines (SQLite for development/testing, PostgreSQL for production) using SQLModel and generic JSON abstraction layers.
+* Maintain relational data-model portability across the supported backends by using SQLModel/SQLAlchemy and compatibility types so the same domain schema works in SQLite for local development/testing and PostgreSQL in production.
+* Keep operator tooling and local maintenance workflows OS-independent by using Python or other cross-platform interfaces for canonical project automation.
 * Verify image asset integrity via SHA-256 file hashing (`file_hash`) while storing binary assets on the local filesystem.
 * Standardize all data validation, API parsing, and database models on **Pydantic V2** and **SQLModel**.
 * Support rich historical attribution (multi-author and multi-recipient relationships via `DocumentPerson`).
@@ -19,8 +20,9 @@ The V3 runtime operates as an asynchronous Python application:
 
 * FastAPI + NiceGUI web application process.
 * In-process `asyncio` background task orchestrator for parallel API execution.
-* Relational persistence via SQLModel / SQLAlchemy (SQLite engine in local development/testing, PostgreSQL engine in production).
+* Relational persistence via SQLModel / SQLAlchemy, using SQLite for local development/testing and PostgreSQL as the production persistence target.
 * Pydantic V2 validation layer wrapping API payloads, prompt configurations, and JSON metadata schemas.
+* Cross-platform operator workflows implemented in Python so core local operations run consistently on Windows, Linux, and macOS.
 
 ^^^mermaid
 flowchart LR
@@ -55,9 +57,9 @@ Application lifespan owns runtime setup/teardown:
 
 Responsibilities:
 
-* Batch orchestration and status transitions (`queued` -> `processing` -> `completed` | `partial_success` | `failed`).
+* Batch orchestration and status transitions (`queued` -> `processing` -> `transcribed` | `partial_success` | `failed`).
 * Parallel single-image API execution using `asyncio.gather` bounded by `asyncio.Semaphore`.
-* Page-level prompt construction, logging full `system_prompt` and `user_prompt` to `JobSource`.
+* Resolve prompt configuration at submission time and persist frozen snapshot fields on `Job`.
 * Pydantic schema parsing and validation prior to database storage.
 
 ### Domain & Service Layer
@@ -75,25 +77,25 @@ Responsibilities:
 1. User uploads a folder or batch of images for a `Document`.
 2. System hashes each image file (SHA-256), writes image files to filesystem storage, and creates `Document`, `Job(status='queued')`, and ordered `Source` pages (`page_number = 1..N`).
 3. Worker claims job, sets `Job.status = 'processing'`, and spawns parallel `asyncio` tasks bounded by semaphore.
-4. Each task builds page-specific system/user prompts and calls Vision API for a **single** `Source` image.
+4. Each task reads the frozen prompt snapshot from `Job` and calls Vision API for a **single** `Source` image.
 5. On task completion:
-* Writes a `JobSource` record containing `status='transcribed'`, `raw_transcription`, complete input details (`prompt_name`, `prompt_hash`, `system_prompt`, `user_prompt`, `temperature`, `top_p`), operational `ai_metadata`, and complete unedited `raw_api_response`.
+* Writes a `JobSource` record containing `status='transcribed'`, `raw_transcription`, operational `ai_metadata`, and complete unedited `raw_api_response`.
 * Caches active output text to `Source.raw_transcription`.
 
 
 6. On page failure:
-* Writes `JobSource` record with `status='failed'`, recorded prompt inputs, and `error_detail`.
+* Writes `JobSource` record with `status='failed'` and `error_detail`.
 
 
 7. Once all page tasks resolve:
-* Marks `Job.status` as `completed` (100% success), `partial_success` (at least 1 success, 1 failure), or `failed` (all failed).
+* Marks `Job.status` as `transcribed` (100% success), `partial_success` (at least 1 success, 1 failure), or `failed` (all failed).
 
 
 
 ## Domain Ownership & Invariants
 
 * **Immutable AI Outputs:** `source.raw_transcription` and `job_source.raw_transcription` store original, point-in-time machine output and are immutable.
-* **Complete Input & Output Provenance:** Every `job_source` record contains both the exact input configuration sent to the model and the complete REST response envelope returned.
+* **Complete Input & Output Provenance:** Every `job` stores the exact frozen input configuration sent to the model, and every `job_source` stores per-page output evidence including the complete REST response envelope returned.
 * **Inlined Revisions:** Human corrections occur on `source.revised_text`. UI renders `COALESCE(revised_text, raw_transcription)`.
 * **Sequential Integrity:** Multi-page documents are strictly ordered by `source.page_number ASC`.
 * **Page Execution Isolation:** A failure on one page image does not invalidate successful transcriptions on sister pages in the same batch job.
@@ -103,7 +105,7 @@ Responsibilities:
 * `Document` has many `Source` pages, many `Job` runs, and many `Person` records via `DocumentPerson` junction (`author` or `recipient`).
 * `Source` belongs to one `Document` and can be processed across many `JobSource` executions.
 * `Job` has many `JobSource` execution records.
-* `JobSource` holds page-level prompts, parameters, execution status, and raw response JSON.
+* `JobSource` holds page-level execution status, output text, and raw response JSON.
 
 ## Test Strategy
 
@@ -125,14 +127,14 @@ Responsibilities:
 
 ## Related Local References
 
-- [System Overview](index_v2.md)
+- [System Overview](index_v3.md)
 - [System Design Intent](invariant/intent.md)
 - [Transcription Methodology](invariant/transcription_methodology.md)
 - System Architecture (this document)
-- [System Requirements](requirements_v2.md)
-- [Data model](schema_v2.md)
-- [Error Handling Policy](error_handling_v2.md)
-- [Implementation Plan](implementation_plan_v2.md)
+- [System Requirements](requirements_v3.md)
+- [Data model](schema_v3.md)
+- [Error Handling Policy](error_handling_v3.md)
+- [Implementation Plan](implementation_plan_v3.md)
 
 
 

docs/implementation_plan_v3.md

@@ -9,10 +9,11 @@ Use a fresh database. There will be no migrations, data conversion, legacy compa
 ## Current Project Impact
 
 * `src/transcription/db/models.py` defines the SQLModel tables. It must be updated to match the approved v3 schema (`Document`, `Person`, `DocumentPerson`, `Source`, `Job`, `JobSource`).
-* The v3 target adds input provenance fields (`prompt_name`, `prompt_hash`, `system_prompt`, `user_prompt`, `temperature`, `top_p`) and full output payloads (`raw_api_response`, `ai_metadata`) to `JobSource`.
+* The v3 target adds frozen submission-time prompt snapshot fields (`prompt_name`, `prompt_hash`, `system_prompt`, `user_prompt`, `temperature`, `top_p`) to `Job` and full output payloads (`raw_api_response`, `ai_metadata`) to `JobSource`.
 * The v3 target adds image asset verification fields (`file_hash`, `file_size_bytes`) to `Source`.
-* Database operations must utilize the `JSONBCompat` decorator to remain database-agnostic (SQLite for local development/testing and PostgreSQL for production).
+* Database operations must utilize `JSONBCompat` and the existing SQLModel/SQLAlchemy abstractions to preserve the same logical schema and JSON behavior across the supported backends, while keeping PostgreSQL as the intended production database.
 * Async CRUD lives in `DocumentService`, `JobService`, `TranscriptionService`, and upload helpers. Their queries and relationship loading must be updated for v3 fields.
+* Canonical operator tooling must remain OS-independent; safety workflows such as destructive-test backup and restore should run through Python or other cross-platform entry points rather than platform-specific shells.
 
 ## Implementation
 
@@ -26,28 +27,36 @@ Use a fresh database. There will be no migrations, data conversion, legacy compa
 
 ### 2. Update Data Services and Async Worker Layer
 
-* Update `JobService` and worker tasks (`worker.py`) to construct and save page-level input prompt fields (`prompt_name`, `prompt_hash`, `system_prompt`, `user_prompt`, `temperature`, `top_p`) directly onto `JobSource` records upon execution.
+* Update job creation and worker orchestration so prompt configuration is resolved at submission and frozen onto `Job` (`prompt_name`, `prompt_hash`, `system_prompt`, `user_prompt`, `temperature`, `top_p`) before execution starts.
 * Update `TranscriptionService` and provider adapters to store the complete unedited API REST response dictionary into `job_source.raw_api_response` alongside operational metrics in `job_source.ai_metadata`.
 * Update upload handlers to calculate and store file metadata (`file_hash` via SHA-256, `file_size_bytes`) on `Source` records during file ingestion.
+* Remove legacy single-source compatibility flows so worker paths persist per-page outcomes only through `JobSource` updates.
 
 ### 3. Update Integration Tests and Mock AI Providers
 
 * Update mock provider fixtures in test suites to return realistic complete API response envelopes.
 * Verify test coverage for `JSONBCompat` field writes and reads under SQLite in-memory test databases.
-* Add assertions in async workflow tests to verify page-level prompt provenance and failure isolation on `JobSource`.
+* Add assertions in async workflow tests to verify frozen prompt snapshot fields on `Job`, plus per-page failure isolation and output evidence on `JobSource`.
 
 ### 4. Update the UI for the v3 Schema
 
 * Review the UI components and views displaying document, job, person, and source data so they reference v3 schema properties instead of v2 relationships.
 * Ensure the UI correctly renders `COALESCE(revised_text, raw_transcription)` for page viewing and inline editing.
-* Ensure the "Retry Failed Pages" UI action spawns targeted jobs correctly using page-level `JobSource` failure states.
+* Ensure resubmit actions only queue failed pages and preserve frozen prompt snapshot behavior on the existing `Job`.
 * Consider the guidance in `docs/ui_style_guide.md` when making UI changes so updated views remain consistent with the project’s visual conventions.
 
+### 5. Keep Operational Tooling Portable
+
+* Implement destructive-test backup and restore workflows in Python so the canonical path runs on Windows, Linux, and macOS.
+* Avoid making core developer or recovery procedures depend on PowerShell-only or shell-specific semantics.
+* Keep operational documentation aligned with the cross-platform command path used by the repository.
+
 ## Done When
 
 * A fresh database is created directly from the v3 SQLModel metadata.
-* Full input/output provenance is captured on `JobSource` for every AI execution task.
+* Frozen prompt input provenance is captured on `Job` for each submission, and full per-page output evidence is captured on `JobSource` for every AI execution task.
 * The focused tests and full test suite pass on both SQLite and PostgreSQL backends.
+* Canonical operator workflows required for development and destructive-test recovery run without a Windows-only shell dependency.
 
 ## Out of Scope
 

docs/index_v3.md

@@ -10,8 +10,9 @@ Read [architecture_v3.md](https://www.google.com/search?q=architecture_v3.md) fi
 
 * **Folder & Multi-Image Ingestion:** Upload one or more images that map sequentially (`page_number`) under a single `Document`.
 * **Parallel Async AI Vision Engine:** Concurrently process single-page image transcriptions using Python `asyncio` bounded by rate limiters.
-* **Portable Relational Storage:** Flexible relational persistence using SQLModel and SQLAlchemy supporting SQLite during local development/testing and PostgreSQL in production.
-* **Complete Auditability & Provenance:** Capture full input prompts (`system_prompt`, `user_prompt`), hyperparameters (`temperature`, `top_p`), operational metrics (`ai_metadata`), and full provider response envelopes (`raw_api_response`) on every page execution (`JobSource`).
+* **Portable Relational Storage:** SQLModel and SQLAlchemy preserve a portable relational model across the supported backends, with SQLite for local development/testing and PostgreSQL as the production database target.
+* **Cross-Platform Operations:** Canonical developer and recovery workflows run through Python-based, OS-independent tooling rather than platform-specific shell scripts.
+* **Complete Auditability & Provenance:** Capture frozen submission-time input prompts (`system_prompt`, `user_prompt`) and hyperparameters (`temperature`, `top_p`) on `Job`, plus per-page operational metrics (`ai_metadata`) and full provider response envelopes (`raw_api_response`) on `JobSource`.
 * **Asset Integrity Tracking:** Calculate and store cryptographic hashes (SHA-256) and file sizes on `Source` image records while preserving clean filesystem storage.
 * **Pydantic V2 Validation:** End-to-end type safety, DB row mapping, and JSON payload validation.
 * **Historical Person Management:** Track authors and recipients across documents with rich biographical entities (`Person`).

docs/requirements_v3.md

@@ -9,15 +9,16 @@ This document captures the **Version 3 baseline requirements** for the productio
 | REQ-0 | System | Provide end-to-end multi-page document transcription with persistent, inspectable async job states. | demonstration |
 | REQ-1 | Functional | Allow users to upload multi-image batches as sequential `Source` pages under a `Document`. | test |
 | REQ-2 | Functional | Process multi-page jobs asynchronously using an `asyncio` worker pool bounded by rate limits. | test |
-| REQ-3 | Functional | Persist page-level execution parameters, full input prompts, and output responses (`system_prompt`, `user_prompt`, `raw_transcription`, `ai_metadata`, `raw_api_response`) on `JobSource`. | test |
+| REQ-3 | Functional | Persist frozen submission-time execution parameters and full input prompts (`system_prompt`, `user_prompt`, `prompt_name`, `prompt_hash`, `temperature`, `top_p`) on `Job`, and persist page-level output responses (`raw_transcription`, `ai_metadata`, `raw_api_response`) on `JobSource`. | test |
 | REQ-4 | Functional | Support job states (`queued`, `processing`, `completed`, `partial_success`, `failed`) and page states (`pending`, `transcribed`, `failed`). | inspection |
 | REQ-5 | Functional | Allow users to manage historical `Person` records and link multiple authors/recipients to a `Document` via `DocumentPerson`. | test |
 | REQ-6 | Functional | Maintain immutable original machine output on `Source.raw_transcription` while permitting inline human edits on `Source.revised_text`. | test |
-| REQ-7 | Data Constraint | Support relational persistence via SQLModel/SQLAlchemy across database backends (SQLite for local testing/development and PostgreSQL for production). | inspection |
+| REQ-7 | Data Constraint | Use SQLModel/SQLAlchemy to preserve a portable relational domain model and compatible data shape across the supported backends, with SQLite for local development/testing and PostgreSQL as the production system of record. | inspection |
 | REQ-8 | Data Constraint | Validate all API requests, database rows, and JSON structures using Pydantic V2 schemas and SQLModel. | test |
 | REQ-9 | Interface | Render multi-page transcriptions sequentially by `page_number` in the web UI with author/recipient metadata. | demonstration |
-| REQ-10 | Operations | Allow operators to retry only failed pages for jobs in a `partial_success` state. | test |
+| REQ-10 | Operations | Allow operators to resubmit only failed pages for queued reprocessing while preserving the frozen prompt snapshot on the existing `Job`. | test |
 | REQ-11 | Data Constraint | Calculate and store cryptographic file hashes (SHA-256) and file sizes for uploaded source images to track asset integrity. | test |
+| REQ-12 | Operations Constraint | Keep core development, testing, restore, and recovery workflows OS-independent across Windows, Linux, and macOS; do not require a platform-specific shell for canonical project processes. | inspection |
 
 ## Element Satisfaction Mapping
 
@@ -26,6 +27,7 @@ This document captures the **Version 3 baseline requirements** for the productio
 * **WORKER (asyncio):** Satisfies REQ-2, REQ-3, REQ-4, REQ-10.
 * **PERSISTENCE (SQLModel/SQLAlchemy):** Satisfies REQ-3, REQ-6, REQ-7, REQ-11.
 * **MODELS (Pydantic V2 / SQLModel):** Satisfies REQ-8.
+* **OPERATIONS TOOLING (Python / OS-neutral automation):** Satisfies REQ-12.
 
 ---
 

docs/schema_v3.md

@@ -1,6 +1,6 @@
 # Database Schema (Version 3)
 
-This document describes the relational schema for the transcription platform. It incorporates multi-image batch orchestration, page-level execution tracking, many-to-many author/recipient attribution, input prompt provenance capture, and raw API payload evidence for archival auditing.
+This document describes the relational schema for the transcription platform. It incorporates multi-image batch orchestration, page-level execution tracking, many-to-many author/recipient attribution, submission-time prompt snapshot capture, and raw API payload evidence for archival auditing.
 
 The schema uses generic JSON columns compatible with SQLite in local development and PostgreSQL native JSONB/UUID types in production.
 
@@ -54,6 +54,12 @@ JOB {
     INTEGER retry_count
     TEXT provider
     TEXT model
+    TEXT prompt_name
+    TEXT prompt_hash
+    TEXT system_prompt
+    TEXT user_prompt
+    FLOAT temperature
+    FLOAT top_p
     TIMESTAMPTZ date_created
     TIMESTAMPTZ date_updated
 }
@@ -79,12 +85,6 @@ JOB_SOURCE {
     UUID source_id FK
     VARCHAR status "pending | transcribed | failed"
     TEXT raw_transcription
-    TEXT prompt_name
-    TEXT prompt_hash
-    TEXT system_prompt
-    TEXT user_prompt
-    FLOAT temperature
-    FLOAT top_p
     JSONB ai_metadata
     JSONB raw_api_response
     TEXT error_detail
@@ -104,7 +104,7 @@ SOURCE ||--o{ JOB_SOURCE : "processed_in"
 ### Page-Level Execution & AI Outputs
 
 * **Execution Granularity:** Every single image execution attempt by an AI model produces a dedicated record in `job_source`.
-* **Page-Level Input Provenance:** Every `job_source` execution captures its exact hyperparameters (`temperature`, `top_p`), prompt identifier details (`prompt_name`, `prompt_hash`), and full prompt text strings (`system_prompt`, `user_prompt`) used for that specific page call.
+* **Submission Snapshot Provenance:** Every `job` captures the frozen prompt identifier details (`prompt_name`, `prompt_hash`), full prompt text strings (`system_prompt`, `user_prompt`), and hyperparameters (`temperature`, `top_p`) at submission time.
 * **Point-in-Time Output Auditability:** `job_source.raw_api_response` stores the complete, unedited provider REST response envelope for that specific image page call. `job_source.ai_metadata` stores spatial bounding boxes, normalized token usage, latency, and cost details for fast querying.
 * **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

@@ -270,8 +270,8 @@ class JobService(ServiceBase):
             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 def resubmit_failed_sources(self, *, job_id: UUID, session: AsyncSession | None = None) -> int:
+        """Reset failed source executions and queue the job for reprocessing."""
         async with self._session_scope(session) as _session:
             query = (
                 select(Job)
@@ -292,12 +292,12 @@ 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.TRANSCRIBED]
+            candidates = [job_source for job_source in job.job_sources if job_source.status == JobSourceStatus.FAILED]
             if not candidates:
                 raise JobResubmitBlockedError(
-                    "Job has no non-transcribed sources to resubmit",
+                    "Job has no failed sources to resubmit",
                     category=ErrorCategory.VALIDATION,
-                    suggestion="Only failed or pending sources can be resubmitted.",
+                    suggestion="Only failed sources can be resubmitted.",
                 )
 
             now = datetime.now(UTC)

src/transcription/services/transcription.py

@@ -346,90 +346,6 @@ class TranscriptionService(ServiceBase):
             result = await _session.exec(query)
             return result.all()
 
-    async def transcribe_document(
-        self,
-        image_path: str | Path,
-        job_id: UUID,
-        *,
-        prompt_name: str = DEFAULT_PROMPT_FILE,
-        session: AsyncSession | None = None,
-    ) -> None:
-        """Transcribe a local image using the configured prompt and provider."""
-        result = await transcribe_document_image(
-            image_path=image_path,
-            prompt_name=prompt_name,
-            settings=self.settings,
-            provider=self.provider,
-        )
-        await self.update_job_transcription(
-            job_id=job_id,
-            text=result.text,
-            error_detail=None,
-            provider=result.provider,
-            model=result.model,
-            prompt_name=result.prompt_name,
-            prompt_hash=result.prompt_hash,
-            system_prompt=result.system_prompt,
-            user_prompt=result.user_prompt,
-            temperature=result.temperature,
-            top_p=result.top_p,
-            session=session,
-        )
-
-    async def update_job_transcription(
-        self,
-        *,
-        job_id: UUID,
-        text: str | None,
-        error_detail: str | None = None,
-        provider: str | None = None,
-        model: str | None = None,
-        prompt_name: str = DEFAULT_PROMPT_FILE,
-        prompt_hash: str | None = None,
-        system_prompt: str | None = None,
-        user_prompt: str | None = None,
-        temperature: float | None = None,
-        top_p: float | None = None,
-        session: AsyncSession | None = None,
-    ) -> 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:
-                raise TranscriptionNotFoundError(
-                    f"Job with id {job_id} not found",
-                    category=ErrorCategory.NOT_FOUND,
-                    suggestion="Verify the job id 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.date_updated = datetime.now(UTC)
-
-            source = await _session.exec(
-                select(Source)
-                .where(Source.document_id == job.document_id)
-                .order_by(Source.page_number)  # pyright: ignore[reportArgumentType]
-            )
-            source_row = source.first()
-            if source_row is not None:
-                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,
-                    session=_session,
-                )
-
-            await self._finalize(session=_session, caller_session=session, refresh=(job,))
-            return job
-
     async def update_job_source_transcription(
         self,
         *,

src/transcription/services/workflows.py

@@ -213,144 +213,6 @@ async def process_next_queued_job(
     return True
 
 
-async def _finalize_transcribed(
-    *,
-    job: Job,
-    services: ServiceBundle,
-    result: TranscriptionResult,
-    session: AsyncSession | None = None,
-) -> Job:
-    """Transaction B: job transcription output + TRANSCRIBED in one commit."""
-    if session is None:
-        async with services.jobs._session_scope() as local_session:
-            await services.transcriptions.update_job_transcription(
-                job_id=job.id,
-                text=result.text,
-                error_detail=None,
-                provider=result.provider,
-                model=result.model,
-                prompt_name=result.prompt_name,
-                session=local_session,
-            )
-            updated_job = await services.jobs.mark_job_status(
-                job.id,
-                JobStatus.TRANSCRIBED,
-                session=local_session,
-            )
-            await local_session.commit()
-            return updated_job
-
-    await services.transcriptions.update_job_transcription(
-        job_id=job.id,
-        text=result.text,
-        error_detail=None,
-        provider=result.provider,
-        model=result.model,
-        prompt_name=result.prompt_name,
-        session=session,
-    )
-    updated_job = await services.jobs.mark_job_status(
-        job.id,
-        JobStatus.TRANSCRIBED,
-        session=session,
-    )
-    await session.commit()
-    return updated_job
-
-
-async def _finalize_retry(
-    *,
-    job: Job,
-    services: ServiceBundle,
-    error: AppError,
-    settings: Settings,
-    session: AsyncSession | None = None,
-) -> Job:
-    """Transaction C: job error detail + QUEUED + retry increment in one commit."""
-    if session is None:
-        async with services.jobs._session_scope() as local_session:
-            await services.transcriptions.update_job_transcription(
-                job_id=job.id,
-                text=None,
-                error_detail=format_error_detail(error),
-                prompt_name=DEFAULT_PROMPT_FILE,
-                session=local_session,
-            )
-            updated_job = await services.jobs.update_job_state(
-                job_id=job.id,
-                status=JobStatus.QUEUED,
-                retry_count_increment=1,
-                session=local_session,
-            )
-            await local_session.commit()
-    else:
-        await services.transcriptions.update_job_transcription(
-            job_id=job.id,
-            text=None,
-            error_detail=format_error_detail(error),
-            prompt_name=DEFAULT_PROMPT_FILE,
-            session=session,
-        )
-        updated_job = await services.jobs.update_job_state(
-            job_id=job.id,
-            status=JobStatus.QUEUED,
-            retry_count_increment=1,
-            session=session,
-        )
-        await session.commit()
-
-    if settings.worker_retry_backoff_seconds > 0:
-        await asyncio.sleep(settings.worker_retry_backoff_seconds)
-    return updated_job
-
-
-async def _finalize_failed(
-    *,
-    job: Job,
-    services: ServiceBundle,
-    error: AppError,
-    session: AsyncSession | None = None,
-) -> Job:
-    """Transaction B: job error detail + FAILED in one commit."""
-    if session is None:
-        async with services.jobs._session_scope() as local_session:
-            await services.transcriptions.update_job_transcription(
-                job_id=job.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,
-                JobStatus.FAILED,
-                session=local_session,
-            )
-            await local_session.commit()
-            return updated_job
-
-    await services.transcriptions.update_job_transcription(
-        job_id=job.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,
-        JobStatus.FAILED,
-        session=session,
-    )
-    await session.commit()
-    return updated_job
-
-
-def _resolve_primary_source(job: Job) -> Source | None:
-    if not job.job_sources:
-        return 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:

src/transcription/ui/pages/jobs_page.py

@@ -219,7 +219,7 @@ def register_page() -> None:  # noqa: PLR0915
             ui.label("Job not found").classes("text-h6 ui-text-danger p-4")
             return
 
-        non_transcribed_count = sum(1 for js in job.job_sources if js.status != JobSourceStatus.TRANSCRIBED)
+        failed_count = sum(1 for js in job.job_sources if js.status == JobSourceStatus.FAILED)
 
         with ui.column().classes("w-full max-w-xl mx-auto p-4 gap-4"):
             page_header("Resubmit Job")
@@ -227,14 +227,14 @@ 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("Non-Transcribed Sources:", str(non_transcribed_count))
+                metadata_row("Failed Sources:", str(failed_count))
                 ui.label(
-                    "Resubmit queues all non-transcribed linked sources. New results overwrite prior page-level results."
+                    "Resubmit queues only failed 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)
+                    resubmitted_count = await jobs_service.resubmit_failed_sources(job_id=job.id)
                 except JobResubmitBlockedError as exc:
                     ui.notify(exc.message, type="warning")
                     return

tests/services/test_job_service.py

@@ -294,7 +294,7 @@ class TestJobService:
         assert pending_entry.error_detail == "Cancelled by user"
 
     @pytest.mark.asyncio
-    async def test_resubmit_non_transcribed_sources_resets_only_non_transcribed(
+    async def test_resubmit_failed_sources_resets_only_failed(
         self,
         job_service: JobService,
         document_service: DocumentService,
@@ -349,7 +349,7 @@ class TestJobService:
             )
             await session.commit()
 
-        count = await job_service.resubmit_non_transcribed_sources(job_id=job.id)
+        count = await job_service.resubmit_failed_sources(job_id=job.id)
         assert count == 1
 
         refreshed = await job_service.read_job(job_id=job.id)
@@ -364,7 +364,63 @@ class TestJobService:
         assert transcribed_entry.status == JobSourceStatus.TRANSCRIBED
 
     @pytest.mark.asyncio
-    async def test_resubmit_non_transcribed_sources_blocks_when_processing(
+    async def test_resubmit_failed_sources_blocks_when_only_pending_or_transcribed(
+        self,
+        job_service: JobService,
+        document_service: DocumentService,
+    ):
+        document = Document(id=uuid4(), name="resubmit-no-failed-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-pending.jpg",
+                filename="stored-resubmit-pending.jpg",
+                file_path="/uploads/stored-resubmit-pending.jpg",
+                file_hash="1" * 64,
+                file_size_bytes=1,
+            )
+            source_two = Source(
+                document_id=document.id,
+                page_number=2,
+                upload_name="resubmit-done.jpg",
+                filename="stored-resubmit-done.jpg",
+                file_path="/uploads/stored-resubmit-done.jpg",
+                file_hash="2" * 64,
+                file_size_bytes=1,
+                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.PENDING,
+                )
+            )
+            session.add(
+                JobSource(
+                    job_id=job.id,
+                    source_id=source_two.id,
+                    status=JobSourceStatus.TRANSCRIBED,
+                    raw_transcription="done text",
+                )
+            )
+            await session.commit()
+
+        with pytest.raises(JobResubmitBlockedError):
+            await job_service.resubmit_failed_sources(job_id=job.id)
+
+    @pytest.mark.asyncio
+    async def test_resubmit_failed_sources_blocks_when_processing(
         self,
         job_service: JobService,
         document_service: DocumentService,
@@ -376,7 +432,7 @@ class TestJobService:
         await job_service.create_job(job=job)
 
         with pytest.raises(JobResubmitBlockedError):
-            await job_service.resubmit_non_transcribed_sources(job_id=job.id)
+            await job_service.resubmit_failed_sources(job_id=job.id)
 
     @pytest.mark.asyncio
     async def test_cancel_job_blocks_transcribed_terminal_jobs(

tests/ui/test_jobs_page.py

@@ -130,7 +130,7 @@ class TestJobsPageRendering:
 
         assert response.status_code == 200
         assert "Resubmit Job" in response.text
-        assert "Non-Transcribed Sources:" in response.text
+        assert "Failed Sources:" in response.text
         assert "Resubmit now" in response.text
 
     @pytest.mark.asyncio

tools/run_destructive_tests.py

@@ -0,0 +1,251 @@
+from __future__ import annotations
+
+import argparse
+import ctypes
+import os
+from ctypes import wintypes
+from datetime import datetime
+from pathlib import Path
+import shlex
+import shutil
+import subprocess
+import sys
+
+
+RETRY_CANCEL_CHOICES = {"n", "no", "c", "cancel", "a", "abort", "q", "quit"}
+
+
+def show_phase(title: str) -> None:
+    print()
+    print(f"========== {title} ==========")
+
+
+def test_file_unlocked(path: Path) -> bool:
+    if not path.exists():
+        return True
+
+    if os.name == "nt":
+        return _test_file_unlocked_windows(path)
+
+    return _test_file_unlocked_posix(path)
+
+
+def _test_file_unlocked_windows(path: Path) -> bool:
+    generic_read = 0x80000000
+    generic_write = 0x40000000
+    open_existing = 3
+    file_attribute_normal = 0x80
+    invalid_handle_value = wintypes.HANDLE(-1).value
+
+    kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
+    kernel32.CreateFileW.argtypes = [
+        wintypes.LPCWSTR,
+        wintypes.DWORD,
+        wintypes.DWORD,
+        wintypes.LPVOID,
+        wintypes.DWORD,
+        wintypes.DWORD,
+        wintypes.HANDLE,
+    ]
+    kernel32.CreateFileW.restype = wintypes.HANDLE
+    kernel32.CloseHandle.argtypes = [wintypes.HANDLE]
+    kernel32.CloseHandle.restype = wintypes.BOOL
+
+    handle = kernel32.CreateFileW(
+        str(path),
+        generic_read | generic_write,
+        0,
+        None,
+        open_existing,
+        file_attribute_normal,
+        None,
+    )
+    if handle == invalid_handle_value:
+        return False
+
+    kernel32.CloseHandle(handle)
+    return True
+
+
+def _test_file_unlocked_posix(path: Path) -> bool:
+    import fcntl
+
+    fd = os.open(path, os.O_RDWR)
+    try:
+        try:
+            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError:
+            return False
+        else:
+            fcntl.flock(fd, fcntl.LOCK_UN)
+            return True
+    finally:
+        os.close(fd)
+
+
+def wait_for_restore_preflight(db_file_path: Path) -> bool:
+    attempt = 1
+    while not test_file_unlocked(db_file_path):
+        show_phase("Restore Preflight")
+        print(f"WARNING: Restore preflight blocked: database appears to be in use: {db_file_path}")
+        print("Close conflicting applications (for example DB Browser for SQLite, uvicorn, or any process using this DB).")
+        print("Restore is paused and waiting for your input.")
+        answer = input(
+            f"Attempt {attempt}: type 'retry' to check again, or 'cancel' to skip restore: "
+        ).strip()
+        print(f"Input received: '{answer}'")
+        if answer.lower() in RETRY_CANCEL_CHOICES:
+            return False
+        attempt += 1
+        print("Re-checking database lock now...")
+
+    if attempt > 1:
+        print("Restore preflight passed: database lock released.")
+    return True
+
+
+def restore_backup(backup_path: Path, data_path: Path) -> None:
+    if not backup_path.exists():
+        raise FileNotFoundError(f"Backup path not found: {backup_path}")
+
+    if data_path.exists():
+        shutil.rmtree(data_path)
+
+    shutil.copytree(backup_path, data_path)
+    print(f"Restored data from backup: {backup_path}")
+
+
+def parse_args(argv: list[str]) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Run potentially destructive tests with backup/restore protection."
+    )
+    parser.add_argument("--auto-restore", action="store_true", help="Restore immediately after successful tests.")
+    parser.add_argument("--keep-backup", action="store_true", help="Keep the backup even after a successful restore.")
+    parser.add_argument(
+        "--skip-restore-prompt",
+        action="store_true",
+        help="Do not prompt after successful tests; keep the backup for a later manual restore.",
+    )
+    parser.add_argument(
+        "--restore-from",
+        help="Restore from an existing backup name or absolute backup path instead of running tests.",
+    )
+    parser.add_argument("command", nargs=argparse.REMAINDER, help="Command to run after '--'.")
+    args = parser.parse_args(argv)
+
+    if args.restore_from and args.command:
+        parser.error("--restore-from cannot be combined with a test command.")
+
+    if not args.restore_from and not args.command:
+        parser.error("A test command is required unless --restore-from is provided.")
+
+    return args
+
+
+def normalize_command(command: list[str]) -> list[str]:
+    if command and command[0] == "--":
+        return command[1:]
+    return command
+
+
+def run_command(command: list[str]) -> int:
+    show_phase("Test Phase")
+    print(f"Running command: {shlex.join(command)}")
+    try:
+        completed = subprocess.run(command, check=False)
+    except OSError as exc:
+        print(f"ERROR: {exc}", file=sys.stderr)
+        return 1
+    return completed.returncode
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv or sys.argv[1:])
+    repo_root = Path(__file__).resolve().parent.parent
+    data_dir = repo_root / "data"
+    backup_root = repo_root / ".test-backups"
+    db_path = data_dir / "transcription.db"
+
+    if not data_dir.exists():
+        raise FileNotFoundError(f"Data directory not found: {data_dir}")
+
+    backup_root.mkdir(parents=True, exist_ok=True)
+
+    if args.restore_from:
+        restore_path = Path(args.restore_from)
+        if not restore_path.is_absolute():
+            restore_path = backup_root / restore_path
+
+        show_phase("Restore Phase")
+        if not wait_for_restore_preflight(db_path):
+            print(f"Restore cancelled. Backup preserved at: {restore_path}")
+            return 1
+
+        restore_backup(restore_path, data_dir)
+        return 0
+
+    command = normalize_command(list(args.command))
+    if not command:
+        raise ValueError("No test command provided.")
+
+    show_phase("Backup Phase")
+    if not test_file_unlocked(db_path):
+        print(f"WARNING: Backup preflight warning: database appears to be in use: {db_path}")
+        print("WARNING: Proceeding with backup, but hot backups can capture an in-flight state.")
+
+    timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
+    backup_name = f"data-backup-{timestamp}"
+    backup_path = backup_root / backup_name
+    shutil.copytree(data_dir, backup_path)
+    print(f"Created backup: {backup_path}")
+
+    test_exit_code = run_command(command)
+    if test_exit_code != 0:
+        show_phase("Post-Test")
+        print(f"Tests failed (exit code {test_exit_code}).")
+        print(f"Backup preserved at: {backup_path}")
+        print("Inspect state, then restore manually if needed:")
+        print(f"  uv run python tools/run_destructive_tests.py --restore-from {backup_name}")
+        return test_exit_code
+
+    should_restore = False
+    if args.auto_restore:
+        should_restore = True
+    elif args.skip_restore_prompt:
+        should_restore = False
+    else:
+        answer = input("Tests passed. Restore data backup now? [y/N] ").strip().lower()
+        if answer in {"y", "yes"}:
+            should_restore = True
+
+    if not should_restore:
+        print(f"Restore skipped by user. Backup kept at: {backup_path}")
+        print("Restore later with:")
+        print(f"  uv run python tools/run_destructive_tests.py --restore-from {backup_name}")
+        return 0
+
+    show_phase("Restore Phase")
+    if not wait_for_restore_preflight(db_path):
+        print(f"Restore cancelled. Backup preserved at: {backup_path}")
+        return 1
+
+    try:
+        restore_backup(backup_path, data_dir)
+        if args.keep_backup:
+            print(f"Kept backup: {backup_path}")
+        else:
+            shutil.rmtree(backup_path)
+            print(f"Deleted backup: {backup_path}")
+    except Exception as exc:
+        print("WARNING: Restore failed. Your current data remains unchanged.")
+        print(f"WARNING: {exc}")
+        print("Likely cause: another process has data/transcription.db open.")
+        print("Stop the process and retry restore with:")
+        print(f"  uv run python tools/run_destructive_tests.py --restore-from {backup_name}")
+        return 1
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())