Add layout OCR pipeline and clean replica generation

app/logic/document_analysis.py

@@ -0,0 +1,207 @@
+from __future__ import annotations
+
+from typing import Any
+
+from sqlalchemy import func
+from sqlalchemy.orm import Session
+
+from app.models.document_analysis_version import DocumentAnalysisVersion
+from app.logic.layout_ocr import run_layout_ocr
+
+
+def _flatten_layout_lines(layout_json: dict | None) -> list[dict[str, Any]]:
+    if not layout_json:
+        return []
+    lines: list[dict[str, Any]] = []
+    for page in layout_json.get("pages", []) or []:
+        for line in page.get("lines", []) or []:
+            if isinstance(line, dict):
+                lines.append(line)
+    return lines
+
+
+def _layout_has_any_text(layout_json: dict | None) -> bool:
+    for line in _flatten_layout_lines(layout_json):
+        if (line.get("text") or "").strip():
+            return True
+    return False
+
+
+def _layout_has_usable_bboxes(layout_json: dict | None) -> bool:
+    for line in _flatten_layout_lines(layout_json):
+        bbox = line.get("bbox")
+        if (
+            isinstance(bbox, (list, tuple))
+            and len(bbox) == 4
+            and all(v is not None for v in bbox)
+        ):
+            return True
+    return False
+
+
+def _build_canonical_analysis_from_document(document) -> dict[str, Any]:
+    text_versions = sorted(
+        getattr(document, "text_versions", []) or [],
+        key=lambda tv: ((tv.version_number or 0), getattr(tv, "created_at", None) or 0),
+        reverse=True,
+    )
+
+    raw_ocr = next((tv for tv in text_versions if getattr(tv, "is_current", False) and tv.version_type == "raw_ocr"), None)
+    reviewed = next((tv for tv in text_versions if getattr(tv, "is_current", False) and tv.version_type == "reviewed"), None)
+
+    source_tv = reviewed or raw_ocr
+    layout_json = getattr(source_tv, "layout_json", None) if source_tv else None
+
+    extracted = None
+    extracted_rows = getattr(document, "extracted_fields", None) or []
+    if extracted_rows:
+        extracted = extracted_rows[0]
+
+    analysis = {
+        "schema_version": 1,
+        "analysis_type": "canonical",
+        "document_info": {
+            "document_id": document.document_id,
+            "document_type": getattr(document, "document_type", None),
+            "mime_type": getattr(document, "mime_type", None),
+            "current_path": getattr(document, "current_path", None),
+            "source_path": getattr(document, "source_path", None),
+            "canonical_filename": getattr(document, "canonical_filename", None),
+        },
+        "text_source": {
+            "raw_ocr_version_id": getattr(raw_ocr, "id", None) if raw_ocr else None,
+            "reviewed_version_id": getattr(reviewed, "id", None) if reviewed else None,
+            "active_version_id": getattr(source_tv, "id", None) if source_tv else None,
+            "active_version_type": getattr(source_tv, "version_type", None) if source_tv else None,
+        },
+        "pages": (layout_json or {}).get("pages", []) if isinstance(layout_json, dict) else [],
+        "semantic_candidates": {
+            "merchant": getattr(extracted, "merchant_normalized", None) if extracted else None,
+            "merchant_raw": getattr(extracted, "merchant_raw", None) if extracted else None,
+            "transaction_date": str(getattr(extracted, "transaction_date", None)) if extracted and getattr(extracted, "transaction_date", None) else None,
+            "total": str(getattr(extracted, "total", None)) if extracted and getattr(extracted, "total", None) is not None else None,
+            "tax": str(getattr(extracted, "tax", None)) if extracted and getattr(extracted, "tax", None) is not None else None,
+            "subtotal": str(getattr(extracted, "subtotal", None)) if extracted and getattr(extracted, "subtotal", None) is not None else None,
+        },
+        "quality": {
+            "text_present": _layout_has_any_text(layout_json),
+            "usable_layout": _layout_has_usable_bboxes(layout_json),
+            "usable_word_boxes": False,
+            "issues": [],
+        },
+    }
+
+    if not analysis["quality"]["text_present"]:
+        analysis["quality"]["issues"].append("no_text_in_layout")
+    if not analysis["quality"]["usable_layout"]:
+        analysis["quality"]["issues"].append("no_usable_bboxes")
+
+    return analysis
+
+
+def get_current_document_analysis(db: Session, document) -> DocumentAnalysisVersion | None:
+    return (
+        db.query(DocumentAnalysisVersion)
+        .filter(
+            DocumentAnalysisVersion.document_id == document.id,
+            DocumentAnalysisVersion.is_current.is_(True),
+        )
+        .order_by(DocumentAnalysisVersion.version_number.desc(), DocumentAnalysisVersion.id.desc())
+        .first()
+    )
+
+
+def ensure_document_analysis(db: Session, document, require_layout: bool = True) -> DocumentAnalysisVersion:
+    current = get_current_document_analysis(db, document)
+    if current and current.analysis_json:
+        quality = (current.analysis_json or {}).get("quality", {}) or {}
+        if not require_layout or quality.get("usable_layout"):
+            return current
+
+    analysis_json = _build_canonical_analysis_from_document(document)
+    quality = analysis_json.get("quality", {}) or {}
+
+    if require_layout and not quality.get("usable_layout"):
+        raise ValueError("document_analysis_missing_usable_layout")
+
+    db.query(DocumentAnalysisVersion).filter(
+        DocumentAnalysisVersion.document_id == document.id,
+        DocumentAnalysisVersion.is_current.is_(True),
+    ).update({"is_current": False}, synchronize_session=False)
+
+    next_version = (
+        db.query(func.max(DocumentAnalysisVersion.version_number))
+        .filter(DocumentAnalysisVersion.document_id == document.id)
+        .scalar()
+        or 0
+    ) + 1
+
+    row = DocumentAnalysisVersion(
+        document_id=document.id,
+        version_number=next_version,
+        analysis_type="canonical",
+        is_current=True,
+        created_by="ensure_document_analysis",
+        engine_name="internal_existing_ocr_adapter",
+        engine_version="v1",
+        model_name=None,
+        prompt_version=None,
+        quality_score=1.0 if quality.get("usable_layout") else 0.5 if quality.get("text_present") else 0.0,
+        quality_note=None,
+        quality_flags=quality.get("issues", []),
+        analysis_json=analysis_json,
+    )
+    db.add(row)
+    db.commit()
+    db.refresh(row)
+    return row
+
+
+
+def build_layout_ocr_analysis_for_document(document) -> dict[str, Any]:
+    current_path = getattr(document, "current_path", None)
+    if not current_path:
+        raise ValueError("Document has no current_path")
+
+    result = run_layout_ocr(current_path)
+    analysis_json = result.to_analysis_json()
+
+    pages = analysis_json.get("pages", []) or []
+    text_lines = []
+    usable_layout = False
+
+    for page in pages:
+        for line in page.get("lines", []) or []:
+            line_text = (line.get("text") or "").strip()
+            if line_text:
+                text_lines.append(line_text)
+            bbox = line.get("bbox")
+            if isinstance(bbox, (list, tuple)) and len(bbox) == 4 and all(v is not None for v in bbox):
+                usable_layout = True
+
+    issues: list[str] = []
+    if not text_lines:
+        issues.append("no_text_detected")
+    if not usable_layout:
+        issues.append("no_usable_bboxes")
+
+    analysis_json["text_source"] = {
+        "active_version_id": None,
+        "raw_ocr_version_id": None,
+        "reviewed_version_id": None,
+        "active_version_type": "layout_ocr",
+    }
+    analysis_json["quality"] = {
+        "text_present": bool(text_lines),
+        "usable_layout": usable_layout,
+        "usable_word_boxes": usable_layout,
+        "issues": issues,
+    }
+    analysis_json["text_content"] = "\n".join(text_lines)
+
+    analysis_json["engine"] = {
+        "name": result.engine_name,
+        "version": result.engine_version,
+    }
+
+    return analysis_json

