Merge pull request #470 from christianlouis/copilot/add-pdfa-export-option

feat(pdfa): PDF/A archival conversion with FreeTSA timestamping and per-provider upload control

.env.demo

@@ -325,7 +325,7 @@ WEBHOOK_ENABLED=True
 
 # Uptime Kuma
 UPTIME_KUMA_URL=https://status.example.com/api/push/abcdef123456?status=up
-UPTIME_KUMA_PING_INTERVAL=5
+UPTIME_KUMA_PING_INTERVAL=5
 
 # **Full-Text Search (Meilisearch)**
 # URL for the Meilisearch instance.
@@ -347,9 +347,32 @@ SHOW_DEDUPLICATION_STEP=True
 # Minimum cosine similarity score (0–1) for two documents to be flagged as
 # near-duplicates.  0.85 means 85 % semantic overlap.  Lower = more matches.
 NEAR_DUPLICATE_THRESHOLD=0.85
+
+# **PDF/A Archival Conversion**
+# When enabled, PDF/A copies of both the original ingested file and the processed
+# file are created and saved alongside the standard copies.  This may double or
+# triple storage but provides better legal coverage with time-stamped archival copies.
+# Uses ocrmypdf with Ghostscript for the conversion.
+ENABLE_PDFA_CONVERSION=false
+# PDF/A format variant: 1 = PDF/A-1b, 2 = PDF/A-2b (default), 3 = PDF/A-3b
+PDFA_FORMAT=2
+# Upload original-file PDF/A variant to all configured storage providers
+PDFA_UPLOAD_ORIGINAL=false
+# Upload processed-file PDF/A variant to all configured storage providers
+PDFA_UPLOAD_PROCESSED=false
+# Subfolder name appended to each provider's folder for PDF/A uploads
+# e.g. if Dropbox folder is '/Documents' this puts PDF/A files into '/Documents/pdfa'
+PDFA_UPLOAD_FOLDER=pdfa
+# Google Drive folder ID for PDF/A uploads (uses folder IDs, not paths)
+# Leave empty to use the same folder as regular uploads
+GOOGLE_DRIVE_PDFA_FOLDER_ID=
+# RFC 3161 timestamping of PDF/A files (creates .tsr proof-of-existence files)
+PDFA_TIMESTAMP_ENABLED=false
+# Timestamp Authority URL (default: FreeTSA, a free RFC 3161 TSA)
+PDFA_TIMESTAMP_URL=https://freetsa.org/tsr
 # Model used to generate text embeddings for document similarity.
 # Must be supported by your OpenAI-compatible API endpoint.
 EMBEDDING_MODEL=text-embedding-3-small
 # Maximum tokens to send to the embedding model.  Set below the model's
 # context window (e.g. 8000 for an 8192-token model).
-EMBEDDING_MAX_TOKENS=8000
+EMBEDDING_MAX_TOKENS=8000

app/config.py

@@ -222,6 +222,69 @@ class Settings(BaseSettings):
 
     # Feature flags
     allow_file_delete: bool = True  # Default to allowing file deletion from database
