feat(pipelines): add custom processing pipeline engine

- Add Pipeline and PipelineStep models with user-specific ownership
- Add pipeline_id FK column to FileRecord
- Migration 017_add_pipelines (batch mode for SQLite FK compat)
- Pipeline CRUD API at /api/pipelines with step management endpoints
- Reorder steps PUT endpoint placed before parameterised {step_id} routes
- POST /api/files/{id}/assign-pipeline for per-file pipeline assignment
- Admin-only POST /api/pipelines/admin/system for system-level pipelines
- Management UI at /pipelines (Jinja2 + Alpine.js + Tailwind)
- Pipelines link added to desktop and mobile navigation
- 41 new tests in tests/test_api_pipelines.py (all passing)

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>

app/api/__init__.py

@@ -17,6 +17,7 @@ from app.api.google_drive import router as google_drive_router
 from app.api.logs import router as logs_router
 from app.api.onedrive import router as onedrive_router
 from app.api.openai import router as openai_router
+from app.api.pipelines import router as pipelines_router
 from app.api.plans import router as plans_router
 from app.api.process import router as process_router
 from app.api.queue import router as queue_router
@@ -60,3 +61,4 @@ router.include_router(webhooks_router)
 router.include_router(database_router)
 router.include_router(subscriptions_router)
 router.include_router(plans_router)
+router.include_router(pipelines_router)

app/api/files.py

@@ -1559,3 +1559,75 @@ def assign_owner(request: Request, db: DbSession, owner_id: str = Query(...), fi
         "updated_count": updated,
         "owner_id": owner_id,
     }
+
+
+# ---------------------------------------------------------------------------
+# Pipeline assignment
+# ---------------------------------------------------------------------------
+
+
+@router.post("/files/{file_id}/assign-pipeline")
+@require_login
+def assign_pipeline_to_file(
+    request: Request,
+    file_id: int,
+    db: DbSession,
+    pipeline_id: int | None = None,
+):
+    """Assign (or remove) a processing pipeline from a file.
+
+    Path Parameters:
+        file_id: The file to update.
+
+    Query / Body Parameters:
+        pipeline_id: The pipeline to assign.  Pass ``null`` or omit to clear the
+            assignment (the system default will be used for future processing).
+
+    Returns:
+        A summary dict with the file_id and updated pipeline_id.
+
+    Raises:
+        HTTPException 404: If the file or pipeline does not exist / is not
+            accessible to the current user.
+    """
+    from app.auth import get_current_user
+    from app.models import Pipeline
+
+    user = get_current_user(request)
+    # Derive user identity the same way the pipelines API does (_get_user_id)
+    if user:
+        user_id: str = user.get("preferred_username") or user.get("email") or user.get("id") or "anonymous"
+    else:
+        user_id = "anonymous"
+
+    is_admin_user = bool(user and user.get("is_admin"))
+
+    file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first()
+    if not file_record:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
+
+    # Non-admins may only update files they own (or unowned files in single-user mode)
+    owner_id = get_current_owner_id(request)
+    if not is_admin_user and file_record.owner_id is not None and file_record.owner_id != owner_id:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
+
+    if pipeline_id is not None:
+        pipeline = db.query(Pipeline).filter(Pipeline.id == pipeline_id).first()
+        if not pipeline:
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Pipeline not found")
+        # Check access: users can only assign their own pipelines or system pipelines (owner_id=None)
+        if not is_admin_user and pipeline.owner_id is not None and pipeline.owner_id != user_id:
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Pipeline not found")
+
+    file_record.pipeline_id = pipeline_id
+
+    try:
+        db.commit()
+        db.refresh(file_record)
+    except Exception as exc:
+        db.rollback()
+        logger.exception(f"Failed to assign pipeline to file id={file_id}: {exc}")
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to assign pipeline")
+
+    logger.info(f"Pipeline {pipeline_id!r} assigned to file id={file_id}")
+    return {"file_id": file_id, "pipeline_id": file_record.pipeline_id}

app/api/pipelines.py

