Merge branch 'main' of https://github.com/christianlouis/DocuElevate into copilot/fix-pricing-page-issues

app/api/__init__.py

@@ -19,6 +19,7 @@ from app.api.logs import router as logs_router
 from app.api.onboarding import router as onboarding_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
@@ -64,3 +65,4 @@ router.include_router(subscriptions_router)
 router.include_router(plans_router)
 router.include_router(onboarding_router)
 router.include_router(billing_router)
+router.include_router(pipelines_router)

app/api/files.py

@@ -1559,3 +1559,71 @@ 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, get_current_user_id
+    from app.models import Pipeline
+
+    user = get_current_user(request)
+    user_id: str = get_current_user_id(request)
+
+    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,886 @@
+"""
+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, get_current_user_id, 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.
+
+    Delegates to :func:`app.auth.get_current_user_id` so the same fallback
+    logic ("anonymous") is used consistently throughout the application.
+    """
+    return get_current_user_id(request)
+
+
+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")
+@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}
+        )
+
+
+# ---------------------------------------------------------------------------
+# Default system pipeline seeding
+# ---------------------------------------------------------------------------
+
+# The steps that make up the standard document-processing workflow.  The order
+# here mirrors what the existing Celery-based pipeline executes for every
+# uploaded file.
+_DEFAULT_PIPELINE_STEPS: list[tuple[str, str]] = [
+    ("convert_to_pdf", "Convert to PDF"),
+    ("check_duplicates", "Check for Duplicates"),
+    ("ocr", "OCR Processing"),
+    ("extract_metadata", "Extract Metadata"),
+    ("embed_metadata", "Embed Metadata into PDF"),
+    ("compute_embedding", "Compute Text Embedding"),
+    ("send_to_destinations", "Send to Storage Destinations"),
+]
+
+#: Human-readable name shown in the management UI for the auto-seeded pipeline.
+DEFAULT_PIPELINE_NAME = "Standard Processing Pipeline"
+
+
+def seed_default_pipeline(db: Session) -> int:
+    """Ensure a system-owned default pipeline exists in the database.
+
+    This function is idempotent — it is a no-op when any system pipeline
+    (``owner_id IS NULL``) already exists.  It is intended to be called once
+    at application startup (in ``app.main.lifespan``) so that the pipeline
+    management UI always shows the default workflow that mirrors the existing
+    Celery-based processing steps.
+
+    The created pipeline:
+
+    * ``owner_id = None`` — owned by the system, visible to all users
+    * ``is_default = True`` — selected automatically for new documents
+    * Steps (in order): convert_to_pdf → check_duplicates → ocr →
+      extract_metadata → embed_metadata → compute_embedding →
+      send_to_destinations
+
+    Args:
+        db: An active SQLAlchemy session.
+
+    Returns:
+        ``1`` if a new pipeline was created, ``0`` if one already existed.
+    """
+    try:
+        if db.query(Pipeline).filter(Pipeline.owner_id.is_(None)).count() > 0:
+            return 0
+    except Exception:
+        # Table may not exist yet during the very first migration run.
+        return 0
+
+    pipeline = Pipeline(
+        owner_id=None,
+        name=DEFAULT_PIPELINE_NAME,
+        description=(
+            "The standard document processing workflow: PDF conversion, "
+            "duplicate detection, OCR, metadata extraction and embedding, "
+            "semantic embeddings, and final distribution to storage destinations."
+        ),
+        is_default=True,
+        is_active=True,
+    )
+    db.add(pipeline)
+    try:
+        db.flush()  # Assign pipeline.id without committing yet
+    except Exception as exc:  # pragma: no cover
+        db.rollback()
+        logger.error(f"Failed to create default pipeline: {exc}")
+        return 0
+
+    for pos, (step_type, label) in enumerate(_DEFAULT_PIPELINE_STEPS):
+        db.add(
+            PipelineStep(
+                pipeline_id=pipeline.id,
+                position=pos,
+                step_type=step_type,
+                label=label,
+                enabled=True,
+            )
+        )
+
+    try:
+        db.commit()
+        logger.info("Seeded default system pipeline: '%s' (id=%d)", DEFAULT_PIPELINE_NAME, pipeline.id)
+    except Exception as exc:  # pragma: no cover
+        db.rollback()
+        logger.error(f"Failed to seed default pipeline steps: {exc}")
+        return 0
+
+    return 1

app/auth.py

@@ -53,6 +53,27 @@ def get_current_user(request: Request):
     return request.session.get("user")
 
 
