transcription/docs/mvp/mvp-step4.md

Step 4: services/upload.py + worker.py

Objective

Implement the MVP upload and background-processing pipeline so the system can:

1. Save uploaded files into UPLOAD_DIR
2. Create Document + Job(status="queued")
3. Process queued jobs in a worker loop:
   • queued -> processing
   • call Step 3 transcription service
   • persist Transcript
   • finalize as transcribed or failed

This step advances MVP Feature 1 + Feature 2 and supports REQ-1, REQ-2, REQ-3, REQ-4, REQ-6.

---

Scope

In scope

• src/transcription/services/upload.py
• src/transcription/worker.py
• Upload persistence logic and initial job creation
• Worker polling and single-job lifecycle execution
• Deterministic test coverage for upload + worker (default suite)

Out of scope

• UI integration and pages (Step 5)
• Queue infrastructure beyond in-process loop
• Async DB/session architecture refactor
• Broad production hardening beyond MVP needs

---

Planned Deliverables

Source files

• src/transcription/services/upload.py
• src/transcription/worker.py
• src/transcription/services/__init__.py (export updates as needed)

Test files

• tests/services/test_upload.py
• tests/services/test_worker.py

Optional external lane (already present pattern)

• reuse external marker for live-provider checks where appropriate
• keep external out of default lane

---

Required MCP Prompt References (for test workflow)

Apply these resources directly during Step 4 test creation:

1. resource://catalog/prompts/pytest-scaffold
2. resource://prompts/pytest-scaffold/document
3. resource://catalog/prompts/pytest-fill-scaffold
4. resource://prompts/pytest-fill-scaffold/document

And (as referenced by those prompts) apply relevant pytest skill references for:

• naming/hierarchy
• marker defaults
• SQLAlchemy sync testing behavior where applicable

---

Design Decisions

1. Upload service owns initial file + record creation

   • Writes file, creates Document, creates queued Job, returns IDs/path.

2. Worker owns lifecycle transitions

   • Worker is the single owner of queued -> processing -> terminal job state changes.

3. Worker uses Step 3 service boundary

   • Worker calls transcribe_document_image(...); no provider-specific SDK logic in worker.

4. Failure information is always persisted

   • On failure: store Transcript(text=None, error_detail=...) and set Job.status=failed.

5. Loop remains simple and stoppable

   • In-process polling loop with stop event and poll interval for MVP simplicity and testability.

---

Task-by-Task Execution Checklist

Phase A — Implement upload service (src/transcription/services/upload.py)

• Create UploadError exception
• Create UploadJobResult dataclass with:
  • document_id
  • job_id
  • stored_path
  • original_filename
• Add filename safety handling:
  • normalize to basename
  • avoid path traversal
  • collision-safe stored name (e.g., UUID prefix/suffix)
• Validate upload payload:
  • non-empty bytes required
  • extension in supported set (.jpg/.jpeg/.png/.tif/.tiff/.pdf)
• Ensure upload directory exists (mkdir(parents=True, exist_ok=True))
• Write file bytes to UPLOAD_DIR
• Persist DB records in one transaction:
  • Document(filename, file_path)
  • Job(document_id=..., status=queued)
• Return UploadJobResult
• Add logging for success/failure boundaries

---

Phase B — Implement worker core (src/transcription/worker.py)

• Add process_next_queued_job(...) -> bool
• Fetch oldest queued job
• Return False when no queued jobs exist
• Transition picked job to processing and update timestamp
• Resolve associated Document.file_path
• Call transcribe_document_image(image_path=...)
• On success:
  • insert/update transcript text
  • clear error detail
  • mark job transcribed
  • update timestamp
• On failure:
  • insert/update transcript with text=None, error_detail=...
  • mark job failed
  • update timestamp
• Commit terminal state and return True
• Add logs around job pickup, transition, and terminal outcome

---

Phase C — Implement worker loop (src/transcription/worker.py)

• Add run_worker_loop(...)
• Accept configurable stop event/signal
• Accept configurable poll interval
• Repeatedly call process_next_queued_job
• Sleep only when queue is empty
• Exit cleanly when stop event is set

---

Phase D — Exports

• Update src/transcription/services/__init__.py to expose upload APIs
• Keep existing transcription exports intact

---

Phase E — Tests via MCP scaffold -> fill flow

E1 Scaffold (structure only)

Use scaffold prompt workflow first for:

• src/transcription/services/upload.py
• src/transcription/worker.py

Expected scaffold targets:

• tests/services/test_upload.py
• tests/services/test_worker.py

Scaffold rules:

• Class hierarchy + method names + one-line docstrings only
• No assertions or implementation details in scaffold phase
• Keep method names concise and behavior-focused

Validation:

• uv run pytest --collect-only -q

E2 Fill scaffold (implementation)

Use fill prompt workflow for:

• tests/services/test_upload.py
• tests/services/test_worker.py
• stack: sqlalchemy-sync (or mixed if combining pure + DB behaviors)
• marker lane preference: unit and integration as appropriate
• strategy: minimal deterministic implementation

Fill rules (invariants):

• Preserve scaffold class names, method names, and one-line docstrings
• Do not rename/re-nest scaffolded tests unless explicitly approved
• One behavior target per test
• Minimal mocking; mock only network/nondeterministic boundaries

Suggested test coverage:

tests/services/test_upload.py

• creates file + document + queued job (integration)
• rejects empty bytes (unit)
• rejects unsupported extension (unit)
• writes collision-safe unique filename (integration)
• persisted job status is queued (integration)

tests/services/test_worker.py

• returns False when queue empty (integration)
• transitions queued -> processing -> transcribed on success (integration)
• stores transcript text on success (integration)
• transitions to failed and stores error_detail on failure (integration)
• updates existing transcript instead of duplicate create (integration)
• worker loop exits when stop event set (unit)

---

Marker Strategy

• unit: pure logic tests (filename handling, loop stop behavior, validation logic)
• integration: DB + service orchestration tests (SQLite/session/contracts)
• external: opt-in live provider tests only (not part of default Step 4 lane)

No new marker needed; reuse existing marker registration.

---

Validation Sequence (strict order)

• uv run pytest --collect-only -q
• uv run pytest -m unit -q (if unit tests touched)
• uv run pytest tests/services/test_upload.py -q
• uv run pytest tests/services/test_worker.py -q
• uv run pytest -q

---

Reporting Requirements (after implementation)

Implementation report must include:

1. Files created/updated
2. Fixture and marker decisions
3. MCP references used and why
4. Validation command results
5. Remaining risks/open questions (only blockers)

---

Guardrails

• Keep Step 4 independent from UI concerns.
• Do not call provider SDK directly from worker.
• Do not silently swallow exceptions.
• Always persist terminal job outcome.
• Keep default suite deterministic and fast.
• Preserve scaffold invariants during fill phase.

---

Definition of Done (Step 4)

Step 4 is complete when:

• Upload service writes file and creates Document + queued Job
• Worker processes queued jobs end-to-end using Step 3 transcription service
• Success path persists transcript text and sets transcribed
• Failure path persists error detail and sets failed
• Queue-empty path returns cleanly
• New tests pass and full suite is green (uv run pytest -q)
• Output report includes MCP reference usage + validation evidence