@@ -0,0 +1,793 @@
+"""
+Pipelines API endpoints.
+
+Provides full CRUD for processing pipelines and their steps.  Pipelines are
+user-specific: regular users can only manage their own pipelines, while admins
+can also create and manage *system default* pipelines (owner_id = NULL) that
+are visible to all users.
+
+Built-in step types are exposed via GET /api/pipelines/step-types so that UIs
+can render the correct configuration form without hard-coding the catalogue.
+"""
+
+import json
+import logging
+from typing import Annotated, Any
+
+from fastapi import APIRouter, Body, Depends, HTTPException, Request, status
+from pydantic import BaseModel, Field
+from sqlalchemy.orm import Session
+
+from app.auth import get_current_user, require_login
+from app.database import get_db
+from app.models import Pipeline, PipelineStep
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/pipelines", tags=["pipelines"])
+
+DbSession = Annotated[Session, Depends(get_db)]
+
+# ---------------------------------------------------------------------------
+# Built-in step type catalogue
+# ---------------------------------------------------------------------------
+
+PIPELINE_STEP_TYPES: dict[str, dict[str, Any]] = {
+    "convert_to_pdf": {
+        "label": "Convert to PDF",
+        "description": "Convert non-PDF documents to PDF format using Gotenberg.",
+        "config_schema": {},
+    },
+    "check_duplicates": {
+        "label": "Check for Duplicates",
+        "description": "Compare file hash against existing documents to detect duplicates.",
+        "config_schema": {},
+    },
+    "ocr": {
+        "label": "OCR Processing",
+        "description": "Extract text using Azure Document Intelligence or local Tesseract.",
+        "config_schema": {
+            "force_cloud_ocr": {
+                "type": "boolean",
+                "default": False,
+                "description": "Always use cloud OCR even if the PDF already has embedded text.",
+            }
+        },
+    },
+    "extract_metadata": {
+        "label": "Metadata Extraction",
+        "description": "Extract structured metadata (document type, sender, recipient, tags) using AI.",
+        "config_schema": {},
+    },
+    "embed_metadata": {
+        "label": "Embed Metadata into PDF",
+        "description": "Write the extracted metadata into the PDF document properties.",
+        "config_schema": {},
+    },
+    "compute_embedding": {
+        "label": "Compute Text Embedding",
+        "description": "Compute semantic text embeddings for full-text and similarity search.",
+        "config_schema": {},
+    },
+    "send_to_destinations": {
+        "label": "Send to Storage Destinations",
+        "description": "Upload the processed document to all configured storage destinations.",
+        "config_schema": {},
+    },
+    "classify": {
+        "label": "Document Classification",
+        "description": "Classify the document type using AI without full metadata extraction.",
+        "config_schema": {},
+    },
+}
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+MAX_STEPS_PER_PIPELINE = 50
+MAX_NAME_LENGTH = 255
+
+
+def _get_user_id(request: Request) -> str:
+    """Return a stable user identifier from the session."""
+    user = get_current_user(request)
+    if user:
+        return user.get("preferred_username") or user.get("email") or user.get("id", "anonymous")
+    return "anonymous"
+
+
+def _is_admin(request: Request) -> bool:
+    """Return True if the current session user is an admin."""
+    user = get_current_user(request)
+    return bool(user and user.get("is_admin"))
+
+
+def _can_access_pipeline(pipeline: Pipeline, user_id: str, admin: bool) -> bool:
+    """Return True if the user may read or write this pipeline."""
+    # System pipelines (owner_id=NULL) are readable by everyone; only admins can write
+    if pipeline.owner_id is None:
+        return True
+    # Own pipeline
+    return pipeline.owner_id == user_id or admin
+
+
+def _can_write_pipeline(pipeline: Pipeline, user_id: str, admin: bool) -> bool:
+    """Return True if the user may create/update/delete this pipeline."""
+    if pipeline.owner_id is None:
+        return admin
+    return pipeline.owner_id == user_id or admin
+
+
+def _serialize_step(step: PipelineStep) -> dict[str, Any]:
+    return {
+        "id": step.id,
+        "pipeline_id": step.pipeline_id,
+        "position": step.position,
+        "step_type": step.step_type,
+        "label": step.label,
+        "config": json.loads(step.config) if step.config else {},
+        "enabled": step.enabled,
+        "created_at": step.created_at.isoformat() if step.created_at else None,
+        "updated_at": step.updated_at.isoformat() if step.updated_at else None,
+    }
+
+
+def _serialize_pipeline(pipeline: Pipeline, include_steps: bool = False, db: Session | None = None) -> dict[str, Any]:
+    data: dict[str, Any] = {
+        "id": pipeline.id,
+        "owner_id": pipeline.owner_id,
+        "name": pipeline.name,
+        "description": pipeline.description,
+        "is_default": pipeline.is_default,
+        "is_active": pipeline.is_active,
+        "created_at": pipeline.created_at.isoformat() if pipeline.created_at else None,
+        "updated_at": pipeline.updated_at.isoformat() if pipeline.updated_at else None,
+    }
+    if include_steps and db is not None:
+        steps = (
+            db.query(PipelineStep).filter(PipelineStep.pipeline_id == pipeline.id).order_by(PipelineStep.position).all()
+        )
+        data["steps"] = [_serialize_step(s) for s in steps]
+    return data
+
+
+# ---------------------------------------------------------------------------
+# Pydantic schemas
+# ---------------------------------------------------------------------------
+
+
+class PipelineCreate(BaseModel):
+    """Body for creating a pipeline."""
+
+    name: str = Field(..., max_length=MAX_NAME_LENGTH, description="Human-readable pipeline name")
+    description: str | None = Field(default=None, max_length=4096)
+    is_default: bool = Field(default=False)
+    is_active: bool = Field(default=True)
+
+
+class PipelineUpdate(BaseModel):
+    """Body for updating a pipeline (all fields optional)."""
+
+    name: str | None = Field(default=None, max_length=MAX_NAME_LENGTH)
+    description: str | None = Field(default=None, max_length=4096)
+    is_default: bool | None = None
+    is_active: bool | None = None
+
+
+class PipelineStepCreate(BaseModel):
+    """Body for adding a step to a pipeline."""
+
+    step_type: str = Field(..., description="One of the recognised step type keys")
+    label: str | None = Field(default=None, max_length=MAX_NAME_LENGTH)
+    config: dict[str, Any] = Field(default_factory=dict)
+    enabled: bool = Field(default=True)
+    position: int | None = Field(default=None, ge=0, description="Insertion position; appended at end if omitted")
+
+
+class PipelineStepUpdate(BaseModel):
+    """Body for updating a pipeline step (all fields optional)."""
+
+    step_type: str | None = None
+    label: str | None = Field(default=None, max_length=MAX_NAME_LENGTH)
+    config: dict[str, Any] | None = None
+    enabled: bool | None = None
+    position: int | None = Field(default=None, ge=0)
+
+
+# ---------------------------------------------------------------------------
+# Step-types catalogue endpoint (no auth required — it's public metadata)
+# ---------------------------------------------------------------------------
+
+
+@router.get("/step-types")
+def list_step_types() -> dict[str, Any]:
+    """Return the catalogue of built-in pipeline step types.
+
+    Returns:
+        A mapping of step_type key → metadata (label, description, config_schema).
+    """
+    return PIPELINE_STEP_TYPES
+
+
+# ---------------------------------------------------------------------------
+# Pipeline CRUD
+# ---------------------------------------------------------------------------
+
+
+@router.get("")
+@require_login
+def list_pipelines(request: Request, db: DbSession) -> list[dict[str, Any]]:
+    """List pipelines visible to the current user.
+
+    Regular users see: their own pipelines + system pipelines (owner_id=NULL).
+    Admins see: all pipelines from all users.
+
+    Returns:
+        A list of pipeline objects (without steps — use GET /pipelines/{id} for steps).
+    """
+    user_id = _get_user_id(request)
+    admin = _is_admin(request)
+
+    if admin:
+        pipelines = db.query(Pipeline).order_by(Pipeline.owner_id.nullsfirst(), Pipeline.name).all()
+    else:
+        pipelines = (
+            db.query(Pipeline)
+            .filter((Pipeline.owner_id == user_id) | (Pipeline.owner_id.is_(None)))
+            .order_by(Pipeline.owner_id.nullsfirst(), Pipeline.name)
+            .all()
+        )
+
+    return [_serialize_pipeline(p) for p in pipelines]
+
+
+@router.post("", status_code=status.HTTP_201_CREATED)
+@require_login
+def create_pipeline(request: Request, db: DbSession, body: PipelineCreate) -> dict[str, Any]:
+    """Create a new pipeline for the current user.
+
+    Admins can create system default pipelines by passing ``owner_id=null``
+    via the body — however, that is handled implicitly: to create a system
+    pipeline, call ``POST /api/admin/pipelines`` (admin endpoint) instead.
+    Regular users always get their own user_id as owner.
+
+    Returns:
+        The created pipeline object.
+
+    Raises:
+        HTTPException 409: If a pipeline with the same name already exists for this owner.
+    """
+    user_id = _get_user_id(request)
+
+    name = body.name.strip() if body.name else ""
+    if not name:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail="name is required",
+        )
+
+    # Enforce unique name per owner
+    existing = db.query(Pipeline).filter(Pipeline.owner_id == user_id, Pipeline.name == name).first()
+    if existing:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=f"A pipeline named '{name}' already exists",
+        )
+
+    # If this pipeline is marked as default, unset the existing default for this user
+    if body.is_default:
+        _unset_default(db, user_id)
+
+    pipeline = Pipeline(
+        owner_id=user_id,
+        name=name,
+        description=body.description,
+        is_default=body.is_default,
+        is_active=body.is_active,
+    )
+    try:
+        db.add(pipeline)
+        db.commit()
+        db.refresh(pipeline)
+    except Exception as exc:
+        db.rollback()
+        logger.exception(f"Failed to create pipeline user={user_id}: {exc}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to create pipeline",
+        )
+
+    logger.info(f"Pipeline created: id={pipeline.id}, owner={user_id}, name={name!r}")
+    return _serialize_pipeline(pipeline)
+
+
+@router.get("/{pipeline_id}")
+@require_login
+def get_pipeline(pipeline_id: int, request: Request, db: DbSession) -> dict[str, Any]:
+    """Return a single pipeline with its steps.
+
+    Path Parameters:
+        pipeline_id: The ID of the pipeline.
+
+    Returns:
+        The pipeline object including its ordered steps.
+
+    Raises:
+        HTTPException 404: If the pipeline does not exist or is not accessible.
+    """
+    user_id = _get_user_id(request)
+    admin = _is_admin(request)
+
+    pipeline = db.query(Pipeline).filter(Pipeline.id == pipeline_id).first()
+    if not pipeline or not _can_access_pipeline(pipeline, user_id, admin):
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Pipeline not found")
+
+    return _serialize_pipeline(pipeline, include_steps=True, db=db)
+
+
+@router.put("/{pipeline_id}")
+@require_login
+def update_pipeline(pipeline_id: int, request: Request, db: DbSession, body: PipelineUpdate) -> dict[str, Any]:
+    """Update a pipeline's metadata.
+
+    Path Parameters:
+        pipeline_id: The ID of the pipeline to update.
+
+    Returns:
+        The updated pipeline object.
+
+    Raises:
+        HTTPException 403: If the caller does not own this pipeline.
+        HTTPException 404: If the pipeline does not exist.
+        HTTPException 409: If the new name conflicts with an existing pipeline.
+    """
+    user_id = _get_user_id(request)
+    admin = _is_admin(request)
+
+    pipeline = db.query(Pipeline).filter(Pipeline.id == pipeline_id).first()
+    if not pipeline or not _can_access_pipeline(pipeline, user_id, admin):
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Pipeline not found")
+
+    if not _can_write_pipeline(pipeline, user_id, admin):
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Cannot modify this pipeline")
+
+    if body.name is not None:
+        new_name = body.name.strip()
+        if not new_name:
+            raise HTTPException(
+                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+                detail="name must not be empty",
+            )
+        if new_name != pipeline.name:
+            conflict = (
+                db.query(Pipeline)
+                .filter(Pipeline.owner_id == pipeline.owner_id, Pipeline.name == new_name, Pipeline.id != pipeline_id)
+                .first()
+            )
+            if conflict:
+                raise HTTPException(
+                    status_code=status.HTTP_409_CONFLICT,
+                    detail=f"A pipeline named '{new_name}' already exists",
+                )
+        pipeline.name = new_name
+
+    if body.description is not None:
+        pipeline.description = body.description
+
+    if body.is_active is not None:
+        pipeline.is_active = body.is_active
+
+    if body.is_default is not None:
+        if body.is_default and not pipeline.is_default:
+            _unset_default(db, pipeline.owner_id)
+        pipeline.is_default = body.is_default
+
+    try:
+        db.commit()
+        db.refresh(pipeline)
+    except Exception as exc:
+        db.rollback()
+        logger.exception(f"Failed to update pipeline id={pipeline_id}: {exc}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to update pipeline",
+        )
+
+    logger.info(f"Pipeline updated: id={pipeline_id}, user={user_id}")
+    return _serialize_pipeline(pipeline, include_steps=True, db=db)
+
+
+@router.delete("/{pipeline_id}", status_code=status.HTTP_204_NO_CONTENT)
+@require_login
+def delete_pipeline(pipeline_id: int, request: Request, db: DbSession) -> None:
+    """Delete a pipeline and all its steps.
+
+    Path Parameters:
+        pipeline_id: The ID of the pipeline to delete.
+
+    Raises:
+        HTTPException 403: If the caller does not own this pipeline.
+        HTTPException 404: If the pipeline does not exist.
+    """
+    user_id = _get_user_id(request)
+    admin = _is_admin(request)
+
+    pipeline = db.query(Pipeline).filter(Pipeline.id == pipeline_id).first()
+    if not pipeline or not _can_access_pipeline(pipeline, user_id, admin):
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Pipeline not found")
+
+    if not _can_write_pipeline(pipeline, user_id, admin):
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Cannot delete this pipeline")
+
+    try:
+        db.query(PipelineStep).filter(PipelineStep.pipeline_id == pipeline_id).delete()
+        db.delete(pipeline)
+        db.commit()
+    except Exception as exc:
+        db.rollback()
+        logger.exception(f"Failed to delete pipeline id={pipeline_id}: {exc}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to delete pipeline",
+        )
+
+    logger.info(f"Pipeline deleted: id={pipeline_id}, user={user_id}")
+
+
+# ---------------------------------------------------------------------------
+# Admin-only: create system (owner_id=NULL) pipeline
+# ---------------------------------------------------------------------------
+
+
+@router.post("/admin/system", status_code=status.HTTP_201_CREATED, tags=["admin-pipelines"])
+@require_login
+def create_system_pipeline(request: Request, db: DbSession, body: PipelineCreate) -> dict[str, Any]:
+    """Create a system-level (owner_id=NULL) default pipeline.  Admin only.
+
+    System pipelines are visible to all users and can be set as the global
+    default.  Only admins may create them.
+
+    Returns:
+        The created system pipeline.
+
+    Raises:
+        HTTPException 403: If the caller is not an admin.
+        HTTPException 409: If a system pipeline with the same name already exists.
+    """
+    if not _is_admin(request):
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required")
+
+    name = body.name.strip() if body.name else ""
+    if not name:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail="name is required",
+        )
+
+    existing = db.query(Pipeline).filter(Pipeline.owner_id.is_(None), Pipeline.name == name).first()
+    if existing:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=f"A system pipeline named '{name}' already exists",
+        )
+
+    if body.is_default:
+        _unset_default(db, None)
+
+    pipeline = Pipeline(
+        owner_id=None,
+        name=name,
+        description=body.description,
+        is_default=body.is_default,
+        is_active=body.is_active,
+    )
+    try:
+        db.add(pipeline)
+        db.commit()
+        db.refresh(pipeline)
+    except Exception as exc:
+        db.rollback()
+        logger.exception(f"Failed to create system pipeline: {exc}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to create system pipeline",
+        )
+
+    logger.info(f"System pipeline created: id={pipeline.id}, name={name!r}")
+    return _serialize_pipeline(pipeline)
+
+
+# ---------------------------------------------------------------------------
+# Step management
+# ---------------------------------------------------------------------------
+
+
+@router.post("/{pipeline_id}/steps", status_code=status.HTTP_201_CREATED)
+@require_login
+def add_step(pipeline_id: int, request: Request, db: DbSession, body: PipelineStepCreate) -> dict[str, Any]:
+    """Add a step to a pipeline.
+
+    Steps are automatically appended at the end unless an explicit ``position``
+    is supplied.  All existing steps at or after the insertion position are
+    shifted forward by one.
+
+    Path Parameters:
+        pipeline_id: The pipeline to add the step to.
+
+    Returns:
+        The created step object.
+
+    Raises:
+        HTTPException 403: If the caller cannot modify this pipeline.
+        HTTPException 404: If the pipeline does not exist.
+        HTTPException 422: If the step_type is not recognised.
+        HTTPException 409: If the maximum number of steps per pipeline is reached.
+    """
+    user_id = _get_user_id(request)
+    admin = _is_admin(request)
+
+    pipeline = db.query(Pipeline).filter(Pipeline.id == pipeline_id).first()
+    if not pipeline or not _can_access_pipeline(pipeline, user_id, admin):
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Pipeline not found")
+
+    if not _can_write_pipeline(pipeline, user_id, admin):
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Cannot modify this pipeline")
+
+    if body.step_type not in PIPELINE_STEP_TYPES:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=f"Unknown step type '{body.step_type}'. Valid types: {sorted(PIPELINE_STEP_TYPES)}",
+        )
+
+    current_count = db.query(PipelineStep).filter(PipelineStep.pipeline_id == pipeline_id).count()
+    if current_count >= MAX_STEPS_PER_PIPELINE:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=f"Maximum of {MAX_STEPS_PER_PIPELINE} steps per pipeline reached",
+        )
+
+    # Determine insertion position
+    if body.position is None:
+        max_pos = (
+            db.query(PipelineStep.position)
+            .filter(PipelineStep.pipeline_id == pipeline_id)
+            .order_by(PipelineStep.position.desc())
+            .first()
+        )
+        insert_pos = (max_pos[0] + 1) if max_pos else 0
+    else:
+        insert_pos = body.position
+        # Shift existing steps
+        steps_to_shift = (
+            db.query(PipelineStep)
+            .filter(PipelineStep.pipeline_id == pipeline_id, PipelineStep.position >= insert_pos)
+            .all()
+        )
+        for s in steps_to_shift:
+            s.position += 1
+
+    step = PipelineStep(
+        pipeline_id=pipeline_id,
+        position=insert_pos,
+        step_type=body.step_type,
+        label=body.label,
+        config=json.dumps(body.config) if body.config else None,
+        enabled=body.enabled,
+    )
+    try:
+        db.add(step)
+        db.commit()
+        db.refresh(step)
+    except Exception as exc:
+        db.rollback()
+        logger.exception(f"Failed to add step to pipeline id={pipeline_id}: {exc}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to add step",
+        )
+
+    logger.info(f"Step added: pipeline={pipeline_id}, step_type={body.step_type!r}, pos={insert_pos}")
+    return _serialize_step(step)
+
+
+@router.put("/{pipeline_id}/steps/reorder", tags=["pipelines"])
+@require_login
+def reorder_steps(
+    pipeline_id: int,
+    request: Request,
+    db: DbSession,
+    step_ids: list[int] = Body(..., description="Ordered list of step IDs representing the new order"),
+) -> list[dict[str, Any]]:
+    """Replace the step order for a pipeline.
+
+    Provide a complete ordered list of *all* step IDs.  Their ``position``
+    values will be reassigned 0, 1, 2, … in the given order.
+
+    Path Parameters:
+        pipeline_id: The pipeline whose steps are being reordered.
+
+    Returns:
+        The updated, ordered list of step objects.
+
+    Raises:
+        HTTPException 422: If the provided list does not contain exactly the
+            current set of step IDs for this pipeline.
+    """
+    user_id = _get_user_id(request)
+    admin = _is_admin(request)
+
+    pipeline = db.query(Pipeline).filter(Pipeline.id == pipeline_id).first()
+    if not pipeline or not _can_access_pipeline(pipeline, user_id, admin):
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Pipeline not found")
+
+    if not _can_write_pipeline(pipeline, user_id, admin):
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Cannot modify this pipeline")
+
+    existing_steps = db.query(PipelineStep).filter(PipelineStep.pipeline_id == pipeline_id).all()
+    existing_ids = {s.id for s in existing_steps}
+
+    if set(step_ids) != existing_ids or len(step_ids) != len(existing_ids):
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail="step_ids must contain exactly the current step IDs for this pipeline",
+        )
+
+    step_map = {s.id: s for s in existing_steps}
+    for pos, sid in enumerate(step_ids):
+        step_map[sid].position = pos
+
+    try:
+        db.commit()
+    except Exception as exc:
+        db.rollback()
+        logger.exception(f"Failed to reorder steps for pipeline id={pipeline_id}: {exc}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to reorder steps",
+        )
+
+    updated = (
+        db.query(PipelineStep).filter(PipelineStep.pipeline_id == pipeline_id).order_by(PipelineStep.position).all()
+    )
+    return [_serialize_step(s) for s in updated]
+
+
+@router.put("/{pipeline_id}/steps/{step_id}")
+@require_login
+def update_step(
+    pipeline_id: int, step_id: int, request: Request, db: DbSession, body: PipelineStepUpdate
+) -> dict[str, Any]:
+    """Update an existing pipeline step.
+
+    Path Parameters:
+        pipeline_id: The owning pipeline.
+        step_id: The step to update.
+
+    Returns:
+        The updated step object.
+    """
+    user_id = _get_user_id(request)
+    admin = _is_admin(request)
+
+    pipeline = db.query(Pipeline).filter(Pipeline.id == pipeline_id).first()
+    if not pipeline or not _can_access_pipeline(pipeline, user_id, admin):
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Pipeline not found")
+
+    if not _can_write_pipeline(pipeline, user_id, admin):
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Cannot modify this pipeline")
+
+    step = db.query(PipelineStep).filter(PipelineStep.id == step_id, PipelineStep.pipeline_id == pipeline_id).first()
+    if not step:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Step not found")
+
+    if body.step_type is not None:
+        if body.step_type not in PIPELINE_STEP_TYPES:
+            raise HTTPException(
+                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+                detail=f"Unknown step type '{body.step_type}'",
+            )
+        step.step_type = body.step_type
+
+    if body.label is not None:
+        step.label = body.label
+
+    if body.config is not None:
+        step.config = json.dumps(body.config)
+
+    if body.enabled is not None:
+        step.enabled = body.enabled
+
+    if body.position is not None and body.position != step.position:
+        old_pos = step.position
+        new_pos = body.position
+        if new_pos > old_pos:
+            # Moving down: shift intervening steps up
+            db.query(PipelineStep).filter(
+                PipelineStep.pipeline_id == pipeline_id,
+                PipelineStep.position > old_pos,
+                PipelineStep.position <= new_pos,
+                PipelineStep.id != step_id,
+            ).update({"position": PipelineStep.position - 1})
+        else:
+            # Moving up: shift intervening steps down
+            db.query(PipelineStep).filter(
+                PipelineStep.pipeline_id == pipeline_id,
+                PipelineStep.position >= new_pos,
+                PipelineStep.position < old_pos,
+                PipelineStep.id != step_id,
+            ).update({"position": PipelineStep.position + 1})
+        step.position = new_pos
+
+    try:
+        db.commit()
+        db.refresh(step)
+    except Exception as exc:
+        db.rollback()
+        logger.exception(f"Failed to update step id={step_id}: {exc}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to update step",
+        )
+
+    return _serialize_step(step)
+
+
+@router.delete("/{pipeline_id}/steps/{step_id}", status_code=status.HTTP_204_NO_CONTENT)
+@require_login
+def delete_step(pipeline_id: int, step_id: int, request: Request, db: DbSession) -> None:
+    """Delete a step from a pipeline.
+
+    Path Parameters:
+        pipeline_id: The owning pipeline.
+        step_id: The step to delete.
+    """
+    user_id = _get_user_id(request)
+    admin = _is_admin(request)
+
+    pipeline = db.query(Pipeline).filter(Pipeline.id == pipeline_id).first()
+    if not pipeline or not _can_access_pipeline(pipeline, user_id, admin):
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Pipeline not found")
+
+    if not _can_write_pipeline(pipeline, user_id, admin):
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Cannot modify this pipeline")
+
+    step = db.query(PipelineStep).filter(PipelineStep.id == step_id, PipelineStep.pipeline_id == pipeline_id).first()
+    if not step:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Step not found")
+
+    deleted_pos = step.position
+    try:
+        db.delete(step)
+        # Compact remaining step positions
+        db.query(PipelineStep).filter(
+            PipelineStep.pipeline_id == pipeline_id,
+            PipelineStep.position > deleted_pos,
+        ).update({"position": PipelineStep.position - 1})
+        db.commit()
+    except Exception as exc:
+        db.rollback()
+        logger.exception(f"Failed to delete step id={step_id}: {exc}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to delete step",
+        )
+
+    logger.info(f"Step deleted: id={step_id}, pipeline={pipeline_id}")
+
+
+# ---------------------------------------------------------------------------
+# Helper: unset default flag for an owner
+# ---------------------------------------------------------------------------
+
+
+def _unset_default(db: Session, owner_id: str | None) -> None:
+    """Clear the is_default flag on all pipelines for the given owner."""
+    if owner_id is None:
+        db.query(Pipeline).filter(Pipeline.owner_id.is_(None), Pipeline.is_default.is_(True)).update(
+            {"is_default": False}
+        )
+    else:
+        db.query(Pipeline).filter(Pipeline.owner_id == owner_id, Pipeline.is_default.is_(True)).update(
+            {"is_default": False}
+        )