+def get_current_user_id(request: Request) -> str:
+    """Return a stable string identifier for the authenticated user.
+
+    Falls back to ``"anonymous"`` when no user is in the session (e.g. when
+    AUTH_ENABLED=False in single-user mode).  The returned value is consistent
+    with how pipeline and file ownership is stored in the database.
+
+    Priority order: ``preferred_username`` → ``email`` → ``id`` → ``"anonymous"``.
+
+    Args:
+        request: The current FastAPI request.
+
+    Returns:
+        A non-empty string identifying the current user.
+    """
+    user = get_current_user(request)
+    if not user or not isinstance(user, dict):
+        return "anonymous"
+    return user.get("preferred_username") or user.get("email") or user.get("id") or "anonymous"
+
+
 def require_login(func):
     if not AUTH_ENABLED:
         return func  # no-op

app/main.py

@@ -113,6 +113,20 @@ async def lifespan(app: FastAPI):
     except Exception:
         logging.debug("Subscription plan seeding skipped — DB may not be ready yet")  # noqa: S110
 
+    # Seed the default system pipeline (mirrors the current hardcoded processing
+    # workflow) so it is immediately visible in the Pipelines management UI.
+    try:
+        from app.api.pipelines import seed_default_pipeline as _seed_pipeline
+        from app.database import SessionLocal as _SessionLocal  # noqa: F811 (re-import for clarity)
+
+        _db_pipeline = _SessionLocal()
+        try:
+            _seed_pipeline(_db_pipeline)
+        finally:
+            _db_pipeline.close()
+    except Exception:
+        logging.debug("Default pipeline seeding skipped — DB may not be ready yet")  # noqa: S110
+
     # Application is now running
     yield
 

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)
 
@@ -293,3 +297,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

@@ -15,6 +15,7 @@ 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.onboarding import router as onboarding_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
@@ -41,3 +42,4 @@ router.include_router(queue_router)
 router.include_router(subscriptions_router)  # Pricing + subscription pages
 router.include_router(plans_router)  # Admin Plan Designer
 router.include_router(onboarding_router)  # User onboarding wizard
+router.include_router(pipelines_router)  # Processing pipelines

app/views/files.py