+
+    # PDF/A archival conversion settings
+    enable_pdfa_conversion: bool = Field(
+        default=False,
+        description=(
+            "Enable PDF/A archival variant generation. When enabled, PDF/A copies of both the "
+            "original ingested file and the processed file are created and saved alongside the "
+            "standard copies. Uses ocrmypdf with Ghostscript for the conversion. "
+            "This may double or triple storage but provides better legal coverage. Default: False."
+        ),
+    )
+    pdfa_format: str = Field(
+        default="2",
+        description=(
+            "PDF/A format variant to produce. Passed to ocrmypdf --output-type pdfa-N. "
+            "Valid values: '1' (PDF/A-1b), '2' (PDF/A-2b), '3' (PDF/A-3b). Default: '2'."
+        ),
+    )
+    pdfa_upload_original: bool = Field(
+        default=False,
+        description=(
+            "Upload the original-file PDF/A variant to all configured storage providers. "
+            "Files are placed in the provider's folder + PDFA_UPLOAD_FOLDER subfolder. Default: False."
+        ),
+    )
+    pdfa_upload_processed: bool = Field(
+        default=False,
+        description=(
+            "Upload the processed-file PDF/A variant to all configured storage providers. "
+            "Files are placed in the provider's folder + PDFA_UPLOAD_FOLDER subfolder. Default: False."
+        ),
+    )
+    pdfa_upload_folder: str = Field(
+        default="pdfa",
+        description=(
+            "Subfolder name appended to each storage provider's configured folder for PDF/A uploads. "
+            "For example if Dropbox folder is '/Documents' and this is 'pdfa', PDF/A files go to "
+            "'/Documents/pdfa'. Set to empty string to upload into the same folder. Default: 'pdfa'."
+        ),
+    )
+    google_drive_pdfa_folder_id: str = Field(
+        default="",
+        description=(
+            "Google Drive folder ID for PDF/A uploads. Since Google Drive uses IDs not paths, "
+            "this must be set separately. If empty, uses the standard google_drive_folder_id."
+        ),
+    )
+    pdfa_timestamp_enabled: bool = Field(
+        default=False,
+        description=(
+            "Enable RFC 3161 timestamping of PDF/A files via a Timestamp Authority (TSA). "
+            "Creates a .tsr file alongside each PDF/A file for legal proof of existence. "
+            "Requires openssl binary on PATH. Default: False."
+        ),
+    )
+    pdfa_timestamp_url: str = Field(
+        default="https://freetsa.org/tsr",
+        description=(
+            "URL of the RFC 3161 Timestamp Authority. Default: FreeTSA (https://freetsa.org/tsr). "
+            "Other options: GlobalSign, DigiStamp, or any RFC 3161-compliant TSA."
+        ),
+    )
+
     imap_readonly_mode: bool = Field(
         default=False,
         description=(

app/models.py

@@ -67,6 +67,10 @@ class FileRecord(Base):
     # Human-readable document title from AI metadata
     document_title = Column(String, nullable=True)
 
+    # PDF/A archival variant paths (generated when ENABLE_PDFA_CONVERSION is True)
+    original_pdfa_path = Column(String, nullable=True)  # PDF/A copy of the original ingested file
+    processed_pdfa_path = Column(String, nullable=True)  # PDF/A copy of the processed file
+
     # Pre-computed text embedding vector stored as JSON array of floats
     embedding = Column(Text, nullable=True)
 

app/tasks/convert_to_pdfa.py

@@ -0,0 +1,453 @@
+"""PDF/A archival conversion task.
+
+Converts PDF files to PDF/A format using ocrmypdf (which relies on Ghostscript
+internally).  Two variants are produced when enabled:
+
+1. **Original PDF/A** – an archival copy of the ingested file, providing a
+   time-stamped record of the document as it was upon ingestion.
+2. **Processed PDF/A** – an archival copy of the processed file with embedded
+   metadata.
+
+Both are saved under ``workdir/pdfa/`` and referenced in the database via
+``FileRecord.original_pdfa_path`` and ``FileRecord.processed_pdfa_path``.
+
+When ``PDFA_TIMESTAMP_ENABLED`` is True, each PDF/A file also gets an RFC 3161
+timestamp response (``.tsr``) from a configurable Timestamp Authority (default:
+FreeTSA).  This provides cryptographic proof of the file's existence at a given
+point in time.
+
+.. note::
+
+   PDF/A conversion may alter font rendering (especially OCR text overlays
+   produced by Microsoft Azure Document Intelligence).  This is expected –
+   the PDF/A copies are parallel archival variants, not replacements.
+"""
+
+import logging
+import os
+import shutil
+import subprocess
+
+import requests as http_requests
+
+from app.celery_app import celery
+from app.config import settings
+from app.database import SessionLocal
+from app.models import FileRecord
+from app.tasks.retry_config import BaseTaskWithRetry
+from app.utils import get_unique_filepath_with_counter, log_task_progress
+
+logger = logging.getLogger(__name__)
+
+# Subdirectory structure under workdir for PDF/A copies
+PDFA_ORIGINAL_SUBDIR = os.path.join("pdfa", "original")
+PDFA_PROCESSED_SUBDIR = os.path.join("pdfa", "processed")
+
+
+def _convert_pdf_to_pdfa(input_path: str, output_path: str, pdfa_format: str = "2") -> bool:
+    """Convert a PDF file to PDF/A using ocrmypdf.
+
+    Uses ``ocrmypdf --skip-text --output-type pdfa-N`` so that existing text
+    layers are preserved (not re-OCR'd) while the output is converted to
+    PDF/A via Ghostscript.
+
+    Args:
+        input_path: Absolute path to the source PDF file.
+        output_path: Absolute path for the PDF/A output file.
+        pdfa_format: PDF/A variant ('1', '2', or '3'). Defaults to '2' for PDF/A-2b.
+
+    Returns:
+        True if conversion succeeded, False otherwise.
+    """
+    # Validate format to prevent argument injection via output-type
+    if pdfa_format not in ("1", "2", "3"):
+        logger.error(f"[convert_to_pdfa] Invalid pdfa_format: {pdfa_format}")
+        return False
+
+    ocrmypdf_bin = shutil.which("ocrmypdf")
+    if not ocrmypdf_bin:
+        logger.error("[convert_to_pdfa] ocrmypdf binary not found on PATH")
+        return False
+
+    output_type = f"pdfa-{pdfa_format}"
+
+    cmd = [
+        ocrmypdf_bin,
+        "--skip-text",
+        "--output-type",
+        output_type,
+        "--quiet",
+        "--invalidate-digital-signatures",
+        input_path,
+        output_path,
+    ]
+
+    logger.info(f"[convert_to_pdfa] Running: {' '.join(cmd)}")
+
+    try:
+        proc = subprocess.run(cmd, capture_output=True, text=True, timeout=600, check=False)  # noqa: S603
+    except subprocess.TimeoutExpired:
+        logger.warning("[convert_to_pdfa] ocrmypdf timed out after 600s")
+        return False
+
+    if proc.returncode != 0:
+        stderr_snippet = proc.stderr.strip()[:500] if proc.stderr else ""
+        logger.warning(f"[convert_to_pdfa] ocrmypdf exited with code {proc.returncode}: {stderr_snippet}")
+        return False
+
+    logger.info(f"[convert_to_pdfa] PDF/A file written to {output_path}")
+    return True
+
+
+def _timestamp_file(file_path: str, tsa_url: str) -> str | None:
+    """Create an RFC 3161 timestamp for a file using a Timestamp Authority.
+
+    Uses ``openssl ts`` to create a timestamp request (TSQ) from the file's
+    SHA-256 hash, submits it to the TSA via HTTP POST, and saves the timestamp
+    response (TSR) alongside the file.
+
+    Args:
+        file_path: Absolute path to the file to timestamp.
+        tsa_url: URL of the RFC 3161 Timestamp Authority.
+
+    Returns:
+        Path to the ``.tsr`` file if successful, None otherwise.
+    """
+    openssl_bin = shutil.which("openssl")
+    if not openssl_bin:
+        logger.error("[timestamp] openssl binary not found on PATH")
+        return None
+
+    tsr_path = file_path + ".tsr"
+    tsq_path = file_path + ".tsq"
+
+    try:
+        # Step 1: Create timestamp request
+        cmd = [openssl_bin, "ts", "-query", "-data", file_path, "-sha256", "-no_nonce", "-out", tsq_path]
+        proc = subprocess.run(cmd, capture_output=True, text=True, timeout=30, check=False)  # noqa: S603
+        if proc.returncode != 0:
+            logger.warning(f"[timestamp] openssl ts -query failed: {proc.stderr.strip()[:200]}")
+            return None
+
+        # Step 2: Submit TSQ to the Timestamp Authority
+        with open(tsq_path, "rb") as f:
+            tsq_data = f.read()
+
+        response = http_requests.post(
+            tsa_url,
+            data=tsq_data,
+            headers={"Content-Type": "application/timestamp-query"},
+            timeout=30,
+        )
+
+        if response.status_code != 200:
+            logger.warning(f"[timestamp] TSA returned HTTP {response.status_code} from {tsa_url}")
+            return None
+
+        # Step 3: Save the timestamp response
+        with open(tsr_path, "wb") as f:
+            f.write(response.content)
+
+        logger.info(f"[timestamp] RFC 3161 timestamp saved to {tsr_path}")
+        return tsr_path
+
+    except http_requests.RequestException as e:
+        logger.warning(f"[timestamp] Failed to contact TSA at {tsa_url}: {e}")
+        return None
+    except subprocess.TimeoutExpired:
+        logger.warning("[timestamp] openssl ts timed out")
+        return None
+    finally:
+        # Always clean up the TSQ file
+        if os.path.exists(tsq_path):
+            os.remove(tsq_path)
+
+
+def _compute_pdfa_folder_overrides() -> dict[str, str]:
+    """Compute per-provider folder overrides for PDF/A uploads.
+
+    Appends ``settings.pdfa_upload_folder`` to each provider's configured
+    folder.  For Google Drive (which uses folder IDs), uses the dedicated
+    ``google_drive_pdfa_folder_id`` setting.
+
+    Returns:
+        Dictionary mapping provider names to folder override strings.
+    """
+    subfolder = settings.pdfa_upload_folder
+    overrides: dict[str, str] = {}
+
+    if not subfolder:
+        return overrides
+
+    # Path-based providers: append subfolder
+    for provider, folder_attr in [
+        ("dropbox", "dropbox_folder"),
+        ("nextcloud", "nextcloud_folder"),
+        ("webdav", "webdav_folder"),
+        ("ftp", "ftp_folder"),
+        ("sftp", "sftp_folder"),
+        ("onedrive", "onedrive_folder_path"),
+    ]:
+        base = getattr(settings, folder_attr, "") or ""
+        overrides[provider] = f"{base.rstrip('/')}/{subfolder}" if base else subfolder
+
+    # S3: append subfolder to prefix (trailing slash is required by S3 convention
+    # where "folder" paths are key prefixes, unlike path-based providers above)
+    s3_prefix = getattr(settings, "s3_folder_prefix", "") or ""
+    overrides["s3"] = f"{s3_prefix.rstrip('/')}/{subfolder}/"
+
+    # Google Drive: use dedicated folder ID or fall back to default
+    gdrive_pdfa_id = settings.google_drive_pdfa_folder_id
+    if gdrive_pdfa_id:
+        overrides["google_drive"] = gdrive_pdfa_id
+
+    return overrides
+
+
+@celery.task(base=BaseTaskWithRetry, bind=True)
+def convert_to_pdfa(self, file_id: int) -> dict:
+    """Generate PDF/A archival copies for a processed document.
+
+    Creates PDF/A variants of both the original ingested file and the
+    processed file (with embedded metadata).  Files are saved under
+    ``workdir/pdfa/original/`` and ``workdir/pdfa/processed/`` respectively.
+
+    When timestamping is enabled, each PDF/A file also gets an RFC 3161
+    ``.tsr`` timestamp from the configured TSA.
+
+    Upload of each variant to storage providers is controlled independently
+    by ``pdfa_upload_original`` and ``pdfa_upload_processed``.
+
+    Args:
+        file_id: ID of the FileRecord to create PDF/A copies for.
+
+    Returns:
+        Dictionary with status and file paths.
+    """
+    task_id = self.request.id
+    logger.info(f"[{task_id}] Starting PDF/A conversion for file_id={file_id}")
+    log_task_progress(
+        task_id,
+        "convert_to_pdfa",
+        "in_progress",
+        "Starting PDF/A archival conversion",
+        file_id=file_id,
+    )
+
+    # Fetch file record
+    with SessionLocal() as db:
+        file_record = db.query(FileRecord).filter_by(id=file_id).first()
+        if not file_record:
+            logger.error(f"[{task_id}] FileRecord {file_id} not found")
+            log_task_progress(task_id, "convert_to_pdfa", "failure", "File record not found", file_id=file_id)
+            return {"error": "File record not found", "file_id": file_id}
+
+        original_path = file_record.original_file_path
+        processed_path = file_record.processed_file_path
+
+    pdfa_format = settings.pdfa_format
+    timestamp_enabled = settings.pdfa_timestamp_enabled
+    timestamp_url = settings.pdfa_timestamp_url
+    results = {}
+
+    # --- Convert original file to PDF/A ---
+    if original_path and os.path.exists(original_path):
+        original_pdfa_dir = os.path.join(settings.workdir, PDFA_ORIGINAL_SUBDIR)
+        os.makedirs(original_pdfa_dir, exist_ok=True)
+
+        base_name = os.path.splitext(os.path.basename(original_path))[0]
+        original_pdfa_path = get_unique_filepath_with_counter(original_pdfa_dir, base_name, ".pdf")
+
+        logger.info(f"[{task_id}] Converting original to PDF/A: {original_path} -> {original_pdfa_path}")
+        log_task_progress(
+            task_id,
+            "convert_original_to_pdfa",
+            "in_progress",
+            f"Converting original to PDF/A: {os.path.basename(original_path)}",
+            file_id=file_id,
+        )
+
+        success = _convert_pdf_to_pdfa(original_path, original_pdfa_path, pdfa_format)
+        if success:
+            results["original_pdfa_path"] = original_pdfa_path
+            log_task_progress(
+                task_id,
+                "convert_original_to_pdfa",
+                "success",
+                f"Original PDF/A saved: {os.path.basename(original_pdfa_path)}",
+                file_id=file_id,
+            )
+            # Timestamp the original PDF/A
+            if timestamp_enabled:
+                tsr = _timestamp_file(original_pdfa_path, timestamp_url)
+                if tsr:
+                    log_task_progress(
+                        task_id,
+                        "timestamp_original_pdfa",
+                        "success",
+                        f"Timestamped: {os.path.basename(tsr)}",
+                        file_id=file_id,
+                    )
+                else:
+                    log_task_progress(
+                        task_id,
+                        "timestamp_original_pdfa",
+                        "failure",
+                        "Failed to timestamp original PDF/A",
+                        file_id=file_id,
+                    )
+        else:
+            log_task_progress(
+                task_id,
+                "convert_original_to_pdfa",
+                "failure",
+                "Failed to convert original to PDF/A",
+                file_id=file_id,
+            )
+    else:
+        logger.warning(f"[{task_id}] Original file not found, skipping original PDF/A conversion")
+        log_task_progress(
+            task_id,
+            "convert_original_to_pdfa",
+            "skipped",
+            "Original file not available",
+            file_id=file_id,
+        )
+
+    # --- Convert processed file to PDF/A ---
+    if processed_path and os.path.exists(processed_path):
+        processed_pdfa_dir = os.path.join(settings.workdir, PDFA_PROCESSED_SUBDIR)
+        os.makedirs(processed_pdfa_dir, exist_ok=True)
+
+        base_name = os.path.splitext(os.path.basename(processed_path))[0]
+        processed_pdfa_path = get_unique_filepath_with_counter(processed_pdfa_dir, f"{base_name}-PDFA", ".pdf")
+
+        logger.info(f"[{task_id}] Converting processed to PDF/A: {processed_path} -> {processed_pdfa_path}")
+        log_task_progress(
+            task_id,
+            "convert_processed_to_pdfa",
+            "in_progress",
+            f"Converting processed to PDF/A: {os.path.basename(processed_path)}",
+            file_id=file_id,
+        )
+
+        success = _convert_pdf_to_pdfa(processed_path, processed_pdfa_path, pdfa_format)
+        if success:
+            results["processed_pdfa_path"] = processed_pdfa_path
+            log_task_progress(
+                task_id,
+                "convert_processed_to_pdfa",
+                "success",
+                f"Processed PDF/A saved: {os.path.basename(processed_pdfa_path)}",
+                file_id=file_id,
+            )
+            # Timestamp the processed PDF/A
+            if timestamp_enabled:
+                tsr = _timestamp_file(processed_pdfa_path, timestamp_url)
+                if tsr:
+                    log_task_progress(
+                        task_id,
+                        "timestamp_processed_pdfa",
+                        "success",
+                        f"Timestamped: {os.path.basename(tsr)}",
+                        file_id=file_id,
+                    )
+                else:
+                    log_task_progress(
+                        task_id,
+                        "timestamp_processed_pdfa",
+                        "failure",
+                        "Failed to timestamp processed PDF/A",
+                        file_id=file_id,
+                    )
+        else:
+            log_task_progress(
+                task_id,
+                "convert_processed_to_pdfa",
+                "failure",
+                "Failed to convert processed to PDF/A",
+                file_id=file_id,
+            )
+    else:
+        logger.warning(f"[{task_id}] Processed file not found, skipping processed PDF/A conversion")
+        log_task_progress(
+            task_id,
+            "convert_processed_to_pdfa",
+            "skipped",
+            "Processed file not available",
+            file_id=file_id,
+        )
+
+    # --- Update database with PDF/A paths ---
+    with SessionLocal() as db:
+        file_record = db.query(FileRecord).filter_by(id=file_id).first()
+        if file_record:
+            if "original_pdfa_path" in results:
+                file_record.original_pdfa_path = results["original_pdfa_path"]
+            if "processed_pdfa_path" in results:
+                file_record.processed_pdfa_path = results["processed_pdfa_path"]
+            db.commit()
+            logger.info(f"[{task_id}] Updated database with PDF/A paths")
+
+    # --- Upload PDF/A variants to storage providers ---
+    folder_overrides = _compute_pdfa_folder_overrides()
+
+    if settings.pdfa_upload_original and "original_pdfa_path" in results:
+        from app.tasks.send_to_all import send_to_all_destinations
+
+        logger.info(f"[{task_id}] Uploading original PDF/A to storage providers")
+        log_task_progress(
+            task_id,
+            "upload_original_pdfa",
+            "in_progress",
+            "Uploading original PDF/A to storage providers",
+            file_id=file_id,
+        )
+        send_to_all_destinations.delay(
+            results["original_pdfa_path"],
+            True,
+            file_id,
+            folder_overrides=folder_overrides if folder_overrides else None,
+        )
+        log_task_progress(
+            task_id,
+            "upload_original_pdfa",
+            "success",
+            "Original PDF/A queued for upload",
+            file_id=file_id,
+        )
+
+    if settings.pdfa_upload_processed and "processed_pdfa_path" in results:
+        from app.tasks.send_to_all import send_to_all_destinations
+
+        logger.info(f"[{task_id}] Uploading processed PDF/A to storage providers")
+        log_task_progress(
+            task_id,
+            "upload_processed_pdfa",
+            "in_progress",
+            "Uploading processed PDF/A to storage providers",
+            file_id=file_id,
+        )
+        send_to_all_destinations.delay(
+            results["processed_pdfa_path"],
+            True,
+            file_id,
+            folder_overrides=folder_overrides if folder_overrides else None,
+        )
+        log_task_progress(
+            task_id,
+            "upload_processed_pdfa",
+            "success",
+            "Processed PDF/A queued for upload",
+            file_id=file_id,
+        )
+
+    # --- Final status ---
+    has_any = bool(results)
+    status = "success" if has_any else "failure"
+    message = (
+        f"PDF/A conversion complete ({len(results)} variant(s) created)" if has_any else "No PDF/A variants created"
+    )
+    log_task_progress(task_id, "convert_to_pdfa", status, message, file_id=file_id)
+
+    return {"status": status, "file_id": file_id, **results}

app/tasks/finalize_document_storage.py

@@ -32,7 +32,7 @@ def finalize_document_storage(self, original_file: str, processed_file: str, met
     task_id = self.request.id
     logger.info(f"[{task_id}] Finalizing document storage for {processed_file}")
 
-    # 1. Update Database Status (From Main)
+    # 1. Update Database Status
     log_task_progress(
         task_id,
         "finalize_document_storage",
@@ -41,7 +41,7 @@ def finalize_document_storage(self, original_file: str, processed_file: str, met
         file_id=file_id,
     )
 
-    # Get file_id from database if not provided (fallback logic from Main)
+    # Get file_id from database if not provided (fallback logic)
     if file_id is None:
         with SessionLocal() as db:
             # Only as a last resort, try to find by exact match on local_filename
@@ -50,7 +50,7 @@ def finalize_document_storage(self, original_file: str, processed_file: str, met
             if file_record:
                 file_id = file_record.id
 
-    # 2. Determine Configured Destinations (From Copilot)
+    # 2. Determine Configured Destinations
     # This is needed for the notification message later
     configured_destinations = []
     try:
@@ -65,18 +65,33 @@ def finalize_document_storage(self, original_file: str, processed_file: str, met
         logger.warning(f"[WARNING] Could not determine configured destinations: {e}")
         configured_destinations = ["configured destinations"]
 
-    # 3. Queue Uploads (Merged)
-    # Uses Main branch signature to ensure file_id is passed, but keeps logic structure
+    # 3. Queue Uploads
     logger.info(f"[{task_id}] Queueing uploads to all destinations")
     log_task_progress(
         task_id, "finalize_document_storage", "success", "Queuing uploads to destinations", file_id=file_id
     )
 
     # Note: send_to_all_destinations is asynchronous and queues upload tasks
-    # We pass 'True' (delete_after) and 'file_id' as per Main branch requirements
     send_to_all_destinations.delay(processed_file, True, file_id)
 
-    # 3a. Queue embedding computation so similarity scores are ready for queries
+    # 3a. Trigger PDF/A archival conversion if enabled (from feature branch)
+    if settings.enable_pdfa_conversion:
+        try:
+            from app.tasks.convert_to_pdfa import convert_to_pdfa
+
+            logger.info(f"[{task_id}] PDF/A conversion enabled, queueing archival conversion")
+            log_task_progress(
+                task_id,
+                "finalize_document_storage",
+                "in_progress",
+                "Queueing PDF/A archival conversion",
+                file_id=file_id,
+            )
+            convert_to_pdfa.delay(file_id)
+        except Exception as e:
+            logger.warning(f"[{task_id}] Could not queue PDF/A conversion: {e}")
+
+    # 3b. Queue embedding computation (from main branch)
     if file_id is not None:
         try:
             from app.tasks.compute_embedding import compute_document_embedding
@@ -86,9 +101,7 @@ def finalize_document_storage(self, original_file: str, processed_file: str, met
         except Exception as e:
             logger.warning(f"[{task_id}] Could not queue embedding task: {e}")
 
-    # 4. Send Notification (From Copilot)
-    # Note: This notification is sent after processing is complete but while uploads
-    # are being queued.
+    # 4. Send Notification
     try:
         # Get file information
         file_size = os.path.getsize(processed_file) if os.path.exists(processed_file) else 0

app/tasks/send_to_all.py

@@ -106,7 +106,7 @@ def get_configured_services_from_validator():
 
 
 @celery.task(base=BaseTaskWithRetry, bind=True)
-def send_to_all_destinations(self, file_path: str, use_validator=True, file_id: int = None):
+def send_to_all_destinations(self, file_path: str, use_validator=True, file_id: int = None, folder_overrides=None):
     """
     Distribute a file to all configured storage destinations.
 
@@ -115,6 +115,10 @@ def send_to_all_destinations(self, file_path: str, use_validator=True, file_id:
         use_validator: Whether to use the config validator to determine enabled services
                        (if False, falls back to individual checks)
         file_id: Optional file ID to associate with logs
+        folder_overrides: Optional dict mapping provider names to folder override strings.
+                          When set, the override is passed to the upload task which uses it
+                          instead of the provider's default folder.  Example:
+                          {"dropbox": "/Documents/pdfa", "s3": "docs/pdfa/"}
     """
     task_id = self.request.id
 
@@ -236,7 +240,10 @@ def send_to_all_destinations(self, file_path: str, use_validator=True, file_id:
                 task_id, f"queue_{service_name}", "in_progress", f"Queueing upload to {service_name}", file_id=file_id
             )
             try:
-                task = service["upload_func"].delay(file_path, file_id=file_id)
+                kwargs = {"file_id": file_id}
+                if folder_overrides and service_name in folder_overrides:
+                    kwargs["folder_override"] = folder_overrides[service_name]
+                task = service["upload_func"].delay(file_path, **kwargs)
                 results[f"{service_name}_task_id"] = task.id
                 queued_count += 1
                 log_task_progress(

app/tasks/upload_to_dropbox.py

@@ -103,7 +103,7 @@ def get_dropbox_client():
 
 
 @celery.task(base=UploadTaskWithRetry, bind=True)
-def upload_to_dropbox(self, file_path: str, file_id: int = None):
+def upload_to_dropbox(self, file_path: str, file_id: int = None, folder_override: str = None):
     """
     Upload a file to Dropbox.
 
@@ -147,7 +147,7 @@ def upload_to_dropbox(self, file_path: str, file_id: int = None):
         dbx = get_dropbox_client()
 
         # Calculate remote path based on local file structure
-        remote_base = settings.dropbox_folder or ""
+        remote_base = folder_override if folder_override is not None else (settings.dropbox_folder or "")
         remote_path = extract_remote_path(file_path, settings.workdir, remote_base)
 
         # Function to check if file exists in Dropbox

app/tasks/upload_to_ftp.py

@@ -15,7 +15,7 @@ logger = logging.getLogger(__name__)
 
 
 @celery.task(base=UploadTaskWithRetry, bind=True)
-def upload_to_ftp(self, file_path: str, file_id: int = None):
+def upload_to_ftp(self, file_path: str, file_id: int = None, folder_override: str = None):
     """
     Uploads a file to an FTP server in the configured folder.
 
@@ -97,10 +97,11 @@ def upload_to_ftp(self, file_path: str, file_id: int = None):
             ftp.login(user=settings.ftp_username, passwd=settings.ftp_password)
 
         # Change to target directory if specified
-        if settings.ftp_folder:
+        ftp_folder_setting = folder_override if folder_override is not None else settings.ftp_folder
+        if ftp_folder_setting:
             try:
                 # Try to navigate to the directory, create if it doesn't exist
-                ftp_folder = settings.ftp_folder
+                ftp_folder = ftp_folder_setting
                 # Remove leading slash if present
                 if ftp_folder.startswith("/"):
                     ftp_folder = ftp_folder[1:]
@@ -138,7 +139,7 @@ def upload_to_ftp(self, file_path: str, file_id: int = None):
             "status": "Completed",
             "file": file_path,
             "ftp_host": settings.ftp_host,
-            "ftp_path": f"{settings.ftp_folder}/{filename}" if settings.ftp_folder else filename,
+            "ftp_path": f"{ftp_folder_setting}/{filename}" if ftp_folder_setting else filename,
             "used_tls": used_tls,
         }
 

app/tasks/upload_to_google_drive.py

@@ -153,7 +153,9 @@ def truncate_property_value(key, value, max_bytes=100):
 
 
 @celery.task(base=UploadTaskWithRetry, bind=True)
-def upload_to_google_drive(self, file_path: str, include_metadata=True, file_id: int = None):
+def upload_to_google_drive(
+    self, file_path: str, include_metadata=True, file_id: int = None, folder_override: str = None
+):
     """
     Uploads a file to Google Drive in the configured folder with optional metadata.
 
@@ -201,8 +203,9 @@ def upload_to_google_drive(self, file_path: str, include_metadata=True, file_id:
         }
 
         # If folder ID is specified, set parent folder
-        if settings.google_drive_folder_id:
-            file_metadata["parents"] = [settings.google_drive_folder_id]
+        gdrive_folder_id = folder_override if folder_override is not None else settings.google_drive_folder_id
+        if gdrive_folder_id:
+            file_metadata["parents"] = [gdrive_folder_id]
 
         # Add custom properties if metadata exists
         if metadata:

app/tasks/upload_to_nextcloud.py

@@ -16,7 +16,7 @@ logger = logging.getLogger(__name__)
 
 
 @celery.task(base=UploadTaskWithRetry, bind=True)
-def upload_to_nextcloud(self, file_path: str, file_id: int = None):
+def upload_to_nextcloud(self, file_path: str, file_id: int = None, folder_override: str = None):
     """
     Upload a file to Nextcloud WebDAV.
 
@@ -60,7 +60,9 @@ def upload_to_nextcloud(self, file_path: str, file_id: int = None):
             webdav_url += "/"
 
         # Calculate remote path based on local file structure
-        remote_base = getattr(settings, "nextcloud_folder", "") or ""
+        remote_base = (
+            folder_override if folder_override is not None else (getattr(settings, "nextcloud_folder", "") or "")
+        )
         remote_path = extract_remote_path(file_path, settings.workdir, remote_base)
         full_url = f"{webdav_url}/{remote_path}"
 

app/tasks/upload_to_onedrive.py

@@ -210,7 +210,7 @@ def upload_large_file(file_path, upload_url):
 
 
 @celery.task(base=UploadTaskWithRetry, bind=True)
-def upload_to_onedrive(self, file_path: str, file_id: int = None):
+def upload_to_onedrive(self, file_path: str, file_id: int = None, folder_override: str = None):
     """
     Uploads a file to OneDrive in the configured folder.
 
@@ -248,15 +248,17 @@ def upload_to_onedrive(self, file_path: str, file_id: int = None):
         # Get access token
         access_token = get_onedrive_token()
 
+        onedrive_folder = folder_override if folder_override is not None else settings.onedrive_folder_path
+
         # Create upload session
-        upload_url = create_upload_session(filename, settings.onedrive_folder_path, access_token)
+        upload_url = create_upload_session(filename, onedrive_folder, access_token)
 
         # Upload the file
         result = upload_large_file(file_path, upload_url)
 
         # Log success
         web_url = result.get("webUrl", "Not available")
-        logger.info(f"[{task_id}] Successfully uploaded {filename} to OneDrive at path {settings.onedrive_folder_path}")
+        logger.info(f"[{task_id}] Successfully uploaded {filename} to OneDrive at path {onedrive_folder}")
         logger.info(f"[{task_id}] File accessible at: {web_url}")
         log_task_progress(
             task_id, "upload_to_onedrive", "success", f"Uploaded to OneDrive: {filename}", file_id=file_id
@@ -265,7 +267,7 @@ def upload_to_onedrive(self, file_path: str, file_id: int = None):
         return {
             "status": "Completed",
             "file_path": file_path,
-            "onedrive_path": f"{settings.onedrive_folder_path}/{filename}",
+            "onedrive_path": f"{onedrive_folder}/{filename}",
             "web_url": web_url,
         }
 

app/tasks/upload_to_s3.py

@@ -15,7 +15,7 @@ logger = logging.getLogger(__name__)
 
 
 @celery.task(base=UploadTaskWithRetry, bind=True)
-def upload_to_s3(self, file_path: str, file_id: int = None):
+def upload_to_s3(self, file_path: str, file_id: int = None, folder_override: str = None):
     """
     Uploads a file to Amazon S3 in the configured bucket and folder.
 
@@ -61,9 +61,10 @@ def upload_to_s3(self, file_path: str, file_id: int = None):
         )
 
         # Construct the S3 key (path within the bucket)
-        if settings.s3_folder_prefix:
+        s3_folder = folder_override if folder_override is not None else settings.s3_folder_prefix
+        if s3_folder:
             # Ensure folder prefix ends with a slash
-            folder_prefix = settings.s3_folder_prefix
+            folder_prefix = s3_folder
             if not folder_prefix.endswith("/"):
                 folder_prefix += "/"
             s3_key = f"{folder_prefix}{filename}"

app/tasks/upload_to_sftp.py

@@ -15,7 +15,7 @@ logger = logging.getLogger(__name__)
 
 
 @celery.task(base=UploadTaskWithRetry, bind=True)
-def upload_to_sftp(self, file_path: str, file_id: int = None):
+def upload_to_sftp(self, file_path: str, file_id: int = None, folder_override: str = None):
     """
     Upload a file to an SFTP server.
 
@@ -95,7 +95,7 @@ def upload_to_sftp(self, file_path: str, file_id: int = None):
         sftp = ssh.open_sftp()
 
         # Calculate remote path based on local file structure
-        remote_base = settings.sftp_folder or ""
+        remote_base = folder_override if folder_override is not None else (settings.sftp_folder or "")
         remote_path = extract_remote_path(file_path, settings.workdir, remote_base)
 
         # Ensure the remote path starts with a slash if the base folder does

app/tasks/upload_to_webdav.py

@@ -15,7 +15,7 @@ logger = logging.getLogger(__name__)
 
 
 @celery.task(base=UploadTaskWithRetry, bind=True)
-def upload_to_webdav(self, file_path: str, file_id: int = None):
+def upload_to_webdav(self, file_path: str, file_id: int = None, folder_override: str = None):
     """
     Uploads a file to a WebDAV server in the configured folder.
 
@@ -50,7 +50,7 @@ def upload_to_webdav(self, file_path: str, file_id: int = None):
         raise ValueError(error_msg)
 
     # Construct the full upload URL
-    webdav_folder = settings.webdav_folder or ""
+    webdav_folder = folder_override if folder_override is not None else (settings.webdav_folder or "")
     # Ensure folder doesn't have leading slash if we're joining it to the base URL
     if webdav_folder and webdav_folder.startswith("/"):
         webdav_folder = webdav_folder[1:]

docs/ConfigurationGuide.md

@@ -881,6 +881,102 @@ Near-duplicate detection:
 
 A score of **≥ 0.90** reliably identifies the same document scanned twice. A score of **0.70–0.90** suggests partial content overlap. Adjust `NEAR_DUPLICATE_THRESHOLD` to tune sensitivity.
 
+## PDF/A Archival Conversion
+
+DocuElevate can optionally generate **PDF/A** archival copies of both the
+original ingested file and the processed file.  PDF/A copies are saved as
+parallel variants alongside the standard files—they do **not** replace the
+originals.  This provides better legal coverage by producing time-stamped,
+self-contained archival documents suitable for long-term storage and
+compliance.
+
+The conversion uses **ocrmypdf** (backed by Ghostscript), which is already
+bundled in the Docker images.
+
+> **Note:** PDF/A conversion may alter font rendering, especially for OCR text
+> overlays produced by Microsoft Azure Document Intelligence.  This is expected
+> and is why PDF/A copies are kept as parallel variants rather than
+> replacements.
+
+| Variable                      | Description                                                                                                       | Default                    |
+|-------------------------------|-------------------------------------------------------------------------------------------------------------------|----------------------------|
+| `ENABLE_PDFA_CONVERSION`     | Enable PDF/A archival variant generation for both original and processed files.                                     | `false`                    |
+| `PDFA_FORMAT`                | PDF/A format variant: `1` (PDF/A-1b), `2` (PDF/A-2b), `3` (PDF/A-3b).                                             | `2`                        |
+| `PDFA_UPLOAD_ORIGINAL`       | Upload the original-file PDF/A variant to all configured storage providers.                                        | `false`                    |
+| `PDFA_UPLOAD_PROCESSED`      | Upload the processed-file PDF/A variant to all configured storage providers.                                       | `false`                    |
+| `PDFA_UPLOAD_FOLDER`         | Subfolder name appended to each provider's folder for PDF/A uploads.                                               | `pdfa`                     |
+| `GOOGLE_DRIVE_PDFA_FOLDER_ID`| Google Drive folder ID for PDF/A uploads (uses folder IDs, not paths). Empty = use default folder.                 | *(empty)*                  |
+| `PDFA_TIMESTAMP_ENABLED`     | Enable RFC 3161 timestamping of PDF/A files (creates `.tsr` proof-of-existence files).                             | `false`                    |
+| `PDFA_TIMESTAMP_URL`         | URL of the RFC 3161 Timestamp Authority.                                                                           | `https://freetsa.org/tsr`  |
+
+### Storage Layout
+
+When enabled, PDF/A copies are stored under `workdir/pdfa/`:
+
+```
+workdir/
+├── original/          # Immutable copy of ingested file
+├── processed/         # Processed file with embedded metadata
+├── pdfa/
+│   ├── original/      # PDF/A copy of the ingested file
+│   │   └── *.pdf.tsr  # RFC 3161 timestamps (when timestamping enabled)
+│   └── processed/     # PDF/A copy of the processed file (with -PDFA suffix)
+│       └── *.pdf.tsr  # RFC 3161 timestamps (when timestamping enabled)
+└── tmp/               # Temporary processing area
+```
+
+### Per-Provider Folder Overrides
+
+When uploading PDF/A files to storage providers, DocuElevate appends the
+`PDFA_UPLOAD_FOLDER` value as a subfolder to each provider's configured folder.
+For example:
+
+| Provider     | Regular Folder              | PDF/A Upload Folder              |
+|--------------|-----------------------------|----------------------------------|
+| Dropbox      | `/Documents`                | `/Documents/pdfa`                |
+| S3           | `docs/uploads/`             | `docs/uploads/pdfa/`             |
+| Nextcloud    | `/Files`                    | `/Files/pdfa`                    |
+| OneDrive     | `Documents/Uploads`         | `Documents/Uploads/pdfa`         |
+| Google Drive | *(folder ID)*               | `GOOGLE_DRIVE_PDFA_FOLDER_ID`    |
+
+Set `PDFA_UPLOAD_FOLDER` to an empty string to upload PDF/A files into the
+same folder as regular uploads.
+
+### RFC 3161 Timestamping
+
+When `PDFA_TIMESTAMP_ENABLED=true`, each PDF/A file is timestamped using
+the configured TSA (default: [FreeTSA](https://freetsa.org)).  This creates
+a `.tsr` file alongside each PDF/A file, providing cryptographic proof that
+the document existed at a specific point in time.
+
+Requires `openssl` on the PATH (included in Docker images).
+
+**Other TSA options:**
+- **GlobalSign** – enterprise, eIDAS qualified
+- **DigiStamp** – high assurance, legal
+- **IdenTrust** – legal, free with certificate purchase
+
+### Configuration Example
+
+```bash
+# Enable PDF/A archival copies
+ENABLE_PDFA_CONVERSION=true
+
+# Use PDF/A-2b format (default, recommended for most use cases)
+PDFA_FORMAT=2
+
+# Upload both original and processed PDF/A to providers
+PDFA_UPLOAD_ORIGINAL=true
+PDFA_UPLOAD_PROCESSED=true
+
+# PDF/A files go into a 'pdfa' subfolder on each provider
+PDFA_UPLOAD_FOLDER=pdfa
+
+# Enable RFC 3161 timestamping via FreeTSA
+PDFA_TIMESTAMP_ENABLED=true
+PDFA_TIMESTAMP_URL=https://freetsa.org/tsr
+```
+
 ## Performance & Caching
 
 DocuElevate automatically optimizes database access and uses Redis as a

migrations/versions/011_add_pdfa_paths.py

@@ -0,0 +1,29 @@
+"""Add PDF/A archival variant path columns to files table
+
+Revision ID: 011_add_pdfa_paths
+Revises: 010_add_embedding_column
+Create Date: 2026-03-02
+
+"""
+
+from typing import Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "011_add_pdfa_paths"
+down_revision: Union[str, None] = "010_add_embedding_column"
+depends_on: Union[str, None] = None
+
+
+def upgrade() -> None:
+    """Add PDF/A variant path columns to files table."""
+    op.add_column("files", sa.Column("original_pdfa_path", sa.String(), nullable=True))
+    op.add_column("files", sa.Column("processed_pdfa_path", sa.String(), nullable=True))
+
+
+def downgrade() -> None:
+    """Remove PDF/A variant path columns from files table."""
+    op.drop_column("files", "processed_pdfa_path")
+    op.drop_column("files", "original_pdfa_path")

tests/test_convert_to_pdfa.py

@@ -0,0 +1,560 @@
+"""Unit tests for app/tasks/convert_to_pdfa.py module."""
+
+import subprocess
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from app.tasks.convert_to_pdfa import (
+    PDFA_ORIGINAL_SUBDIR,
+    PDFA_PROCESSED_SUBDIR,
+    _compute_pdfa_folder_overrides,
+    _convert_pdf_to_pdfa,
+    _timestamp_file,
+    convert_to_pdfa,
+)
+
+
+@pytest.mark.unit
+class TestConvertPdfToPdfa:
+    """Tests for the _convert_pdf_to_pdfa helper function."""
+
+    @patch("app.tasks.convert_to_pdfa.subprocess.run")
+    @patch("app.tasks.convert_to_pdfa.shutil.which", return_value="/usr/bin/ocrmypdf")
+    def test_successful_conversion(self, mock_which, mock_run):
+        """Test successful PDF to PDF/A conversion."""
+        mock_run.return_value = MagicMock(returncode=0, stderr="")
+
+        result = _convert_pdf_to_pdfa("/input.pdf", "/output.pdf", "2")
+
+        assert result is True
+        mock_which.assert_called_once_with("ocrmypdf")
+        mock_run.assert_called_once()
+
+        cmd = mock_run.call_args[0][0]
+        assert cmd[0] == "/usr/bin/ocrmypdf"
+        assert "--skip-text" in cmd
+        assert "--output-type" in cmd
+        assert "pdfa-2" in cmd
+        assert "--quiet" in cmd
+        assert "--invalidate-digital-signatures" in cmd
+        assert "/input.pdf" in cmd
+        assert "/output.pdf" in cmd
+
+    @patch("app.tasks.convert_to_pdfa.shutil.which", return_value=None)
+    def test_ocrmypdf_not_found(self, mock_which):
+        """Test returns False when ocrmypdf binary is not on PATH."""
+        result = _convert_pdf_to_pdfa("/input.pdf", "/output.pdf")
+        assert result is False
+
+    @patch("app.tasks.convert_to_pdfa.subprocess.run")
+    @patch("app.tasks.convert_to_pdfa.shutil.which", return_value="/usr/bin/ocrmypdf")
+    def test_conversion_failure(self, mock_which, mock_run):
+        """Test returns False when ocrmypdf exits with non-zero code."""
+        mock_run.return_value = MagicMock(returncode=1, stderr="Some error occurred")
+        result = _convert_pdf_to_pdfa("/input.pdf", "/output.pdf")
+        assert result is False
+
+    @patch("app.tasks.convert_to_pdfa.subprocess.run")
+    @patch("app.tasks.convert_to_pdfa.shutil.which", return_value="/usr/bin/ocrmypdf")
+    def test_conversion_timeout(self, mock_which, mock_run):
+        """Test returns False when ocrmypdf times out."""
+        mock_run.side_effect = subprocess.TimeoutExpired(cmd="ocrmypdf", timeout=600)
+        result = _convert_pdf_to_pdfa("/input.pdf", "/output.pdf")
+        assert result is False
+
+    @patch("app.tasks.convert_to_pdfa.subprocess.run")
+    @patch("app.tasks.convert_to_pdfa.shutil.which", return_value="/usr/bin/ocrmypdf")
+    def test_pdfa_format_variants(self, mock_which, mock_run):
+        """Test different PDF/A format variants are passed correctly."""
+        mock_run.return_value = MagicMock(returncode=0, stderr="")
+        for fmt in ("1", "2", "3"):
+            _convert_pdf_to_pdfa("/input.pdf", "/output.pdf", fmt)
+            cmd = mock_run.call_args[0][0]
+            assert f"pdfa-{fmt}" in cmd
+
+    def test_invalid_pdfa_format_rejected(self):
+        """Test that invalid PDF/A format values are rejected."""
+        result = _convert_pdf_to_pdfa("/input.pdf", "/output.pdf", "invalid")
+        assert result is False
+
+    @patch("app.tasks.convert_to_pdfa.subprocess.run")
+    @patch("app.tasks.convert_to_pdfa.shutil.which", return_value="/usr/bin/ocrmypdf")
+    def test_conversion_failure_empty_stderr(self, mock_which, mock_run):
+        """Test handles empty stderr on failure."""
+        mock_run.return_value = MagicMock(returncode=2, stderr="")
+        result = _convert_pdf_to_pdfa("/input.pdf", "/output.pdf")
+        assert result is False
+
+
+@pytest.mark.unit
+class TestTimestampFile:
+    """Tests for the _timestamp_file helper function."""
+
+    @patch("app.tasks.convert_to_pdfa.os.path.exists", return_value=False)
+    @patch("app.tasks.convert_to_pdfa.http_requests.post")
+    @patch("app.tasks.convert_to_pdfa.subprocess.run")
+    @patch("app.tasks.convert_to_pdfa.shutil.which", return_value="/usr/bin/openssl")
+    def test_successful_timestamp(self, mock_which, mock_run, mock_post, mock_exists):
+        """Test successful RFC 3161 timestamping."""
+        mock_run.return_value = MagicMock(returncode=0)
+        mock_post.return_value = MagicMock(status_code=200, content=b"tsr-data")
+
+        with patch("builtins.open", MagicMock()):
+            result = _timestamp_file("/test.pdf", "https://freetsa.org/tsr")
+
+        assert result == "/test.pdf.tsr"
+        mock_which.assert_called_once_with("openssl")
+        mock_run.assert_called_once()
+        mock_post.assert_called_once()
+
+    @patch("app.tasks.convert_to_pdfa.shutil.which", return_value=None)
+    def test_openssl_not_found(self, mock_which):
+        """Test returns None when openssl is not on PATH."""
+        result = _timestamp_file("/test.pdf", "https://freetsa.org/tsr")
+        assert result is None
+
+    @patch("app.tasks.convert_to_pdfa.os.path.exists", return_value=False)
+    @patch("app.tasks.convert_to_pdfa.subprocess.run")
+    @patch("app.tasks.convert_to_pdfa.shutil.which", return_value="/usr/bin/openssl")
+    def test_openssl_ts_query_fails(self, mock_which, mock_run, mock_exists):
+        """Test returns None when openssl ts -query fails."""
+        mock_run.return_value = MagicMock(returncode=1, stderr="error")
+        result = _timestamp_file("/test.pdf", "https://freetsa.org/tsr")
+        assert result is None
+
+    @patch("app.tasks.convert_to_pdfa.os.path.exists", return_value=False)
+    @patch("app.tasks.convert_to_pdfa.http_requests.post")
+    @patch("app.tasks.convert_to_pdfa.subprocess.run")
+    @patch("app.tasks.convert_to_pdfa.shutil.which", return_value="/usr/bin/openssl")
+    def test_tsa_returns_error(self, mock_which, mock_run, mock_post, mock_exists):
+        """Test returns None when TSA returns non-200 status."""
+        mock_run.return_value = MagicMock(returncode=0)
+        mock_post.return_value = MagicMock(status_code=500)
+
+        with patch("builtins.open", MagicMock()):
+            result = _timestamp_file("/test.pdf", "https://freetsa.org/tsr")
+
+        assert result is None
+
+    @patch("app.tasks.convert_to_pdfa.os.path.exists", return_value=False)
+    @patch("app.tasks.convert_to_pdfa.http_requests.post")
+    @patch("app.tasks.convert_to_pdfa.subprocess.run")
+    @patch("app.tasks.convert_to_pdfa.shutil.which", return_value="/usr/bin/openssl")
+    def test_tsa_network_error(self, mock_which, mock_run, mock_post, mock_exists):
+        """Test returns None on network error contacting TSA."""
+        import requests
+
+        mock_run.return_value = MagicMock(returncode=0)
+        mock_post.side_effect = requests.ConnectionError("Network error")
+
+        with patch("builtins.open", MagicMock()):
+            result = _timestamp_file("/test.pdf", "https://freetsa.org/tsr")
+
+        assert result is None
+
+    @patch("app.tasks.convert_to_pdfa.os.path.exists", return_value=False)
+    @patch("app.tasks.convert_to_pdfa.subprocess.run")
+    @patch("app.tasks.convert_to_pdfa.shutil.which", return_value="/usr/bin/openssl")
+    def test_openssl_timeout(self, mock_which, mock_run, mock_exists):
+        """Test returns None when openssl times out."""
+        mock_run.side_effect = subprocess.TimeoutExpired(cmd="openssl", timeout=30)
+        result = _timestamp_file("/test.pdf", "https://freetsa.org/tsr")
+        assert result is None
+
+
+@pytest.mark.unit
+class TestComputePdfaFolderOverrides:
+    """Tests for _compute_pdfa_folder_overrides helper."""
+
+    @patch("app.tasks.convert_to_pdfa.settings")
+    def test_appends_subfolder_to_providers(self, mock_settings):
+        """Test subfolder is appended to each provider's folder."""
+        mock_settings.pdfa_upload_folder = "pdfa"
+        mock_settings.dropbox_folder = "/Documents"
+        mock_settings.nextcloud_folder = "/Files"
+        mock_settings.webdav_folder = "/webdav"
+        mock_settings.ftp_folder = "/uploads"
+        mock_settings.sftp_folder = "/remote"
+        mock_settings.onedrive_folder_path = "Documents/Uploads"
+        mock_settings.s3_folder_prefix = "docs/"
+        mock_settings.google_drive_pdfa_folder_id = "gdrive-pdfa-folder-id"
+
+        overrides = _compute_pdfa_folder_overrides()
+
+        assert overrides["dropbox"] == "/Documents/pdfa"
+        assert overrides["nextcloud"] == "/Files/pdfa"
+        assert overrides["webdav"] == "/webdav/pdfa"
+        assert overrides["ftp"] == "/uploads/pdfa"
+        assert overrides["sftp"] == "/remote/pdfa"
+        assert overrides["onedrive"] == "Documents/Uploads/pdfa"
+        assert overrides["s3"] == "docs/pdfa/"
+        assert overrides["google_drive"] == "gdrive-pdfa-folder-id"
+
+    @patch("app.tasks.convert_to_pdfa.settings")
+    def test_empty_subfolder_returns_empty(self, mock_settings):
+        """Test empty subfolder returns empty overrides dict."""
+        mock_settings.pdfa_upload_folder = ""
+        overrides = _compute_pdfa_folder_overrides()
+        assert overrides == {}
+
+    @patch("app.tasks.convert_to_pdfa.settings")
+    def test_no_gdrive_pdfa_id_excluded(self, mock_settings):
+        """Test Google Drive excluded when no dedicated folder ID set."""
+        mock_settings.pdfa_upload_folder = "archive"
+        mock_settings.dropbox_folder = "/docs"
+        mock_settings.nextcloud_folder = ""
+        mock_settings.webdav_folder = ""
+        mock_settings.ftp_folder = ""
+        mock_settings.sftp_folder = ""
+        mock_settings.onedrive_folder_path = ""
+        mock_settings.s3_folder_prefix = ""
+        mock_settings.google_drive_pdfa_folder_id = ""
+
+        overrides = _compute_pdfa_folder_overrides()
+
+        assert overrides["dropbox"] == "/docs/archive"
+        assert "google_drive" not in overrides
+
+
+@pytest.mark.unit
+class TestConvertToPdfaTask:
+    """Tests for the convert_to_pdfa Celery task."""
+
+    def _mock_settings(self, mock_settings):
+        """Set standard mock settings for PDF/A tests."""
+        mock_settings.workdir = "/workdir"
+        mock_settings.pdfa_format = "2"
+        mock_settings.pdfa_upload_original = False
+        mock_settings.pdfa_upload_processed = False
+        mock_settings.pdfa_timestamp_enabled = False
+        mock_settings.pdfa_timestamp_url = "https://freetsa.org/tsr"
+
+    @patch("app.tasks.convert_to_pdfa.settings")
+    @patch("app.tasks.convert_to_pdfa._convert_pdf_to_pdfa", return_value=True)
+    @patch("app.tasks.convert_to_pdfa.get_unique_filepath_with_counter")
+    @patch("app.tasks.convert_to_pdfa.os.makedirs")
+    @patch("app.tasks.convert_to_pdfa.os.path.exists", return_value=True)
+    @patch("app.tasks.convert_to_pdfa.log_task_progress")
+    @patch("app.tasks.convert_to_pdfa.SessionLocal")
+    def test_successful_conversion_both_files(
+        self,
+        mock_session_local,
+        mock_log,
+        mock_exists,
+        mock_makedirs,
+        mock_unique_path,
+        mock_convert,
+        mock_settings,
+    ):
+        """Test successful PDF/A conversion of both original and processed files."""
+        self._mock_settings(mock_settings)
+        mock_unique_path.side_effect = [
+            "/workdir/pdfa/original/test.pdf",
+            "/workdir/pdfa/processed/test-PDFA.pdf",
+        ]
+
+        mock_db = MagicMock()
+        mock_session_local.return_value.__enter__.return_value = mock_db
+        mock_record = MagicMock()
+        mock_record.original_file_path = "/workdir/original/test.pdf"
+        mock_record.processed_file_path = "/workdir/processed/test.pdf"
+        mock_db.query.return_value.filter_by.return_value.first.return_value = mock_record
+
+        convert_to_pdfa.request.id = "test-task-id"
+        result = convert_to_pdfa.__wrapped__(file_id=1)
+
+        assert result["status"] == "success"
+        assert result["file_id"] == 1
+        assert "original_pdfa_path" in result
+        assert "processed_pdfa_path" in result
+        assert mock_convert.call_count == 2
+
+    @patch("app.tasks.convert_to_pdfa.settings")
+    @patch("app.tasks.convert_to_pdfa.log_task_progress")
+    @patch("app.tasks.convert_to_pdfa.SessionLocal")
+    def test_file_record_not_found(self, mock_session_local, mock_log, mock_settings):
+        """Test returns error when file record is not found."""
+        mock_db = MagicMock()
+        mock_session_local.return_value.__enter__.return_value = mock_db
+        mock_db.query.return_value.filter_by.return_value.first.return_value = None
+
+        convert_to_pdfa.request.id = "test-task-id"
+        result = convert_to_pdfa.__wrapped__(file_id=999)
+
+        assert "error" in result
+        assert result["file_id"] == 999
+
+    @patch("app.tasks.convert_to_pdfa.settings")
+    @patch("app.tasks.convert_to_pdfa._convert_pdf_to_pdfa", return_value=False)
+    @patch("app.tasks.convert_to_pdfa.get_unique_filepath_with_counter")
+    @patch("app.tasks.convert_to_pdfa.os.makedirs")
+    @patch("app.tasks.convert_to_pdfa.os.path.exists", return_value=True)
+    @patch("app.tasks.convert_to_pdfa.log_task_progress")
+    @patch("app.tasks.convert_to_pdfa.SessionLocal")
+    def test_conversion_failure_both_files(
+        self,
+        mock_session_local,
+        mock_log,
+        mock_exists,
+        mock_makedirs,
+        mock_unique_path,
+        mock_convert,
+        mock_settings,
+    ):
+        """Test handles failure when both conversions fail."""
+        self._mock_settings(mock_settings)
+        mock_unique_path.side_effect = [
+            "/workdir/pdfa/original/test.pdf",
+            "/workdir/pdfa/processed/test-PDFA.pdf",
+        ]
+
+        mock_db = MagicMock()
+        mock_session_local.return_value.__enter__.return_value = mock_db
+        mock_record = MagicMock()
+        mock_record.original_file_path = "/workdir/original/test.pdf"
+        mock_record.processed_file_path = "/workdir/processed/test.pdf"
+        mock_db.query.return_value.filter_by.return_value.first.return_value = mock_record
+
+        convert_to_pdfa.request.id = "test-task-id"
+        result = convert_to_pdfa.__wrapped__(file_id=1)
+
+        assert result["status"] == "failure"
+
+    @patch("app.tasks.convert_to_pdfa.settings")
+    @patch("app.tasks.convert_to_pdfa.log_task_progress")
+    @patch("app.tasks.convert_to_pdfa.SessionLocal")
+    def test_skips_missing_files(self, mock_session_local, mock_log, mock_settings):
+        """Test skips conversion when original/processed files don't exist."""
+        self._mock_settings(mock_settings)
+        mock_db = MagicMock()
+        mock_session_local.return_value.__enter__.return_value = mock_db
+        mock_record = MagicMock()
+        mock_record.original_file_path = None
+        mock_record.processed_file_path = None
+        mock_db.query.return_value.filter_by.return_value.first.return_value = mock_record
+
+        convert_to_pdfa.request.id = "test-task-id"
+        result = convert_to_pdfa.__wrapped__(file_id=1)
+        assert result["status"] == "failure"
+
+    @patch("app.tasks.convert_to_pdfa._compute_pdfa_folder_overrides", return_value={"dropbox": "/docs/pdfa"})
+    @patch("app.tasks.send_to_all.send_to_all_destinations")
+    @patch("app.tasks.convert_to_pdfa.settings")
+    @patch("app.tasks.convert_to_pdfa._convert_pdf_to_pdfa", return_value=True)
+    @patch("app.tasks.convert_to_pdfa.get_unique_filepath_with_counter")
+    @patch("app.tasks.convert_to_pdfa.os.makedirs")
+    @patch("app.tasks.convert_to_pdfa.os.path.exists", return_value=True)
+    @patch("app.tasks.convert_to_pdfa.log_task_progress")
+    @patch("app.tasks.convert_to_pdfa.SessionLocal")
+    def test_uploads_processed_pdfa_when_enabled(
+        self,
+        mock_session_local,
+        mock_log,
+        mock_exists,
+        mock_makedirs,
+        mock_unique_path,
+        mock_convert,
+        mock_settings,
+        mock_send_all,
+        mock_overrides,
+    ):
+        """Test uploads processed PDF/A with folder overrides when enabled."""
+        self._mock_settings(mock_settings)
+        mock_settings.pdfa_upload_processed = True
+        mock_unique_path.side_effect = [
+            "/workdir/pdfa/original/test.pdf",
+            "/workdir/pdfa/processed/test-PDFA.pdf",
+        ]
+
+        mock_db = MagicMock()
+        mock_session_local.return_value.__enter__.return_value = mock_db
+        mock_record = MagicMock()
+        mock_record.original_file_path = "/workdir/original/test.pdf"
+        mock_record.processed_file_path = "/workdir/processed/test.pdf"
+        mock_db.query.return_value.filter_by.return_value.first.return_value = mock_record
+
+        convert_to_pdfa.request.id = "test-task-id"
+        convert_to_pdfa.__wrapped__(file_id=42)
+
+        mock_send_all.delay.assert_called_once_with(
+            "/workdir/pdfa/processed/test-PDFA.pdf",
+            True,
+            42,
+            folder_overrides={"dropbox": "/docs/pdfa"},
+        )
+
+    @patch("app.tasks.convert_to_pdfa._compute_pdfa_folder_overrides", return_value={"s3": "docs/pdfa/"})
+    @patch("app.tasks.send_to_all.send_to_all_destinations")
+    @patch("app.tasks.convert_to_pdfa.settings")
+    @patch("app.tasks.convert_to_pdfa._convert_pdf_to_pdfa", return_value=True)
+    @patch("app.tasks.convert_to_pdfa.get_unique_filepath_with_counter")
+    @patch("app.tasks.convert_to_pdfa.os.makedirs")
+    @patch("app.tasks.convert_to_pdfa.os.path.exists", return_value=True)
+    @patch("app.tasks.convert_to_pdfa.log_task_progress")
+    @patch("app.tasks.convert_to_pdfa.SessionLocal")
+    def test_uploads_original_pdfa_when_enabled(
+        self,
+        mock_session_local,
+        mock_log,
+        mock_exists,
+        mock_makedirs,
+        mock_unique_path,
+        mock_convert,
+        mock_settings,
+        mock_send_all,
+        mock_overrides,
+    ):
+        """Test uploads original PDF/A to providers when pdfa_upload_original is True."""
+        self._mock_settings(mock_settings)
+        mock_settings.pdfa_upload_original = True
+        mock_unique_path.side_effect = [
+            "/workdir/pdfa/original/test.pdf",
+            "/workdir/pdfa/processed/test-PDFA.pdf",
+        ]
+
+        mock_db = MagicMock()
+        mock_session_local.return_value.__enter__.return_value = mock_db
+        mock_record = MagicMock()
+        mock_record.original_file_path = "/workdir/original/test.pdf"
+        mock_record.processed_file_path = "/workdir/processed/test.pdf"
+        mock_db.query.return_value.filter_by.return_value.first.return_value = mock_record
+
+        convert_to_pdfa.request.id = "test-task-id"
+        convert_to_pdfa.__wrapped__(file_id=42)
+
+        # Only original should be uploaded
+        mock_send_all.delay.assert_called_once()
+        call_args = mock_send_all.delay.call_args
+        assert call_args[0][0] == "/workdir/pdfa/original/test.pdf"
+
+    @patch("app.tasks.convert_to_pdfa.settings")
+    @patch("app.tasks.convert_to_pdfa._convert_pdf_to_pdfa", return_value=True)
+    @patch("app.tasks.convert_to_pdfa.get_unique_filepath_with_counter")
+    @patch("app.tasks.convert_to_pdfa.os.makedirs")
+    @patch("app.tasks.convert_to_pdfa.os.path.exists", return_value=True)
+    @patch("app.tasks.convert_to_pdfa.log_task_progress")
+    @patch("app.tasks.convert_to_pdfa.SessionLocal")
+    def test_does_not_upload_when_disabled(
+        self,
+        mock_session_local,
+        mock_log,
+        mock_exists,
+        mock_makedirs,
+        mock_unique_path,
+        mock_convert,
+        mock_settings,
+    ):
+        """Test does not upload PDF/A when both upload flags are False."""
+        self._mock_settings(mock_settings)
+        mock_unique_path.side_effect = [
+            "/workdir/pdfa/original/test.pdf",
+            "/workdir/pdfa/processed/test-PDFA.pdf",
+        ]
+
+        mock_db = MagicMock()
+        mock_session_local.return_value.__enter__.return_value = mock_db
+        mock_record = MagicMock()
+        mock_record.original_file_path = "/workdir/original/test.pdf"
+        mock_record.processed_file_path = "/workdir/processed/test.pdf"
+        mock_db.query.return_value.filter_by.return_value.first.return_value = mock_record
+
+        convert_to_pdfa.request.id = "test-task-id"
+        result = convert_to_pdfa.__wrapped__(file_id=1)
+        assert result["status"] == "success"
+
+    @patch("app.tasks.convert_to_pdfa._timestamp_file", return_value="/workdir/pdfa/original/test.pdf.tsr")
+    @patch("app.tasks.convert_to_pdfa.settings")
+    @patch("app.tasks.convert_to_pdfa._convert_pdf_to_pdfa", return_value=True)
+    @patch("app.tasks.convert_to_pdfa.get_unique_filepath_with_counter")
+    @patch("app.tasks.convert_to_pdfa.os.makedirs")
+    @patch("app.tasks.convert_to_pdfa.os.path.exists", return_value=True)
+    @patch("app.tasks.convert_to_pdfa.log_task_progress")
+    @patch("app.tasks.convert_to_pdfa.SessionLocal")
+    def test_timestamping_when_enabled(
+        self,
+        mock_session_local,
+        mock_log,
+        mock_exists,
+        mock_makedirs,
+        mock_unique_path,
+        mock_convert,
+        mock_settings,
+        mock_timestamp,
+    ):
+        """Test RFC 3161 timestamping is called when enabled."""
+        self._mock_settings(mock_settings)
+        mock_settings.pdfa_timestamp_enabled = True
+        mock_unique_path.side_effect = [
+            "/workdir/pdfa/original/test.pdf",
+            "/workdir/pdfa/processed/test-PDFA.pdf",
+        ]
+
+        mock_db = MagicMock()
+        mock_session_local.return_value.__enter__.return_value = mock_db
+        mock_record = MagicMock()
+        mock_record.original_file_path = "/workdir/original/test.pdf"
+        mock_record.processed_file_path = "/workdir/processed/test.pdf"
+        mock_db.query.return_value.filter_by.return_value.first.return_value = mock_record
+
+        convert_to_pdfa.request.id = "test-task-id"
+        result = convert_to_pdfa.__wrapped__(file_id=1)
+
+        assert result["status"] == "success"
+        assert mock_timestamp.call_count == 2
+
+    @patch("app.tasks.convert_to_pdfa.settings")
+    @patch("app.tasks.convert_to_pdfa._convert_pdf_to_pdfa")
+    @patch("app.tasks.convert_to_pdfa.get_unique_filepath_with_counter")
+    @patch("app.tasks.convert_to_pdfa.os.makedirs")
+    @patch("app.tasks.convert_to_pdfa.os.path.exists")
+    @patch("app.tasks.convert_to_pdfa.log_task_progress")
+    @patch("app.tasks.convert_to_pdfa.SessionLocal")
+    def test_partial_success_original_only(
+        self,
+        mock_session_local,
+        mock_log,
+        mock_exists,
+        mock_makedirs,
+        mock_unique_path,
+        mock_convert,
+        mock_settings,
+    ):
+        """Test partial success when only original conversion succeeds."""
+        self._mock_settings(mock_settings)
+
+        def exists_side_effect(path):
+            return "/original/" in path
+
+        mock_exists.side_effect = exists_side_effect
+        mock_unique_path.return_value = "/workdir/pdfa/original/test.pdf"
+        mock_convert.return_value = True
+
+        mock_db = MagicMock()
+        mock_session_local.return_value.__enter__.return_value = mock_db
+        mock_record = MagicMock()
+        mock_record.original_file_path = "/workdir/original/test.pdf"
+        mock_record.processed_file_path = "/workdir/processed/test.pdf"
+        mock_db.query.return_value.filter_by.return_value.first.return_value = mock_record
+
+        convert_to_pdfa.request.id = "test-task-id"
+        result = convert_to_pdfa.__wrapped__(file_id=1)
+
+        assert result["status"] == "success"
+        assert "original_pdfa_path" in result
+        assert "processed_pdfa_path" not in result
+
+
+@pytest.mark.unit
+class TestPdfaSubdirectoryConstants:
+    """Tests for PDF/A subdirectory constants."""
+
+    def test_original_subdir(self):
+        """Test PDFA_ORIGINAL_SUBDIR is correct."""
+        assert "pdfa" in PDFA_ORIGINAL_SUBDIR
+        assert "original" in PDFA_ORIGINAL_SUBDIR
+
+    def test_processed_subdir(self):
+        """Test PDFA_PROCESSED_SUBDIR is correct."""
+        assert "pdfa" in PDFA_PROCESSED_SUBDIR
+        assert "processed" in PDFA_PROCESSED_SUBDIR

tests/test_finalize_storage.py

@@ -107,6 +107,7 @@ class TestFinalizeDocumentStorage:
                     ):
                         with patch("app.tasks.finalize_document_storage.settings") as mock_settings:
                             mock_settings.workdir = "/tmp"
+                            mock_settings.enable_pdfa_conversion = False
 
                             finalize_document_storage.request.id = "test-task-id"