app/models.py

@@ -6,6 +6,7 @@ from app.database import Base
 
 # Foreign key constants
 _FILES_ID_FK = "files.id"
+_PIPELINES_ID_FK = "pipelines.id"
 
 
 class DocumentMetadata(Base):
@@ -79,6 +80,9 @@ class FileRecord(Base):
     # Pre-computed text embedding vector stored as JSON array of floats
     embedding = Column(Text, nullable=True)
 
+    # Processing pipeline assigned to this file (NULL = use system default)
+    pipeline_id = Column(Integer, ForeignKey(_PIPELINES_ID_FK), nullable=True, index=True)
+
     # Timestamp when we inserted this record
     created_at = Column(DateTime(timezone=True), server_default=func.now(), index=True)
 
@@ -259,3 +263,69 @@ class SubscriptionPlan(Base):
 
     created_at = Column(DateTime(timezone=True), server_default=func.now())
     updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
+
+
+class Pipeline(Base):
+    """User-defined processing pipeline: an ordered set of steps.
+
+    Pipelines are user-specific.  A pipeline with ``owner_id = NULL`` is a
+    *system default* pipeline that only admins may create.  Regular users
+    create pipelines under their own ``owner_id``.  When a file has no
+    explicit pipeline assigned, the active system default is used.
+    """
+
+    __tablename__ = "pipelines"
+
+    id = Column(Integer, primary_key=True, index=True)
+
+    # Owner of this pipeline.  NULL = system/admin pipeline visible to everyone.
+    owner_id = Column(String, nullable=True, index=True)
+
+    # Human-readable name (unique per owner)
+    name = Column(String(255), nullable=False)
+
+    # Optional description
+    description = Column(Text, nullable=True)
+
+    # When True this pipeline is the default for new files belonging to the owner
+    # (or the global default when owner_id is NULL).  Only one pipeline per
+    # owner may be active default at a time — enforced at the application level.
+    is_default = Column(Boolean, nullable=False, default=False)
+
+    # Soft-disable without deleting
+    is_active = Column(Boolean, nullable=False, default=True)
+
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+    updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
+
+
+class PipelineStep(Base):
+    """A single step in a processing pipeline.
+
+    Steps are executed in ascending ``position`` order.  Each step has a
+    ``step_type`` that maps to a built-in processing action and an optional
+    ``config`` JSON blob with step-specific parameters.
+    """
+
+    __tablename__ = "pipeline_steps"
+
+    id = Column(Integer, primary_key=True, index=True)
+    pipeline_id = Column(Integer, ForeignKey(_PIPELINES_ID_FK), nullable=False, index=True)
+
+    # Execution order within the pipeline (lower = earlier)
+    position = Column(Integer, nullable=False, default=0)
+
+    # One of the recognised step types (see PIPELINE_STEP_TYPES in pipelines.py)
+    step_type = Column(String(100), nullable=False)
+
+    # Optional human-readable label override (defaults to step_type label)
+    label = Column(String(255), nullable=True)
+
+    # JSON-encoded step-specific configuration dict
+    config = Column(Text, nullable=True)
+
+    # When False this step is skipped during execution
+    enabled = Column(Boolean, nullable=False, default=True)
+
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+    updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())

app/views/__init__.py

@@ -14,6 +14,7 @@ from app.views.general import router as general_router
 from app.views.google_drive import router as google_drive_router
 from app.views.license_routes import router as license_router  # Add the license router
 from app.views.onedrive import router as onedrive_router
+from app.views.pipelines import router as pipelines_router  # Processing pipelines
 from app.views.plans import router as plans_router  # Admin Plan Designer
 from app.views.queue import router as queue_router
 from app.views.search import router as search_router
@@ -39,3 +40,4 @@ router.include_router(search_router)
 router.include_router(queue_router)
 router.include_router(subscriptions_router)  # Pricing + subscription pages
 router.include_router(plans_router)  # Admin Plan Designer
+router.include_router(pipelines_router)  # Processing pipelines

app/views/pipelines.py

@@ -0,0 +1,34 @@
+"""Pipelines view: management UI for processing pipelines."""
+
+import logging
+
+from fastapi import HTTPException, Request, status
+
+from app.views.base import APIRouter, require_login, settings, templates
+
+logger = logging.getLogger(__name__)
+router = APIRouter()
+
+
+@router.get("/pipelines")
+@require_login
+async def pipelines_page(request: Request):
+    """Processing pipeline management page for the current user.
+
+    Regular users manage their own pipelines.  Admins additionally have access
+    to system-level pipelines through the same UI.
+    """
+    try:
+        return templates.TemplateResponse(
+            "pipelines.html",
+            {
+                "request": request,
+                "app_version": settings.version,
+            },
+        )
+    except Exception as exc:
+        logger.error(f"Error loading pipelines page: {exc}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to load pipelines page",
+        )

frontend/templates/base.html

@@ -63,6 +63,7 @@
             <a href="/upload" class="text-gray-700 hover:text-gray-900" {% if request and request.url.path == '/upload' %}aria-current="page"{% endif %}>Upload</a>
             <a href="/files" class="text-gray-700 hover:text-gray-900" {% if request and request.url.path == '/files' %}aria-current="page"{% endif %}>Files</a>
             <a href="/search" class="text-gray-700 hover:text-gray-900" {% if request and request.url.path == '/search' %}aria-current="page"{% endif %}>Search</a>