app/logic/document_outputs.py

@@ -72,6 +72,10 @@ from sqlalchemy.orm import Session
 from app.core.config import FIELD_ENRICHED_ROOT, OCR_CORRECTED_ROOT
 from app.models.document import Document
 from app.models.document_version import DocumentVersion
+
+from app.models.document_replica_layout_version import DocumentReplicaLayoutVersion
+from app.models.document_replica_output import DocumentReplicaOutput
+from app.models.document_replica_review_state import DocumentReplicaReviewState
 from app.models.text_version import TextVersion
 
 
@@ -442,12 +446,20 @@ def create_ocr_corrected_pdf_version(db: Session, document: Document, output_pat
                 if not text_line:
                     continue
 
-                left, top, right, bottom = line["bbox"]
-
+                bbox = line.get("bbox")
+                if not bbox or not isinstance(bbox, (list, tuple)) or len(bbox) != 4:
+                    continue
+                try:
+                    left, top, right, bottom = [float(v) for v in bbox]
+                except (TypeError, ValueError):
+                    continue
+                if right <= left or bottom <= top:
+                    continue
                 pdf_x = left * scale_x
                 pdf_y = page_h - (bottom * scale_y)
                 box_width = max(10.0, (right - left) * scale_x)
                 box_height = max(6.0, (bottom - top) * scale_y)
+                box_height = max(6.0, (bottom - top) * scale_y)
 
                 font_size = _fit_font_size(text_line, box_width, box_height)
 
@@ -685,19 +697,10 @@ def save_ocr_corrected_pdf_current(db: Session, document: Document, output_path:
     except Exception:
         share_path_value = None
 
-    document.share_path = share_path_value
-    document.current_path = str(out_path)
-    document.canonical_filename = out_path.name
-    document.sha256_current = file_hash
-    db.add(document)
-
+    # Replica outputs are non-destructive exports for now.
+    # Do not replace the primary/current document path.
     db.commit()
 
-    keep_paths = {str(out_path)}
-    if document.share_path:
-        keep_paths.add(str(document.share_path))
-    _prune_old_saved_files(db, document, keep_paths)
-
 
 def save_field_enriched_pdf_current(db: Session, document: Document, output_path: Path) -> None:
     if not document.current_path:
@@ -736,3 +739,312 @@ def save_field_enriched_pdf_current(db: Session, document: Document, output_path
     if document.share_path:
         keep_paths.add(str(document.share_path))
     _prune_old_saved_files(db, document, keep_paths)
+
+
+def _next_replica_layout_version_number(db: Session, document_id: int) -> int:
+    return (
+        db.query(func.max(DocumentReplicaLayoutVersion.version_number))
+        .filter(DocumentReplicaLayoutVersion.document_id == document_id)
+        .scalar()
+        or 0
+    ) + 1
+
+
+def _get_current_replica_review_state(document: Document) -> DocumentReplicaReviewState | None:
+    rows = getattr(document, "replica_review_states", None) or []
+
+
+def _layout_has_any_text(layout_json: dict | None) -> bool:
+    if not layout_json:
+        return False
+    for page in layout_json.get("pages", []):
+        for line in page.get("lines", []):
+            if (line.get("text") or "").strip():
+                return True
+    return False
+
+
+def _layout_has_usable_bboxes(layout_json: dict | None) -> bool:
+    if not layout_json:
+        return False
+    for page in layout_json.get("pages", []):
+        for line in page.get("lines", []):
+            bbox = line.get("bbox")
+            if (
+                isinstance(bbox, (list, tuple))
+                and len(bbox) == 4
+                and all(v is not None for v in bbox)
+            ):
+                return True
+    return False
+    return rows[0] if rows else None
+
+
+def _get_replica_source_context(document: Document):
+    if not document.current_path:
+        raise ValueError("Document has no current_path")
+
+    current_file = Path(document.current_path)
+    if not current_file.exists():
+        raise FileNotFoundError(f"Current file not found: {current_file}")
+
+    raw_ocr = _latest_current_text_version(document, "raw_ocr")
+    reviewed = _latest_current_text_version(document, "reviewed")
+
+    if current_file.suffix.lower() != ".pdf":
+        raise ValueError("Replica PDF generation currently supports PDFs only")
+
+    if reviewed is not None and _layout_has_usable_bboxes(reviewed.layout_json):
+        return current_file, raw_ocr, reviewed, reviewed.layout_json, "reviewed"
+
+    if raw_ocr is not None and _layout_has_usable_bboxes(raw_ocr.layout_json):
+        return current_file, raw_ocr, reviewed, raw_ocr.layout_json, "raw_ocr"
+
+    if reviewed is not None and _layout_has_any_text(reviewed.layout_json):
+        return current_file, raw_ocr, reviewed, reviewed.layout_json, "reviewed_text_only"
+
+    if raw_ocr is not None and _layout_has_any_text(raw_ocr.layout_json):
+        return current_file, raw_ocr, reviewed, raw_ocr.layout_json, "raw_text_only"
+
+    return current_file, raw_ocr, reviewed, {"pages": []}, "no_layout"
+
+def build_replica_layout(document: Document, mode: str = "shared") -> dict:
+    current_file, raw_ocr, reviewed, source_layout, layout_source = _get_replica_source_context(document)
+    reader = PdfReader(str(current_file))
+
+    pages = []
+    page_layouts = {page["page"]: page for page in source_layout.get("pages", [])}
+
+    for page_num, pdf_page in enumerate(reader.pages, start=1):
+        page_w = float(pdf_page.mediabox.width)
+        page_h = float(pdf_page.mediabox.height)
+        page_layout = page_layouts.get(page_num, {"lines": []})
+        src_w = float(page_layout.get("image_width") or 1.0)
+        src_h = float(page_layout.get("image_height") or 1.0)
+        scale_x = page_w / src_w
+        scale_y = page_h / src_h
+
+        line_entries = []
+        for line in page_layout.get("lines", []):
+            text_line = (line.get("text") or "").strip()
+            if not text_line:
+                continue
+
+            bbox = line.get("bbox")
+            if not bbox or not isinstance(bbox, (list, tuple)) or len(bbox) != 4:
+                continue
+            try:
+                left, top, right, bottom = [float(v) for v in bbox]
+            except (TypeError, ValueError):
+                continue
+            if right <= left or bottom <= top:
+                continue
+            pdf_x = left * scale_x
+            pdf_y = page_h - (bottom * scale_y)
+            box_width = max(10.0, (right - left) * scale_x)
+            box_height = max(6.0, (bottom - top) * scale_y)
+            font_size = _fit_font_size(text_line, box_width, box_height)
+
+            line_entries.append(
+                {
+                    "text": text_line,
+                    "bbox_source": [left, top, right, bottom],
+                    "pdf_x": pdf_x,
+                    "pdf_y": pdf_y,
+                    "box_width": box_width,
+                    "box_height": box_height,
+                    "font_family_guess": "Helvetica",
+                    "font_size_guess": font_size,
+                    "text_color_guess": "#000000",
+                    "text_render_mode_clean": 0,
+                    "text_render_mode_scan_backed": 3,
+                }
+            )
+
+        pages.append(
+            {
+                "page": page_num,
+                "page_width": page_w,
+                "page_height": page_h,
+                "image_width": src_w,
+                "image_height": src_h,
+                "lines": line_entries,
+            }
+        )
+
+    return {
+        "schema_version": 1,
+        "mode_source": mode,
+        "current_path": str(current_file),
+        "text_version_source": {
+            "raw_ocr_version_id": raw_ocr.id if raw_ocr else None,
+            "reviewed_version_id": reviewed.id if reviewed else None,
+        },
+        "pages": pages,
+    }
+
+
+def _save_replica_layout_version(
+    db: Session,
+    document: Document,
+    layout_json: dict,
+    mode: str,
+    created_by: str = "save_replica_pdf",
+) -> DocumentReplicaLayoutVersion:
+    db.query(DocumentReplicaLayoutVersion).filter(
+        DocumentReplicaLayoutVersion.document_id == document.id,
+        DocumentReplicaLayoutVersion.is_current == True,  # noqa: E712
+    ).update({"is_current": False}, synchronize_session=False)
+
+    version = DocumentReplicaLayoutVersion(
+        document_id=document.id,
+        version_number=_next_replica_layout_version_number(db, document.id),
+        version_type="heuristic",
+        render_mode_source=mode,
+        is_current=True,
+        created_by=created_by,
+        quality_flags=[],
+        inference_metadata_json={"pipeline": "heuristic_replica_v1", "mode": mode},
+        layout_json=layout_json,
+    )
+    db.add(version)
+    db.flush()
+
+    state = _get_current_replica_review_state(document)
+    if state is None:
+        state = DocumentReplicaReviewState(document_id=document.id)
+        db.add(state)
+
+    state.current_replica_layout_version_id = version.id
+    state.is_reviewed = False
+    state.is_approved = False
+    state.needs_manual_adjustment = False
+    state.needs_model_retry = False
+    db.flush()
+
+    return version
+
+
+def _render_replica_pdf_from_layout(
+    current_file: Path,
+    layout_json: dict,
+    out_path: Path,
+    mode: str,
+) -> None:
+    reader = PdfReader(str(current_file))
+    out_path.parent.mkdir(parents=True, exist_ok=True)
+
+    with tempfile.TemporaryDirectory() as tmpdirname:
+        tmpdir = Path(tmpdirname)
+        images = _render_pdf_page_images(current_file, tmpdir)
+        overlay_pdf_path = tmpdir / "replica.pdf"
+        c = None
+
+        pages = {page["page"]: page for page in layout_json.get("pages", [])}
+
+        for page_num, img_path in enumerate(images, start=1):
+            pdf_page = reader.pages[page_num - 1]
+            page_w = float(pdf_page.mediabox.width)
+            page_h = float(pdf_page.mediabox.height)
+
+            if c is None:
+                c = canvas.Canvas(str(overlay_pdf_path), pagesize=(page_w, page_h))
+            else:
+                c.setPageSize((page_w, page_h))
+
+            if mode == "scan_backed":
+                c.drawImage(ImageReader(str(img_path)), 0, 0, width=page_w, height=page_h)
+
+            page_layout = pages.get(page_num, {"lines": []})
+
+            for line in page_layout.get("lines", []):
+                text_line = (line.get("text") or "").strip()
+                if not text_line:
+                    continue
+
+                text_obj = c.beginText()
+                text_obj.setTextRenderMode(3 if mode == "scan_backed" else 0)
+                text_obj.setFont(line.get("font_family_guess") or "Helvetica", float(line.get("font_size_guess") or 10))
+                text_obj.setTextOrigin(float(line["pdf_x"]), float(line["pdf_y"]) + 1)
+                text_obj.textLine(text_line)
+                c.drawText(text_obj)
+
+            c.showPage()
+
+        if c is None:
+            raise ValueError("Failed to build replica PDF")
+
+        c.save()
+        shutil.copy2(overlay_pdf_path, out_path)
+
+    compress_pdf_with_ghostscript(out_path)
+
+
+def save_replica_pdf(db: Session, document: Document, output_path: Path, mode: str) -> None:
+    if mode not in {"clean", "scan_backed"}:
+        raise ValueError(f"Unsupported replica mode: {mode}")
+
+    current_file, _, _, _, _ = _get_replica_source_context(document)
+    out_path = Path(output_path)
+    out_path = out_path.with_name(re.sub(r"_v\d+(?=\.[^.]+$)", "", out_path.name))
+
+    stem = re.sub(r"(_replica_clean|_replica_scan_backed)$", "", out_path.stem)
+    suffix = out_path.suffix or ".pdf"
+
+    if mode == "clean":
+        out_path = out_path.with_name(f"{stem}_replica_clean{suffix}")
+    else:
+        out_path = out_path.with_name(f"{stem}_replica_scan_backed{suffix}")
+
+    out_path.parent.mkdir(parents=True, exist_ok=True)
+
+    requested_mode = mode
+    actual_mode = mode
+
+    layout_json = build_replica_layout(document, mode=mode)
+
+    page_lines = []
+    for page in (layout_json.get("pages") or []):
+        page_lines.extend(page.get("lines") or [])
+
+    if mode == "clean" and not page_lines:
+        raise ValueError("clean_replica_has_no_renderable_lines")
+    if mode == "clean":
+        has_text = False
+        for page in layout_json.get("pages", []):
+            if page.get("lines"):
+                has_text = True
+                break
+        if not has_text:
+            actual_mode = "scan_backed"
+            out_path = out_path.with_name(f"{stem}_replica_scan_backed{suffix}")
+            layout_json = build_replica_layout(document, mode="scan_backed")
+
+    layout_version = _save_replica_layout_version(db, document, layout_json, mode=actual_mode)
+
+    _render_replica_pdf_from_layout(current_file, layout_json, out_path, mode=actual_mode)
+
+    file_hash = sha256_for_file(out_path)
+    file_size = out_path.stat().st_size
+
+    try:
+        mirror_path = _mirror_to_secondary_owner(document, out_path)
+        share_path_value = str(mirror_path) if mirror_path else None
+    except Exception:
+        share_path_value = None
+
+    output = DocumentReplicaOutput(
+        document_id=document.id,
+        replica_layout_version_id=layout_version.id,
+        output_type=actual_mode,
+        file_path=str(out_path),
+        sha256=file_hash,
+        file_size_bytes=file_size,
+        created_by="save_replica_pdf",
+        render_settings_json={"requested_mode": requested_mode, "actual_mode": actual_mode},
+    )
+    db.add(output)
+
+    # Replica outputs are non-destructive exports.
+    # Do not replace the primary/current document path or prune sibling files.
+    db.commit()

app/logic/layout_ocr.py

@@ -0,0 +1,153 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+import fitz
+import pytesseract
+from pdf2image import convert_from_path
+from PIL import Image
+
+
+@dataclass
+class LayoutOCRResult:
+    engine_name: str
+    engine_version: str
+    pages: list[dict[str, Any]]
+
+    def to_analysis_json(self) -> dict[str, Any]:
+        return {
+            "schema_version": 1,
+            "analysis_type": "canonical",
+            "engine": {
+                "name": self.engine_name,
+                "version": self.engine_version,
+            },
+            "pages": self.pages,
+        }
+
+
+def _group_words_into_lines(words: list[dict[str, Any]], y_tol: float = 12.0) -> list[dict[str, Any]]:
+    if not words:
+        return []
+
+    words = sorted(words, key=lambda w: (w["bbox"][1], w["bbox"][0]))
+    groups: list[list[dict[str, Any]]] = []
+
+    for word in words:
+        placed = False
+        wy = word["bbox"][1]
+        for group in groups:
+            gy = sum(item["bbox"][1] for item in group) / len(group)
+            if abs(wy - gy) <= y_tol:
+                group.append(word)
+                placed = True
+                break
+        if not placed:
+            groups.append([word])
+
+    lines: list[dict[str, Any]] = []
+    for group in groups:
+        group = sorted(group, key=lambda w: w["bbox"][0])
+        text = " ".join((w.get("text") or "").strip() for w in group).strip()
+        if not text:
+            continue
+        left = min(w["bbox"][0] for w in group)
+        top = min(w["bbox"][1] for w in group)
+        right = max(w["bbox"][2] for w in group)
+        bottom = max(w["bbox"][3] for w in group)
+        avg_height = max(1.0, sum((w["bbox"][3] - w["bbox"][1]) for w in group) / len(group))
+        lines.append(
+            {
+                "text": text,
+                "bbox": [left, top, right, bottom],
+                "confidence": None,
+                "font_family_guess": "Helvetica",
+                "font_size_guess": max(6.0, avg_height * 0.75),
+                "text_color_guess": "#000000",
+                "words": group,
+            }
+        )
+    return lines
+
+
+def run_layout_ocr(pdf_path: str | Path, dpi: int = 300) -> LayoutOCRResult:
+    pdf_path = Path(pdf_path)
+    if not pdf_path.exists():
+        raise FileNotFoundError(f"PDF not found: {pdf_path}")
+
+    doc = fitz.open(pdf_path)
+    pil_pages = convert_from_path(str(pdf_path), dpi=dpi)
+
+    pages: list[dict[str, Any]] = []
+
+    for idx, (pdf_page, pil_img) in enumerate(zip(doc, pil_pages), start=1):
+        page_w = float(pdf_page.rect.width)
+        page_h = float(pdf_page.rect.height)
+
+        if not isinstance(pil_img, Image.Image):
+            raise ValueError(f"Rendered page {idx} is not a PIL image")
+
+        img_w, img_h = pil_img.size
+        scale_x = page_w / float(img_w)
+        scale_y = page_h / float(img_h)
+
+        data = pytesseract.image_to_data(
+            pil_img,
+            output_type=pytesseract.Output.DICT,
+            config="--oem 3 --psm 6",
+        )
+
+        words: list[dict[str, Any]] = []
+        n = len(data.get("text", []))
+        for i in range(n):
+            text = (data["text"][i] or "").strip()
+            if not text:
+                continue
+
+            try:
+                conf = float(data["conf"][i])
+            except Exception:
+                conf = None
+
+            left_px = float(data["left"][i])
+            top_px = float(data["top"][i])
+            width_px = float(data["width"][i])
+            height_px = float(data["height"][i])
+
+            if width_px <= 0 or height_px <= 0:
+                continue
+
+            left = left_px * scale_x
+            top = top_px * scale_y
+            right = (left_px + width_px) * scale_x
+            bottom = (top_px + height_px) * scale_y
+
+            words.append(
+                {
+                    "text": text,
+                    "bbox": [left, top, right, bottom],
+                    "confidence": conf,
+                }
+            )
+
+        lines = _group_words_into_lines(words)
+
+        pages.append(
+            {
+                "page": idx,
+                "page_width": page_w,
+                "page_height": page_h,
+                "image_width": page_w,
+                "image_height": page_h,
+                "lines": lines,
+                "words": words,
+            }
+        )
+
+    return LayoutOCRResult(
+        engine_name="tesseract_layout",
+        engine_version=str(pytesseract.get_tesseract_version()),
+        pages=pages,
+    )

app/models/__init__.py

@@ -8,6 +8,7 @@ from app.models.document_additional_field import DocumentAdditionalField
 from app.models.document_preset import DocumentPreset
 
 __all__ = [
+    "DocumentAnalysisVersion",
     "Document",
     "DocumentVersion",
     "TextVersion",
@@ -18,3 +19,8 @@ __all__ = [
     "DocumentPreset",
 ]
 from app.models.document_naming_field import DocumentNamingField
+from app.models.document_replica_layout_version import DocumentReplicaLayoutVersion
+from app.models.document_replica_output import DocumentReplicaOutput
+from app.models.document_replica_review_state import DocumentReplicaReviewState
+import app.models.document_analysis_version
+from app.models.document_analysis_version import DocumentAnalysisVersion

app/models/document.py

@@ -105,3 +105,20 @@ class Document(Base):
         cascade="all, delete-orphan",
         uselist=False,
     )
+
+    replica_layout_versions: Mapped[list["DocumentReplicaLayoutVersion"]] = relationship(
+        back_populates="document",
+        cascade="all, delete-orphan",
+        order_by="DocumentReplicaLayoutVersion.version_number",
+    )
+    replica_outputs: Mapped[list["DocumentReplicaOutput"]] = relationship(
+        back_populates="document",
+        cascade="all, delete-orphan",
+    )
+    analysis_versions: Mapped[list["DocumentAnalysisVersion"]] = relationship(
+        "DocumentAnalysisVersion", back_populates="document", cascade="all, delete-orphan"
+    )
+    replica_review_states: Mapped[list["DocumentReplicaReviewState"]] = relationship(
+        back_populates="document",
+        cascade="all, delete-orphan",
+    )

app/models/document_analysis_version.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, Text, func
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from app.db.base import Base
+
+
+class DocumentAnalysisVersion(Base):
+    __tablename__ = "document_analysis_versions"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    document_id: Mapped[int] = mapped_column(ForeignKey("documents.id", ondelete="CASCADE"), index=True, nullable=False)
+
+    version_number: Mapped[int] = mapped_column(Integer, nullable=False)
+    analysis_type: Mapped[str] = mapped_column(String(50), nullable=False, default="canonical")
+    is_current: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
+
+    created_by: Mapped[str | None] = mapped_column(String(100), nullable=True)
+    engine_name: Mapped[str | None] = mapped_column(String(100), nullable=True)
+    engine_version: Mapped[str | None] = mapped_column(String(100), nullable=True)
+    model_name: Mapped[str | None] = mapped_column(String(200), nullable=True)
+    prompt_version: Mapped[str | None] = mapped_column(String(100), nullable=True)
+
+    quality_score: Mapped[float | None] = mapped_column(nullable=True)
+    quality_note: Mapped[str | None] = mapped_column(Text, nullable=True)
+    quality_flags: Mapped[dict | list | None] = mapped_column(JSONB, nullable=True)
+    analysis_json: Mapped[dict | None] = mapped_column(JSONB, nullable=True)
+
+    created_at: Mapped[DateTime] = mapped_column(DateTime(timezone=True), server_default=func.now(), nullable=False)
+
+    document = relationship("Document", back_populates="analysis_versions")

app/models/document_replica_layout_version.py

@@ -0,0 +1,37 @@
+from sqlalchemy import Boolean, Column, DateTime, ForeignKey, Integer, JSON, String, Text
+from sqlalchemy.orm import relationship
+from sqlalchemy.sql import func
+
+from app.db.base import Base
+
+
+class DocumentReplicaLayoutVersion(Base):
+    __tablename__ = "document_replica_layout_versions"
+
+    id = Column(Integer, primary_key=True, index=True)
+    document_id = Column(Integer, ForeignKey("documents.id"), nullable=False, index=True)
+
+    version_number = Column(Integer, nullable=False)
+    version_type = Column(String, nullable=False, default="heuristic")
+    render_mode_source = Column(String, nullable=False, default="shared")
+    is_current = Column(Boolean, nullable=False, default=True)
+
+    created_by = Column(String, nullable=True)
+    derived_from_text_version_id = Column(Integer, ForeignKey("text_versions.id"), nullable=True)
+    derived_from_replica_layout_version_id = Column(Integer, ForeignKey("document_replica_layout_versions.id"), nullable=True)
+
+    model_name = Column(String, nullable=True)
+    model_version = Column(String, nullable=True)
+    prompt_version = Column(String, nullable=True)
+
+    quality_score = Column(String, nullable=True)
+    quality_note = Column(Text, nullable=True)
+    quality_flags = Column(JSON, nullable=True)
+    inference_metadata_json = Column(JSON, nullable=True)
+    layout_json = Column(JSON, nullable=False)
+
+    created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
+
+    document = relationship("Document", back_populates="replica_layout_versions")
+    outputs = relationship("DocumentReplicaOutput", back_populates="replica_layout_version", cascade="all, delete-orphan")
+    parent_layout_version = relationship("DocumentReplicaLayoutVersion", remote_side=[id])

app/models/document_replica_output.py

@@ -0,0 +1,25 @@
+from sqlalchemy import Column, DateTime, ForeignKey, Integer, JSON, String
+from sqlalchemy.orm import relationship
+from sqlalchemy.sql import func
+
+from app.db.base import Base
+
+
+class DocumentReplicaOutput(Base):
+    __tablename__ = "document_replica_outputs"
+
+    id = Column(Integer, primary_key=True, index=True)
+    document_id = Column(Integer, ForeignKey("documents.id"), nullable=False, index=True)
+    replica_layout_version_id = Column(Integer, ForeignKey("document_replica_layout_versions.id"), nullable=False, index=True)
+
+    output_type = Column(String, nullable=False)
+    file_path = Column(String, nullable=False)
+    sha256 = Column(String, nullable=True)
+    file_size_bytes = Column(Integer, nullable=True)
+    created_by = Column(String, nullable=True)
+    render_settings_json = Column(JSON, nullable=True)
+
+    created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
+
+    document = relationship("Document", back_populates="replica_outputs")
+    replica_layout_version = relationship("DocumentReplicaLayoutVersion", back_populates="outputs")

app/models/document_replica_review_state.py

@@ -0,0 +1,27 @@
+from sqlalchemy import Boolean, Column, DateTime, ForeignKey, Integer, Text
+from sqlalchemy.orm import relationship
+from sqlalchemy.sql import func
+
+from app.db.base import Base
+
+
+class DocumentReplicaReviewState(Base):
+    __tablename__ = "document_replica_review_states"
+
+    id = Column(Integer, primary_key=True, index=True)
+    document_id = Column(Integer, ForeignKey("documents.id"), nullable=False, index=True)
+    current_replica_layout_version_id = Column(Integer, ForeignKey("document_replica_layout_versions.id"), nullable=True)
+
+    is_reviewed = Column(Boolean, nullable=False, default=False)
+    is_approved = Column(Boolean, nullable=False, default=False)
+    needs_model_retry = Column(Boolean, nullable=False, default=False)
+    needs_manual_adjustment = Column(Boolean, nullable=False, default=False)
+
+    reviewed_by = Column(Text, nullable=True)
+    review_note = Column(Text, nullable=True)
+
+    reviewed_at = Column(DateTime(timezone=True), nullable=True)
+    created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
+    updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False)
+
+    document = relationship("Document", back_populates="replica_review_states")

app/routes/documents.py

@@ -22,6 +22,7 @@ from app.db.deps import get_db
 from app.logic.document_outputs import (
     save_field_enriched_pdf_current,
     save_ocr_corrected_pdf_current,
+    save_replica_pdf,
 )
 from app.logic.storage_paths import build_proposed_storage_path
 from app.logic.extraction import (
@@ -34,6 +35,9 @@ from app.logic.extraction import (
     _replace_document_line_items,
 )
 from app.logic.ingest import compute_quality_score, rerun_ocr_for_document
+from app.models.document_analysis_version import DocumentAnalysisVersion
+from app.logic.document_analysis import build_layout_ocr_analysis_for_document
+from app.logic.layout_ocr import run_layout_ocr
 from app.models.document import Document
 from app.models.document_line_item import DocumentLineItem
 from app.models.document_line_item_set import DocumentLineItemSet
@@ -945,17 +949,93 @@ def save_document_type_route(
 
 @router.post("/{document_id}/rerun-ocr", response_class=RedirectResponse)
 def rerun_ocr(document_id: str, db: Session = Depends(get_db)):
-    document = db.query(Document).filter(Document.document_id == document_id).first()
+    document = (
+        db.query(Document)
+        .options(
+            selectinload(Document.text_versions),
+            selectinload(Document.analysis_versions),
+        )
+        .filter(Document.document_id == document_id)
+        .first()
+    )
     if document is None:
         return RedirectResponse(url="/documents/", status_code=303)
 
     try:
-        rerun_ocr_for_document(db, document)
+        if not document.current_path:
+            return RedirectResponse(
+                url=f"/documents/{document.document_id}?error=rerun_ocr_failed&tab=ocr-review",
+                status_code=303,
+            )
+
+        layout_result = run_layout_ocr(document.current_path)
+        analysis_json = build_layout_ocr_analysis_for_document(document)
+        text_content = analysis_json.get("text_content") or ""
+
+        for row in getattr(document, "text_versions", []) or []:
+            if getattr(row, "is_current", False):
+                row.is_current = False
+
+        next_version = (
+            max((getattr(v, "version_number", 0) or 0) for v in getattr(document, "text_versions", []) or []) + 1
+            if getattr(document, "text_versions", None) else 1
+        )
+
+        text_row = TextVersion(
+            document_id=document.id,
+            version_number=next_version,
+            version_type="raw_ocr",
+            text_content=text_content,
+            created_by="rerun_ocr_layout",
+            is_current=True,
+            ocr_engine=layout_result.engine_name,
+            ocr_engine_version=layout_result.engine_version,
+            rerun_source="layout_ocr",
+            quality_score=0.9 if analysis_json.get("quality", {}).get("usable_layout") else 0.5,
+            quality_flags=analysis_json.get("quality", {}).get("issues", []),
+            quality_note="Layout OCR generated line and word boxes for replica workflow.",
+            layout_json={"pages": analysis_json.get("pages", [])},
+        )
+        db.add(text_row)
+        db.flush()
+
+        for row in getattr(document, "analysis_versions", []) or []:
+            if getattr(row, "is_current", False):
+                row.is_current = False
+
+        next_analysis_version = (
+            max((getattr(v, "version_number", 0) or 0) for v in getattr(document, "analysis_versions", []) or []) + 1
+            if getattr(document, "analysis_versions", None) else 1
+        )
+
+        analysis_row = DocumentAnalysisVersion(
+            document_id=document.id,
+            version_number=next_analysis_version,
+            analysis_type="canonical",
+            is_current=True,
+            created_by="rerun_ocr_layout",
+            engine_name=layout_result.engine_name,
+            engine_version=layout_result.engine_version,
+            quality_score=0.9 if analysis_json.get("quality", {}).get("usable_layout") else 0.5,
+            quality_flags=analysis_json.get("quality", {}).get("issues", []),
+            quality_note="Canonical analysis refreshed from layout OCR result.",
+            analysis_json=analysis_json,
+        )
+        db.add(analysis_row)
+        db.commit()
+
     except Exception:
-        return RedirectResponse(url=f"/documents/{document.document_id}?error=rerun_ocr_failed", status_code=303)
-
-    return RedirectResponse(url=f"/documents/{document.document_id}?editor_source=raw&tab=ocr-review", status_code=303)
+        traceback.print_exc()
+        db.rollback()
+        return RedirectResponse(
+            url=f"/documents/{document.document_id}?error=rerun_ocr_failed&tab=ocr-review",
+            status_code=303,
+        )
 
+    return RedirectResponse(
+        url=f"/documents/{document.document_id}?success=rerun_ocr&editor_source=raw&tab=ocr-review",
+        status_code=303,
+    )
 
 @router.post("/{document_id}/save-ocr-corrected-pdf", response_class=RedirectResponse)
 def save_ocr_corrected_pdf(document_id: str, db: Session = Depends(get_db)):
@@ -1003,6 +1083,38 @@ def move_to_trash(document_id: str, db: Session = Depends(get_db)):
 
 
 
+
+def _resolve_document_output_path(document, output_path: str = "") -> Path:
+    save_root = get_default_save_root()
+    naming_row = document.naming_fields[0] if getattr(document, "naming_fields", None) else None
+
+    default_output_path = Path(
+        build_proposed_storage_path(
+            document=document,
+            save_root=save_root,
+            naming_row=naming_row,
+        )
+    )
+    default_output_path = default_output_path.with_name(
+        re.sub(r"(?:_v\d+|_\d+)(?=\.[^.]+$)", "", default_output_path.name)
+    )
+    if default_output_path.suffix.lower() != ".pdf":
+        default_output_path = default_output_path.with_suffix(".pdf")
+
+    output_path_raw = (output_path or "").strip()
+    output_path_obj = Path(output_path_raw) if output_path_raw else default_output_path
+
+    if output_path_obj.suffix.lower() != ".pdf":
+        output_path_obj = output_path_obj.with_suffix(".pdf")
+
+    allowed_root = Path(save_root).resolve()
+    resolved_parent = output_path_obj.parent.resolve()
+    if allowed_root != resolved_parent and allowed_root not in resolved_parent.parents:
+        raise ValueError("invalid_output_path")
+
+    output_path_obj.parent.mkdir(parents=True, exist_ok=True)
+    return output_path_obj
+
 @router.post("/{document_id}/save-pdf", response_class=RedirectResponse)
 def save_pdf(document_id: str, output_path: str = Form(""), db: Session = Depends(get_db)):
     if not _storage_available():
@@ -1024,41 +1136,14 @@ def save_pdf(document_id: str, output_path: str = Form(""), db: Session = Depend
     if document is None:
         return RedirectResponse(url="/documents/", status_code=303)
 
-    save_root = get_default_save_root()
-    naming_row = document.naming_fields[0] if getattr(document, "naming_fields", None) else None
-
-    default_output_path = Path(
-        build_proposed_storage_path(
-            document=document,
-            save_root=save_root,
-            naming_row=naming_row,
-        )
-    )
-    default_output_path = default_output_path.with_name(
-        re.sub(r"(?:_v\d+|_\d+)(?=\.[^.]+$)", "", default_output_path.name)
-    )
-    if default_output_path.suffix.lower() != ".pdf":
-        default_output_path = default_output_path.with_suffix(".pdf")
-
-    output_path_raw = (output_path or "").strip()
-    if output_path_raw:
-        output_path_obj = Path(output_path_raw)
-    else:
-        output_path_obj = default_output_path
-
-    if output_path_obj.suffix.lower() != ".pdf":
-        output_path_obj = output_path_obj.with_suffix(".pdf")
-
-    allowed_root = Path(save_root).resolve()
-    resolved_parent = output_path_obj.parent.resolve()
-    if allowed_root != resolved_parent and allowed_root not in resolved_parent.parents:
+    try:
+        output_path_obj = _resolve_document_output_path(document, output_path)
+    except ValueError:
         return RedirectResponse(
             url=f"/documents/{document.document_id}?error=invalid_output_path",
             status_code=303,
         )
 
-    output_path_obj.parent.mkdir(parents=True, exist_ok=True)
-
     has_extracted = bool(getattr(document, "extracted_fields", None))
     has_additional = bool(getattr(document, "additional_fields", None))
 
@@ -1079,6 +1164,91 @@ def save_pdf(document_id: str, output_path: str = Form(""), db: Session = Depend
 
 
 
+
+@router.post("/{document_id}/save-replica-pdf", response_class=RedirectResponse)
+def save_replica_pdf_clean(document_id: str, output_path: str = Form(""), db: Session = Depends(get_db)):
+    if not _storage_available():
+        return RedirectResponse(url=f"/documents/{document_id}?error=storage_unavailable", status_code=303)
+
+    document = (
+        db.query(Document)
+        .options(
+            selectinload(Document.text_versions),
+            selectinload(Document.naming_fields),
+            selectinload(Document.replica_review_states),
+            selectinload(Document.replica_outputs),
+            selectinload(Document.extracted_fields),
+            selectinload(Document.analysis_versions),
+        )
+        .filter(Document.document_id == document_id)
+        .first()
+    )
+    if document is None:
+        return RedirectResponse(url="/documents/", status_code=303)
+
+    try:
+        output_path_obj = _resolve_document_output_path(document, output_path)
+        save_replica_pdf(db, document, output_path_obj, mode="clean")
+        return RedirectResponse(
+            url=f"/documents/{document.document_id}?success=saved_replica_pdf&tab=ocr-review&viewer_source=replica",
+            status_code=303,
+        )
+    except ValueError as e:
+        msg = str(e)
+        if "invalid_output_path" in msg:
+            return RedirectResponse(
+                url=f"/documents/{document.document_id}?error=invalid_output_path",
+                status_code=303,
+            )
+        if "document_analysis_missing_usable_layout" in msg or "clean_replica_has_no_renderable_lines" in msg:
+            return RedirectResponse(
+                url=f"/documents/{document.document_id}?error=clean_replica_requires_layout_ocr&tab=ocr-review&viewer_source=scan",
+                status_code=303,
+            )
+
+        traceback.print_exc()
+        return RedirectResponse(
+            url=f"/documents/{document.document_id}?error=save_replica_pdf_failed&tab=ocr-review",
+            status_code=303,
+        )
+    except Exception:
+        traceback.print_exc()
+        return RedirectResponse(
+            url=f"/documents/{document.document_id}?error=save_replica_pdf_failed&tab=ocr-review",
+            status_code=303,
+        )
+
+@router.post("/{document_id}/save-replica-pdf-scan-backed", response_class=RedirectResponse)
+def save_replica_pdf_scan_backed(document_id: str, output_path: str = Form(""), db: Session = Depends(get_db)):
+    if not _storage_available():
+        return RedirectResponse(url=f"/documents/{document_id}?error=storage_unavailable", status_code=303)
+
+    document = (
+        db.query(Document)
+        .options(
+            selectinload(Document.text_versions),
+            selectinload(Document.naming_fields),
+            selectinload(Document.replica_review_states),
+        )
+        .filter(Document.document_id == document_id)
+        .first()
+    )
+    if document is None:
+        return RedirectResponse(url="/documents/", status_code=303)
+
+    try:
+        output_path_obj = _resolve_document_output_path(document, output_path)
+        save_replica_pdf(db, document, output_path_obj, mode="scan_backed")
+    except ValueError as e:
+        if "invalid_output_path" in str(e):
+            return RedirectResponse(url=f"/documents/{document.document_id}?error=invalid_output_path", status_code=303)
+        return RedirectResponse(url=f"/documents/{document.document_id}?error=save_replica_pdf_scan_backed_failed&tab=ocr-review", status_code=303)
+    except Exception:
+        traceback.print_exc()
+        return RedirectResponse(url=f"/documents/{document.document_id}?error=save_replica_pdf_scan_backed_failed&tab=ocr-review", status_code=303)
+
+    return RedirectResponse(url=f"/documents/{document.document_id}?success=saved_replica_pdf_scan_backed&tab=ocr-review", status_code=303)
+
 @router.post("/{document_id}/save-field-enriched-pdf", response_class=RedirectResponse)
 def save_field_enriched_pdf(document_id: str, db: Session = Depends(get_db)):
     return RedirectResponse(
@@ -1459,12 +1629,13 @@ async def save_line_items(
     )
 
 @router.get("/{document_id}/preview-file")
-def document_preview_file(document_id: str, db: Session = Depends(get_db)):
+def document_preview_file(document_id: str, path: str | None = None, db: Session = Depends(get_db)):
     document = db.query(Document).filter(Document.document_id == document_id).first()
-    if document is None or not document.current_path:
+    resolved_path = path or (document.current_path if document else None)
+    if document is None or not resolved_path:
         return HTMLResponse(content="Preview file not found", status_code=404)
 
-    path_obj = Path(document.current_path)
+    path_obj = Path(resolved_path)
     if not path_obj.exists() or not path_obj.is_file():
         return HTMLResponse(content="Preview file not found", status_code=404)
 
@@ -1472,8 +1643,26 @@ def document_preview_file(document_id: str, db: Session = Depends(get_db)):
     return FileResponse(path=str(path_obj), media_type=media_type, filename=path_obj.name, headers={"Content-Disposition": "inline; filename=\"" + path_obj.name + "\""})
 
 
+
+def _get_latest_replica_output(document, output_type: str):
+    outputs = getattr(document, "replica_outputs", None) or []
+    matches = [row for row in outputs if getattr(row, "output_type", None) == output_type]
+    matches.sort(key=lambda x: getattr(x, "created_at", None) or 0, reverse=True)
+    return matches[0] if matches else None
+
+
+def _build_preview_url_for_path(request: Request, document_id: str, path_value: str | None):
+    if not path_value:
+        return None
+    path_obj = Path(path_value)
+    if not path_obj.exists() or not path_obj.is_file():
+        return None
+    from urllib.parse import quote
+    base = str(request.url_for("document_preview_file", document_id=document_id))
+    return f"{base}?path={quote(str(path_obj))}&v={int(path_obj.stat().st_mtime)}"
+
 @router.get("/{document_id}", response_class=HTMLResponse)
-def document_detail(document_id: str, request: Request, queue: str | None = None, db: Session = Depends(get_db)):
+def document_detail(document_id: str, request: Request, queue: str | None = None, viewer_source: str = "scan", db: Session = Depends(get_db)):
     current_user = getattr(request.state, "current_user", None)
     document = (
         db.query(Document)
@@ -1511,12 +1700,26 @@ def document_detail(document_id: str, request: Request, queue: str | None = None
     actual_line_count = len(review_text_value.splitlines()) if review_text_value else 0
     line_numbers = list(range(1, max(actual_line_count, expected_line_count) + 1))
 
-    file_url = None
+    replica_clean_output = _get_latest_replica_output(document, "clean")
+    replica_scan_backed_output = _get_latest_replica_output(document, "scan_backed")
+
+    scan_path = document.current_path
+    replica_path = replica_clean_output.file_path if replica_clean_output and replica_clean_output.file_path else None
+    replica_scan_backed_path = replica_scan_backed_output.file_path if replica_scan_backed_output and replica_scan_backed_output.file_path else None
+
+    effective_viewer_source = viewer_source or "scan"
+    preview_path = scan_path
+
+    if effective_viewer_source == "replica" and replica_path:
+        preview_path = replica_path
+    elif effective_viewer_source == "replica_scan_backed" and replica_scan_backed_path:
+        preview_path = replica_scan_backed_path
+    else:
+        effective_viewer_source = "scan"
+        preview_path = scan_path
+
     storage_available = _storage_available()
-    if document.current_path:
-        current_path = Path(document.current_path)
-        if current_path.exists() and current_path.is_file():
-            file_url = str(request.url_for("document_preview_file", document_id=document.document_id))
+    file_url = _build_preview_url_for_path(request, document.document_id, preview_path)
 
     app_url = str(request.url_for("document_detail", document_id=document.document_id))
     error = request.query_params.get("error")
@@ -1615,6 +1818,9 @@ def document_detail(document_id: str, request: Request, queue: str | None = None
             "review_text_value": review_text_value,
             "file_url": file_url,
             "storage_available": storage_available,
+            "viewer_source": effective_viewer_source,
+            "replica_clean_output": replica_clean_output,
+            "replica_scan_backed_output": replica_scan_backed_output,
             "version_rows": version_rows,
             "current_line_item_version": current_line_item_version,
             "ocr_version_options": ocr_version_options,

app/static/app.css

@@ -6231,3 +6231,40 @@ table {
   }
 }
 /* ===== end line item queue card polish ===== */
+
+
+
+.preview-card-header {
+  display: flex;
+  align-items: center;
+  justify-content: space-between;
+  gap: 0.75rem;
+  flex-wrap: wrap;
+}
+
+.preview-source-toggle {
+  display: flex;
+  gap: 0.45rem;
+  flex-wrap: wrap;
+}
+
+.preview-source-link {
+  display: inline-flex;
+  align-items: center;
+  justify-content: center;
+  min-height: 2rem;
+  padding: 0.35rem 0.7rem;
+  border: 1px solid #d7dce5;
+  border-radius: 999px;
+  background: #fff;
+  color: #334155;
+  text-decoration: none;
+  font-size: 0.82rem;
+  line-height: 1;
+}
+
+.preview-source-link.active {
+  background: #0f172a;
+  border-color: #0f172a;
+  color: #fff;
+}

app/templates/base.html

@@ -4,7 +4,7 @@
     <meta charset="UTF-8">
     <meta name="viewport" content="width=device-width, initial-scale=1">
     <title>{% block title %}Document Processor{% endblock %}</title>
-    <link rel="stylesheet" href="/static/app.css?v=171">
+    <link rel="stylesheet" href="/static/app.css?v=174">
     <link rel="stylesheet" href="/static/app-shell.css?v=158">
 </head>
 <body>

app/templates/documents/detail.html

@@ -73,14 +73,26 @@ document.addEventListener("DOMContentLoaded", () => {
     <div class="success-message">OCR rerun successfully.</div>
 {% elif success == "regenerated_line_items" %}
     <div class="success-message">Line items regenerated successfully.</div>
+{% elif success == "saved_replica_pdf" %}
+    <div class="success-message">Replica PDF saved.</div>
+{% elif success == "saved_replica_pdf_scan_backed" %}
+    <div class="success-message">Scan-backed replica PDF saved.</div>
 {% elif success == "saved_reviewed_ocr" %}
     <div class="success-message">Reviewed OCR saved.</div>
+{% elif success == "saved_replica_pdf" %}
+    <div class="success-message">Replica PDF saved.</div>
+{% elif success == "saved_replica_pdf_scan_backed" %}
+    <div class="success-message">Scan-backed replica PDF saved.</div>
 {% elif success == "saved_reviewed_ocr" %}
     <div class="success-message">Reviewed OCR saved.</div>
 {% elif error == "rerun_ocr_failed" %}
             <div class="error-box">OCR rerun failed.</div>
         {% elif error == "deprecated_pdf_route_disabled" %}
         <div class="error-box">This deprecated PDF save route has been disabled. Use Save Document instead.</div>
+    {% elif error == "save_replica_pdf_failed" %}
+        <div class="error-box">Could not save replica PDF.</div>
+    {% elif error == "save_replica_pdf_scan_backed_failed" %}
+        <div class="error-box">Could not save scan-backed replica PDF.</div>
     {% elif error == "save_field_enriched_failed" %}
             <div class="error-box">Could not save field-enriched PDF.</div>
         {% endif %}
@@ -163,6 +175,14 @@ document.addEventListener("DOMContentLoaded", () => {
                             <button type="button" id="toggle-path-edit" class="top-pill-button">Edit path</button>
                         </div>
 </form>
+<div class="button-row" style="margin-top:0.6rem;">
+    <form method="post" action="/documents/{{ document.document_id }}/save-replica-pdf" style="display:inline;">
+        <button type="submit">Save Replica PDF</button>
+    </form>
+    <form method="post" action="/documents/{{ document.document_id }}/save-replica-pdf-scan-backed" style="display:inline;">
+        <button type="submit">Save Replica PDF (Scan-backed)</button>
+    </form>
+</div>
 
                 </div>
 
@@ -200,7 +220,20 @@ document.addEventListener("DOMContentLoaded", () => {
 <div class="workspace-grid">
             <section>
                 <div class="card preview-card">
+                    <div class="preview-card-header">
                         <h2 class="card-title">Document preview</h2>
+
+                <div class="preview-source-toggle">
+                    <a class="preview-source-link{% if viewer_source == 'scan' %} active{% endif %}" href="/documents/{{ document.document_id }}?tab={{ active_tab }}&viewer_source=scan">Scan</a>
+                    {% if replica_clean_output %}
+                    <a class="preview-source-link{% if viewer_source == 'replica' %} active{% endif %}" href="/documents/{{ document.document_id }}?tab={{ active_tab }}&viewer_source=replica">Replica</a>
+                    {% endif %}
+                    {% if replica_scan_backed_output %}
+                    <a class="preview-source-link{% if viewer_source == 'replica_scan_backed' %} active{% endif %}" href="/documents/{{ document.document_id }}?tab={{ active_tab }}&viewer_source=replica_scan_backed">Replica (Scan-backed)</a>
+                    {% endif %}
+                </div>
+
+                    </div>
                     {% if not storage_available %}
                         <p class="empty-state">Storage mount unavailable. Preview is temporarily unavailable.</p>
                     {% elif file_url %}