From e6549277c6361ac3fabae0dbcd58dd938851bd5f Mon Sep 17 00:00:00 2001 From: Jim Lancaster <40281233+zoltan57@users.noreply.github.com> Date: Sun, 9 Aug 2026 10:51:30 -0500 Subject: [PATCH] V3 Updated V3 core documents. Added data folder backup/restore before/after running destructive tests. --- .gitignore | 3 + README.md | 44 ++++ docs/architecture_v3.md | 34 +-- docs/implementation_plan_v3.md | 21 +- docs/index_v3.md | 5 +- docs/requirements_v3.md | 8 +- docs/schema_v3.md | 16 +- src/transcription/services/jobs.py | 10 +- src/transcription/services/transcription.py | 84 ------- src/transcription/services/workflows.py | 138 ----------- src/transcription/ui/pages/jobs_page.py | 8 +- tests/services/test_job_service.py | 64 ++++- tests/ui/test_jobs_page.py | 2 +- tools/run_destructive_tests.py | 251 ++++++++++++++++++++ 14 files changed, 417 insertions(+), 271 deletions(-) create mode 100644 tools/run_destructive_tests.py diff --git a/.gitignore b/.gitignore index a665861..8879741 100644 --- a/.gitignore +++ b/.gitignore @@ -18,3 +18,6 @@ wheels/ # Document images uploads/* data/* + +# Local destructive-test backups +.test-backups/ diff --git a/README.md b/README.md index 29e2be2..d8876e6 100644 --- a/README.md +++ b/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. diff --git a/docs/architecture_v3.md b/docs/architecture_v3.md index 5500935..f859732 100644 --- a/docs/architecture_v3.md +++ b/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) diff --git a/docs/implementation_plan_v3.md b/docs/implementation_plan_v3.md index 2b49167..87ce1f1 100644 --- a/docs/implementation_plan_v3.md +++ b/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 diff --git a/docs/index_v3.md b/docs/index_v3.md index 3f90a73..cbac3b1 100644 --- a/docs/index_v3.md +++ b/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`). diff --git a/docs/requirements_v3.md b/docs/requirements_v3.md index dc26d69..9a01c7c 100644 --- a/docs/requirements_v3.md +++ b/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. --- diff --git a/docs/schema_v3.md b/docs/schema_v3.md index ab30411..5bb69cd 100644 --- a/docs/schema_v3.md +++ b/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. diff --git a/src/transcription/services/jobs.py b/src/transcription/services/jobs.py index ccdb07f..5a3b207 100644 --- a/src/transcription/services/jobs.py +++ b/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) diff --git a/src/transcription/services/transcription.py b/src/transcription/services/transcription.py index 041ceae..3bfe990 100644 --- a/src/transcription/services/transcription.py +++ b/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, *, diff --git a/src/transcription/services/workflows.py b/src/transcription/services/workflows.py index a1e9ada..5a73626 100644 --- a/src/transcription/services/workflows.py +++ b/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: diff --git a/src/transcription/ui/pages/jobs_page.py b/src/transcription/ui/pages/jobs_page.py index 344b3a5..163d5f3 100644 --- a/src/transcription/ui/pages/jobs_page.py +++ b/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 diff --git a/tests/services/test_job_service.py b/tests/services/test_job_service.py index 6f04d97..8996263 100644 --- a/tests/services/test_job_service.py +++ b/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( diff --git a/tests/ui/test_jobs_page.py b/tests/ui/test_jobs_page.py index 52b54f1..497dbf3 100644 --- a/tests/ui/test_jobs_page.py +++ b/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 diff --git a/tools/run_destructive_tests.py b/tools/run_destructive_tests.py new file mode 100644 index 0000000..4de9355 --- /dev/null +++ b/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()) \ No newline at end of file