+            <a href="/pipelines" class="text-gray-700 hover:text-gray-900" {% if request and request.url.path == '/pipelines' %}aria-current="page"{% endif %}>Pipelines</a>
             <a href="/pricing" class="text-gray-700 hover:text-gray-900" {% if request and request.url.path == '/pricing' %}aria-current="page"{% endif %}>Pricing</a>
 
             <!-- Admin dropdown (shown only for admin users via JS) -->
@@ -175,6 +176,7 @@
               <a href="/upload" class="block px-3 py-3 rounded-md text-base font-medium text-gray-700 hover:text-gray-900 hover:bg-gray-50">Upload</a>
               <a href="/files" class="block px-3 py-3 rounded-md text-base font-medium text-gray-700 hover:text-gray-900 hover:bg-gray-50">Files</a>
               <a href="/search" class="block px-3 py-3 rounded-md text-base font-medium text-gray-700 hover:text-gray-900 hover:bg-gray-50">Search</a>
+              <a href="/pipelines" class="block px-3 py-3 rounded-md text-base font-medium text-gray-700 hover:text-gray-900 hover:bg-gray-50">Pipelines</a>
               <a href="/pricing" class="block px-3 py-3 rounded-md text-base font-medium text-gray-700 hover:text-gray-900 hover:bg-gray-50">Pricing</a>
 
               <!-- Admin section in mobile menu (shown only for admin users via JS) -->

frontend/templates/pipelines.html

