Sprint 7: persistence layer and job orchestration

Complete persistence and job lifecycle management:

Persistence:
- file_store.py: artifact storage abstraction for all job outputs (raw
payload, canonical JSON, ALTO XML, PAGE XML, viewer JSON, events log,
input images). Provider profile CRUD. Handles local and /data paths
- db.py: SQLite with auto-creating schema for jobs and providers tables.
Job CRUD with full state serialization (status, timestamps, warnings,
artifact flags). Provider record storage

Job system:
- models.py: Job model with 5-state machine (queued→running→succeeded/
failed/partial_success). Duration tracking, summary view, artifact flags
- events.py: EventLog with context-manager step() that auto-times each
pipeline phase. skip() for non-eligible exports. Failure capture.
13 named JobSteps matching §23.1
- service.py: JobService orchestrates the full pipeline:
1.receive_file → 5.save_raw → 6.normalize → 7.enrich → 8.validate →
9.compute_readiness → 10.export_alto → 11.export_page → 13.persist
Exports are conditional on eligibility. Failed jobs captured with error
and events. Enricher pipeline configurable

34 new tests (file_store: 15, database: 8, events: 7, job service: 5
integration tests including full lifecycle, failure capture, events
logging, and canonical JSON validation from disk).
411 total tests passing.

https://claude.ai/code/session_01Cuzvc9Pjfo5u46eT3ta2Cg

src/app/jobs/events.py

@@ -0,0 +1,103 @@
+"""Job events — structured log of each pipeline step."""
+
+from __future__ import annotations
+
+import time
+from contextlib import contextmanager
+from datetime import datetime, timezone
+from enum import Enum
+from typing import Generator
+
+from pydantic import BaseModel, Field
+
+
+class JobStep(str, Enum):
+    """Named steps in the processing pipeline."""
+
+    RECEIVE_FILE = "receive_file"
+    CREATE_CONTEXT = "create_context"
+    RESOLVE_PROVIDER = "resolve_provider"
+    EXECUTE_RUNTIME = "execute_runtime"
+    SAVE_RAW = "save_raw"
+    NORMALIZE = "normalize"
+    ENRICH = "enrich"
+    VALIDATE = "validate"
+    COMPUTE_READINESS = "compute_readiness"
+    EXPORT_ALTO = "export_alto"
+    EXPORT_PAGE = "export_page"
+    BUILD_VIEWER = "build_viewer"
+    PERSIST = "persist"
+
+
+class JobEvent(BaseModel):
+    """A single pipeline step event."""
+
+    step: JobStep
+    status: str = "started"  # started, completed, failed, skipped
+    started_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+    completed_at: datetime | None = None
+    duration_ms: float | None = None
+    message: str | None = None
+    error: str | None = None
+
+
+class EventLog:
+    """Collects events during job execution."""
+
+    def __init__(self) -> None:
+        self._events: list[JobEvent] = []
+
+    @property
+    def events(self) -> list[JobEvent]:
+        return list(self._events)
+
+    def add(self, event: JobEvent) -> None:
+        self._events.append(event)
+
+    @contextmanager
+    def step(self, step: JobStep) -> Generator[JobEvent, None, None]:
+        """Context manager that auto-times a pipeline step."""
+        event = JobEvent(step=step, status="started")
+        self._events.append(event)
+        t0 = time.monotonic()
+        try:
+            yield event
+            elapsed = (time.monotonic() - t0) * 1000
+            # Since JobEvent is a Pydantic model, update via index
+            idx = len(self._events) - 1
+            self._events[idx] = event.model_copy(update={
+                "status": "completed",
+                "completed_at": datetime.now(timezone.utc),
+                "duration_ms": elapsed,
+            })
+        except Exception as exc:
+            elapsed = (time.monotonic() - t0) * 1000
+            idx = len(self._events) - 1
+            self._events[idx] = event.model_copy(update={
+                "status": "failed",
+                "completed_at": datetime.now(timezone.utc),
+                "duration_ms": elapsed,
+                "error": str(exc),
+            })
+            raise
+
+    def skip(self, step: JobStep, reason: str) -> None:
+        """Record a skipped step."""
+        self._events.append(JobEvent(
+            step=step,
+            status="skipped",
+            completed_at=datetime.now(timezone.utc),
+            message=reason,
+        ))
+
+    def to_dicts(self) -> list[dict]:
+        """Serialize all events for persistence."""
+        return [e.model_dump(mode="json") for e in self._events]
+
+    @property
+    def has_failures(self) -> bool:
+        return any(e.status == "failed" for e in self._events)
+
+    @property
+    def total_duration_ms(self) -> float:
+        return sum(e.duration_ms or 0 for e in self._events)

src/app/jobs/models.py

