Merge pull request #580 from christianlouis/copilot/add-conditional-routing

feat: add conditional routing based on document type and metadata

app/api/__init__.py

@@ -34,6 +34,7 @@ from app.api.plans import router as plans_router
 from app.api.process import router as process_router
 from app.api.profile import router as profile_router
 from app.api.queue import router as queue_router
+from app.api.routing_rules import router as routing_rules_router
 from app.api.saved_searches import router as saved_searches_router
 from app.api.scheduled_jobs import router as scheduled_jobs_router
 from app.api.search import router as search_router
@@ -85,6 +86,7 @@ router.include_router(onboarding_router)
 router.include_router(billing_router)
 router.include_router(pipelines_router)
 router.include_router(profile_router)
+router.include_router(routing_rules_router)
 router.include_router(imap_accounts_router)
 router.include_router(imap_profiles_router)
 router.include_router(integrations_router)

app/api/routing_rules.py

@@ -0,0 +1,468 @@
+"""Routing rules API endpoints.
+
+Provides full CRUD for pipeline routing rules that conditionally assign
+documents to pipelines based on document properties (file type, category,
+metadata fields, size, etc.).
+
+Rules are evaluated in ascending ``position`` order.  The first rule whose
+condition matches wins and routes the document to the specified target
+pipeline.  If no rule matches, the caller falls back to the owner's (or
+system) default pipeline.
+"""
+
+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 require_login
+from app.database import get_db
+from app.models import Pipeline, PipelineRoutingRule
+from app.utils.routing_engine import (
+    BUILTIN_FIELDS,
+    VALID_OPERATORS,
+    _evaluate_condition,
+    _resolve_field,
+)
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/routing-rules", tags=["routing-rules"])
+
+DbSession = Annotated[Session, Depends(get_db)]
+
+MAX_RULES_PER_OWNER = 100
+MAX_NAME_LENGTH = 255
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _get_user_id(request: Request) -> str:
+    """Return the authenticated user identifier."""
+    user = getattr(request.state, "user", None)
+    if user:
+        if isinstance(user, dict):
+            return user.get("sub", user.get("email", "anonymous"))
+        return getattr(user, "sub", getattr(user, "email", "anonymous"))
+    return "anonymous"
+
+
+def _is_admin(request: Request) -> bool:
+    """Return ``True`` when the current user has admin privileges."""
+    user = getattr(request.state, "user", None)
+    if not user:
+        return False
+    groups = user.get("groups", []) if isinstance(user, dict) else getattr(user, "groups", [])
+    return "admin" in groups
+
+
+def _can_access_rule(rule: PipelineRoutingRule, user_id: str, admin: bool) -> bool:
+    """Check whether the user is allowed to read this rule."""
+    if admin:
+        return True
+    return rule.owner_id == user_id
+
+
+def _can_write_rule(rule: PipelineRoutingRule, user_id: str, admin: bool) -> bool:
+    """Check whether the user is allowed to modify this rule."""
+    if rule.owner_id is None:
+        return admin
+    return rule.owner_id == user_id
+
+
+def _validate_field(field: str) -> None:
+    """Raise 422 if the field name is invalid."""
+    if field in BUILTIN_FIELDS:
+        return
+    if field.startswith("metadata.") and len(field) > len("metadata."):
+        return
+    raise HTTPException(
+        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+        detail=(
+            f"Invalid field '{field}'. "
+            f"Valid built-in fields: {sorted(BUILTIN_FIELDS)}. "
+            "For AI metadata, use 'metadata.<key>'."
+        ),
+    )
+
+
+def _validate_operator(operator: str) -> None:
+    """Raise 422 if the operator is not recognised."""
+    if operator not in VALID_OPERATORS:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=f"Invalid operator '{operator}'. Valid operators: {sorted(VALID_OPERATORS)}",
+        )
+
+
+def _serialize_rule(rule: PipelineRoutingRule) -> dict[str, Any]:
+    """Serialize a routing rule to a JSON-compatible dict."""
+    return {
+        "id": rule.id,
+        "owner_id": rule.owner_id,
+        "name": rule.name,
+        "position": rule.position,
+        "field": rule.field,
+        "operator": rule.operator,
+        "value": rule.value,
+        "target_pipeline_id": rule.target_pipeline_id,
+        "is_active": rule.is_active,
+        "created_at": rule.created_at.isoformat() if rule.created_at else None,
+        "updated_at": rule.updated_at.isoformat() if rule.updated_at else None,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Pydantic request models
+# ---------------------------------------------------------------------------
+
+
+class RoutingRuleCreate(BaseModel):
+    """Request body for creating a routing rule."""
+
+    name: str = Field(..., min_length=1, max_length=MAX_NAME_LENGTH)
+    field: str = Field(..., min_length=1, max_length=255)
+    operator: str = Field(..., min_length=1, max_length=50)
+    value: str = Field(..., max_length=1024)
+    target_pipeline_id: int
+    position: int | None = None
+    is_active: bool = True
+
+
+class RoutingRuleUpdate(BaseModel):
+    """Request body for updating a routing rule."""
+
+    name: str | None = Field(None, min_length=1, max_length=MAX_NAME_LENGTH)
+    field: str | None = Field(None, min_length=1, max_length=255)
+    operator: str | None = Field(None, min_length=1, max_length=50)
+    value: str | None = Field(None, max_length=1024)
+    target_pipeline_id: int | None = None
+    position: int | None = None
+    is_active: bool | None = None
+
+
+class RoutingRuleEvaluateRequest(BaseModel):
+    """Request body for dry-run rule evaluation."""
+
+    file_type: str | None = None
+    filename: str | None = None
+    size: int | None = None
+    document_type: str | None = None
+    metadata: dict[str, Any] | None = None
+
+
+# ---------------------------------------------------------------------------
+# Endpoints
+# ---------------------------------------------------------------------------
+
+
+@router.get("")
+@require_login
+def list_routing_rules(request: Request, db: DbSession) -> list[dict[str, Any]]:
+    """List all routing rules accessible by the current user.
+
+    Returns the user's own rules plus any system-wide rules (``owner_id=NULL``).
+    Rules are sorted by position.
+    """
+    user_id = _get_user_id(request)
+
+    rules = (
+        db.query(PipelineRoutingRule)
+        .filter((PipelineRoutingRule.owner_id == user_id) | (PipelineRoutingRule.owner_id.is_(None)))
+        .order_by(
+            PipelineRoutingRule.owner_id.is_(None).asc(),
+            PipelineRoutingRule.position.asc(),
+        )
+        .all()
+    )
+
+    return [_serialize_rule(r) for r in rules]
+
+
+@router.post("", status_code=status.HTTP_201_CREATED)
+@require_login
+def create_routing_rule(request: Request, db: DbSession, body: RoutingRuleCreate) -> dict[str, Any]:
+    """Create a new routing rule for the current user.
+
+    Returns:
+        The created routing rule.
+
+    Raises:
+        HTTPException 422: If the field or operator is invalid.
+        HTTPException 404: If the target pipeline does not exist.
+        HTTPException 409: If the maximum number of rules is reached.
+    """
+    user_id = _get_user_id(request)
+
+    _validate_field(body.field)
+    _validate_operator(body.operator)
+
+    # Verify target pipeline exists and is accessible.
+    pipeline = db.query(Pipeline).filter(Pipeline.id == body.target_pipeline_id).first()
+    if not pipeline:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Target pipeline {body.target_pipeline_id} not found",
+        )
+
+    # Enforce per-owner limit.
+    count = db.query(PipelineRoutingRule).filter(PipelineRoutingRule.owner_id == user_id).count()
+    if count >= MAX_RULES_PER_OWNER:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=f"Maximum of {MAX_RULES_PER_OWNER} routing rules per user reached",
+        )
+
+    # Auto-assign position if not specified.
+    position = body.position
+    if position is None:
+        max_pos = (
+            db.query(PipelineRoutingRule.position)
+            .filter(PipelineRoutingRule.owner_id == user_id)
+            .order_by(PipelineRoutingRule.position.desc())
+            .first()
+        )
+        position = (max_pos[0] + 1) if max_pos else 0
+
+    rule = PipelineRoutingRule(
+        owner_id=user_id,
+        name=body.name.strip(),
+        position=position,
+        field=body.field,
+        operator=body.operator,
+        value=body.value,
+        target_pipeline_id=body.target_pipeline_id,
+        is_active=body.is_active,
+    )
+
+    try:
+        db.add(rule)
+        db.commit()
+        db.refresh(rule)
+    except Exception:
+        db.rollback()
+        logger.exception("Failed to create routing rule for user=%s", user_id)
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to create routing rule",
+        )
+
+    logger.info("Routing rule created: id=%s, user=%s", rule.id, user_id)
+    return _serialize_rule(rule)
+
+
+@router.get("/operators")
+def list_operators() -> dict[str, Any]:
+    """Return the list of supported operators and fields.
+
+    This is a public endpoint (no auth required) so that UIs can populate
+    dropdowns without hard-coding the catalogue.
+    """
+    return {
+        "operators": sorted(VALID_OPERATORS),
+        "builtin_fields": sorted(BUILTIN_FIELDS),
+        "metadata_prefix": "metadata.",
+    }
+
+
+@router.post("/evaluate")
+@require_login
+def evaluate_rules(request: Request, db: DbSession, body: RoutingRuleEvaluateRequest) -> dict[str, Any]:
+    """Dry-run rule evaluation against the provided document properties.
+
+    Returns the first matching rule and target pipeline (if any), or
+    indicates that no rule matched (default pipeline will be used).
+    """
+    user_id = _get_user_id(request)
+
+    doc_props: dict[str, Any] = {
+        "file_type": body.file_type,
+        "filename": body.filename,
+        "size": body.size,
+        "document_type": body.document_type,
+        "metadata": body.metadata or {},
+    }
+
+    rules = (
+        db.query(PipelineRoutingRule)
+        .filter(
+            PipelineRoutingRule.is_active.is_(True),
+            (PipelineRoutingRule.owner_id == user_id) | (PipelineRoutingRule.owner_id.is_(None)),
+        )
+        .order_by(
+            PipelineRoutingRule.owner_id.is_(None).asc(),
+            PipelineRoutingRule.position.asc(),
+        )
+        .all()
+    )
+
+    for rule in rules:
+        actual = _resolve_field(rule.field, doc_props)
+        if _evaluate_condition(actual, rule.operator, rule.value):
+            pipeline = db.query(Pipeline).filter(Pipeline.id == rule.target_pipeline_id).first()
+            return {
+                "matched": True,
+                "rule": _serialize_rule(rule),
+                "target_pipeline": {
+                    "id": pipeline.id,
+                    "name": pipeline.name,
+                    "is_active": pipeline.is_active,
+                }
+                if pipeline
+                else None,
+            }
+
+    return {"matched": False, "rule": None, "target_pipeline": None}
+
+
+@router.put("/reorder")
+@require_login
+def reorder_routing_rules(
+    request: Request,
+    db: DbSession,
+    rule_ids: list[int] = Body(..., embed=True),
+) -> list[dict[str, Any]]:
+    """Reorder the caller's routing rules.
+
+    Expects a JSON body ``{"rule_ids": [3, 1, 2]}`` where the list
+    contains the IDs of the caller's rules in the desired order.
+    """
+    user_id = _get_user_id(request)
+
+    rules = (
+        db.query(PipelineRoutingRule)
+        .filter(PipelineRoutingRule.owner_id == user_id, PipelineRoutingRule.id.in_(rule_ids))
+        .all()
+    )
+
+    rule_map = {r.id: r for r in rules}
+
+    if len(rule_map) != len(rule_ids) or set(rule_map.keys()) != set(rule_ids):
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail="rule_ids must contain exactly the IDs of your routing rules",
+        )
+
+    for pos, rid in enumerate(rule_ids):
+        rule_map[rid].position = pos
+
+    try:
+        db.commit()
+    except Exception:
+        db.rollback()
+        logger.exception("Failed to reorder routing rules for user=%s", user_id)
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to reorder routing rules",
+        )
+
+    ordered = sorted(rules, key=lambda r: r.position)
+    return [_serialize_rule(r) for r in ordered]
+
+
+@router.get("/{rule_id}")
+@require_login
+def get_routing_rule(rule_id: int, request: Request, db: DbSession) -> dict[str, Any]:
+    """Return a single routing rule by ID."""
+    user_id = _get_user_id(request)
+    admin = _is_admin(request)
+
+    rule = db.query(PipelineRoutingRule).filter(PipelineRoutingRule.id == rule_id).first()
+    if not rule or not _can_access_rule(rule, user_id, admin):
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Routing rule not found")
+
+    return _serialize_rule(rule)
+
+
+@router.put("/{rule_id}")
+@require_login
+def update_routing_rule(rule_id: int, request: Request, db: DbSession, body: RoutingRuleUpdate) -> dict[str, Any]:
+    """Update a routing rule.
+
+    Only the fields present in the request body are updated.
+    """
+    user_id = _get_user_id(request)
+    admin = _is_admin(request)
+
+    rule = db.query(PipelineRoutingRule).filter(PipelineRoutingRule.id == rule_id).first()
+    if not rule or not _can_access_rule(rule, user_id, admin):
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Routing rule not found")
+
+    if not _can_write_rule(rule, user_id, admin):
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Cannot modify this rule")
+
+    if body.field is not None:
+        _validate_field(body.field)
+        rule.field = body.field
+
+    if body.operator is not None:
+        _validate_operator(body.operator)
+        rule.operator = body.operator
+
+    if body.value is not None:
+        rule.value = body.value
+
+    if body.name is not None:
+        rule.name = body.name.strip()
+
+    if body.target_pipeline_id is not None:
+        pipeline = db.query(Pipeline).filter(Pipeline.id == body.target_pipeline_id).first()
+        if not pipeline:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail=f"Target pipeline {body.target_pipeline_id} not found",
+            )
+        rule.target_pipeline_id = body.target_pipeline_id
+
+    if body.position is not None:
+        rule.position = body.position
+
+    if body.is_active is not None:
+        rule.is_active = body.is_active
+
+    try:
+        db.commit()
+        db.refresh(rule)
+    except Exception:
+        db.rollback()
+        logger.exception("Failed to update routing rule id=%s", rule_id)
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to update routing rule",
+        )
+
+    logger.info("Routing rule updated: id=%s, user=%s", rule_id, user_id)
+    return _serialize_rule(rule)
+
+
+@router.delete("/{rule_id}", status_code=status.HTTP_204_NO_CONTENT)
+@require_login
+def delete_routing_rule(rule_id: int, request: Request, db: DbSession) -> None:
+    """Delete a routing rule."""
+    user_id = _get_user_id(request)
+    admin = _is_admin(request)
+
+    rule = db.query(PipelineRoutingRule).filter(PipelineRoutingRule.id == rule_id).first()
+    if not rule or not _can_access_rule(rule, user_id, admin):
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Routing rule not found")
+
+    if not _can_write_rule(rule, user_id, admin):
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Cannot modify this rule")
+
+    try:
+        db.delete(rule)
+        db.commit()
+    except Exception:
+        db.rollback()
+        logger.exception("Failed to delete routing rule id=%s", rule_id)
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to delete routing rule",
+        )
+
+    logger.info("Routing rule deleted: id=%s, user=%s", rule_id, user_id)