@@ -0,0 +1,857 @@
+{% extends "base.html" %}
+{% block title %}Processing Pipelines – DocuElevate{% endblock %}
+
+{% block content %}
+<div class="container mx-auto px-4 py-8" x-data="pipelinesApp()" x-init="init()">
+
+  <!-- ── Header ─────────────────────────────────────────────────────────────── -->
+  <div class="mb-6 flex flex-col sm:flex-row sm:items-center sm:justify-between gap-3">
+    <div>
+      <h1 class="text-3xl font-bold text-gray-900 dark:text-white flex items-center gap-2">
+        <i class="fas fa-project-diagram text-blue-500" aria-hidden="true"></i>
+        Processing Pipelines
+      </h1>
+      <p class="text-gray-500 dark:text-gray-400 text-sm mt-1">
+        Define and manage custom document processing workflows.
+        System pipelines (created by admins) are shown with a
+        <span class="text-xs font-semibold text-purple-700 bg-purple-50 border border-purple-200 rounded px-1">System</span>
+        badge and are visible to all users.
+      </p>
+    </div>
+    <button
+      type="button"
+      @click="openCreatePipelineModal()"
+      class="inline-flex items-center px-4 py-2 bg-blue-600 hover:bg-blue-700 text-white text-sm font-medium rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-offset-2 focus:ring-blue-500"
+      style="min-height:44px;"
+    >
+      <i class="fas fa-plus mr-2" aria-hidden="true"></i> New Pipeline
+    </button>
+  </div>
+
+  <!-- ── Alert banner ───────────────────────────────────────────────────────── -->
+  <div x-show="alert.show" x-transition class="mb-4" role="alert" :aria-live="alert.type === 'error' ? 'assertive' : 'polite'">
+    <div
+      :class="alert.type === 'success'
+        ? 'bg-green-50 border-green-400 text-green-800'
+        : 'bg-red-50 border-red-400 text-red-800'"
+      class="border-l-4 p-4 rounded dark:bg-opacity-10"
+    >
+      <p class="font-semibold" x-text="alert.title"></p>
+      <p class="text-sm" x-text="alert.message"></p>
+    </div>
+  </div>
+
+  <!-- ── Loading ────────────────────────────────────────────────────────────── -->
+  <template x-if="loading">
+    <div class="text-center py-12 text-gray-400">
+      <i class="fas fa-spinner fa-spin text-3xl mb-3" aria-hidden="true"></i>
+      <p>Loading pipelines…</p>
+    </div>
+  </template>
+
+  <!-- ── Empty state ────────────────────────────────────────────────────────── -->
+  <template x-if="!loading && pipelines.length === 0">
+    <div class="text-center py-16 bg-white dark:bg-gray-800 rounded-lg shadow">
+      <i class="fas fa-project-diagram text-5xl text-gray-300 mb-4" aria-hidden="true"></i>
+      <h2 class="text-xl font-semibold text-gray-700 dark:text-gray-300 mb-2">No pipelines yet</h2>
+      <p class="text-gray-500 dark:text-gray-400 mb-6">Create your first pipeline to define custom document processing workflows.</p>
+      <button
+        type="button"
+        @click="openCreatePipelineModal()"
+        class="inline-flex items-center px-5 py-2.5 bg-blue-600 hover:bg-blue-700 text-white text-sm font-medium rounded-md"
+        style="min-height:44px;"
+      >
+        <i class="fas fa-plus mr-2" aria-hidden="true"></i> Create Pipeline
+      </button>
+    </div>
+  </template>
+
+  <!-- ── Pipeline cards ─────────────────────────────────────────────────────── -->
+  <template x-if="!loading && pipelines.length > 0">
+    <div class="space-y-4">
+      <template x-for="pipeline in pipelines" :key="pipeline.id">
+        <div class="bg-white dark:bg-gray-800 shadow rounded-lg overflow-hidden">
+
+          <!-- Card header -->
+          <div class="px-5 py-4 flex flex-col sm:flex-row sm:items-center sm:justify-between gap-3 border-b border-gray-100 dark:border-gray-700">
+            <div class="flex items-center gap-2 flex-wrap">
+              <h2 class="text-lg font-semibold text-gray-900 dark:text-white" x-text="pipeline.name"></h2>
+              <!-- Badges -->
+              <template x-if="pipeline.owner_id === null">
+                <span class="text-xs font-semibold text-purple-700 bg-purple-50 border border-purple-200 rounded px-1.5 py-0.5">System</span>
+              </template>
+              <template x-if="pipeline.is_default">
+                <span class="text-xs font-semibold text-green-700 bg-green-50 border border-green-200 rounded px-1.5 py-0.5">Default</span>
+              </template>
+              <template x-if="!pipeline.is_active">
+                <span class="text-xs font-semibold text-gray-500 bg-gray-100 border border-gray-200 rounded px-1.5 py-0.5">Inactive</span>
+              </template>
+            </div>
+
+            <!-- Actions -->
+            <div class="flex items-center gap-2 flex-shrink-0">
+              <button
+                type="button"
+                @click="toggleSteps(pipeline)"
+                class="inline-flex items-center px-3 py-1.5 text-xs font-medium text-gray-600 dark:text-gray-300 bg-gray-100 dark:bg-gray-700 hover:bg-gray-200 dark:hover:bg-gray-600 rounded-md"
+                style="min-height:36px;"
+                :aria-expanded="pipeline._expanded"
+                :aria-label="`${pipeline._expanded ? 'Collapse' : 'Expand'} steps for ${pipeline.name}`"
+              >
+                <i class="fas fa-list-ul mr-1" aria-hidden="true"></i>
+                <span x-text="(pipeline.steps || []).length + ' step' + ((pipeline.steps || []).length !== 1 ? 's' : '')"></span>
+                <i :class="pipeline._expanded ? 'fa-chevron-up' : 'fa-chevron-down'" class="fas ml-1 text-xs" aria-hidden="true"></i>
+              </button>
+              <button
+                type="button"
+                @click="openEditPipelineModal(pipeline)"
+                class="inline-flex items-center px-3 py-1.5 text-xs font-medium text-blue-600 bg-blue-50 hover:bg-blue-100 rounded-md"
+                style="min-height:36px;"
+                :aria-label="`Edit pipeline ${pipeline.name}`"
+              >
+                <i class="fas fa-pencil-alt mr-1" aria-hidden="true"></i> Edit
+              </button>
+              <button
+                type="button"
+                @click="confirmDeletePipeline(pipeline)"
+                class="inline-flex items-center px-3 py-1.5 text-xs font-medium text-red-600 bg-red-50 hover:bg-red-100 rounded-md"
+                style="min-height:36px;"
+                :aria-label="`Delete pipeline ${pipeline.name}`"
+              >
+                <i class="fas fa-trash mr-1" aria-hidden="true"></i> Delete
+              </button>
+            </div>
+          </div>
+
+          <!-- Description -->
+          <template x-if="pipeline.description">
+            <p class="px-5 pt-3 text-sm text-gray-500 dark:text-gray-400" x-text="pipeline.description"></p>
+          </template>
+
+          <!-- Steps (collapsible) -->
+          <div x-show="pipeline._expanded" x-transition class="px-5 py-4">
+
+            <!-- Step list -->
+            <template x-if="(pipeline.steps || []).length === 0">
+              <p class="text-sm text-gray-400 italic mb-3">No steps defined. Add a step to start building your pipeline.</p>
+            </template>
+
+            <template x-if="(pipeline.steps || []).length > 0">
+              <ol class="space-y-2 mb-4" role="list" :aria-label="`Steps for ${pipeline.name}`">
+                <template x-for="(step, idx) in (pipeline.steps || [])" :key="step.id">
+                  <li class="flex items-center justify-between bg-gray-50 dark:bg-gray-700 rounded-md px-4 py-2.5 gap-3">
+                    <div class="flex items-center gap-3 min-w-0">
+                      <span class="flex-shrink-0 w-6 h-6 rounded-full bg-blue-100 text-blue-600 text-xs font-bold flex items-center justify-center" x-text="idx + 1" aria-hidden="true"></span>
+                      <div class="min-w-0">
+                        <p class="text-sm font-medium text-gray-800 dark:text-gray-200 truncate" x-text="step.label || stepTypeLabel(step.step_type)"></p>
+                        <p class="text-xs text-gray-400 truncate" x-text="step.step_type"></p>
+                      </div>
+                      <template x-if="!step.enabled">
+                        <span class="text-xs text-gray-400 bg-gray-200 rounded px-1.5 py-0.5 flex-shrink-0">Disabled</span>
+                      </template>
+                    </div>
+                    <div class="flex items-center gap-1.5 flex-shrink-0">
+                      <button
+                        type="button"
+                        @click="openEditStepModal(pipeline, step)"
+                        class="p-1.5 text-blue-500 hover:bg-blue-50 dark:hover:bg-blue-900 rounded"
+                        style="min-height:36px;min-width:36px;"
+                        :aria-label="`Edit step ${step.label || step.step_type}`"
+                      ><i class="fas fa-pencil-alt text-xs" aria-hidden="true"></i></button>
+                      <button
+                        type="button"
+                        @click="moveStep(pipeline, idx, -1)"
+                        :disabled="idx === 0"
+                        class="p-1.5 text-gray-400 hover:bg-gray-100 dark:hover:bg-gray-600 rounded disabled:opacity-30"
+                        style="min-height:36px;min-width:36px;"
+                        :aria-label="`Move step up: ${step.label || step.step_type}`"
+                      ><i class="fas fa-arrow-up text-xs" aria-hidden="true"></i></button>
+                      <button
+                        type="button"
+                        @click="moveStep(pipeline, idx, 1)"
+                        :disabled="idx === (pipeline.steps || []).length - 1"
+                        class="p-1.5 text-gray-400 hover:bg-gray-100 dark:hover:bg-gray-600 rounded disabled:opacity-30"
+                        style="min-height:36px;min-width:36px;"
+                        :aria-label="`Move step down: ${step.label || step.step_type}`"
+                      ><i class="fas fa-arrow-down text-xs" aria-hidden="true"></i></button>
+                      <button
+                        type="button"
+                        @click="confirmDeleteStep(pipeline, step)"
+                        class="p-1.5 text-red-400 hover:bg-red-50 dark:hover:bg-red-900 rounded"
+                        style="min-height:36px;min-width:36px;"
+                        :aria-label="`Delete step ${step.label || step.step_type}`"
+                      ><i class="fas fa-times text-xs" aria-hidden="true"></i></button>
+                    </div>
+                  </li>
+                </template>
+              </ol>
+            </template>
+
+            <!-- Add step button -->
+            <button
+              type="button"
+              @click="openAddStepModal(pipeline)"
+              class="inline-flex items-center px-3 py-1.5 text-xs font-medium text-green-600 bg-green-50 hover:bg-green-100 rounded-md border border-green-200"
+              style="min-height:36px;"
+              :aria-label="`Add step to ${pipeline.name}`"
+            >
+              <i class="fas fa-plus mr-1" aria-hidden="true"></i> Add Step
+            </button>
+          </div>
+
+        </div>
+      </template>
+    </div>
+  </template>
+
+
+  <!-- ═══════════════════════════════════════════════════════════════════════ -->
+  <!-- Modal: Create / Edit Pipeline                                         -->
+  <!-- ═══════════════════════════════════════════════════════════════════════ -->
+  <div
+    x-show="pipelineModal.open"
+    x-transition:enter="transition ease-out duration-200"
+    x-transition:enter-start="opacity-0"
+    x-transition:enter-end="opacity-100"
+    x-transition:leave="transition ease-in duration-150"
+    x-transition:leave-start="opacity-100"
+    x-transition:leave-end="opacity-0"
+    class="fixed inset-0 z-50 flex items-center justify-center bg-black bg-opacity-50"
+    role="dialog"
+    aria-modal="true"
+    :aria-labelledby="pipelineModal.editMode ? 'editPipelineTitle' : 'createPipelineTitle'"
+    @keydown.escape.window="closePipelineModal()"
+  >
+    <div class="bg-white dark:bg-gray-800 rounded-lg shadow-xl w-full max-w-md mx-4 p-6" @click.stop>
+      <h2
+        :id="pipelineModal.editMode ? 'editPipelineTitle' : 'createPipelineTitle'"
+        class="text-xl font-semibold text-gray-900 dark:text-white mb-5"
+        x-text="pipelineModal.editMode ? 'Edit Pipeline' : 'New Pipeline'"
+      ></h2>
+
+      <form @submit.prevent="savePipeline()" novalidate>
+        <div class="space-y-4">
+
+          <!-- Name -->
+          <div>
+            <label for="pipelineName" class="block text-sm font-medium text-gray-700 dark:text-gray-300 mb-1">
+              Name <span aria-hidden="true" class="text-red-500">*</span>
+            </label>
+            <input
+              id="pipelineName"
+              type="text"
+              x-model="pipelineModal.form.name"
+              maxlength="255"
+              required
+              class="w-full border border-gray-300 dark:border-gray-600 rounded-md px-3 py-2 text-sm focus:outline-none focus:ring-2 focus:ring-blue-400 dark:bg-gray-700 dark:text-white"
+              placeholder="My pipeline"
+              autocomplete="off"
+            />
+          </div>
+
+          <!-- Description -->
+          <div>
+            <label for="pipelineDesc" class="block text-sm font-medium text-gray-700 dark:text-gray-300 mb-1">Description</label>
+            <textarea
+              id="pipelineDesc"
+              x-model="pipelineModal.form.description"
+              rows="2"
+              maxlength="4096"
+              class="w-full border border-gray-300 dark:border-gray-600 rounded-md px-3 py-2 text-sm focus:outline-none focus:ring-2 focus:ring-blue-400 dark:bg-gray-700 dark:text-white resize-none"
+              placeholder="Optional description"
+            ></textarea>
+          </div>
+
+          <!-- Flags -->
+          <div class="flex flex-col gap-2">
+            <label class="inline-flex items-center gap-2 cursor-pointer">
+              <input
+                type="checkbox"
+                x-model="pipelineModal.form.is_default"
+                class="rounded border-gray-300 text-blue-600 focus:ring-blue-400"
+              />
+              <span class="text-sm text-gray-700 dark:text-gray-300">Set as my default pipeline</span>
+            </label>
+            <label class="inline-flex items-center gap-2 cursor-pointer">
+              <input
+                type="checkbox"
+                x-model="pipelineModal.form.is_active"
+                class="rounded border-gray-300 text-blue-600 focus:ring-blue-400"
+              />
+              <span class="text-sm text-gray-700 dark:text-gray-300">Active</span>
+            </label>
+          </div>
+
+          <!-- System pipeline option (admin only, create only) -->
+          <template x-if="isAdmin && !pipelineModal.editMode">
+            <label class="inline-flex items-center gap-2 cursor-pointer">
+              <input
+                type="checkbox"
+                x-model="pipelineModal.form.system"
+                class="rounded border-gray-300 text-purple-600 focus:ring-purple-400"
+              />
+              <span class="text-sm text-gray-700 dark:text-gray-300">System pipeline (visible to all users)</span>
+            </label>
+          </template>
+
+          <!-- Error message -->
+          <p x-show="pipelineModal.error" class="text-sm text-red-600" x-text="pipelineModal.error" role="alert"></p>
+        </div>
+
+        <div class="flex justify-end gap-3 mt-6">
+          <button
+            type="button"
+            @click="closePipelineModal()"
+            class="px-4 py-2 text-sm font-medium text-gray-700 dark:text-gray-300 bg-white dark:bg-gray-700 border border-gray-300 dark:border-gray-600 rounded-md hover:bg-gray-50"
+            style="min-height:44px;"
+          >Cancel</button>
+          <button
+            type="submit"
+            :disabled="pipelineModal.saving"
+            class="px-4 py-2 text-sm font-medium text-white bg-blue-600 hover:bg-blue-700 disabled:opacity-60 rounded-md"
+            style="min-height:44px;"
+          >
+            <template x-if="pipelineModal.saving">
+              <i class="fas fa-spinner fa-spin mr-1" aria-hidden="true"></i>
+            </template>
+            <span x-text="pipelineModal.editMode ? 'Save Changes' : 'Create'"></span>
+          </button>
+        </div>
+      </form>
+    </div>
+  </div>
+
+
+  <!-- ═══════════════════════════════════════════════════════════════════════ -->
+  <!-- Modal: Add / Edit Step                                                -->
+  <!-- ═══════════════════════════════════════════════════════════════════════ -->
+  <div
+    x-show="stepModal.open"
+    x-transition:enter="transition ease-out duration-200"
+    x-transition:enter-start="opacity-0"
+    x-transition:enter-end="opacity-100"
+    x-transition:leave="transition ease-in duration-150"
+    x-transition:leave-start="opacity-100"
+    x-transition:leave-end="opacity-0"
+    class="fixed inset-0 z-50 flex items-center justify-center bg-black bg-opacity-50"
+    role="dialog"
+    aria-modal="true"
+    :aria-labelledby="stepModal.editMode ? 'editStepTitle' : 'addStepTitle'"
+    @keydown.escape.window="closeStepModal()"
+  >
+    <div class="bg-white dark:bg-gray-800 rounded-lg shadow-xl w-full max-w-lg mx-4 p-6" @click.stop>
+      <h2
+        :id="stepModal.editMode ? 'editStepTitle' : 'addStepTitle'"
+        class="text-xl font-semibold text-gray-900 dark:text-white mb-5"
+        x-text="stepModal.editMode ? 'Edit Step' : 'Add Step'"
+      ></h2>
+
+      <form @submit.prevent="saveStep()" novalidate>
+        <div class="space-y-4">
+
+          <!-- Step type -->
+          <div>
+            <label for="stepType" class="block text-sm font-medium text-gray-700 dark:text-gray-300 mb-1">
+              Step Type <span aria-hidden="true" class="text-red-500">*</span>
+            </label>
+            <select
+              id="stepType"
+              x-model="stepModal.form.step_type"
+              :disabled="stepModal.editMode"
+              class="w-full border border-gray-300 dark:border-gray-600 rounded-md px-3 py-2 text-sm focus:outline-none focus:ring-2 focus:ring-blue-400 dark:bg-gray-700 dark:text-white"
+            >
+              <option value="">— select —</option>
+              <template x-for="[key, meta] in Object.entries(stepTypes)" :key="key">
+                <option :value="key" x-text="meta.label"></option>
+              </template>
+            </select>
+            <template x-if="stepModal.form.step_type && stepTypes[stepModal.form.step_type]">
+              <p class="mt-1 text-xs text-gray-400" x-text="stepTypes[stepModal.form.step_type].description"></p>
+            </template>
+          </div>
+
+          <!-- Custom label -->
+          <div>
+            <label for="stepLabel" class="block text-sm font-medium text-gray-700 dark:text-gray-300 mb-1">
+              Custom Label <span class="text-gray-400 font-normal">(optional)</span>
+            </label>
+            <input
+              id="stepLabel"
+              type="text"
+              x-model="stepModal.form.label"
+              maxlength="255"
+              class="w-full border border-gray-300 dark:border-gray-600 rounded-md px-3 py-2 text-sm focus:outline-none focus:ring-2 focus:ring-blue-400 dark:bg-gray-700 dark:text-white"
+              placeholder="Override the default step name"
+              autocomplete="off"
+            />
+          </div>
+
+          <!-- force_cloud_ocr (only shown for ocr step) -->
+          <template x-if="stepModal.form.step_type === 'ocr'">
+            <label class="inline-flex items-center gap-2 cursor-pointer">
+              <input
+                type="checkbox"
+                x-model="stepModal.form.config.force_cloud_ocr"
+                class="rounded border-gray-300 text-blue-600 focus:ring-blue-400"
+              />
+              <span class="text-sm text-gray-700 dark:text-gray-300">Force cloud OCR (skip local text extraction)</span>
+            </label>
+          </template>
+
+          <!-- Enabled -->
+          <label class="inline-flex items-center gap-2 cursor-pointer">
+            <input
+              type="checkbox"
+              x-model="stepModal.form.enabled"
+              class="rounded border-gray-300 text-blue-600 focus:ring-blue-400"
+            />
+            <span class="text-sm text-gray-700 dark:text-gray-300">Enabled</span>
+          </label>
+
+          <!-- Error -->
+          <p x-show="stepModal.error" class="text-sm text-red-600" x-text="stepModal.error" role="alert"></p>
+        </div>
+
+        <div class="flex justify-end gap-3 mt-6">
+          <button
+            type="button"
+            @click="closeStepModal()"
+            class="px-4 py-2 text-sm font-medium text-gray-700 dark:text-gray-300 bg-white dark:bg-gray-700 border border-gray-300 dark:border-gray-600 rounded-md hover:bg-gray-50"
+            style="min-height:44px;"
+          >Cancel</button>
+          <button
+            type="submit"
+            :disabled="stepModal.saving"
+            class="px-4 py-2 text-sm font-medium text-white bg-blue-600 hover:bg-blue-700 disabled:opacity-60 rounded-md"
+            style="min-height:44px;"
+          >
+            <template x-if="stepModal.saving">
+              <i class="fas fa-spinner fa-spin mr-1" aria-hidden="true"></i>
+            </template>
+            <span x-text="stepModal.editMode ? 'Save Changes' : 'Add Step'"></span>
+          </button>
+        </div>
+      </form>
+    </div>
+  </div>
+
+
+  <!-- ═══════════════════════════════════════════════════════════════════════ -->
+  <!-- Confirm delete dialog                                                 -->
+  <!-- ═══════════════════════════════════════════════════════════════════════ -->
+  <div
+    x-show="confirmModal.open"
+    x-transition:enter="transition ease-out duration-150"
+    x-transition:enter-start="opacity-0"
+    x-transition:enter-end="opacity-100"
+    x-transition:leave="transition ease-in duration-100"
+    x-transition:leave-start="opacity-100"
+    x-transition:leave-end="opacity-0"
+    class="fixed inset-0 z-50 flex items-center justify-center bg-black bg-opacity-50"
+    role="alertdialog"
+    aria-modal="true"
+    aria-labelledby="confirmTitle"
+    @keydown.escape.window="confirmModal.open = false"
+  >
+    <div class="bg-white dark:bg-gray-800 rounded-lg shadow-xl w-full max-w-sm mx-4 p-6" @click.stop>
+      <h2 id="confirmTitle" class="text-lg font-semibold text-gray-900 dark:text-white mb-2" x-text="confirmModal.title"></h2>
+      <p class="text-sm text-gray-600 dark:text-gray-400 mb-6" x-text="confirmModal.message"></p>
+      <div class="flex justify-end gap-3">
+        <button
+          type="button"
+          @click="confirmModal.open = false"
+          class="px-4 py-2 text-sm font-medium text-gray-700 dark:text-gray-300 bg-white dark:bg-gray-700 border border-gray-300 dark:border-gray-600 rounded-md hover:bg-gray-50"
+          style="min-height:44px;"
+        >Cancel</button>
+        <button
+          type="button"
+          @click="confirmModal.action(); confirmModal.open = false"
+          class="px-4 py-2 text-sm font-medium text-white bg-red-600 hover:bg-red-700 rounded-md"
+          style="min-height:44px;"
+        >Delete</button>
+      </div>
+    </div>
+  </div>
+
+</div>
+{% endblock %}
+
+{% block head_extra %}
+<script>
+function pipelinesApp() {
+  return {
+    loading: true,
+    pipelines: [],
+    stepTypes: {},
+    isAdmin: false,
+
+    alert: { show: false, type: 'success', title: '', message: '' },
+
+    pipelineModal: {
+      open: false,
+      editMode: false,
+      pipelineId: null,
+      saving: false,
+      error: '',
+      form: { name: '', description: '', is_default: false, is_active: true, system: false },
+    },
+
+    stepModal: {
+      open: false,
+      editMode: false,
+      pipeline: null,
+      stepId: null,
+      saving: false,
+      error: '',
+      form: { step_type: '', label: '', config: {}, enabled: true },
+    },
+
+    confirmModal: {
+      open: false,
+      title: '',
+      message: '',
+      action: () => {},
+    },
+
+    // ── Initialisation ────────────────────────────────────────────────────
+
+    async init() {
+      await Promise.all([this.fetchStepTypes(), this.fetchCurrentUser()]);
+      await this.fetchPipelines();
+    },
+
+    async fetchCurrentUser() {
+      try {
+        const r = await fetch('/api/auth/whoami');
+        if (r.ok) {
+          const u = await r.json();
+          this.isAdmin = !!u.is_admin;
+        }
+      } catch (_) {}
+    },
+
+    async fetchStepTypes() {
+      try {
+        const r = await fetch('/api/pipelines/step-types');
+        if (r.ok) this.stepTypes = await r.json();
+      } catch (_) {}
+    },
+
+    async fetchPipelines() {
+      this.loading = true;
+      try {
+        const r = await fetch('/api/pipelines');
+        if (!r.ok) throw new Error(await r.text());
+        const list = await r.json();
+        // Fetch steps for each pipeline
+        const detailed = await Promise.all(list.map(p => this.fetchPipeline(p.id)));
+        this.pipelines = detailed.map(p => ({ ...p, _expanded: false }));
+      } catch (err) {
+        this.showAlert('error', 'Failed to load pipelines', err.message || String(err));
+      } finally {
+        this.loading = false;
+      }
+    },
+
+    async fetchPipeline(id) {
+      const r = await fetch(`/api/pipelines/${id}`);
+      if (!r.ok) throw new Error(`Pipeline ${id} not found`);
+      return r.json();
+    },
+
+    // ── Helpers ───────────────────────────────────────────────────────────
+
+    stepTypeLabel(key) {
+      return this.stepTypes[key] ? this.stepTypes[key].label : key;
+    },
+
+    showAlert(type, title, message) {
+      this.alert = { show: true, type, title, message };
+      setTimeout(() => { this.alert.show = false; }, 6000);
+    },
+
+    toggleSteps(pipeline) {
+      pipeline._expanded = !pipeline._expanded;
+    },
+
+    // ── Pipeline Modal ────────────────────────────────────────────────────
+
+    openCreatePipelineModal() {
+      this.pipelineModal = {
+        open: true,
+        editMode: false,
+        pipelineId: null,
+        saving: false,
+        error: '',
+        form: { name: '', description: '', is_default: false, is_active: true, system: false },
+      };
+    },
+
+    openEditPipelineModal(pipeline) {
+      this.pipelineModal = {
+        open: true,
+        editMode: true,
+        pipelineId: pipeline.id,
+        saving: false,
+        error: '',
+        form: {
+          name: pipeline.name,
+          description: pipeline.description || '',
+          is_default: pipeline.is_default,
+          is_active: pipeline.is_active,
+          system: false,
+        },
+      };
+    },
+
+    closePipelineModal() {
+      this.pipelineModal.open = false;
+    },
+
+    async savePipeline() {
+      const fm = this.pipelineModal.form;
+      if (!fm.name.trim()) {
+        this.pipelineModal.error = 'Pipeline name is required.';
+        return;
+      }
+      this.pipelineModal.error = '';
+      this.pipelineModal.saving = true;
+
+      try {
+        let r;
+        if (this.pipelineModal.editMode) {
+          r = await fetch(`/api/pipelines/${this.pipelineModal.pipelineId}`, {
+            method: 'PUT',
+            headers: this._jsonHeaders(),
+            body: JSON.stringify({
+              name: fm.name.trim(),
+              description: fm.description || null,
+              is_default: fm.is_default,
+              is_active: fm.is_active,
+            }),
+          });
+        } else {
+          const url = fm.system ? '/api/pipelines/admin/system' : '/api/pipelines';
+          r = await fetch(url, {
+            method: 'POST',
+            headers: this._jsonHeaders(),
+            body: JSON.stringify({
+              name: fm.name.trim(),
+              description: fm.description || null,
+              is_default: fm.is_default,
+              is_active: fm.is_active,
+            }),
+          });
+        }
+        if (!r.ok) {
+          const e = await r.json().catch(() => ({}));
+          throw new Error(e.detail || `HTTP ${r.status}`);
+        }
+        const saved = await r.json();
+        if (this.pipelineModal.editMode) {
+          const idx = this.pipelines.findIndex(p => p.id === saved.id);
+          if (idx !== -1) {
+            const expanded = this.pipelines[idx]._expanded;
+            this.pipelines[idx] = { ...saved, steps: this.pipelines[idx].steps || [], _expanded: expanded };
+          }
+        } else {
+          this.pipelines.push({ ...saved, steps: [], _expanded: false });
+        }
+        this.closePipelineModal();
+        this.showAlert('success', 'Pipeline saved', `"${saved.name}" has been ${this.pipelineModal.editMode ? 'updated' : 'created'}.`);
+      } catch (err) {
+        this.pipelineModal.error = err.message || 'Unknown error';
+      } finally {
+        this.pipelineModal.saving = false;
+      }
+    },
+
+    confirmDeletePipeline(pipeline) {
+      this.confirmModal = {
+        open: true,
+        title: 'Delete pipeline?',
+        message: `Are you sure you want to permanently delete "${pipeline.name}" and all its steps? This cannot be undone.`,
+        action: () => this.deletePipeline(pipeline),
+      };
+    },
+
+    async deletePipeline(pipeline) {
+      try {
+        const r = await fetch(`/api/pipelines/${pipeline.id}`, { method: 'DELETE', headers: this._csrfHeaders() });
+        if (!r.ok) {
+          const e = await r.json().catch(() => ({}));
+          throw new Error(e.detail || `HTTP ${r.status}`);
+        }
+        this.pipelines = this.pipelines.filter(p => p.id !== pipeline.id);
+        this.showAlert('success', 'Pipeline deleted', `"${pipeline.name}" has been deleted.`);
+      } catch (err) {
+        this.showAlert('error', 'Delete failed', err.message || String(err));
+      }
+    },
+
+    // ── Step Modal ────────────────────────────────────────────────────────
+
+    openAddStepModal(pipeline) {
+      this.stepModal = {
+        open: true,
+        editMode: false,
+        pipeline,
+        stepId: null,
+        saving: false,
+        error: '',
+        form: { step_type: '', label: '', config: {}, enabled: true },
+      };
+    },
+
+    openEditStepModal(pipeline, step) {
+      this.stepModal = {
+        open: true,
+        editMode: true,
+        pipeline,
+        stepId: step.id,
+        saving: false,
+        error: '',
+        form: {
+          step_type: step.step_type,
+          label: step.label || '',
+          config: { ...(step.config || {}) },
+          enabled: step.enabled,
+        },
+      };
+    },
+
+    closeStepModal() {
+      this.stepModal.open = false;
+    },
+
+    async saveStep() {
+      const fm = this.stepModal.form;
+      if (!fm.step_type) {
+        this.stepModal.error = 'Step type is required.';
+        return;
+      }
+      this.stepModal.error = '';
+      this.stepModal.saving = true;
+      const pid = this.stepModal.pipeline.id;
+
+      try {
+        let r;
+        if (this.stepModal.editMode) {
+          r = await fetch(`/api/pipelines/${pid}/steps/${this.stepModal.stepId}`, {
+            method: 'PUT',
+            headers: this._jsonHeaders(),
+            body: JSON.stringify({
+              step_type: fm.step_type,
+              label: fm.label || null,
+              config: fm.config || {},
+              enabled: fm.enabled,
+            }),
+          });
+        } else {
+          r = await fetch(`/api/pipelines/${pid}/steps`, {
+            method: 'POST',
+            headers: this._jsonHeaders(),
+            body: JSON.stringify({
+              step_type: fm.step_type,
+              label: fm.label || null,
+              config: fm.config || {},
+              enabled: fm.enabled,
+            }),
+          });
+        }
+        if (!r.ok) {
+          const e = await r.json().catch(() => ({}));
+          throw new Error(e.detail || `HTTP ${r.status}`);
+        }
+        const savedStep = await r.json();
+        // Refresh the pipeline's steps
+        const updated = await this.fetchPipeline(pid);
+        const pIdx = this.pipelines.findIndex(p => p.id === pid);
+        if (pIdx !== -1) {
+          const expanded = this.pipelines[pIdx]._expanded;
+          this.pipelines[pIdx] = { ...updated, _expanded: expanded };
+        }
+        this.closeStepModal();
+        this.showAlert('success', 'Step saved', `Step "${savedStep.label || savedStep.step_type}" has been ${this.stepModal.editMode ? 'updated' : 'added'}.`);
+      } catch (err) {
+        this.stepModal.error = err.message || 'Unknown error';
+      } finally {
+        this.stepModal.saving = false;
+      }
+    },
+
+    async moveStep(pipeline, idx, direction) {
+      const steps = pipeline.steps || [];
+      const newIdx = idx + direction;
+      if (newIdx < 0 || newIdx >= steps.length) return;
+      // Build the new order
+      const reordered = [...steps];
+      [reordered[idx], reordered[newIdx]] = [reordered[newIdx], reordered[idx]];
+      const ids = reordered.map(s => s.id);
+      try {
+        const r = await fetch(`/api/pipelines/${pipeline.id}/steps/reorder`, {
+          method: 'PUT',
+          headers: this._jsonHeaders(),
+          body: JSON.stringify(ids),
+        });
+        if (!r.ok) {
+          const e = await r.json().catch(() => ({}));
+          throw new Error(e.detail || `HTTP ${r.status}`);
+        }
+        const updated = await this.fetchPipeline(pipeline.id);
+        const pIdx = this.pipelines.findIndex(p => p.id === pipeline.id);
+        if (pIdx !== -1) {
+          this.pipelines[pIdx] = { ...updated, _expanded: true };
+        }
+      } catch (err) {
+        this.showAlert('error', 'Reorder failed', err.message || String(err));
+      }
+    },
+
+    confirmDeleteStep(pipeline, step) {
+      this.confirmModal = {
+        open: true,
+        title: 'Delete step?',
+        message: `Remove "${step.label || step.step_type}" from pipeline "${pipeline.name}"?`,
+        action: () => this.deleteStep(pipeline, step),
+      };
+    },
+
+    async deleteStep(pipeline, step) {
+      try {
+        const r = await fetch(`/api/pipelines/${pipeline.id}/steps/${step.id}`, {
+          method: 'DELETE',
+          headers: this._csrfHeaders(),
+        });
+        if (!r.ok) {
+          const e = await r.json().catch(() => ({}));
+          throw new Error(e.detail || `HTTP ${r.status}`);
+        }
+        const updated = await this.fetchPipeline(pipeline.id);
+        const pIdx = this.pipelines.findIndex(p => p.id === pipeline.id);
+        if (pIdx !== -1) {
+          this.pipelines[pIdx] = { ...updated, _expanded: true };
+        }
+        this.showAlert('success', 'Step deleted', `"${step.label || step.step_type}" removed.`);
+      } catch (err) {
+        this.showAlert('error', 'Delete failed', err.message || String(err));
+      }
+    },
+
+    // ── Fetch helpers ─────────────────────────────────────────────────────
+
+    _csrfToken() {
+      return document.querySelector('meta[name="csrf-token"]')?.content || '';
+    },
+
+    _csrfHeaders() {
+      return { 'X-CSRF-Token': this._csrfToken() };
+    },
+
+    _jsonHeaders() {
+      return { 'Content-Type': 'application/json', 'X-CSRF-Token': this._csrfToken() };
+    },
+  };
+}
+</script>
+{% endblock %}