@@ -0,0 +1,69 @@
+"""Job models — represents a single OCR processing run."""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from enum import Enum
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+
+class JobStatus(str, Enum):
+    """State machine for job lifecycle."""
+
+    QUEUED = "queued"
+    RUNNING = "running"
+    SUCCEEDED = "succeeded"
+    FAILED = "failed"
+    PARTIAL_SUCCESS = "partial_success"
+
+
+class Job(BaseModel):
+    """A single OCR processing job."""
+
+    job_id: str = Field(min_length=1)
+    status: JobStatus = JobStatus.QUEUED
+
+    # Input
+    provider_id: str = Field(min_length=1)
+    provider_family: str = Field(min_length=1)
+    source_filename: str | None = None
+    image_width: int | None = Field(default=None, gt=0)
+    image_height: int | None = Field(default=None, gt=0)
+
+    # Results
+    has_raw_payload: bool = False
+    has_canonical: bool = False
+    has_alto: bool = False
+    has_page_xml: bool = False
+    has_viewer: bool = False
+
+    # Timing
+    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+    started_at: datetime | None = None
+    completed_at: datetime | None = None
+
+    # Errors and warnings
+    error: str | None = None
+    warnings: list[str] = Field(default_factory=list)
+
+    @property
+    def duration_ms(self) -> float | None:
+        if self.started_at and self.completed_at:
+            return (self.completed_at - self.started_at).total_seconds() * 1000
+        return None
+
+    def to_summary(self) -> dict[str, Any]:
+        """Lightweight summary for list views."""
+        return {
+            "job_id": self.job_id,
+            "status": self.status.value,
+            "provider_id": self.provider_id,
+            "source_filename": self.source_filename,
+            "created_at": self.created_at.isoformat(),
+            "duration_ms": self.duration_ms,
+            "has_alto": self.has_alto,
+            "has_page_xml": self.has_page_xml,
+            "error": self.error,
+        }

src/app/jobs/service.py

@@ -0,0 +1,223 @@
+"""Job service — orchestrates the full processing pipeline.
+
+Pipeline steps (§23.1):
+  1. receive file
+  2. create execution context
+  3. resolve provider
+  4. execute runtime (skipped in V1 — raw payload provided externally)
+  5. save raw payload
+  6. normalize
+  7. enrich
+  8. validate
+  9. compute readiness
+  10. export ALTO
+  11. export PAGE
+  12. build viewer (placeholder)
+  13. persist
+"""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+
+from src.app.domain.models import CanonicalDocument, RawProviderPayload
+from src.app.domain.models.geometry import GeometryContext
+from src.app.enrichers import EnricherPipeline
+from src.app.enrichers.bbox_repair_light import BboxRepairLightEnricher
+from src.app.enrichers.hyphenation_basic import HyphenationBasicEnricher
+from src.app.enrichers.lang_propagation import LangPropagationEnricher
+from src.app.enrichers.polygon_to_bbox import PolygonToBboxEnricher
+from src.app.enrichers.reading_order_simple import ReadingOrderSimpleEnricher
+from src.app.enrichers.text_consistency import TextConsistencyEnricher
+from src.app.jobs.events import EventLog, JobStep
+from src.app.jobs.models import Job, JobStatus
+from src.app.normalization.pipeline import normalize
+from src.app.persistence.db import Database
+from src.app.persistence.file_store import FileStore
+from src.app.policies.document_policy import DocumentPolicy
+from src.app.policies.export_policy import check_alto_export, check_page_export
+from src.app.serializers.alto_xml import serialize_alto
+from src.app.serializers.page_xml import serialize_page_xml
+from src.app.validators.export_eligibility_validator import compute_export_eligibility
+from src.app.validators.structural_validator import validate_structure
+
+
+def _default_enricher_pipeline() -> EnricherPipeline:
+    return EnricherPipeline([
+        PolygonToBboxEnricher(),
+        BboxRepairLightEnricher(),
+        LangPropagationEnricher(),
+        ReadingOrderSimpleEnricher(),
+        HyphenationBasicEnricher(),
+        TextConsistencyEnricher(),
+    ])
+
+
+class JobService:
+    """Orchestrates the full document processing pipeline."""
+
+    def __init__(self, db: Database, file_store: FileStore) -> None:
+        self._db = db
+        self._store = file_store
+
+    def create_job(
+        self,
+        provider_id: str,
+        provider_family: str,
+        source_filename: str | None = None,
+    ) -> Job:
+        """Create a new job record."""
+        job = Job(
+            job_id=f"job_{uuid.uuid4().hex[:12]}",
+            provider_id=provider_id,
+            provider_family=provider_family,
+            source_filename=source_filename,
+        )
+        self._db.save_job(job)
+        return job
+
+    def run_job(
+        self,
+        job: Job,
+        raw_payload: RawProviderPayload,
+        image_width: int,
+        image_height: int,
+        *,
+        image_path: Path | None = None,
+        policy: DocumentPolicy | None = None,
+        enricher_pipeline: EnricherPipeline | None = None,
+    ) -> Job:
+        """Execute the full pipeline for a job.
+
+        In V1, the raw_payload is provided externally (runtime execution
+        is not yet integrated). The pipeline runs synchronously.
+        """
+        if policy is None:
+            policy = DocumentPolicy()
+        if enricher_pipeline is None:
+            enricher_pipeline = _default_enricher_pipeline()
+
+        events = EventLog()
+        job = job.model_copy(update={
+            "status": JobStatus.RUNNING,
+            "started_at": datetime.now(timezone.utc),
+            "image_width": image_width,
+            "image_height": image_height,
+        })
+        self._db.save_job(job)
+
+        canonical: CanonicalDocument | None = None
+        warnings: list[str] = []
+
+        try:
+            # Step 1: receive file
+            with events.step(JobStep.RECEIVE_FILE):
+                if image_path and image_path.exists():
+                    self._store.save_input_image(job.job_id, image_path)
+
+            # Step 5: save raw payload
+            with events.step(JobStep.SAVE_RAW):
+                self._store.save_raw_payload(
+                    job.job_id, raw_payload.model_dump(mode="json")
+                )
+                job = job.model_copy(update={"has_raw_payload": True})
+
+            # Step 6: normalize
+            with events.step(JobStep.NORMALIZE):
+                geo_ctx = GeometryContext(
+                    source_width=image_width, source_height=image_height
+                )
+                canonical = normalize(
+                    raw_payload,
+                    family=job.provider_family,
+                    geometry_context=geo_ctx,
+                    document_id=job.job_id,
+                    source_filename=job.source_filename,
+                )
+
+            # Step 7: enrich
+            with events.step(JobStep.ENRICH):
+                canonical = enricher_pipeline.run(canonical, policy)
+
+            # Step 8: validate
+            with events.step(JobStep.VALIDATE):
+                struct_report = validate_structure(
+                    canonical, bbox_tolerance=policy.bbox_containment_tolerance
+                )
+                for entry in struct_report.warnings:
+                    warnings.append(f"[structural] {entry.path}: {entry.message}")
+                for entry in struct_report.errors:
+                    warnings.append(f"[structural:ERROR] {entry.path}: {entry.message}")
+
+            # Step 9: compute readiness + export eligibility
+            with events.step(JobStep.COMPUTE_READINESS):
+                eligibility = compute_export_eligibility(canonical, policy)
+
+            # Save canonical
+            self._store.save_canonical(
+                job.job_id, canonical.model_dump(mode="json")
+            )
+            job = job.model_copy(update={"has_canonical": True})
+
+            # Step 10: export ALTO
+            alto_decision = check_alto_export(eligibility, policy)
+            if alto_decision.allowed:
+                with events.step(JobStep.EXPORT_ALTO):
+                    alto_bytes = serialize_alto(canonical)
+                    self._store.save_alto(job.job_id, alto_bytes)
+                    job = job.model_copy(update={"has_alto": True})
+            else:
+                events.skip(JobStep.EXPORT_ALTO, alto_decision.reason)
+
+            # Step 11: export PAGE
+            page_decision = check_page_export(eligibility, policy)
+            if page_decision.allowed:
+                with events.step(JobStep.EXPORT_PAGE):
+                    page_bytes = serialize_page_xml(canonical)
+                    self._store.save_page_xml(job.job_id, page_bytes)
+                    job = job.model_copy(update={"has_page_xml": True})
+            else:
+                events.skip(JobStep.EXPORT_PAGE, page_decision.reason)
+
+            # Step 12: build viewer (placeholder — full impl in Sprint 10)
+            events.skip(JobStep.BUILD_VIEWER, "Not yet implemented")
+
+            # Step 13: persist final state
+            with events.step(JobStep.PERSIST):
+                self._store.save_events(job.job_id, events.to_dicts())
+
+            # Determine final status
+            if job.has_alto or job.has_page_xml:
+                if job.has_alto and job.has_page_xml:
+                    final_status = JobStatus.SUCCEEDED
+                else:
+                    final_status = JobStatus.PARTIAL_SUCCESS
+            else:
+                final_status = JobStatus.PARTIAL_SUCCESS
+
+            job = job.model_copy(update={
+                "status": final_status,
+                "completed_at": datetime.now(timezone.utc),
+                "warnings": warnings,
+            })
+
+        except Exception as exc:
+            job = job.model_copy(update={
+                "status": JobStatus.FAILED,
+                "completed_at": datetime.now(timezone.utc),
+                "error": str(exc),
+                "warnings": warnings,
+            })
+            # Save events even on failure
+            self._store.save_events(job.job_id, events.to_dicts())
+
+        self._db.save_job(job)
+        return job
+
+    def get_job(self, job_id: str) -> Job | None:
+        return self._db.get_job(job_id)
+
+    def list_jobs(self, limit: int = 100, offset: int = 0) -> list[Job]:
+        return self._db.list_jobs(limit, offset)