app/models.py

@@ -7,6 +7,7 @@ from app.database import Base
 # Foreign key constants
 _FILES_ID_FK = "files.id"
 _PIPELINES_ID_FK = "pipelines.id"
+_ROUTING_RULES_TABLE = "pipeline_routing_rules"
 
 
 class DocumentMetadata(Base):
@@ -970,3 +971,53 @@ class ComplianceTemplate(Base):
     applied_by = Column(String(255), nullable=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 PipelineRoutingRule(Base):
+    """Conditional routing rule that assigns documents to pipelines.
+
+    Rules are evaluated in ascending ``position`` order for a given owner.
+    The first rule whose condition matches the document properties wins and
+    the document is routed to ``target_pipeline_id``.  If no rule matches,
+    the caller falls back to the owner's (or system) default pipeline.
+
+    Supported fields:
+        file_type, document_type, category, filename, size, and any key
+        inside the AI-extracted metadata JSON (prefixed ``metadata.``).
+
+    Supported operators:
+        equals, not_equals, contains, not_contains, regex, gt, lt, gte, lte.
+    """
+
+    __tablename__ = _ROUTING_RULES_TABLE
+
+    id = Column(Integer, primary_key=True, index=True)
+
+    # Owner of this rule.  NULL = system-wide rule (admin only).
+    owner_id = Column(String, nullable=True, index=True)
+
+    # Human-readable label for the rule.
+    name = Column(String(255), nullable=False)
+
+    # Evaluation order (lower = earlier).  First matching rule wins.
+    position = Column(Integer, nullable=False, default=0)
+
+    # The document property to evaluate.
+    # Built-in: file_type, document_type, category, filename, size.
+    # For AI metadata fields, use the "metadata.<key>" prefix.
+    field = Column(String(255), nullable=False)
+
+    # Comparison operator.
+    operator = Column(String(50), nullable=False)
+
+    # Value to compare against (always stored as text; cast as needed).
+    value = Column(String(1024), nullable=False)
+
+    # Target pipeline when the condition matches.
+    target_pipeline_id = Column(Integer, ForeignKey(_PIPELINES_ID_FK), nullable=False, index=True)
+
+    # 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())

app/utils/routing_engine.py

@@ -0,0 +1,223 @@
+"""Routing engine for conditional document-to-pipeline assignment.
+
+Evaluates a set of :class:`PipelineRoutingRule` rows against document
+properties and returns the first matching target pipeline (if any).
+
+Supported document fields
+-------------------------
+* ``file_type`` – MIME type of the file (e.g. ``application/pdf``)
+* ``filename``  – original filename
+* ``size``      – file size in bytes (numeric comparison)
+* ``document_type`` – AI-classified document type (e.g. ``Invoice``)
+* ``category``  – alias for ``document_type``
+* ``metadata.<key>`` – arbitrary key inside the AI-extracted JSON metadata
+
+Supported comparison operators
+------------------------------
+* ``equals`` / ``not_equals``
+* ``contains`` / ``not_contains`` (substring match, case-insensitive)
+* ``regex`` (Python ``re`` full-match, case-insensitive)
+* ``gt`` / ``lt`` / ``gte`` / ``lte`` (numeric comparison)
+"""
+
+import json
+import logging
+import re
+from typing import Any
+
+from sqlalchemy.orm import Session
+
+from app.models import Pipeline, PipelineRoutingRule
+
+logger = logging.getLogger(__name__)
+
+# Operators recognised by the engine.
+VALID_OPERATORS: frozenset[str] = frozenset(
+    {
+        "equals",
+        "not_equals",
+        "contains",
+        "not_contains",
+        "regex",
+        "gt",
+        "lt",
+        "gte",
+        "lte",
+    }
+)
+
+# Fields that are resolved directly from the FileRecord.
+BUILTIN_FIELDS: frozenset[str] = frozenset(
+    {
+        "file_type",
+        "filename",
+        "size",
+        "document_type",
+        "category",
+    }
+)
+
+
+def _resolve_field(field: str, doc_props: dict[str, Any]) -> Any:
+    """Resolve a *field* name to its actual value from *doc_props*.
+
+    ``doc_props`` is expected to contain top-level keys that mirror the
+    built-in field names **plus** a ``metadata`` dict with the parsed
+    AI metadata JSON.
+    """
+    if field == "category":
+        # ``category`` is an alias for ``document_type``.
+        field = "document_type"
+
+    if field.startswith("metadata."):
+        meta_key = field[len("metadata.") :]
+        metadata = doc_props.get("metadata") or {}
+        return metadata.get(meta_key)
+
+    return doc_props.get(field)
+
+
+def _to_float(value: Any) -> float | None:
+    """Try to convert *value* to a float for numeric comparison."""
+    if value is None:
+        return None
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def _evaluate_condition(actual: Any, operator: str, expected: str) -> bool:
+    """Return ``True`` when *actual* satisfies *operator* against *expected*.
+
+    All string comparisons are case-insensitive.  Numeric operators (``gt``,
+    ``lt``, ``gte``, ``lte``) attempt to cast both sides to ``float``.
+    """
+    if actual is None:
+        # If the document property is missing, the rule cannot match
+        # (except for ``not_equals`` / ``not_contains`` which should match).
+        if operator == "not_equals":
+            return True
+        if operator == "not_contains":
+            return True
+        return False
+
+    actual_str = str(actual).lower()
+    expected_lower = expected.lower()
+
+    if operator == "equals":
+        return actual_str == expected_lower
+    if operator == "not_equals":
+        return actual_str != expected_lower
+    if operator == "contains":
+        return expected_lower in actual_str
+    if operator == "not_contains":
+        return expected_lower not in actual_str
+    if operator == "regex":
+        try:
+            return bool(re.fullmatch(expected, str(actual), flags=re.IGNORECASE))
+        except re.error:
+            logger.warning("Invalid regex in routing rule: %s", expected)
+            return False
+
+    # Numeric operators
+    actual_num = _to_float(actual)
+    expected_num = _to_float(expected)
+    if actual_num is None or expected_num is None:
+        return False
+
+    if operator == "gt":
+        return actual_num > expected_num
+    if operator == "lt":
+        return actual_num < expected_num
+    if operator == "gte":
+        return actual_num >= expected_num
+    if operator == "lte":
+        return actual_num <= expected_num
+
+    return False
+
+
+def build_document_properties(file_record: Any) -> dict[str, Any]:
+    """Build the property dict that the engine evaluates against.
+
+    Args:
+        file_record: A :class:`FileRecord` instance (or any object with the
+            same attributes).
+
+    Returns:
+        A dict with ``file_type``, ``filename``, ``size``, ``document_type``,
+        and ``metadata`` keys.
+    """
+    metadata: dict[str, Any] = {}
+    raw_meta = getattr(file_record, "ai_metadata", None)
+    if raw_meta:
+        try:
+            metadata = json.loads(raw_meta) if isinstance(raw_meta, str) else raw_meta
+        except (json.JSONDecodeError, TypeError):
+            metadata = {}
+
+    return {
+        "file_type": getattr(file_record, "mime_type", None),
+        "filename": getattr(file_record, "original_filename", None),
+        "size": getattr(file_record, "file_size", None),
+        "document_type": metadata.get("document_type"),
+        "metadata": metadata,
+    }
+
+
+def evaluate_routing_rules(
+    db: Session,
+    owner_id: str | None,
+    doc_props: dict[str, Any],
+) -> Pipeline | None:
+    """Evaluate routing rules and return the first matching pipeline.
+
+    Rules are fetched for the given *owner_id* **plus** any system-wide rules
+    (``owner_id IS NULL``).  Owner rules are evaluated first (by position),
+    then system rules.
+
+    Args:
+        db: Active database session.
+        owner_id: The document owner's identifier (may be ``None``).
+        doc_props: Document property dict as produced by
+            :func:`build_document_properties`.
+
+    Returns:
+        The first matching :class:`Pipeline`, or ``None`` when no rule
+        matches (caller should fall back to the default pipeline).
+    """
+    # Fetch active rules for the owner + system rules, ordered by position.
+    rules = (
+        db.query(PipelineRoutingRule)
+        .filter(
+            PipelineRoutingRule.is_active.is_(True),
+            (PipelineRoutingRule.owner_id == owner_id) | (PipelineRoutingRule.owner_id.is_(None)),
+        )
+        .order_by(
+            # Owner-specific rules take priority over system rules.
+            PipelineRoutingRule.owner_id.is_(None).asc(),
+            PipelineRoutingRule.position.asc(),
+        )
+        .all()
+    )
+
+    for rule in rules:
+        actual = _resolve_field(rule.field, doc_props)
+        if _evaluate_condition(actual, rule.operator, rule.value):
+            pipeline = db.query(Pipeline).filter(Pipeline.id == rule.target_pipeline_id).first()
+            if pipeline and pipeline.is_active:
+                logger.info(
+                    "Routing rule matched: rule_id=%s, name=%s, target_pipeline=%s",
+                    rule.id,
+                    rule.name,
+                    rule.target_pipeline_id,
+                )
+                return pipeline
+            logger.warning(
+                "Routing rule %s matched but target pipeline %s is inactive or missing",
+                rule.id,
+                rule.target_pipeline_id,
+            )
+
+    return None

docs/API.md

@@ -1960,6 +1960,160 @@ Pass no `pipeline_id` query parameter (or omit it) to clear the assignment.
 ```
 
 
+## Routing Rules
+
+Routing rules let you conditionally assign documents to different pipelines
+based on file properties such as type, size, filename, or AI-extracted
+metadata.  Rules are evaluated in **position order** (lowest first); the first
+rule that matches wins.  If no rule matches, the system falls back to the
+owner's (or global) default pipeline.
+
+### Supported operators and fields
+
+```bash
+GET /api/routing-rules/operators
+```
+
+Returns the catalogue of valid operators and built-in fields so UIs can
+populate dropdowns without hard-coding values.
+
+**Response (200):**
+```json
+{
+  "operators": ["contains", "equals", "gt", "gte", "lt", "lte", "not_contains", "not_equals", "regex"],
+  "builtin_fields": ["category", "document_type", "file_type", "filename", "size"],
+  "metadata_prefix": "metadata."
+}
+```
+
+> **Tip:** For AI metadata fields use the `metadata.` prefix, e.g.
+> `metadata.sender`, `metadata.amount`.
+
+### List routing rules
+
+```bash
+GET /api/routing-rules
+```
+
+Returns the current user's rules **plus** any system-wide rules
+(`owner_id = null`), ordered by position.
+
+**Response (200):**
+```json
+[
+  {
+    "id": 1,
+    "owner_id": "alice",
+    "name": "Route invoices",
+    "position": 0,
+    "field": "document_type",
+    "operator": "equals",
+    "value": "Invoice",
+    "target_pipeline_id": 3,
+    "is_active": true,
+    "created_at": "2026-03-09T12:00:00+00:00",
+    "updated_at": "2026-03-09T12:00:00+00:00"
+  }
+]
+```
+
+### Create routing rule
+
+```bash
+POST /api/routing-rules
+Content-Type: application/json
+
+{
+  "name": "Route invoices",
+  "field": "document_type",
+  "operator": "equals",
+  "value": "Invoice",
+  "target_pipeline_id": 3
+}
+```
+
+Optional fields: `position` (auto-assigned if omitted), `is_active` (default `true`).
+
+**Response (201 Created):** The created rule object.
+
+### Get routing rule
+
+```bash
+GET /api/routing-rules/{rule_id}
+```
+
+**Response (200):** A single rule object.
+
+### Update routing rule
+
+```bash
+PUT /api/routing-rules/{rule_id}
+Content-Type: application/json
+
+{ "name": "Renamed rule", "operator": "contains", "is_active": false }
+```
+
+Only the supplied fields are updated.
+
+**Response (200):** The updated rule object.
+
+### Delete routing rule
+
+```bash
+DELETE /api/routing-rules/{rule_id}
+```
+
+Returns **204 No Content**.
+
+### Reorder routing rules
+
+```bash
+PUT /api/routing-rules/reorder
+Content-Type: application/json
+
+{ "rule_ids": [3, 1, 2] }
+```
+
+Provide the complete ordered list of your rule IDs.  Positions are reassigned
+0, 1, 2, … in the given order.
+
+### Evaluate rules (dry run)
+
+```bash
+POST /api/routing-rules/evaluate
+Content-Type: application/json
+
+{
+  "file_type": "application/pdf",
+  "filename": "invoice_2024.pdf",
+  "size": 204800,
+  "document_type": "Invoice",
+  "metadata": { "sender": "Acme Corp" }
+}
+```
+
+Tests which rule (if any) would match the given properties **without**
+actually routing a document.
+
+**Response (200) – match found:**
+```json
+{
+  "matched": true,
+  "rule": { "id": 1, "name": "Route invoices", "..." : "..." },
+  "target_pipeline": { "id": 3, "name": "Invoice Pipeline", "is_active": true }
+}
+```
+
+**Response (200) – no match:**
+```json
+{
+  "matched": false,
+  "rule": null,
+  "target_pipeline": null
+}
+```
+
+
 ## API Tokens
 
 Personal API tokens allow programmatic access to the DocuElevate API without

docs/UserGuide.md

@@ -626,6 +626,58 @@ Pass no `pipeline_id` to clear the assignment and fall back to the system defaul
 
 Admins can create **system pipelines** that appear in every user's pipeline list. These can be set as the global default so all users benefit from a consistent processing baseline. Navigate to **Pipelines** and check the **System pipeline** box when creating a new one (admin only).
 
+### Conditional routing rules
+
+Routing rules automatically assign incoming documents to the right pipeline
+based on their properties — no manual pipeline selection required.
+
+**How it works:**
+
+1. Define one or more routing rules via the API
+   (`POST /api/routing-rules`).
+2. Each rule specifies a **field** to inspect, an **operator** (condition),
+   a **value** to compare against, and a **target pipeline**.
+3. When a document is processed, rules are evaluated **in position order**
+   (lowest first).  The first matching rule wins and the document is routed
+   to that pipeline.
+4. If no rule matches, the document is processed by the default pipeline.
+
+**Available fields:**
+
+| Field | Description |
+|-------|-------------|
+| `file_type` | MIME type, e.g. `application/pdf` |
+| `filename` | Original filename |
+| `size` | File size in bytes |
+| `document_type` | AI-classified type (Invoice, Contract, …) |
+| `category` | Alias for `document_type` |
+| `metadata.<key>` | Any key from the AI-extracted metadata JSON |
+
+**Available operators:**
+
+| Operator | Description |
+|----------|-------------|
+| `equals` / `not_equals` | Exact match (case-insensitive) |
+| `contains` / `not_contains` | Substring match (case-insensitive) |
+| `regex` | Full Python regex match (case-insensitive) |
+| `gt` / `lt` / `gte` / `lte` | Numeric comparison (greater/less than) |
+
+**Example:** Route invoices to one pipeline and large files to another:
+
+```
+Rule 1: field=document_type, operator=equals, value=Invoice, target_pipeline=3
+Rule 2: field=size,          operator=gt,     value=1048576, target_pipeline=5
+```
+
+With first-match-wins logic, an invoice of any size matches Rule 1 and is
+routed to pipeline 3.  A non-invoice file larger than 1 MB matches Rule 2
+and is routed to pipeline 5.  Everything else falls back to the default
+pipeline.
+
+You can test your rules without actually routing a document using the
+**evaluate** endpoint (`POST /api/routing-rules/evaluate`).  For the full
+API reference, see [API Documentation](API.md#routing-rules).
+
 ## API Access
 
 For programmatic access, DocuElevate provides a comprehensive REST API:

migrations/env.py

@@ -33,6 +33,7 @@ from app.models import (  # noqa: F401
     LocalUser,
     MobileDevice,
     Pipeline,
+    PipelineRoutingRule,
     PipelineStep,
     ProcessingLog,
     SavedSearch,

migrations/versions/035_add_routing_rules.py

@@ -0,0 +1,46 @@
+"""Add pipeline_routing_rules table for conditional document routing.
+
+Revision ID: 035_add_routing_rules
+Revises: 034_add_user_profile_settings
+Create Date: 2026-03-09
+"""
+
+from typing import Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "035_add_routing_rules"
+down_revision: Union[str, None] = "034_add_user_profile_settings"
+depends_on: Union[str, None] = None
+
+
+def upgrade() -> None:
+    """Create pipeline_routing_rules table."""
+    op.create_table(
+        "pipeline_routing_rules",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("owner_id", sa.String(), nullable=True),
+        sa.Column("name", sa.String(255), nullable=False),
+        sa.Column("position", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("field", sa.String(255), nullable=False),
+        sa.Column("operator", sa.String(50), nullable=False),
+        sa.Column("value", sa.String(1024), nullable=False),
+        sa.Column("target_pipeline_id", sa.Integer(), nullable=False),
+        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()),
+        sa.PrimaryKeyConstraint("id"),
+        sa.ForeignKeyConstraint(["target_pipeline_id"], ["pipelines.id"]),
+    )
+    op.create_index("ix_routing_rules_id", "pipeline_routing_rules", ["id"])
+    op.create_index("ix_routing_rules_owner_id", "pipeline_routing_rules", ["owner_id"])
+    op.create_index("ix_routing_rules_target_pipeline_id", "pipeline_routing_rules", ["target_pipeline_id"])
+
+
+def downgrade() -> None:
+    """Drop pipeline_routing_rules table."""
+    op.drop_index("ix_routing_rules_target_pipeline_id", "pipeline_routing_rules")
+    op.drop_index("ix_routing_rules_owner_id", "pipeline_routing_rules")
+    op.drop_index("ix_routing_rules_id", "pipeline_routing_rules")
+    op.drop_table("pipeline_routing_rules")

tests/conftest.py

@@ -66,6 +66,7 @@ from app.models import (  # noqa: F401, E402
     DocumentMetadata,
     FileRecord,
     Pipeline,
+    PipelineRoutingRule,
     PipelineStep,
     ProcessingLog,
     SavedSearch,

tests/test_routing_rules.py

@@ -0,0 +1,803 @@
+"""Tests for the routing rules API and routing engine.
+
+Covers CRUD operations for routing rules, rule evaluation (dry-run and engine),
+operator logic, access control, and edge cases.
+"""
+
+import json
+
+import pytest
+
+from app.models import FileRecord, Pipeline, PipelineRoutingRule
+from app.utils.routing_engine import (
+    BUILTIN_FIELDS,
+    VALID_OPERATORS,
+    _evaluate_condition,
+    _resolve_field,
+    _to_float,
+    build_document_properties,
+    evaluate_routing_rules,
+)
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_pipeline(db_session, name="Test Pipeline", owner_id="testuser", is_active=True):
+    """Insert a minimal Pipeline and return it."""
+    p = Pipeline(owner_id=owner_id, name=name, is_default=False, is_active=is_active)
+    db_session.add(p)
+    db_session.commit()
+    db_session.refresh(p)
+    return p
+
+
+def _make_rule(db_session, target_pipeline_id, **kwargs):
+    """Insert a PipelineRoutingRule and return it."""
+    defaults = {
+        "owner_id": "testuser",
+        "name": "Test Rule",
+        "position": 0,
+        "field": "file_type",
+        "operator": "equals",
+        "value": "application/pdf",
+        "is_active": True,
+    }
+    defaults.update(kwargs)
+    defaults["target_pipeline_id"] = target_pipeline_id
+    rule = PipelineRoutingRule(**defaults)
+    db_session.add(rule)
+    db_session.commit()
+    db_session.refresh(rule)
+    return rule
+
+
+def _make_file_record(db_session, **kwargs):
+    """Insert a minimal FileRecord and return it."""
+    defaults = {
+        "owner_id": "testuser",
+        "filehash": "abc123",
+        "original_filename": "test.pdf",
+        "local_filename": "/tmp/test.pdf",
+        "file_size": 1024,
+        "mime_type": "application/pdf",
+    }
+    defaults.update(kwargs)
+    fr = FileRecord(**defaults)
+    db_session.add(fr)
+    db_session.commit()
+    db_session.refresh(fr)
+    return fr
+
+
+# ===========================================================================
+# Unit tests – routing engine
+# ===========================================================================
+
+
+@pytest.mark.unit
+class TestResolveField:
+    """Tests for _resolve_field()."""
+
+    def test_builtin_field(self):
+        """Built-in fields are resolved directly from the dict."""
+        props = {"file_type": "application/pdf", "size": 1024}
+        assert _resolve_field("file_type", props) == "application/pdf"
+        assert _resolve_field("size", props) == 1024
+
+    def test_category_alias(self):
+        """'category' is an alias for 'document_type'."""
+        props = {"document_type": "Invoice"}
+        assert _resolve_field("category", props) == "Invoice"
+
+    def test_metadata_field(self):
+        """'metadata.<key>' resolves from the nested metadata dict."""
+        props = {"metadata": {"sender": "Acme Corp", "amount": 100.50}}
+        assert _resolve_field("metadata.sender", props) == "Acme Corp"
+        assert _resolve_field("metadata.amount", props) == 100.50
+
+    def test_missing_metadata_key(self):
+        """Missing metadata key returns None."""
+        props = {"metadata": {"sender": "Acme"}}
+        assert _resolve_field("metadata.missing_key", props) is None
+
+    def test_missing_metadata_dict(self):
+        """Missing metadata dict returns None."""
+        props = {}
+        assert _resolve_field("metadata.sender", props) is None
+
+    def test_missing_builtin_field(self):
+        """Missing built-in field returns None."""
+        props = {}
+        assert _resolve_field("file_type", props) is None
+
+
+@pytest.mark.unit
+class TestToFloat:
+    """Tests for _to_float()."""
+
+    def test_int_value(self):
+        assert _to_float(42) == 42.0
+
+    def test_float_value(self):
+        assert _to_float(3.14) == 3.14
+
+    def test_string_number(self):
+        assert _to_float("100") == 100.0
+
+    def test_none_returns_none(self):
+        assert _to_float(None) is None
+
+    def test_non_numeric_string_returns_none(self):
+        assert _to_float("not-a-number") is None
+
+
+@pytest.mark.unit
+class TestEvaluateCondition:
+    """Tests for _evaluate_condition()."""
+
+    def test_equals_match(self):
+        assert _evaluate_condition("application/pdf", "equals", "application/pdf") is True
+
+    def test_equals_case_insensitive(self):
+        assert _evaluate_condition("Application/PDF", "equals", "application/pdf") is True
+
+    def test_equals_no_match(self):
+        assert _evaluate_condition("image/png", "equals", "application/pdf") is False
+
+    def test_not_equals_match(self):
+        assert _evaluate_condition("image/png", "not_equals", "application/pdf") is True
+
+    def test_not_equals_no_match(self):
+        assert _evaluate_condition("application/pdf", "not_equals", "application/pdf") is False
+
+    def test_contains_match(self):
+        assert _evaluate_condition("invoice_2024.pdf", "contains", "invoice") is True
+
+    def test_contains_case_insensitive(self):
+        assert _evaluate_condition("INVOICE_2024.PDF", "contains", "invoice") is True
+
+    def test_contains_no_match(self):
+        assert _evaluate_condition("receipt.pdf", "contains", "invoice") is False
+
+    def test_not_contains_match(self):
+        assert _evaluate_condition("receipt.pdf", "not_contains", "invoice") is True
+
+    def test_not_contains_no_match(self):
+        assert _evaluate_condition("invoice_2024.pdf", "not_contains", "invoice") is False
+
+    def test_regex_match(self):
+        assert _evaluate_condition("invoice_2024.pdf", "regex", r"invoice_\d+\.pdf") is True
+
+    def test_regex_no_match(self):
+        assert _evaluate_condition("receipt.pdf", "regex", r"invoice_\d+\.pdf") is False
+
+    def test_regex_case_insensitive(self):
+        assert _evaluate_condition("INVOICE_2024.PDF", "regex", r"invoice_\d+\.pdf") is True
+
+    def test_regex_invalid_pattern(self):
+        """Invalid regex should return False, not raise."""
+        assert _evaluate_condition("test", "regex", r"[invalid") is False
+
+    def test_gt(self):
+        assert _evaluate_condition(2048, "gt", "1024") is True
+        assert _evaluate_condition(1024, "gt", "1024") is False
+
+    def test_lt(self):
+        assert _evaluate_condition(512, "lt", "1024") is True
+        assert _evaluate_condition(1024, "lt", "1024") is False
+
+    def test_gte(self):
+        assert _evaluate_condition(1024, "gte", "1024") is True
+        assert _evaluate_condition(2048, "gte", "1024") is True
+        assert _evaluate_condition(512, "gte", "1024") is False
+
+    def test_lte(self):
+        assert _evaluate_condition(1024, "lte", "1024") is True
+        assert _evaluate_condition(512, "lte", "1024") is True
+        assert _evaluate_condition(2048, "lte", "1024") is False
+
+    def test_none_actual_returns_false(self):
+        """When the actual value is None, most operators return False."""
+        assert _evaluate_condition(None, "equals", "test") is False
+        assert _evaluate_condition(None, "contains", "test") is False
+        assert _evaluate_condition(None, "regex", "test") is False
+        assert _evaluate_condition(None, "gt", "10") is False
+
+    def test_none_actual_not_equals_returns_true(self):
+        """not_equals should return True when actual is None."""
+        assert _evaluate_condition(None, "not_equals", "test") is True
+
+    def test_none_actual_not_contains_returns_true(self):
+        """not_contains should return True when actual is None."""
+        assert _evaluate_condition(None, "not_contains", "test") is True
+
+    def test_non_numeric_gt_returns_false(self):
+        """Non-numeric values should return False for numeric operators."""
+        assert _evaluate_condition("abc", "gt", "100") is False
+
+    def test_unknown_operator_returns_false(self):
+        """Unknown operator should return False."""
+        assert _evaluate_condition("test", "unknown_op", "test") is False
+
+
+@pytest.mark.unit
+class TestBuildDocumentProperties:
+    """Tests for build_document_properties()."""
+
+    def test_basic_properties(self):
+        """Properties are extracted from FileRecord attributes."""
+        fr = _MockFileRecord(
+            mime_type="application/pdf",
+            original_filename="test.pdf",
+            file_size=2048,
+            ai_metadata=json.dumps({"document_type": "Invoice", "sender": "Acme"}),
+        )
+        props = build_document_properties(fr)
+        assert props["file_type"] == "application/pdf"
+        assert props["filename"] == "test.pdf"
+        assert props["size"] == 2048
+        assert props["document_type"] == "Invoice"
+        assert props["metadata"]["sender"] == "Acme"
+
+    def test_no_metadata(self):
+        """When ai_metadata is None, metadata is an empty dict."""
+        fr = _MockFileRecord(mime_type="image/png", original_filename="img.png", file_size=512, ai_metadata=None)
+        props = build_document_properties(fr)
+        assert props["metadata"] == {}
+        assert props["document_type"] is None
+
+    def test_invalid_metadata_json(self):
+        """Invalid JSON in ai_metadata should result in empty metadata."""
+        fr = _MockFileRecord(
+            mime_type="application/pdf",
+            original_filename="test.pdf",
+            file_size=1024,
+            ai_metadata="not-json",
+        )
+        props = build_document_properties(fr)
+        assert props["metadata"] == {}
+
+
+class _MockFileRecord:
+    """Lightweight stand-in for FileRecord in unit tests."""
+
+    def __init__(self, **kwargs):
+        for k, v in kwargs.items():
+            setattr(self, k, v)
+
+
+# ===========================================================================
+# Unit tests – evaluate_routing_rules (DB-backed)
+# ===========================================================================
+
+
+@pytest.mark.unit
+class TestEvaluateRoutingRules:
+    """Tests for evaluate_routing_rules() with a real DB session."""
+
+    def test_first_match_wins(self, db_session):
+        """The first matching rule (by position) is used."""
+        p1 = _make_pipeline(db_session, name="Pipeline A")
+        p2 = _make_pipeline(db_session, name="Pipeline B")
+
+        _make_rule(db_session, p1.id, position=0, field="file_type", operator="equals", value="application/pdf")
+        _make_rule(db_session, p2.id, position=1, field="file_type", operator="equals", value="application/pdf")
+
+        doc = {"file_type": "application/pdf", "metadata": {}}
+        result = evaluate_routing_rules(db_session, "testuser", doc)
+        assert result is not None
+        assert result.id == p1.id
+
+    def test_no_match_returns_none(self, db_session):
+        """When no rule matches, None is returned (caller uses default)."""
+        p = _make_pipeline(db_session, name="Pipeline A")
+        _make_rule(db_session, p.id, field="file_type", operator="equals", value="image/png")
+
+        doc = {"file_type": "application/pdf", "metadata": {}}
+        result = evaluate_routing_rules(db_session, "testuser", doc)
+        assert result is None
+
+    def test_inactive_rule_skipped(self, db_session):
+        """Inactive rules are not evaluated."""
+        p = _make_pipeline(db_session, name="Pipeline A")
+        _make_rule(db_session, p.id, is_active=False)
+
+        doc = {"file_type": "application/pdf", "metadata": {}}
+        result = evaluate_routing_rules(db_session, "testuser", doc)
+        assert result is None
+
+    def test_inactive_pipeline_skipped(self, db_session):
+        """Matching rule with inactive target pipeline is skipped."""
+        p = _make_pipeline(db_session, name="Inactive Pipeline", is_active=False)
+        _make_rule(db_session, p.id)
+
+        doc = {"file_type": "application/pdf", "metadata": {}}
+        result = evaluate_routing_rules(db_session, "testuser", doc)
+        assert result is None
+
+    def test_system_rules_evaluated_after_user_rules(self, db_session):
+        """System rules (owner_id=NULL) are evaluated after user-specific rules."""
+        p_user = _make_pipeline(db_session, name="User Pipeline")
+        p_system = _make_pipeline(db_session, name="System Pipeline", owner_id=None)
+
+        # System rule at position 0, user rule at position 1 — user should still win.
+        _make_rule(
+            db_session,
+            p_system.id,
+            owner_id=None,
+            position=0,
+            field="file_type",
+            operator="equals",
+            value="application/pdf",
+            name="System Rule",
+        )
+        _make_rule(
+            db_session,
+            p_user.id,
+            owner_id="testuser",
+            position=1,
+            field="file_type",
+            operator="equals",
+            value="application/pdf",
+            name="User Rule",
+        )
+
+        doc = {"file_type": "application/pdf", "metadata": {}}
+        result = evaluate_routing_rules(db_session, "testuser", doc)
+        assert result is not None
+        assert result.id == p_user.id
+
+    def test_metadata_field_routing(self, db_session):
+        """Rules can match on metadata.* fields."""
+        p = _make_pipeline(db_session, name="Invoice Pipeline")
+        _make_rule(
+            db_session,
+            p.id,
+            field="metadata.sender",
+            operator="contains",
+            value="acme",
+        )
+
+        doc = {"file_type": "application/pdf", "metadata": {"sender": "Acme Corporation"}}
+        result = evaluate_routing_rules(db_session, "testuser", doc)
+        assert result is not None
+        assert result.id == p.id
+
+    def test_size_routing(self, db_session):
+        """Rules can match on file size with numeric comparison."""
+        p = _make_pipeline(db_session, name="Large File Pipeline")
+        _make_rule(
+            db_session,
+            p.id,
+            field="size",
+            operator="gt",
+            value="1048576",
+        )
+
+        # 2 MB file
+        doc = {"size": 2097152, "metadata": {}}
+        result = evaluate_routing_rules(db_session, "testuser", doc)
+        assert result is not None
+        assert result.id == p.id
+
+    def test_regex_routing(self, db_session):
+        """Rules can match using regex on filename."""
+        p = _make_pipeline(db_session, name="Invoice Pipeline")
+        _make_rule(
+            db_session,
+            p.id,
+            field="filename",
+            operator="regex",
+            value=r"invoice_\d{4}.*",
+        )
+
+        doc = {"filename": "invoice_2024_q1.pdf", "metadata": {}}
+        result = evaluate_routing_rules(db_session, "testuser", doc)
+        assert result is not None
+        assert result.id == p.id
+
+
+# ===========================================================================
+# API tests – operators endpoint (public)
+# ===========================================================================
+
+
+@pytest.mark.unit
+class TestOperatorsEndpoint:
+    """Tests for the /api/routing-rules/operators catalogue endpoint."""
+
+    def test_operators_returns_lists(self, client):
+        """GET /api/routing-rules/operators returns operators and fields."""
+        r = client.get("/api/routing-rules/operators")
+        assert r.status_code == 200
+        data = r.json()
+        assert "operators" in data
+        assert "builtin_fields" in data
+        assert "metadata_prefix" in data
+        assert set(data["operators"]) == VALID_OPERATORS
+        assert set(data["builtin_fields"]) == BUILTIN_FIELDS
+
+
+# ===========================================================================
+# API tests – CRUD
+# ===========================================================================
+
+
+@pytest.mark.unit
+class TestRoutingRuleCRUD:
+    """Tests for the routing rules CRUD API endpoints."""
+
+    def test_create_rule(self, client, db_session):
+        """POST /api/routing-rules creates a new rule."""
+        p = _make_pipeline(db_session, name="Target Pipeline")
+        r = client.post(
+            "/api/routing-rules",
+            json={
+                "name": "Route PDFs",
+                "field": "file_type",
+                "operator": "equals",
+                "value": "application/pdf",
+                "target_pipeline_id": p.id,
+            },
+        )
+        assert r.status_code == 201
+        data = r.json()
+        assert data["name"] == "Route PDFs"
+        assert data["field"] == "file_type"
+        assert data["operator"] == "equals"
+        assert data["target_pipeline_id"] == p.id
+        assert data["is_active"] is True
+
+    def test_create_rule_invalid_field(self, client, db_session):
+        """POST with an invalid field returns 422."""
+        p = _make_pipeline(db_session, name="Target")
+        r = client.post(
+            "/api/routing-rules",
+            json={
+                "name": "Bad field",
+                "field": "invalid_field",
+                "operator": "equals",
+                "value": "test",
+                "target_pipeline_id": p.id,
+            },
+        )
+        assert r.status_code == 422
+
+    def test_create_rule_invalid_operator(self, client, db_session):
+        """POST with an invalid operator returns 422."""
+        p = _make_pipeline(db_session, name="Target")
+        r = client.post(
+            "/api/routing-rules",
+            json={
+                "name": "Bad op",
+                "field": "file_type",
+                "operator": "invalid_op",
+                "value": "test",
+                "target_pipeline_id": p.id,
+            },
+        )
+        assert r.status_code == 422
+
+    def test_create_rule_missing_pipeline(self, client):
+        """POST referencing a nonexistent pipeline returns 404."""
+        r = client.post(
+            "/api/routing-rules",
+            json={
+                "name": "No pipeline",
+                "field": "file_type",
+                "operator": "equals",
+                "value": "application/pdf",
+                "target_pipeline_id": 99999,
+            },
+        )
+        assert r.status_code == 404
+
+    def test_list_rules(self, client, db_session):
+        """GET /api/routing-rules returns the user's rules."""
+        p = _make_pipeline(db_session, name="Pipeline")
+        _make_rule(db_session, p.id, name="Rule 1", position=0, owner_id="anonymous")
+        _make_rule(db_session, p.id, name="Rule 2", position=1, owner_id="anonymous")
+
+        r = client.get("/api/routing-rules")
+        assert r.status_code == 200
+        data = r.json()
+        assert len(data) >= 2
+        names = [d["name"] for d in data]
+        assert "Rule 1" in names
+        assert "Rule 2" in names
+
+    def test_get_rule(self, client, db_session):
+        """GET /api/routing-rules/{id} returns a specific rule."""
+        p = _make_pipeline(db_session, name="Pipeline")
+        rule = _make_rule(db_session, p.id, name="My Rule", owner_id="anonymous")
+
+        r = client.get(f"/api/routing-rules/{rule.id}")
+        assert r.status_code == 200
+        assert r.json()["name"] == "My Rule"
+
+    def test_get_rule_not_found(self, client):
+        """GET /api/routing-rules/{id} returns 404 for nonexistent rule."""
+        r = client.get("/api/routing-rules/99999")
+        assert r.status_code == 404
+
+    def test_update_rule(self, client, db_session):
+        """PUT /api/routing-rules/{id} updates a rule."""
+        p = _make_pipeline(db_session, name="Pipeline")
+        rule = _make_rule(db_session, p.id, name="Old Name", owner_id="anonymous")
+
+        r = client.put(
+            f"/api/routing-rules/{rule.id}",
+            json={"name": "New Name", "operator": "contains"},
+        )
+        assert r.status_code == 200
+        data = r.json()
+        assert data["name"] == "New Name"
+        assert data["operator"] == "contains"
+
+    def test_update_rule_invalid_field(self, client, db_session):
+        """PUT with invalid field returns 422."""
+        p = _make_pipeline(db_session, name="Pipeline")
+        rule = _make_rule(db_session, p.id, owner_id="anonymous")
+
+        r = client.put(f"/api/routing-rules/{rule.id}", json={"field": "bad_field"})
+        assert r.status_code == 422
+
+    def test_update_rule_invalid_operator(self, client, db_session):
+        """PUT with invalid operator returns 422."""
+        p = _make_pipeline(db_session, name="Pipeline")
+        rule = _make_rule(db_session, p.id, owner_id="anonymous")
+
+        r = client.put(f"/api/routing-rules/{rule.id}", json={"operator": "bad_op"})
+        assert r.status_code == 422
+
+    def test_update_rule_missing_pipeline(self, client, db_session):
+        """PUT with nonexistent target pipeline returns 404."""
+        p = _make_pipeline(db_session, name="Pipeline")
+        rule = _make_rule(db_session, p.id, owner_id="anonymous")
+
+        r = client.put(f"/api/routing-rules/{rule.id}", json={"target_pipeline_id": 99999})
+        assert r.status_code == 404
+
+    def test_delete_rule(self, client, db_session):
+        """DELETE /api/routing-rules/{id} removes a rule."""
+        p = _make_pipeline(db_session, name="Pipeline")
+        rule = _make_rule(db_session, p.id, owner_id="anonymous")
+
+        r = client.delete(f"/api/routing-rules/{rule.id}")
+        assert r.status_code == 204
+
+        # Verify it's gone
+        r = client.get(f"/api/routing-rules/{rule.id}")
+        assert r.status_code == 404
+
+    def test_delete_rule_not_found(self, client):
+        """DELETE for nonexistent rule returns 404."""
+        r = client.delete("/api/routing-rules/99999")
+        assert r.status_code == 404
+
+
+# ===========================================================================
+# API tests – reorder
+# ===========================================================================
+
+
+@pytest.mark.unit
+class TestReorderRoutingRules:
+    """Tests for the PUT /api/routing-rules/reorder endpoint."""
+
+    def test_reorder_rules(self, client, db_session):
+        """Reordering updates the position of rules."""
+        p = _make_pipeline(db_session, name="Pipeline")
+        r1 = _make_rule(db_session, p.id, name="A", position=0, owner_id="anonymous")
+        r2 = _make_rule(db_session, p.id, name="B", position=1, owner_id="anonymous")
+
+        r = client.put(
+            "/api/routing-rules/reorder",
+            json={"rule_ids": [r2.id, r1.id]},
+        )
+        assert r.status_code == 200
+        data = r.json()
+        assert data[0]["id"] == r2.id
+        assert data[0]["position"] == 0
+        assert data[1]["id"] == r1.id
+        assert data[1]["position"] == 1
+
+    def test_reorder_invalid_ids(self, client, db_session):
+        """Reorder with invalid IDs returns 422."""
+        r = client.put(
+            "/api/routing-rules/reorder",
+            json={"rule_ids": [99999]},
+        )
+        assert r.status_code == 422
+
+
+# ===========================================================================
+# API tests – evaluate (dry-run)
+# ===========================================================================
+
+
+@pytest.mark.unit
+class TestEvaluateEndpoint:
+    """Tests for the POST /api/routing-rules/evaluate dry-run endpoint."""
+
+    def test_evaluate_match(self, client, db_session):
+        """Evaluate returns the matching rule and target pipeline."""
+        p = _make_pipeline(db_session, name="Invoice Pipeline")
+        _make_rule(
+            db_session,
+            p.id,
+            name="PDF Route",
+            field="file_type",
+            operator="equals",
+            value="application/pdf",
+            owner_id="anonymous",
+        )
+
+        r = client.post(
+            "/api/routing-rules/evaluate",
+            json={"file_type": "application/pdf"},
+        )
+        assert r.status_code == 200
+        data = r.json()
+        assert data["matched"] is True
+        assert data["rule"]["name"] == "PDF Route"
+        assert data["target_pipeline"]["id"] == p.id
+
+    def test_evaluate_no_match(self, client, db_session):
+        """Evaluate returns matched=False when no rule applies."""
+        p = _make_pipeline(db_session, name="Pipeline")
+        _make_rule(
+            db_session,
+            p.id,
+            name="PNG Route",
+            field="file_type",
+            operator="equals",
+            value="image/png",
+            owner_id="anonymous",
+        )
+
+        r = client.post(
+            "/api/routing-rules/evaluate",
+            json={"file_type": "application/pdf"},
+        )
+        assert r.status_code == 200
+        data = r.json()
+        assert data["matched"] is False
+        assert data["rule"] is None
+
+    def test_evaluate_with_metadata(self, client, db_session):
+        """Evaluate with metadata.* fields works."""
+        p = _make_pipeline(db_session, name="Invoice Pipeline")
+        _make_rule(
+            db_session,
+            p.id,
+            name="Invoice Route",
+            field="metadata.sender",
+            operator="contains",
+            value="acme",
+            owner_id="anonymous",
+        )
+
+        r = client.post(
+            "/api/routing-rules/evaluate",
+            json={"metadata": {"sender": "Acme Corporation"}},
+        )
+        assert r.status_code == 200
+        data = r.json()
+        assert data["matched"] is True
+
+    def test_evaluate_with_size(self, client, db_session):
+        """Evaluate with size comparisons works."""
+        p = _make_pipeline(db_session, name="Large Pipeline")
+        _make_rule(
+            db_session,
+            p.id,
+            name="Large Files",
+            field="size",
+            operator="gt",
+            value="1000000",
+            owner_id="anonymous",
+        )
+
+        r = client.post(
+            "/api/routing-rules/evaluate",
+            json={"size": 2000000},
+        )
+        assert r.status_code == 200
+        assert r.json()["matched"] is True
+
+
+# ===========================================================================
+# API tests – metadata field validation
+# ===========================================================================
+
+
+@pytest.mark.unit
+class TestFieldValidation:
+    """Tests for field validation in routing rule creation."""
+
+    def test_metadata_prefix_accepted(self, client, db_session):
+        """Fields with 'metadata.' prefix are valid."""
+        p = _make_pipeline(db_session, name="Pipeline")
+        r = client.post(
+            "/api/routing-rules",
+            json={
+                "name": "Metadata Rule",
+                "field": "metadata.sender",
+                "operator": "equals",
+                "value": "test",
+                "target_pipeline_id": p.id,
+            },
+        )
+        assert r.status_code == 201
+
+    def test_bare_metadata_rejected(self, client, db_session):
+        """Just 'metadata.' without a key suffix is invalid."""
+        p = _make_pipeline(db_session, name="Pipeline")
+        r = client.post(
+            "/api/routing-rules",
+            json={
+                "name": "Bad Metadata",
+                "field": "metadata.",
+                "operator": "equals",
+                "value": "test",
+                "target_pipeline_id": p.id,
+            },
+        )
+        assert r.status_code == 422
+
+    @pytest.mark.parametrize("field", sorted(BUILTIN_FIELDS))
+    def test_builtin_fields_accepted(self, client, db_session, field):
+        """All built-in fields are accepted."""
+        p = _make_pipeline(db_session, name=f"Pipeline for {field}")
+        r = client.post(
+            "/api/routing-rules",
+            json={
+                "name": f"Rule for {field}",
+                "field": field,
+                "operator": "equals",
+                "value": "test",
+                "target_pipeline_id": p.id,
+            },
+        )
+        assert r.status_code == 201
+
+
+# ===========================================================================
+# API tests – auto-position
+# ===========================================================================
+
+
+@pytest.mark.unit
+class TestAutoPosition:
+    """Tests for automatic position assignment."""
+
+    def test_auto_position_increments(self, client, db_session):
+        """Rules created without position get incrementing positions."""
+        p = _make_pipeline(db_session, name="Pipeline")
+
+        r1 = client.post(
+            "/api/routing-rules",
+            json={
+                "name": "First",
+                "field": "file_type",
+                "operator": "equals",
+                "value": "application/pdf",
+                "target_pipeline_id": p.id,
+            },
+        )
+        r2 = client.post(
+            "/api/routing-rules",
+            json={
+                "name": "Second",
+                "field": "file_type",
+                "operator": "equals",
+                "value": "image/png",
+                "target_pipeline_id": p.id,
+            },
+        )
+        assert r1.status_code == 201
+        assert r2.status_code == 201
+        assert r2.json()["position"] > r1.json()["position"]