migrations/versions/017_add_pipelines.py

@@ -0,0 +1,68 @@
+"""Add pipelines and pipeline_steps tables; add pipeline_id to files
+
+Revision ID: 017_add_pipelines
+Revises: 016_add_userprofile_billing
+Create Date: 2026-03-07
+
+"""
+
+from typing import Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "017_add_pipelines"
+down_revision: Union[str, None] = "016_add_userprofile_billing"
+depends_on: Union[str, None] = None
+
+
+def upgrade() -> None:
+    """Create pipelines, pipeline_steps tables and add pipeline_id FK to files."""
+    op.create_table(
+        "pipelines",
+        sa.Column("id", sa.Integer(), primary_key=True, index=True),
+        sa.Column("owner_id", sa.String(), nullable=True, index=True),
+        sa.Column("name", sa.String(255), nullable=False),
+        sa.Column("description", sa.Text(), nullable=True),
+        sa.Column("is_default", sa.Boolean(), nullable=False, server_default="0"),
+        sa.Column("is_active", sa.Boolean(), nullable=False, server_default="1"),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
+        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
+    )
+
+    op.create_table(
+        "pipeline_steps",
+        sa.Column("id", sa.Integer(), primary_key=True, index=True),
+        sa.Column("pipeline_id", sa.Integer(), sa.ForeignKey("pipelines.id"), nullable=False, index=True),
+        sa.Column("position", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("step_type", sa.String(100), nullable=False),
+        sa.Column("label", sa.String(255), nullable=True),
+        sa.Column("config", sa.Text(), nullable=True),
+        sa.Column("enabled", sa.Boolean(), nullable=False, server_default="1"),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
+        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
+    )
+
+    # Use batch mode for SQLite compatibility when altering the files table.
+    # SQLite does not support adding FK constraints inline via ALTER TABLE, so we
+    # add the plain integer column and define the FK reference at the model level.
+    with op.batch_alter_table("files") as batch_op:
+        batch_op.add_column(
+            sa.Column("pipeline_id", sa.Integer(), nullable=True),
+        )
+        batch_op.create_foreign_key(
+            "fk_files_pipeline_id",
+            "pipelines",
+            ["pipeline_id"],
+            ["id"],
+        )
+
+
+def downgrade() -> None:
+    """Drop pipeline_id from files and remove pipeline tables."""
+    with op.batch_alter_table("files") as batch_op:
+        batch_op.drop_column("pipeline_id")
+
+    op.drop_table("pipeline_steps")
+    op.drop_table("pipelines")

tests/conftest.py

@@ -62,6 +62,8 @@ from app.main import app as fastapi_app  # noqa: E402
 from app.models import (  # noqa: F401, E402
     DocumentMetadata,
     FileRecord,
+    Pipeline,
+    PipelineStep,
     ProcessingLog,
     SavedSearch,
     UserProfile,

tests/test_api_pipelines.py

@@ -0,0 +1,467 @@
+"""Tests for the pipelines API endpoints.
+
+Covers CRUD operations for pipelines and steps, ownership/admin access control,
+step reordering, and the assign-pipeline-to-file endpoint.
+"""
+
+import json
+from unittest.mock import patch
+
+import pytest
+
+from app.models import FileRecord, Pipeline, PipelineStep
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_file(db_session, owner_id=None):
+    """Insert a minimal FileRecord and return it.
+
+    Default owner_id=None so tests work without an authenticated session.
+    """
+    fr = FileRecord(
+        owner_id=owner_id,
+        filehash="abc123",
+        original_filename="test.pdf",
+        local_filename="/tmp/test.pdf",
+        file_size=1024,
+        mime_type="application/pdf",
+    )
+    db_session.add(fr)
+    db_session.commit()
+    db_session.refresh(fr)
+    return fr
+
+
+# ---------------------------------------------------------------------------
+# Unit tests – step-types catalogue
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.unit
+class TestStepTypesCatalogue:
+    """Tests for the step-types read-only catalogue endpoint."""
+
+    def test_step_types_returns_dict(self, client):
+        """GET /api/pipelines/step-types returns a dict of known types."""
+        r = client.get("/api/pipelines/step-types")
+        assert r.status_code == 200
+        data = r.json()
+        assert isinstance(data, dict)
+        # Core built-in types must be present
+        for key in ("convert_to_pdf", "ocr", "extract_metadata", "embed_metadata", "compute_embedding"):
+            assert key in data, f"Expected step type '{key}' in catalogue"
+
+    def test_each_type_has_label_and_description(self, client):
+        """Each step-type entry has at least a label and description."""
+        r = client.get("/api/pipelines/step-types")
+        for key, meta in r.json().items():
+            assert "label" in meta, f"Step type '{key}' missing 'label'"
+            assert "description" in meta, f"Step type '{key}' missing 'description'"
+
+
+# ---------------------------------------------------------------------------
+# Integration tests – Pipeline CRUD
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.integration
+class TestPipelineCRUD:
+    """Full CRUD test-suite for pipeline management."""
+
+    def test_list_pipelines_empty(self, client):
+        """List returns an empty array when no pipelines exist."""
+        r = client.get("/api/pipelines")
+        assert r.status_code == 200
+        assert r.json() == []
+
+    def test_create_pipeline(self, client):
+        """POST /api/pipelines creates a new pipeline."""
+        r = client.post(
+            "/api/pipelines",
+            json={"name": "My Pipeline", "description": "Test description"},
+        )
+        assert r.status_code == 201
+        data = r.json()
+        assert data["name"] == "My Pipeline"
+        assert data["description"] == "Test description"
+        assert data["is_default"] is False
+        assert data["is_active"] is True
+        assert data["id"] is not None
+
+    def test_create_pipeline_duplicate_name_same_owner_rejected(self, client):
+        """Creating two pipelines with the same name is rejected with 409."""
+        client.post("/api/pipelines", json={"name": "Dupe"})
+        r = client.post("/api/pipelines", json={"name": "Dupe"})
+        assert r.status_code == 409
+
+    def test_create_pipeline_empty_name_rejected(self, client):
+        """An empty pipeline name returns 422."""
+        r = client.post("/api/pipelines", json={"name": "   "})
+        assert r.status_code == 422
+
+    def test_get_pipeline_includes_steps(self, client):
+        """GET /api/pipelines/{id} returns the pipeline with a steps array."""
+        created = client.post("/api/pipelines", json={"name": "With Steps"}).json()
+        r = client.get(f"/api/pipelines/{created['id']}")
+        assert r.status_code == 200
+        assert "steps" in r.json()
+        assert r.json()["steps"] == []
+
+    def test_get_pipeline_not_found(self, client):
+        """GET on a non-existent pipeline returns 404."""
+        r = client.get("/api/pipelines/99999")
+        assert r.status_code == 404
+
+    def test_update_pipeline(self, client):
+        """PUT /api/pipelines/{id} updates name and description."""
+        created = client.post("/api/pipelines", json={"name": "Original"}).json()
+        r = client.put(
+            f"/api/pipelines/{created['id']}",
+            json={"name": "Renamed", "description": "New desc"},
+        )
+        assert r.status_code == 200
+        assert r.json()["name"] == "Renamed"
+        assert r.json()["description"] == "New desc"
+
+    def test_update_pipeline_empty_name_rejected(self, client):
+        """Updating a pipeline with an empty name returns 422."""
+        created = client.post("/api/pipelines", json={"name": "Good"}).json()
+        r = client.put(f"/api/pipelines/{created['id']}", json={"name": ""})
+        assert r.status_code == 422
+
+    def test_update_pipeline_name_conflict_rejected(self, client):
+        """Updating a pipeline's name to one already taken returns 409."""
+        client.post("/api/pipelines", json={"name": "Taken"})
+        second = client.post("/api/pipelines", json={"name": "Other"}).json()
+        r = client.put(f"/api/pipelines/{second['id']}", json={"name": "Taken"})
+        assert r.status_code == 409
+
+    def test_delete_pipeline(self, client):
+        """DELETE /api/pipelines/{id} removes the pipeline."""
+        created = client.post("/api/pipelines", json={"name": "Deletable"}).json()
+        r = client.delete(f"/api/pipelines/{created['id']}")
+        assert r.status_code == 204
+        assert client.get(f"/api/pipelines/{created['id']}").status_code == 404
+
+    def test_delete_pipeline_not_found(self, client):
+        """Deleting a non-existent pipeline returns 404."""
+        r = client.delete("/api/pipelines/99999")
+        assert r.status_code == 404
+
+    def test_is_default_flag(self, client):
+        """Setting is_default=True marks the pipeline as default."""
+        r = client.post("/api/pipelines", json={"name": "Default Pipeline", "is_default": True})
+        assert r.status_code == 201
+        assert r.json()["is_default"] is True
+
+    def test_only_one_default_per_owner(self, client):
+        """When a new default is set, the old one is cleared."""
+        first = client.post("/api/pipelines", json={"name": "First Default", "is_default": True}).json()
+        second = client.post("/api/pipelines", json={"name": "Second Default", "is_default": True}).json()
+
+        assert second["is_default"] is True
+        # The first should no longer be default
+        first_updated = client.get(f"/api/pipelines/{first['id']}").json()
+        assert first_updated["is_default"] is False
+
+    def test_list_returns_created_pipeline(self, client):
+        """After creating a pipeline it appears in the list."""
+        client.post("/api/pipelines", json={"name": "Visible"})
+        r = client.get("/api/pipelines")
+        names = [p["name"] for p in r.json()]
+        assert "Visible" in names
+
+
+# ---------------------------------------------------------------------------
+# Integration tests – Step management
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.integration
+class TestPipelineStepManagement:
+    """Tests for adding, updating, deleting, and reordering pipeline steps."""
+
+    def _create_pipeline(self, client, name="Test Pipeline"):
+        return client.post("/api/pipelines", json={"name": name}).json()
+
+    def test_add_step(self, client):
+        """POST /api/pipelines/{id}/steps adds a step."""
+        p = self._create_pipeline(client)
+        r = client.post(f"/api/pipelines/{p['id']}/steps", json={"step_type": "ocr"})
+        assert r.status_code == 201
+        step = r.json()
+        assert step["step_type"] == "ocr"
+        assert step["position"] == 0
+        assert step["enabled"] is True
+
+    def test_add_step_with_config(self, client):
+        """A step can be added with a custom config dict."""
+        p = self._create_pipeline(client)
+        r = client.post(
+            f"/api/pipelines/{p['id']}/steps",
+            json={"step_type": "ocr", "config": {"force_cloud_ocr": True}},
+        )
+        assert r.status_code == 201
+        assert r.json()["config"]["force_cloud_ocr"] is True
+
+    def test_add_step_with_custom_label(self, client):
+        """A step can override the default label."""
+        p = self._create_pipeline(client)
+        r = client.post(
+            f"/api/pipelines/{p['id']}/steps",
+            json={"step_type": "convert_to_pdf", "label": "My Converter"},
+        )
+        assert r.status_code == 201
+        assert r.json()["label"] == "My Converter"
+
+    def test_add_invalid_step_type_rejected(self, client):
+        """An unrecognised step type returns 422."""
+        p = self._create_pipeline(client)
+        r = client.post(f"/api/pipelines/{p['id']}/steps", json={"step_type": "nonexistent_step"})
+        assert r.status_code == 422
+
+    def test_steps_appended_in_order(self, client):
+        """Multiple steps are appended in position order."""
+        p = self._create_pipeline(client)
+        client.post(f"/api/pipelines/{p['id']}/steps", json={"step_type": "convert_to_pdf"})
+        client.post(f"/api/pipelines/{p['id']}/steps", json={"step_type": "ocr"})
+        client.post(f"/api/pipelines/{p['id']}/steps", json={"step_type": "extract_metadata"})
+
+        details = client.get(f"/api/pipelines/{p['id']}").json()
+        types = [s["step_type"] for s in details["steps"]]
+        assert types == ["convert_to_pdf", "ocr", "extract_metadata"]
+
+    def test_update_step(self, client):
+        """PUT /api/pipelines/{id}/steps/{step_id} updates enabled flag."""
+        p = self._create_pipeline(client)
+        step = client.post(f"/api/pipelines/{p['id']}/steps", json={"step_type": "ocr"}).json()
+
+        r = client.put(
+            f"/api/pipelines/{p['id']}/steps/{step['id']}",
+            json={"enabled": False},
+        )
+        assert r.status_code == 200
+        assert r.json()["enabled"] is False
+
+    def test_update_step_not_found(self, client):
+        """Updating a step on the wrong pipeline returns 404."""
+        p = self._create_pipeline(client)
+        r = client.put(f"/api/pipelines/{p['id']}/steps/99999", json={"enabled": False})
+        assert r.status_code == 404
+
+    def test_delete_step(self, client):
+        """DELETE /api/pipelines/{id}/steps/{step_id} removes the step."""
+        p = self._create_pipeline(client)
+        step = client.post(f"/api/pipelines/{p['id']}/steps", json={"step_type": "ocr"}).json()
+
+        r = client.delete(f"/api/pipelines/{p['id']}/steps/{step['id']}")
+        assert r.status_code == 204
+
+        details = client.get(f"/api/pipelines/{p['id']}").json()
+        assert details["steps"] == []
+
+    def test_delete_step_compacts_positions(self, client):
+        """After deleting a step, remaining steps have contiguous positions."""
+        p = self._create_pipeline(client)
+        s1 = client.post(f"/api/pipelines/{p['id']}/steps", json={"step_type": "convert_to_pdf"}).json()
+        client.post(f"/api/pipelines/{p['id']}/steps", json={"step_type": "ocr"})
+        client.post(f"/api/pipelines/{p['id']}/steps", json={"step_type": "extract_metadata"})
+
+        client.delete(f"/api/pipelines/{p['id']}/steps/{s1['id']}")
+
+        details = client.get(f"/api/pipelines/{p['id']}").json()
+        positions = [s["position"] for s in details["steps"]]
+        assert positions == sorted(positions)
+        assert positions[0] == 0
+
+    def test_reorder_steps(self, client):
+        """PUT /api/pipelines/{id}/steps/reorder reorders all steps."""
+        p = self._create_pipeline(client)
+        s1 = client.post(f"/api/pipelines/{p['id']}/steps", json={"step_type": "convert_to_pdf"}).json()
+        s2 = client.post(f"/api/pipelines/{p['id']}/steps", json={"step_type": "ocr"}).json()
+        s3 = client.post(f"/api/pipelines/{p['id']}/steps", json={"step_type": "extract_metadata"}).json()
+
+        # Reverse order
+        r = client.put(
+            f"/api/pipelines/{p['id']}/steps/reorder",
+            json=[s3["id"], s2["id"], s1["id"]],
+        )
+        assert r.status_code == 200
+        types = [s["step_type"] for s in r.json()]
+        assert types == ["extract_metadata", "ocr", "convert_to_pdf"]
+
+    def test_reorder_steps_invalid_ids_rejected(self, client):
+        """Providing wrong step IDs returns 422."""
+        p = self._create_pipeline(client)
+        client.post(f"/api/pipelines/{p['id']}/steps", json={"step_type": "ocr"})
+
+        r = client.put(f"/api/pipelines/{p['id']}/steps/reorder", json=[99999])
+        assert r.status_code == 422
+
+
+# ---------------------------------------------------------------------------
+# Integration tests – System pipeline (admin endpoint)
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.integration
+class TestSystemPipeline:
+    """Tests for the admin system-pipeline creation endpoint."""
+
+    def test_create_system_pipeline_as_admin(self, client):
+        """An admin can create a system pipeline (owner_id=NULL)."""
+        with patch("app.api.pipelines._is_admin", return_value=True):
+            r = client.post(
+                "/api/pipelines/admin/system",
+                json={"name": "Global Default", "is_default": True},
+            )
+        assert r.status_code == 201
+        data = r.json()
+        assert data["owner_id"] is None
+        assert data["is_default"] is True
+
+    def test_create_system_pipeline_as_non_admin_forbidden(self, client):
+        """A non-admin user cannot create a system pipeline."""
+        r = client.post(
+            "/api/pipelines/admin/system",
+            json={"name": "Should Fail"},
+        )
+        assert r.status_code == 403
+
+
+# ---------------------------------------------------------------------------
+# Integration tests – File pipeline assignment
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.integration
+class TestAssignPipelineToFile:
+    """Tests for POST /api/files/{id}/assign-pipeline."""
+
+    def test_assign_pipeline_to_file(self, client, db_session):
+        """Assigning a pipeline to a file stores pipeline_id on the record."""
+        # File with no owner so anonymous test session can access it
+        fr = _make_file(db_session, owner_id=None)
+        pipeline = client.post("/api/pipelines", json={"name": "Assign Test"}).json()
+
+        r = client.post(f"/api/files/{fr.id}/assign-pipeline?pipeline_id={pipeline['id']}")
+        assert r.status_code == 200
+        data = r.json()
+        assert data["file_id"] == fr.id
+        assert data["pipeline_id"] == pipeline["id"]
+
+        db_session.refresh(fr)
+        assert fr.pipeline_id == pipeline["id"]
+
+    def test_clear_pipeline_from_file(self, client, db_session):
+        """Passing no pipeline_id clears the assignment."""
+        fr = _make_file(db_session, owner_id=None)
+        pipeline = client.post("/api/pipelines", json={"name": "Clearable"}).json()
+        client.post(f"/api/files/{fr.id}/assign-pipeline?pipeline_id={pipeline['id']}")
+
+        r = client.post(f"/api/files/{fr.id}/assign-pipeline")
+        assert r.status_code == 200
+        assert r.json()["pipeline_id"] is None
+
+    def test_assign_nonexistent_pipeline_returns_404(self, client, db_session):
+        """Assigning a non-existent pipeline returns 404."""
+        fr = _make_file(db_session, owner_id=None)
+        r = client.post(f"/api/files/{fr.id}/assign-pipeline?pipeline_id=99999")
+        assert r.status_code == 404
+
+    def test_assign_pipeline_to_nonexistent_file_returns_404(self, client):
+        """Assigning a pipeline to a non-existent file returns 404."""
+        r = client.post("/api/files/99999/assign-pipeline?pipeline_id=1")
+        assert r.status_code == 404
+
+
+# ---------------------------------------------------------------------------
+# Unit tests – API helper logic
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.unit
+class TestPipelineAPIHelpers:
+    """Unit tests for API helper functions (no DB required)."""
+
+    def test_serialize_step_returns_expected_keys(self, db_session):
+        """_serialize_step returns all required fields."""
+        from app.api.pipelines import _serialize_step
+
+        # Build a minimal in-memory step
+        p = Pipeline(owner_id="u1", name="P", is_default=False, is_active=True)
+        db_session.add(p)
+        db_session.commit()
+
+        s = PipelineStep(
+            pipeline_id=p.id,
+            position=0,
+            step_type="ocr",
+            label="OCR",
+            config=json.dumps({"force_cloud_ocr": False}),
+            enabled=True,
+        )
+        db_session.add(s)
+        db_session.commit()
+
+        result = _serialize_step(s)
+        for key in ("id", "pipeline_id", "position", "step_type", "label", "config", "enabled"):
+            assert key in result, f"Missing key '{key}' in serialized step"
+        assert result["config"] == {"force_cloud_ocr": False}
+
+    def test_serialize_pipeline_returns_expected_keys(self, db_session):
+        """_serialize_pipeline returns all required fields."""
+        from app.api.pipelines import _serialize_pipeline
+
+        p = Pipeline(owner_id="u1", name="MyPipeline", is_default=True, is_active=True)
+        db_session.add(p)
+        db_session.commit()
+
+        result = _serialize_pipeline(p)
+        for key in ("id", "owner_id", "name", "description", "is_default", "is_active"):
+            assert key in result, f"Missing key '{key}' in serialized pipeline"
+
+    def test_can_access_system_pipeline(self):
+        """Anyone can read a system pipeline (owner_id=None)."""
+        from app.api.pipelines import _can_access_pipeline
+
+        p = Pipeline(owner_id=None, name="System", is_default=False, is_active=True)
+        assert _can_access_pipeline(p, "any_user", admin=False) is True
+
+    def test_cannot_write_system_pipeline_as_regular_user(self):
+        """Regular users cannot modify system pipelines."""
+        from app.api.pipelines import _can_write_pipeline
+
+        p = Pipeline(owner_id=None, name="System", is_default=False, is_active=True)
+        assert _can_write_pipeline(p, "regular_user", admin=False) is False
+
+    def test_admin_can_write_system_pipeline(self):
+        """Admins can modify system pipelines."""
+        from app.api.pipelines import _can_write_pipeline
+
+        p = Pipeline(owner_id=None, name="System", is_default=False, is_active=True)
+        assert _can_write_pipeline(p, "admin", admin=True) is True
+
+    def test_user_can_access_own_pipeline(self):
+        """A user can access pipelines they own."""
+        from app.api.pipelines import _can_access_pipeline
+
+        p = Pipeline(owner_id="user1", name="Mine", is_default=False, is_active=True)
+        assert _can_access_pipeline(p, "user1", admin=False) is True
+
+    def test_user_cannot_access_other_users_pipeline(self):
+        """A regular user cannot access another user's pipeline."""
+        from app.api.pipelines import _can_access_pipeline
+
+        p = Pipeline(owner_id="user1", name="Theirs", is_default=False, is_active=True)
+        assert _can_access_pipeline(p, "user2", admin=False) is False
+
+    def test_admin_can_access_any_pipeline(self):
+        """Admins can access any pipeline regardless of owner."""
+        from app.api.pipelines import _can_access_pipeline
+
+        p = Pipeline(owner_id="user99", name="Private", is_default=False, is_active=True)
+        assert _can_access_pipeline(p, "admin", admin=True) is True