@@ -270,6 +270,9 @@ def file_view_page(request: Request, file_id: int, db: Session = Depends(get_db)
         except Exception:
             step_summary = None
 
+        # Resolve the pipeline assigned to this file (explicit or system default)
+        pipeline_info = _resolve_pipeline(db, file_record)
+
         return templates.TemplateResponse(
             "file_view.html",
             {
@@ -279,6 +282,7 @@ def file_view_page(request: Request, file_id: int, db: Session = Depends(get_db)
                 "original_file_exists": original_file_exists,
                 "processed_file_exists": processed_file_exists,
                 "step_summary": step_summary,
+                "pipeline_info": pipeline_info,
             },
         )
     except Exception as e:
@@ -337,8 +341,11 @@ def file_detail_page(request: Request, file_id: int, db: Session = Depends(get_d
                 except Exception as e:
                     logger.warning(f"Failed to load metadata from {metadata_path}: {e}")
 
-        # Compute processing flow for visualization
-        flow_data = _compute_processing_flow(logs)
+        # Resolve the pipeline assigned to this file (explicit or system default)
+        pipeline_info = _resolve_pipeline(db, file_record)
+
+        # Compute processing flow for visualization — filter to pipeline steps when available
+        flow_data = _compute_processing_flow(logs, pipeline_steps=pipeline_info["steps"] if pipeline_info else None)
 
         # Compute step-aligned summary from status table (preferred) or fallback to logs
         try:
@@ -360,6 +367,7 @@ def file_detail_page(request: Request, file_id: int, db: Session = Depends(get_d
                 "gpt_metadata": gpt_metadata,
                 "flow_data": flow_data,
                 "step_summary": step_summary,
+                "pipeline_info": pipeline_info,
             },
         )
     except Exception as e:
@@ -367,15 +375,99 @@ def file_detail_page(request: Request, file_id: int, db: Session = Depends(get_d
         return templates.TemplateResponse("file_detail.html", {"request": request, "file": None, "error": str(e)})
 
 
-def _compute_processing_flow(logs):
+# ---------------------------------------------------------------------------
+# Pipeline ↔ Celery-log stage mapping
+# ---------------------------------------------------------------------------
+
+# Maps each pipeline step_type to the set of Celery task log stage keys that
+# implement it.  Used to filter the flow visualization when a pipeline is
+# assigned to a file.
+#
+# ⚠️  MAINTENANCE NOTE: When a new step type is added to PIPELINE_STEP_TYPES
+# in app/api/pipelines.py it MUST also be added here, otherwise the flow
+# visualization will silently skip its Celery-task stages for files using that
+# step type.  The test ``TestPipelineInfoInViews::test_step_type_mapping_is_complete``
+# enforces this invariant automatically.
+_STEP_TYPE_TO_STAGES: dict[str, list[str]] = {
+    "convert_to_pdf": ["convert_to_pdf"],
+    "check_duplicates": ["check_for_duplicates"],
+    "ocr": ["check_text", "extract_text", "process_with_ocr"],
+    "extract_metadata": ["extract_metadata_with_gpt"],
+    "embed_metadata": ["embed_metadata_into_pdf"],
+    "compute_embedding": ["compute_embedding"],
+    "send_to_destinations": ["finalize_document_storage", "send_to_all_destinations"],
+    # "classify" is defined in PIPELINE_STEP_TYPES but has no Celery log stages yet.
+    # When a classify task is implemented, add its stage key(s) here.
+    "classify": [],
+}
+
+# These internal bookkeeping stages are always shown in the flow regardless of
+# which pipeline steps are defined.
+_ALWAYS_SHOW_STAGES: frozenset[str] = frozenset({"create_file_record"})
+
+
+def _resolve_pipeline(db: Session, file_record) -> dict | None:
+    """Resolve the pipeline information for a file.
+
+    If the file has an explicit ``pipeline_id``, load that pipeline.
+    Otherwise fall back to the active system-default pipeline
+    (``owner_id IS NULL``, ``is_default=True``).
+
+    Returns a dict with keys:
+        id, name, description, is_default, is_system, is_explicit, steps
+    or ``None`` when no pipeline exists in the database.
+    """
+    from app.models import Pipeline, PipelineStep
+
+    pipeline = None
+    if file_record.pipeline_id:
+        pipeline = db.query(Pipeline).filter(Pipeline.id == file_record.pipeline_id).first()
+
+    if pipeline is None:
+        pipeline = (
+            db.query(Pipeline)
+            .filter(
+                Pipeline.owner_id.is_(None),
+                Pipeline.is_default.is_(True),
+                Pipeline.is_active.is_(True),
+            )
+            .first()
+        )
+
+    if pipeline is None:
+        return None
+
+    steps = db.query(PipelineStep).filter(PipelineStep.pipeline_id == pipeline.id).order_by(PipelineStep.position).all()
+
+    return {
+        "id": pipeline.id,
+        "name": pipeline.name,
+        "description": pipeline.description,
+        "is_default": pipeline.is_default,
+        "is_system": pipeline.owner_id is None,
+        # True when the file has a pipeline explicitly assigned (not inferred default)
+        "is_explicit": bool(file_record.pipeline_id),
+        "steps": steps,
+    }
+
+
+def _compute_processing_flow(logs, pipeline_steps=None):
     """
     Compute the processing flow structure from logs for visualization.
 
     Returns a structured representation of the processing pipeline with branches.
     Detects upload sub-tasks and organizes them as branches under the parent upload stage.
+
+    Args:
+        logs: list of ProcessingLog objects (ordered by timestamp asc)
+        pipeline_steps: optional list of PipelineStep objects for the assigned pipeline.
+            When provided, the set of stages shown is filtered to only those that
+            correspond to the pipeline's enabled steps (plus bookkeeping stages like
+            ``create_file_record`` and any stage that actually ran in the logs).
     """
-    # Define the main processing stages
+    # Define the full catalogue of main processing stages
     stages = {
+        "convert_to_pdf": {"label": "Convert to PDF", "next": ["check_for_duplicates", "create_file_record"]},
         "check_for_duplicates": {"label": "Check for Duplicates", "next": ["create_file_record"]},
         "create_file_record": {"label": "Create File Record", "next": ["check_text"]},
         "check_text": {
@@ -406,6 +498,22 @@ def _compute_processing_flow(logs):
         if "create_file_record" in stages:
             stages["create_file_record"]["next"] = ["check_text"]
 
+    # When a pipeline is assigned, filter stages to only those relevant to the
+    # pipeline's enabled steps plus always-show bookkeeping stages and any stage
+    # that actually produced log entries (so nothing already-run is hidden).
+    if pipeline_steps is not None:
+        # Collect Celery stage keys that the pipeline's enabled steps map to
+        allowed: set[str] = set(_ALWAYS_SHOW_STAGES)
+        for ps in pipeline_steps:
+            if ps.enabled:
+                allowed.update(_STEP_TYPE_TO_STAGES.get(ps.step_type, []))
+        # Pre-scan logs so we can also keep any stage that already ran
+        ran_stages: set[str] = set()
+        for log in logs:
+            ran_stages.add(log.step_name)
+        allowed.update(ran_stages)
+        stages = {k: v for k, v in stages.items() if k in allowed}
+
     # Define upload sub-tasks (branches)
     upload_tasks = {
         "upload_to_dropbox": "Dropbox",

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",
+        )