src/app/persistence/db.py

@@ -0,0 +1,178 @@
+"""SQLite database — lightweight persistence for jobs and providers.
+
+Uses synchronous sqlite3 for V1 simplicity.  The schema auto-creates
+on first access.
+"""
+
+from __future__ import annotations
+
+import json
+import sqlite3
+from pathlib import Path
+from typing import Any
+
+from src.app.jobs.models import Job, JobStatus
+
+_SCHEMA = """
+CREATE TABLE IF NOT EXISTS jobs (
+    job_id TEXT PRIMARY KEY,
+    status TEXT NOT NULL DEFAULT 'queued',
+    provider_id TEXT NOT NULL,
+    provider_family TEXT NOT NULL,
+    source_filename TEXT,
+    image_width INTEGER,
+    image_height INTEGER,
+    has_raw_payload INTEGER NOT NULL DEFAULT 0,
+    has_canonical INTEGER NOT NULL DEFAULT 0,
+    has_alto INTEGER NOT NULL DEFAULT 0,
+    has_page_xml INTEGER NOT NULL DEFAULT 0,
+    has_viewer INTEGER NOT NULL DEFAULT 0,
+    created_at TEXT NOT NULL,
+    started_at TEXT,
+    completed_at TEXT,
+    error TEXT,
+    warnings TEXT NOT NULL DEFAULT '[]'
+);
+
+CREATE TABLE IF NOT EXISTS providers (
+    provider_id TEXT PRIMARY KEY,
+    data TEXT NOT NULL,
+    created_at TEXT NOT NULL,
+    updated_at TEXT NOT NULL
+);
+"""
+
+
+class Database:
+    """Thin wrapper around sqlite3 for jobs and providers."""
+
+    def __init__(self, db_path: Path) -> None:
+        self._path = db_path
+        self._path.parent.mkdir(parents=True, exist_ok=True)
+        self._conn: sqlite3.Connection | None = None
+
+    def connect(self) -> None:
+        self._conn = sqlite3.connect(str(self._path))
+        self._conn.row_factory = sqlite3.Row
+        self._conn.executescript(_SCHEMA)
+
+    def close(self) -> None:
+        if self._conn:
+            self._conn.close()
+            self._conn = None
+
+    @property
+    def conn(self) -> sqlite3.Connection:
+        if self._conn is None:
+            self.connect()
+        return self._conn  # type: ignore[return-value]
+
+    # -- Jobs -----------------------------------------------------------------
+
+    def save_job(self, job: Job) -> None:
+        self.conn.execute(
+            """INSERT OR REPLACE INTO jobs
+               (job_id, status, provider_id, provider_family, source_filename,
+                image_width, image_height, has_raw_payload, has_canonical,
+                has_alto, has_page_xml, has_viewer, created_at, started_at,
+                completed_at, error, warnings)
+               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+            (
+                job.job_id,
+                job.status.value,
+                job.provider_id,
+                job.provider_family,
+                job.source_filename,
+                job.image_width,
+                job.image_height,
+                int(job.has_raw_payload),
+                int(job.has_canonical),
+                int(job.has_alto),
+                int(job.has_page_xml),
+                int(job.has_viewer),
+                job.created_at.isoformat(),
+                job.started_at.isoformat() if job.started_at else None,
+                job.completed_at.isoformat() if job.completed_at else None,
+                job.error,
+                json.dumps(job.warnings),
+            ),
+        )
+        self.conn.commit()
+
+    def get_job(self, job_id: str) -> Job | None:
+        row = self.conn.execute(
+            "SELECT * FROM jobs WHERE job_id = ?", (job_id,)
+        ).fetchone()
+        if row is None:
+            return None
+        return self._row_to_job(row)
+
+    def list_jobs(self, limit: int = 100, offset: int = 0) -> list[Job]:
+        rows = self.conn.execute(
+            "SELECT * FROM jobs ORDER BY created_at DESC LIMIT ? OFFSET ?",
+            (limit, offset),
+        ).fetchall()
+        return [self._row_to_job(r) for r in rows]
+
+    @staticmethod
+    def _row_to_job(row: sqlite3.Row) -> Job:
+        from datetime import datetime
+
+        def _parse_dt(v: str | None) -> Any:
+            if v is None:
+                return None
+            return datetime.fromisoformat(v)
+
+        return Job(
+            job_id=row["job_id"],
+            status=JobStatus(row["status"]),
+            provider_id=row["provider_id"],
+            provider_family=row["provider_family"],
+            source_filename=row["source_filename"],
+            image_width=row["image_width"],
+            image_height=row["image_height"],
+            has_raw_payload=bool(row["has_raw_payload"]),
+            has_canonical=bool(row["has_canonical"]),
+            has_alto=bool(row["has_alto"]),
+            has_page_xml=bool(row["has_page_xml"]),
+            has_viewer=bool(row["has_viewer"]),
+            created_at=_parse_dt(row["created_at"]),
+            started_at=_parse_dt(row["started_at"]),
+            completed_at=_parse_dt(row["completed_at"]),
+            error=row["error"],
+            warnings=json.loads(row["warnings"]),
+        )
+
+    # -- Providers ------------------------------------------------------------
+
+    def save_provider_record(self, provider_id: str, data: dict) -> None:
+        from datetime import datetime, timezone
+
+        now = datetime.now(timezone.utc).isoformat()
+        self.conn.execute(
+            """INSERT OR REPLACE INTO providers (provider_id, data, created_at, updated_at)
+               VALUES (?, ?, COALESCE((SELECT created_at FROM providers WHERE provider_id = ?), ?), ?)""",
+            (provider_id, json.dumps(data, default=str), provider_id, now, now),
+        )
+        self.conn.commit()
+
+    def get_provider_record(self, provider_id: str) -> dict | None:
+        row = self.conn.execute(
+            "SELECT data FROM providers WHERE provider_id = ?", (provider_id,)
+        ).fetchone()
+        if row is None:
+            return None
+        return json.loads(row["data"])
+
+    def list_provider_records(self) -> list[dict]:
+        rows = self.conn.execute(
+            "SELECT data FROM providers ORDER BY created_at"
+        ).fetchall()
+        return [json.loads(r["data"]) for r in rows]
+
+    def delete_provider_record(self, provider_id: str) -> bool:
+        cursor = self.conn.execute(
+            "DELETE FROM providers WHERE provider_id = ?", (provider_id,)
+        )
+        self.conn.commit()
+        return cursor.rowcount > 0

src/app/persistence/file_store.py

@@ -0,0 +1,167 @@
+"""File store — abstraction for persisting job artifacts on disk.
+
+Storage layout:
+    {STORAGE_ROOT}/
+      jobs/
+        {job_id}/
+          input.{ext}        — original uploaded image
+          raw_payload.json   — RawProviderPayload
+          canonical.json     — CanonicalDocument
+          alto.xml           — ALTO XML export
+          page.xml           — PAGE XML export
+          viewer.json        — ViewerProjection
+          events.json        — job events log
+"""
+
+from __future__ import annotations
+
+import json
+import shutil
+from pathlib import Path
+from typing import Any
+
+
+class FileStore:
+    """Manages artifact persistence for jobs."""
+
+    def __init__(self, storage_root: Path) -> None:
+        self._root = storage_root
+        self._jobs_dir = storage_root / "jobs"
+        self._providers_dir = storage_root / "providers"
+
+    def ensure_dirs(self) -> None:
+        """Create required directories."""
+        self._jobs_dir.mkdir(parents=True, exist_ok=True)
+        self._providers_dir.mkdir(parents=True, exist_ok=True)
+
+    # -- Job directory --------------------------------------------------------
+
+    def job_dir(self, job_id: str) -> Path:
+        d = self._jobs_dir / job_id
+        d.mkdir(parents=True, exist_ok=True)
+        return d
+
+    # -- Save artifacts -------------------------------------------------------
+
+    def save_input_image(self, job_id: str, source_path: Path) -> Path:
+        """Copy the input image into the job directory."""
+        dest = self.job_dir(job_id) / f"input{source_path.suffix}"
+        shutil.copy2(source_path, dest)
+        return dest
+
+    def save_json(self, job_id: str, filename: str, data: Any) -> Path:
+        """Save a JSON-serializable object."""
+        dest = self.job_dir(job_id) / filename
+        dest.write_text(json.dumps(data, ensure_ascii=False, indent=2, default=str), encoding="utf-8")
+        return dest
+
+    def save_bytes(self, job_id: str, filename: str, data: bytes) -> Path:
+        """Save raw bytes (e.g. XML)."""
+        dest = self.job_dir(job_id) / filename
+        dest.write_bytes(data)
+        return dest
+
+    def save_raw_payload(self, job_id: str, data: dict) -> Path:
+        return self.save_json(job_id, "raw_payload.json", data)
+
+    def save_canonical(self, job_id: str, data: dict) -> Path:
+        return self.save_json(job_id, "canonical.json", data)
+
+    def save_alto(self, job_id: str, xml_bytes: bytes) -> Path:
+        return self.save_bytes(job_id, "alto.xml", xml_bytes)
+
+    def save_page_xml(self, job_id: str, xml_bytes: bytes) -> Path:
+        return self.save_bytes(job_id, "page.xml", xml_bytes)
+
+    def save_viewer(self, job_id: str, data: dict) -> Path:
+        return self.save_json(job_id, "viewer.json", data)
+
+    def save_events(self, job_id: str, events: list[dict]) -> Path:
+        return self.save_json(job_id, "events.json", events)
+
+    # -- Load artifacts -------------------------------------------------------
+
+    def load_json(self, job_id: str, filename: str) -> Any:
+        """Load a JSON file from the job directory. Returns None if not found."""
+        path = self._jobs_dir / job_id / filename
+        if not path.exists():
+            return None
+        return json.loads(path.read_text(encoding="utf-8"))
+
+    def load_bytes(self, job_id: str, filename: str) -> bytes | None:
+        """Load raw bytes. Returns None if not found."""
+        path = self._jobs_dir / job_id / filename
+        if not path.exists():
+            return None
+        return path.read_bytes()
+
+    def load_raw_payload(self, job_id: str) -> dict | None:
+        return self.load_json(job_id, "raw_payload.json")
+
+    def load_canonical(self, job_id: str) -> dict | None:
+        return self.load_json(job_id, "canonical.json")
+
+    def load_alto(self, job_id: str) -> bytes | None:
+        return self.load_bytes(job_id, "alto.xml")
+
+    def load_page_xml(self, job_id: str) -> bytes | None:
+        return self.load_bytes(job_id, "page.xml")
+
+    def load_viewer(self, job_id: str) -> dict | None:
+        return self.load_json(job_id, "viewer.json")
+
+    def load_events(self, job_id: str) -> list[dict] | None:
+        return self.load_json(job_id, "events.json")
+
+    # -- Input image ----------------------------------------------------------
+
+    def get_input_image_path(self, job_id: str) -> Path | None:
+        """Find the input image for a job (any extension)."""
+        d = self._jobs_dir / job_id
+        if not d.exists():
+            return None
+        for f in d.iterdir():
+            if f.stem == "input" and f.suffix in (".png", ".jpg", ".jpeg", ".tif", ".tiff", ".webp"):
+                return f
+        return None
+
+    # -- Provider profiles ----------------------------------------------------
+
+    def save_provider(self, provider_id: str, data: dict) -> Path:
+        dest = self._providers_dir / f"{provider_id}.json"
+        dest.write_text(json.dumps(data, ensure_ascii=False, indent=2, default=str), encoding="utf-8")
+        return dest
+
+    def load_provider(self, provider_id: str) -> dict | None:
+        path = self._providers_dir / f"{provider_id}.json"
+        if not path.exists():
+            return None
+        return json.loads(path.read_text(encoding="utf-8"))
+
+    def list_providers(self) -> list[str]:
+        if not self._providers_dir.exists():
+            return []
+        return [f.stem for f in self._providers_dir.glob("*.json")]
+
+    def delete_provider(self, provider_id: str) -> bool:
+        path = self._providers_dir / f"{provider_id}.json"
+        if path.exists():
+            path.unlink()
+            return True
+        return False
+
+    # -- Listing --------------------------------------------------------------
+
+    def list_jobs(self) -> list[str]:
+        if not self._jobs_dir.exists():
+            return []
+        return sorted(
+            [d.name for d in self._jobs_dir.iterdir() if d.is_dir()],
+            reverse=True,
+        )
+
+    def job_exists(self, job_id: str) -> bool:
+        return (self._jobs_dir / job_id).exists()
+
+    def job_has_artifact(self, job_id: str, filename: str) -> bool:
+        return (self._jobs_dir / job_id / filename).exists()

tests/integration/test_job_service.py

@@ -0,0 +1,163 @@
+"""Integration test: full job lifecycle via JobService."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from src.app.domain.models import RawProviderPayload
+from src.app.jobs.models import JobStatus
+from src.app.jobs.service import JobService
+from src.app.persistence.db import Database
+from src.app.persistence.file_store import FileStore
+
+
+class TestJobService:
+    def _setup(self, tmp_storage: Path) -> tuple[JobService, Database, FileStore]:
+        db = Database(tmp_storage / "test.db")
+        db.connect()
+        store = FileStore(tmp_storage)
+        store.ensure_dirs()
+        return JobService(db, store), db, store
+
+    def test_full_pipeline_succeeds(self, tmp_storage: Path, fixtures_dir: Path) -> None:
+        svc, db, store = self._setup(tmp_storage)
+
+        with open(fixtures_dir / "paddle_ocr_sample.json") as f:
+            payload = json.load(f)
+
+        raw = RawProviderPayload(
+            provider_id="paddleocr",
+            adapter_id="adapter.word_box_json.v1",
+            runtime_type="local",
+            payload=payload,
+            image_width=2480, image_height=3508,
+        )
+
+        # Create job
+        job = svc.create_job(
+            provider_id="paddleocr",
+            provider_family="word_box_json",
+            source_filename="test.png",
+        )
+        assert job.status == JobStatus.QUEUED
+
+        # Run pipeline
+        result = svc.run_job(job, raw, image_width=2480, image_height=3508)
+
+        assert result.status == JobStatus.SUCCEEDED
+        assert result.has_raw_payload
+        assert result.has_canonical
+        assert result.has_alto
+        assert result.has_page_xml
+        assert result.error is None
+        assert result.duration_ms is not None
+        assert result.duration_ms > 0
+
+        # Verify artifacts on disk
+        assert store.load_raw_payload(result.job_id) is not None
+        assert store.load_canonical(result.job_id) is not None
+        assert store.load_alto(result.job_id) is not None
+        assert store.load_page_xml(result.job_id) is not None
+        assert store.load_events(result.job_id) is not None
+
+        # Verify in database
+        db_job = svc.get_job(result.job_id)
+        assert db_job is not None
+        assert db_job.status == JobStatus.SUCCEEDED
+
+        db.close()
+
+    def test_job_appears_in_list(self, tmp_storage: Path, fixtures_dir: Path) -> None:
+        svc, db, store = self._setup(tmp_storage)
+
+        with open(fixtures_dir / "paddle_ocr_sample.json") as f:
+            payload = json.load(f)
+
+        raw = RawProviderPayload(
+            provider_id="paddleocr", adapter_id="v1", runtime_type="local",
+            payload=payload, image_width=2480, image_height=3508,
+        )
+
+        job = svc.create_job("paddleocr", "word_box_json")
+        svc.run_job(job, raw, image_width=2480, image_height=3508)
+
+        jobs = svc.list_jobs()
+        assert len(jobs) >= 1
+        assert any(j.job_id == job.job_id for j in jobs)
+        db.close()
+
+    def test_failed_job_captured(self, tmp_storage: Path) -> None:
+        svc, db, store = self._setup(tmp_storage)
+
+        raw = RawProviderPayload(
+            provider_id="bad", adapter_id="v1", runtime_type="local",
+            payload={"not": "a list"},  # will cause adapter to fail
+        )
+
+        job = svc.create_job("bad", "word_box_json")
+        result = svc.run_job(job, raw, image_width=100, image_height=100)
+
+        assert result.status == JobStatus.FAILED
+        assert result.error is not None
+        assert "list payload" in result.error
+
+        # Should still be in DB
+        db_job = svc.get_job(result.job_id)
+        assert db_job is not None
+        assert db_job.status == JobStatus.FAILED
+
+        # Events should be saved even on failure
+        events = store.load_events(result.job_id)
+        assert events is not None
+        db.close()
+
+    def test_events_log_all_steps(self, tmp_storage: Path, fixtures_dir: Path) -> None:
+        svc, db, store = self._setup(tmp_storage)
+
+        with open(fixtures_dir / "paddle_ocr_sample.json") as f:
+            payload = json.load(f)
+
+        raw = RawProviderPayload(
+            provider_id="paddleocr", adapter_id="v1", runtime_type="local",
+            payload=payload, image_width=2480, image_height=3508,
+        )
+
+        job = svc.create_job("paddleocr", "word_box_json")
+        svc.run_job(job, raw, image_width=2480, image_height=3508)
+
+        events = store.load_events(job.job_id)
+        assert events is not None
+        steps = [e["step"] for e in events]
+        assert "receive_file" in steps
+        assert "save_raw" in steps
+        assert "normalize" in steps
+        assert "enrich" in steps
+        assert "validate" in steps
+        assert "compute_readiness" in steps
+        assert "export_alto" in steps
+        assert "export_page" in steps
+        assert "persist" in steps
+        db.close()
+
+    def test_canonical_json_on_disk_is_valid(self, tmp_storage: Path, fixtures_dir: Path) -> None:
+        svc, db, store = self._setup(tmp_storage)
+
+        with open(fixtures_dir / "paddle_ocr_sample.json") as f:
+            payload = json.load(f)
+
+        raw = RawProviderPayload(
+            provider_id="paddleocr", adapter_id="v1", runtime_type="local",
+            payload=payload, image_width=2480, image_height=3508,
+        )
+
+        job = svc.create_job("paddleocr", "word_box_json")
+        svc.run_job(job, raw, image_width=2480, image_height=3508)
+
+        # Load canonical from disk and validate it
+        from src.app.domain.models import CanonicalDocument
+        canon_data = store.load_canonical(job.job_id)
+        doc = CanonicalDocument.model_validate(canon_data)
+        assert doc.document_id == job.job_id
+        assert len(doc.pages) == 1
+        db.close()

tests/unit/test_database.py

@@ -0,0 +1,132 @@
+"""Tests for the SQLite database layer."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from src.app.jobs.models import Job, JobStatus
+from src.app.persistence.db import Database
+
+
+class TestDatabase:
+    def test_connect_creates_tables(self, tmp_storage: Path) -> None:
+        db = Database(tmp_storage / "test.db")
+        db.connect()
+        # Should not raise
+        db.conn.execute("SELECT * FROM jobs LIMIT 1")
+        db.conn.execute("SELECT * FROM providers LIMIT 1")
+        db.close()
+
+    def test_save_and_get_job(self, tmp_storage: Path) -> None:
+        db = Database(tmp_storage / "test.db")
+        db.connect()
+
+        job = Job(
+            job_id="job_001",
+            provider_id="paddle",
+            provider_family="word_box_json",
+            source_filename="test.png",
+        )
+        db.save_job(job)
+
+        loaded = db.get_job("job_001")
+        assert loaded is not None
+        assert loaded.job_id == "job_001"
+        assert loaded.status == JobStatus.QUEUED
+        assert loaded.provider_id == "paddle"
+        assert loaded.source_filename == "test.png"
+        db.close()
+
+    def test_update_job(self, tmp_storage: Path) -> None:
+        db = Database(tmp_storage / "test.db")
+        db.connect()
+
+        job = Job(job_id="job_002", provider_id="p", provider_family="f")
+        db.save_job(job)
+
+        updated = job.model_copy(update={
+            "status": JobStatus.SUCCEEDED,
+            "has_alto": True,
+        })
+        db.save_job(updated)
+
+        loaded = db.get_job("job_002")
+        assert loaded is not None
+        assert loaded.status == JobStatus.SUCCEEDED
+        assert loaded.has_alto is True
+        db.close()
+
+    def test_list_jobs(self, tmp_storage: Path) -> None:
+        db = Database(tmp_storage / "test.db")
+        db.connect()
+
+        for i in range(5):
+            db.save_job(Job(job_id=f"job_{i:03d}", provider_id="p", provider_family="f"))
+
+        jobs = db.list_jobs(limit=3)
+        assert len(jobs) == 3
+
+        all_jobs = db.list_jobs()
+        assert len(all_jobs) == 5
+        db.close()
+
+    def test_get_nonexistent_job(self, tmp_storage: Path) -> None:
+        db = Database(tmp_storage / "test.db")
+        db.connect()
+        assert db.get_job("nonexistent") is None
+        db.close()
+
+    def test_job_with_warnings(self, tmp_storage: Path) -> None:
+        db = Database(tmp_storage / "test.db")
+        db.connect()
+
+        job = Job(
+            job_id="job_warn",
+            provider_id="p",
+            provider_family="f",
+            warnings=["bbox overflow", "missing confidence"],
+        )
+        db.save_job(job)
+
+        loaded = db.get_job("job_warn")
+        assert loaded is not None
+        assert len(loaded.warnings) == 2
+        assert "bbox overflow" in loaded.warnings
+        db.close()
+
+    def test_job_with_error(self, tmp_storage: Path) -> None:
+        db = Database(tmp_storage / "test.db")
+        db.connect()
+
+        job = Job(
+            job_id="job_err",
+            provider_id="p",
+            provider_family="f",
+            status=JobStatus.FAILED,
+            error="Provider timeout",
+        )
+        db.save_job(job)
+
+        loaded = db.get_job("job_err")
+        assert loaded is not None
+        assert loaded.status == JobStatus.FAILED
+        assert loaded.error == "Provider timeout"
+        db.close()
+
+    def test_provider_crud(self, tmp_storage: Path) -> None:
+        db = Database(tmp_storage / "test.db")
+        db.connect()
+
+        data = {"provider_id": "paddle", "family": "word_box_json"}
+        db.save_provider_record("paddle", data)
+
+        loaded = db.get_provider_record("paddle")
+        assert loaded is not None
+        assert loaded["provider_id"] == "paddle"
+
+        all_providers = db.list_provider_records()
+        assert len(all_providers) == 1
+
+        assert db.delete_provider_record("paddle")
+        assert db.get_provider_record("paddle") is None
+        db.close()

tests/unit/test_file_store.py

@@ -0,0 +1,103 @@
+"""Tests for the file store."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from src.app.persistence.file_store import FileStore
+
+
+class TestFileStore:
+    def test_ensure_dirs(self, tmp_storage: Path) -> None:
+        store = FileStore(tmp_storage)
+        store.ensure_dirs()
+        assert (tmp_storage / "jobs").exists()
+        assert (tmp_storage / "providers").exists()
+
+    def test_save_and_load_json(self, tmp_storage: Path) -> None:
+        store = FileStore(tmp_storage)
+        data = {"key": "value", "number": 42}
+        store.save_json("job_001", "test.json", data)
+        loaded = store.load_json("job_001", "test.json")
+        assert loaded == data
+
+    def test_save_and_load_bytes(self, tmp_storage: Path) -> None:
+        store = FileStore(tmp_storage)
+        data = b"<alto>test</alto>"
+        store.save_bytes("job_001", "alto.xml", data)
+        loaded = store.load_bytes("job_001", "alto.xml")
+        assert loaded == data
+
+    def test_load_nonexistent_returns_none(self, tmp_storage: Path) -> None:
+        store = FileStore(tmp_storage)
+        assert store.load_json("missing", "test.json") is None
+        assert store.load_bytes("missing", "test.xml") is None
+
+    def test_save_raw_payload(self, tmp_storage: Path) -> None:
+        store = FileStore(tmp_storage)
+        store.save_raw_payload("job_001", {"payload": [1, 2, 3]})
+        loaded = store.load_raw_payload("job_001")
+        assert loaded == {"payload": [1, 2, 3]}
+
+    def test_save_canonical(self, tmp_storage: Path) -> None:
+        store = FileStore(tmp_storage)
+        store.save_canonical("job_001", {"document_id": "doc1"})
+        loaded = store.load_canonical("job_001")
+        assert loaded["document_id"] == "doc1"
+
+    def test_save_alto(self, tmp_storage: Path) -> None:
+        store = FileStore(tmp_storage)
+        xml = b"<?xml version='1.0'?><alto/>"
+        store.save_alto("job_001", xml)
+        assert store.load_alto("job_001") == xml
+
+    def test_save_page_xml(self, tmp_storage: Path) -> None:
+        store = FileStore(tmp_storage)
+        xml = b"<?xml version='1.0'?><PcGts/>"
+        store.save_page_xml("job_001", xml)
+        assert store.load_page_xml("job_001") == xml
+
+    def test_save_events(self, tmp_storage: Path) -> None:
+        store = FileStore(tmp_storage)
+        events = [{"step": "normalize", "status": "completed"}]
+        store.save_events("job_001", events)
+        loaded = store.load_events("job_001")
+        assert loaded == events
+
+    def test_list_jobs(self, tmp_storage: Path) -> None:
+        store = FileStore(tmp_storage)
+        store.save_json("job_aaa", "test.json", {})
+        store.save_json("job_bbb", "test.json", {})
+        jobs = store.list_jobs()
+        assert "job_aaa" in jobs
+        assert "job_bbb" in jobs
+
+    def test_job_exists(self, tmp_storage: Path) -> None:
+        store = FileStore(tmp_storage)
+        assert not store.job_exists("job_001")
+        store.save_json("job_001", "test.json", {})
+        assert store.job_exists("job_001")
+
+    def test_job_has_artifact(self, tmp_storage: Path) -> None:
+        store = FileStore(tmp_storage)
+        store.save_json("job_001", "canonical.json", {})
+        assert store.job_has_artifact("job_001", "canonical.json")
+        assert not store.job_has_artifact("job_001", "alto.xml")
+
+    def test_provider_crud(self, tmp_storage: Path) -> None:
+        store = FileStore(tmp_storage)
+        store.ensure_dirs()
+        store.save_provider("paddle", {"provider_id": "paddle", "family": "word_box_json"})
+        assert store.load_provider("paddle") is not None
+        assert "paddle" in store.list_providers()
+        assert store.delete_provider("paddle")
+        assert store.load_provider("paddle") is None
+
+    def test_save_input_image(self, tmp_storage: Path, tmp_path: Path) -> None:
+        store = FileStore(tmp_storage)
+        img = tmp_path / "test.png"
+        img.write_bytes(b"\x89PNG\r\n\x1a\n")
+        result = store.save_input_image("job_001", img)
+        assert result.exists()
+        assert result.name == "input.png"
+        assert store.get_input_image_path("job_001") == result

tests/unit/test_job_events.py

@@ -0,0 +1,72 @@
+"""Tests for job events and the event log."""
+
+from __future__ import annotations
+
+from src.app.jobs.events import EventLog, JobStep
+
+
+class TestEventLog:
+    def test_step_context_manager(self) -> None:
+        log = EventLog()
+        with log.step(JobStep.NORMALIZE) as event:
+            pass  # simulate work
+        assert len(log.events) == 1
+        assert log.events[0].step == JobStep.NORMALIZE
+        assert log.events[0].status == "completed"
+        assert log.events[0].duration_ms is not None
+        assert log.events[0].duration_ms >= 0
+
+    def test_step_failure(self) -> None:
+        log = EventLog()
+        try:
+            with log.step(JobStep.EXECUTE_RUNTIME):
+                raise RuntimeError("boom")
+        except RuntimeError:
+            pass
+        assert log.events[0].status == "failed"
+        assert log.events[0].error == "boom"
+
+    def test_skip(self) -> None:
+        log = EventLog()
+        log.skip(JobStep.EXPORT_ALTO, "Not eligible")
+        assert log.events[0].status == "skipped"
+        assert log.events[0].message == "Not eligible"
+
+    def test_has_failures(self) -> None:
+        log = EventLog()
+        with log.step(JobStep.NORMALIZE):
+            pass
+        assert not log.has_failures
+
+        try:
+            with log.step(JobStep.VALIDATE):
+                raise ValueError("bad")
+        except ValueError:
+            pass
+        assert log.has_failures
+
+    def test_total_duration(self) -> None:
+        log = EventLog()
+        with log.step(JobStep.NORMALIZE):
+            pass
+        with log.step(JobStep.VALIDATE):
+            pass
+        assert log.total_duration_ms >= 0
+
+    def test_to_dicts(self) -> None:
+        log = EventLog()
+        with log.step(JobStep.NORMALIZE):
+            pass
+        log.skip(JobStep.EXPORT_ALTO, "skip")
+        dicts = log.to_dicts()
+        assert len(dicts) == 2
+        assert dicts[0]["step"] == "normalize"
+        assert dicts[1]["step"] == "export_alto"
+
+    def test_multiple_steps_in_order(self) -> None:
+        log = EventLog()
+        for step in [JobStep.RECEIVE_FILE, JobStep.NORMALIZE, JobStep.VALIDATE]:
+            with log.step(step):
+                pass
+        steps = [e.step for e in log.events]
+        assert steps == [JobStep.RECEIVE_FILE, JobStep.NORMALIZE, JobStep.VALIDATE]