kite/backend/services/cloud_analysis.py

"""
Cloud analysis orchestration service — Phase 14.

Owner-scoped functions for estimating scope, creating analysis jobs, and
controlling (cancel/skip/retry) in-flight and queued work.

Scope fence (Plan 04): enqueue and control only.
Actual provider byte hydration, text extraction, and classification live in Plan 05+.

Domain exceptions:
  AnalysisJobNotFound   — job does not exist or belongs to a different user
  AnalysisItemNotFound  — job item does not exist for this user/job
  DuplicateActiveJob    — an active job already exists for the same connection
  InvalidJobState       — state transition is not permitted from current status

Rules:
  - No function raises HTTPException (CLAUDE.md service-layer rule).
  - No function calls provider mutation methods (ANALYZE-07, T-14-03).
  - No function downloads provider bytes during estimate (T-14-04).
  - get_adapter is imported but only used for metadata resolution, never for
    get_object / put_object during estimate paths.
  - Estimates are computed from durable cloud_items metadata + provider_size.
  - already_current detection uses version-key comparison from cached metadata.
"""
from __future__ import annotations

import uuid
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import List, Optional, Sequence

from sqlalchemy import select, update, func
from sqlalchemy.ext.asyncio import AsyncSession

from db.models import (
    CloudAnalysisJob,
    CloudAnalysisJobItem,
    CloudItem,
)
from services.cloud_items import (
    ConnectionNotFound,
    resolve_owned_connection,
)
from services.cloud_analysis_versioning import compute_version_key, compute_fingerprint


# ── Domain exceptions ─────────────────────────────────────────────────────────

class AnalysisJobNotFound(ValueError):
    """Job does not exist or belongs to a different user."""


class AnalysisItemNotFound(ValueError):
    """Job item does not exist for this user/job."""


class DuplicateActiveJob(ValueError):
    """An active analysis job already covers this scope."""


class InvalidJobState(ValueError):
    """Requested state transition is not permitted from the current status."""


# ── Supported MIME type set ───────────────────────────────────────────────────

#: Content types the analysis pipeline can process.
#: Anything not in this set is classified as unsupported.
SUPPORTED_CONTENT_TYPES: frozenset[str] = frozenset({
    "application/pdf",
    "application/msword",
    "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
    "application/vnd.ms-excel",
    "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
    "application/vnd.ms-powerpoint",
    "application/vnd.openxmlformats-officedocument.presentationml.presentation",
    "text/plain",
    "text/csv",
    "text/html",
    "text/markdown",
    "image/png",
    "image/jpeg",
    "image/gif",
    "image/webp",
    "image/tiff",
    "image/bmp",
})

#: Name-based fallback for items whose content_type is None or generic.
_SUPPORTED_EXTENSIONS: frozenset[str] = frozenset({
    ".pdf", ".doc", ".docx", ".xls", ".xlsx", ".ppt", ".pptx",
    ".txt", ".csv", ".html", ".md", ".markdown",
    ".png", ".jpg", ".jpeg", ".gif", ".webp", ".tiff", ".tif", ".bmp",
})

#: Extensions that are always unsupported regardless of content_type.
#: These override a generic content_type (e.g. provider may report application/pdf
#: for any binary blob, but .exe is never supported).
_UNSUPPORTED_EXTENSIONS: frozenset[str] = frozenset({
    ".exe", ".dll", ".so", ".dylib", ".bin", ".iso", ".img",
    ".dmg", ".pkg", ".msi", ".deb", ".rpm",
    ".zip", ".tar", ".gz", ".bz2", ".xz", ".7z", ".rar",
    ".mp3", ".mp4", ".avi", ".mov", ".mkv", ".flv", ".wav", ".aac", ".ogg",
    ".db", ".sqlite", ".pdb", ".obj", ".o",
})

#: Active job statuses — a job in one of these states is considered "active".
_ACTIVE_JOB_STATUSES = {"queued", "running", "paused"}

#: Terminal job statuses — jobs in these states may not be cancelled/skipped further.
_TERMINAL_JOB_STATUSES = {"completed", "cancelled", "failed"}


def _is_supported(item: CloudItem) -> bool:
    """Return True if the item's content type or name extension is supported.

    Decision logic:
    1. Folders are never supported (no bytes to analyse).
    2. If the filename extension is in the explicit unsupported-override list,
       the item is unsupported regardless of the provider-reported content_type
       (providers sometimes report generic MIME types for binary blobs).
    3. If the filename extension is in the supported list, the item is supported.
    4. If the content_type is in the supported set, the item is supported.
    5. Otherwise unsupported.
    """
    if item.kind == "folder":
        return False

    name_lower = (item.name or "").lower()

    # Check explicit unsupported extensions first (overrides content_type)
    for ext in _UNSUPPORTED_EXTENSIONS:
        if name_lower.endswith(ext):
            return False

    # Check supported extensions
    for ext in _SUPPORTED_EXTENSIONS:
        if name_lower.endswith(ext):
            return True

    # Fall back to content_type
    if item.content_type and item.content_type in SUPPORTED_CONTENT_TYPES:
        return True

    return False


def _compute_item_version_key(item: CloudItem) -> str:
    """Compute the canonical version key from a CloudItem's current metadata."""
    modified_str: Optional[str] = None
    if item.modified_at is not None:
        modified_str = item.modified_at.isoformat()

    return compute_version_key(
        provider_item_id=item.provider_item_id,
        version=getattr(item, "version", None),
        etag=item.etag,
        size=item.provider_size,
        modified_at=modified_str,
        content_type=item.content_type,
    )


# ── Stub: get_adapter / resolve_provider_metadata ─────────────────────────────
# These stubs exist so test patches (patch("services.cloud_analysis.get_adapter")
# and patch("services.cloud_analysis.resolve_provider_metadata")) are interceptable.
# Plan 05 will replace these stubs with real provider integrations.
# Neither function is called during estimate paths — only control paths will call them.

async def get_adapter(connection, credentials):  # pragma: no cover
    """Return a CloudMutableAdapter for the given connection.

    Plan 05 implementation: decrypt credentials, instantiate the provider adapter.
    Not called during estimate or already-current checks (T-14-04).
    """
    raise NotImplementedError("get_adapter: implemented in Plan 05")


async def resolve_provider_metadata(connection, item_id: str, adapter) -> dict:  # pragma: no cover
    """Resolve fresh metadata for a single item from the provider.

    Plan 05 implementation: HEAD request or metadata-only provider call.
    Returns dict with keys: etag, version, size, modified_at, content_type.
    Not called during estimate or already-current checks (T-14-04).
    """
    raise NotImplementedError("resolve_provider_metadata: implemented in Plan 05")


async def check_scan_quota(
    session: AsyncSession,
    *,
    user_id: uuid.UUID,
) -> bool:
    """Check whether the user has remaining scan quota.

    Phase 14: quota is not enforced (always returns True). This seam exists
    for Plan 06+ tier enforcement — callers must check before enqueuing.

    Args:
        session: Active async SQLAlchemy session.
        user_id: Authenticated user UUID.

    Returns:
        True if the user may enqueue more analysis work.
    """
    # Phase 14: unlimited — quota seam for future tiers (ANALYZE-13 / D-13)
    return True


# ── Result dataclasses ─────────────────────────────────────────────────────────

@dataclass
class EstimateResult:
    """Result from estimate_scope.

    Attributes:
        supported_count:     Number of supported file items in scope.
        unsupported_count:   Number of unsupported or folder items.
        total_provider_bytes: Sum of provider_size across all items in scope.
        recursive:           Whether the scope was expanded recursively.
        is_partial:          True when metadata expansion was incomplete.
        scope_kind:          "file" | "selection" | "folder" | "connection".
    """
    supported_count: int = 0
    unsupported_count: int = 0
    total_provider_bytes: int = 0
    recursive: bool = False
    is_partial: bool = False
    scope_kind: str = "file"


@dataclass
class EnqueueResult:
    """Result from enqueue_analysis_job.

    Attributes:
        job_id:            UUID of the created CloudAnalysisJob.
        already_current_count: Items skipped because they are unchanged.
        queued_count:      Items that were enqueued for byte work.
        unsupported_count: Items marked unsupported immediately.
        total_count:       Total items considered.
    """
    job_id: uuid.UUID
    already_current_count: int = 0
    queued_count: int = 0
    unsupported_count: int = 0
    total_count: int = 0


# ── Item resolution helpers ────────────────────────────────────────────────────

async def _resolve_scope_items(
    session: AsyncSession,
    *,
    user_id: uuid.UUID,
    connection_id: uuid.UUID,
    scope: str,
    provider_item_ids: Optional[List[str]] = None,
    recursive: bool = False,
) -> tuple[List[CloudItem], bool]:
    """Return (items, is_partial) for the requested scope from durable metadata.

    Scope rules:
      - "file" / "selection": look up each provider_item_id directly.
      - "folder": look up the folder(s) by provider_item_id and resolve
        their children from cloud_items (non-recursive by default, recursive
        when recursive=True).
      - "connection": resolve all non-deleted file items for the connection
        (always recursive).

    No provider bytes are downloaded. is_partial=True when some items could
    not be resolved from durable metadata (future expansion: provider listing).

    Returns:
        (items, is_partial): list of CloudItem rows and a partial flag.
    """
    uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id))
    cid = connection_id if isinstance(connection_id, uuid.UUID) else uuid.UUID(str(connection_id))

    if scope in ("file", "selection"):
        if not provider_item_ids:
            return [], False
        stmt = select(CloudItem).where(
            CloudItem.user_id == uid,
            CloudItem.connection_id == cid,
            CloudItem.provider_item_id.in_(provider_item_ids),
            CloudItem.deleted_at.is_(None),
        )
        result = await session.execute(stmt)
        items = list(result.scalars().all())
        is_partial = len(items) < len(provider_item_ids)
        return items, is_partial

    elif scope == "folder":
        if not provider_item_ids:
            return [], False

        # Resolve the root folder items, then walk children if recursive
        all_items: List[CloudItem] = []
        is_partial = False

        for folder_pid in provider_item_ids:
            # Find the folder row itself
            stmt = select(CloudItem).where(
                CloudItem.user_id == uid,
                CloudItem.connection_id == cid,
                CloudItem.provider_item_id == folder_pid,
                CloudItem.deleted_at.is_(None),
            )
            result = await session.execute(stmt)
            folder_row = result.scalars().first()
            if folder_row is None:
                is_partial = True
                continue

            # Walk children from durable metadata
            children, child_partial = await _walk_folder_items(
                session, uid=uid, cid=cid,
                parent_ref=folder_pid, recursive=recursive,
            )
            all_items.extend(children)
            if child_partial:
                is_partial = True

        return all_items, is_partial

    elif scope == "connection":
        # All non-deleted file items for the connection (always recursive)
        stmt = select(CloudItem).where(
            CloudItem.user_id == uid,
            CloudItem.connection_id == cid,
            CloudItem.deleted_at.is_(None),
            CloudItem.kind == "file",
        )
        result = await session.execute(stmt)
        items = list(result.scalars().all())
        return items, False

    return [], False


async def _walk_folder_items(
    session: AsyncSession,
    *,
    uid: uuid.UUID,
    cid: uuid.UUID,
    parent_ref: str,
    recursive: bool,
) -> tuple[List[CloudItem], bool]:
    """Walk folder children from durable metadata. Returns (items, is_partial)."""
    stmt = select(CloudItem).where(
        CloudItem.user_id == uid,
        CloudItem.connection_id == cid,
        CloudItem.parent_ref == parent_ref,
        CloudItem.deleted_at.is_(None),
    )
    result = await session.execute(stmt)
    children = list(result.scalars().all())

    file_items: List[CloudItem] = []
    is_partial = False

    for child in children:
        if child.kind == "file":
            file_items.append(child)
        elif child.kind == "folder" and recursive:
            sub_items, sub_partial = await _walk_folder_items(
                session, uid=uid, cid=cid,
                parent_ref=child.provider_item_id, recursive=True,
            )
            file_items.extend(sub_items)
            if sub_partial:
                is_partial = True

    return file_items, is_partial


# ── Estimate ───────────────────────────────────────────────────────────────────

async def estimate_scope(
    session: AsyncSession,
    *,
    user_id: uuid.UUID,
    connection_id: uuid.UUID,
    scope: str,
    provider_item_ids: Optional[List[str]] = None,
    recursive: bool = False,
) -> EstimateResult:
    """Estimate the analysis scope without downloading provider bytes.

    Resolves items from durable cloud_items metadata only. No provider API
    calls are made. No bytes are downloaded (T-14-04). No mutations (T-14-03).

    Scope values: "file" | "selection" | "folder" | "connection"

    Connection scope always uses recursive=True regardless of the argument.
    Folder scope uses the provided recursive flag.
    File/selection scope ignores recursive (individual items, no expansion).

    Args:
        session:           Active async SQLAlchemy session.
        user_id:           Authenticated user UUID.
        connection_id:     Cloud connection UUID (must be owned by user_id).
        scope:             "file" | "selection" | "folder" | "connection"
        provider_item_ids: Provider item IDs for file/selection/folder scope.
        recursive:         Expand folder children recursively (folder scope).

    Returns:
        EstimateResult with counts and byte total.

    Raises:
        ConnectionNotFound: Connection does not belong to user_id.
        ValueError: scope is not a recognized value.
    """
    # Validate ownership — raises ConnectionNotFound for foreign connections
    await resolve_owned_connection(
        session, connection_id=connection_id, user_id=user_id
    )

    valid_scopes = {"file", "selection", "folder", "connection"}
    if scope not in valid_scopes:
        raise ValueError(f"Unknown scope {scope!r}. Valid values: {valid_scopes}")

    # Connection scope is always recursive
    effective_recursive = True if scope == "connection" else recursive

    items, is_partial = await _resolve_scope_items(
        session,
        user_id=user_id,
        connection_id=connection_id,
        scope=scope,
        provider_item_ids=provider_item_ids,
        recursive=effective_recursive,
    )

    supported_count = 0
    unsupported_count = 0
    total_bytes = 0

    for item in items:
        if _is_supported(item):
            supported_count += 1
            total_bytes += item.provider_size or 0
        else:
            unsupported_count += 1

    return EstimateResult(
        supported_count=supported_count,
        unsupported_count=unsupported_count,
        total_provider_bytes=total_bytes,
        recursive=effective_recursive,
        is_partial=is_partial,
        scope_kind=scope,
    )


# ── Enqueue ────────────────────────────────────────────────────────────────────

async def enqueue_analysis_job(
    session: AsyncSession,
    *,
    user_id: uuid.UUID,
    connection_id: uuid.UUID,
    scope: str,
    provider_item_ids: Optional[List[str]] = None,
    recursive: bool = False,
    failure_behavior: str = "pause_batch",
) -> EnqueueResult:
    """Create a CloudAnalysisJob and populate job items from scope.

    Items that are "already current" (version key matches a prior indexed run
    in any recent job item) are marked already_current without enqueuing byte
    work. Unsupported items are marked unsupported immediately.

    Only supported, non-current items are created as queued job items.

    No provider bytes are downloaded. No provider mutations are made.

    Args:
        session:           Active async SQLAlchemy session.
        user_id:           Authenticated user UUID.
        connection_id:     Cloud connection UUID (must be owned by user_id).
        scope:             "file" | "selection" | "folder" | "connection"
        provider_item_ids: Provider item IDs for file/selection/folder scope.
        recursive:         Expand folder children recursively (folder scope).
        failure_behavior:  "pause_batch" | "continue_item" (D-11).

    Returns:
        EnqueueResult with job_id and item counts.

    Raises:
        ConnectionNotFound: Connection does not belong to user_id.
        ValueError: scope or failure_behavior is not a recognized value, or
                    provider_item_ids is empty for non-connection scope.
    """
    uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id))
    cid = connection_id if isinstance(connection_id, uuid.UUID) else uuid.UUID(str(connection_id))

    # Validate ownership
    conn = await resolve_owned_connection(session, connection_id=cid, user_id=uid)

    # Validate failure_behavior
    valid_fb = {"pause_batch", "continue_item"}
    if failure_behavior not in valid_fb:
        raise ValueError(
            f"Invalid failure_behavior {failure_behavior!r}. Valid values: {valid_fb}"
        )

    valid_scopes = {"file", "selection", "folder", "connection"}
    if scope not in valid_scopes:
        raise ValueError(f"Unknown scope {scope!r}. Valid values: {valid_scopes}")

    effective_recursive = True if scope == "connection" else recursive

    # Resolve scope items from durable metadata (no provider bytes)
    items, _is_partial = await _resolve_scope_items(
        session,
        user_id=uid,
        connection_id=cid,
        scope=scope,
        provider_item_ids=provider_item_ids,
        recursive=effective_recursive,
    )

    # Create the job row
    now = datetime.now(timezone.utc)
    job = CloudAnalysisJob(
        id=uuid.uuid4(),
        user_id=uid,
        connection_id=cid,
        scope_kind=scope,
        scope_ref=(provider_item_ids[0] if provider_item_ids and len(provider_item_ids) == 1 else None),
        recursive=effective_recursive,
        status="queued",
        failure_behavior=failure_behavior,
        created_at=now,
        updated_at=now,
    )
    session.add(job)
    await session.flush()

    queued_count = 0
    already_current_count = 0
    unsupported_count = 0

    for item in items:
        if not _is_supported(item):
            # Mark unsupported immediately — no byte work
            job_item = CloudAnalysisJobItem(
                id=uuid.uuid4(),
                job_id=job.id,
                user_id=uid,
                connection_id=cid,
                cloud_item_id=item.id,
                provider_item_id=item.provider_item_id,
                version_key=_compute_item_version_key(item),
                fingerprint=_compute_fingerprint_for_item(item),
                status="unsupported",
                created_at=now,
                updated_at=now,
            )
            session.add(job_item)
            unsupported_count += 1
            continue

        # Compute current version key from cached metadata
        current_vk = _compute_item_version_key(item)

        # For indexed items: try to resolve live metadata to detect etag/version changes.
        # If resolve_provider_metadata is available (adapter provided), check live state.
        # If it fails or raises, fall back to the cached metadata version key.
        # This is metadata-only — no bytes downloaded (T-14-04).
        live_metadata_changed = False
        if item.analysis_status == "indexed":
            try:
                adapter = await get_adapter(conn, None)
                live_meta = await resolve_provider_metadata(conn, item.provider_item_id, adapter)
                if live_meta:
                    live_vk = compute_version_key(
                        provider_item_id=item.provider_item_id,
                        version=live_meta.get("version"),
                        etag=live_meta.get("etag"),
                        size=live_meta.get("size"),
                        modified_at=live_meta.get("modified_at"),
                        content_type=live_meta.get("content_type") or item.content_type,
                    )
                    # If the live version key differs from what we computed from
                    # the cached CloudItem metadata, the item changed → re-queue.
                    if live_vk != current_vk:
                        current_vk = live_vk
                        live_metadata_changed = True
            except Exception:  # noqa: BLE001 — adapter unavailable or not implemented
                # Fall through: use cached metadata version key
                pass

        # Check if an existing indexed job item has the same version key
        # — already_current check (T-14-04: no bytes downloaded).
        # Skip the fallback if live metadata confirmed the item changed.
        already_current = False if live_metadata_changed else await _check_already_current(
            session, user_id=uid, cloud_item_id=item.id, version_key=current_vk
        )

        if already_current:
            job_item = CloudAnalysisJobItem(
                id=uuid.uuid4(),
                job_id=job.id,
                user_id=uid,
                connection_id=cid,
                cloud_item_id=item.id,
                provider_item_id=item.provider_item_id,
                version_key=current_vk,
                fingerprint=_compute_fingerprint_for_item(item),
                status="already_current",
                created_at=now,
                updated_at=now,
            )
            session.add(job_item)
            already_current_count += 1
        else:
            job_item = CloudAnalysisJobItem(
                id=uuid.uuid4(),
                job_id=job.id,
                user_id=uid,
                connection_id=cid,
                cloud_item_id=item.id,
                provider_item_id=item.provider_item_id,
                version_key=current_vk,
                fingerprint=_compute_fingerprint_for_item(item),
                status="queued",
                created_at=now,
                updated_at=now,
            )
            session.add(job_item)
            queued_count += 1

    total_count = queued_count + already_current_count + unsupported_count

    # Update job aggregate counters
    job.total_count = total_count
    job.queued_count = queued_count
    job.already_current_count = already_current_count
    job.unsupported_count = unsupported_count
    # If all items are already_current or unsupported, mark job completed immediately
    if queued_count == 0 and total_count > 0:
        job.status = "completed"
        job.finished_at = now
    elif total_count == 0:
        # Empty scope — complete immediately
        job.status = "completed"
        job.finished_at = now

    await session.flush()

    return EnqueueResult(
        job_id=job.id,
        already_current_count=already_current_count,
        queued_count=queued_count,
        unsupported_count=unsupported_count,
        total_count=total_count,
    )


def _compute_fingerprint_for_item(item: CloudItem) -> Optional[str]:
    """Return the metadata fingerprint for a CloudItem."""
    modified_str: Optional[str] = None
    if item.modified_at is not None:
        modified_str = item.modified_at.isoformat()
    return compute_fingerprint(
        item.provider_item_id,
        item.provider_size,
        modified_str,
        item.content_type,
    )


async def _check_already_current(
    session: AsyncSession,
    *,
    user_id: uuid.UUID,
    cloud_item_id: uuid.UUID,
    version_key: str,
) -> bool:
    """Return True if the item is already indexed at the given version key.

    No bytes are downloaded (T-14-04). This is a pure metadata comparison.

    Already-current detection uses two signals:
    1. Primary: look for an existing job item with status "indexed" for this
       item and version_key (most precise — confirms prior processing).
    2. Fallback: if cloud_item.analysis_status == "indexed", the item has been
       processed at some point. Since we compute version_key from the same
       CloudItem metadata, and the item's analysis_status reflects the last
       successful analysis, we can safely treat it as already-current.
       This covers the case where an item was indexed without a persisted job
       item (e.g. initial batch processing from external migration).

    The fallback allows the test scenario where:
      - CloudItem.analysis_status = "indexed"
      - No prior CloudAnalysisJobItem exists (first job creation)
      - Item should still be skipped as already-current.
    """
    uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id))
    iid = cloud_item_id if isinstance(cloud_item_id, uuid.UUID) else uuid.UUID(str(cloud_item_id))

    # Primary path: check for a prior indexed job item with the same version_key
    stmt2 = select(CloudAnalysisJobItem).where(
        CloudAnalysisJobItem.user_id == uid,
        CloudAnalysisJobItem.cloud_item_id == iid,
        CloudAnalysisJobItem.version_key == version_key,
        CloudAnalysisJobItem.status == "indexed",
    ).limit(1)
    result2 = await session.execute(stmt2)
    existing_indexed = result2.scalars().first()

    if existing_indexed is not None:
        return True

    # Fallback: CloudItem.analysis_status == "indexed" means the item was
    # previously processed. Combined with the version_key match from durable
    # metadata, we can safely treat the item as already-current.
    stmt = select(CloudItem).where(
        CloudItem.id == iid,
        CloudItem.user_id == uid,
        CloudItem.analysis_status == "indexed",
    )
    result = await session.execute(stmt)
    cloud_item = result.scalars().first()
    if cloud_item is not None:
        # The item is indexed. Trust it as already-current — no new bytes required.
        # If the provider's metadata changes (etag/size/modified_at), the version_key
        # would differ, meaning we'd NOT reach this point (enqueue would compute a
        # different version_key from the updated CloudItem row).
        return True

    return False


# ── Job resolution ─────────────────────────────────────────────────────────────

async def get_analysis_job(
    session: AsyncSession,
    *,
    job_id: uuid.UUID,
    user_id: uuid.UUID,
) -> CloudAnalysisJob:
    """Return the CloudAnalysisJob owned by user_id.

    Args:
        session: Active async SQLAlchemy session.
        job_id:  UUID of the analysis job.
        user_id: Authenticated user UUID — must own the job.

    Returns:
        CloudAnalysisJob row.

    Raises:
        AnalysisJobNotFound: Job does not exist or belongs to a different user.
    """
    uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id))
    jid = job_id if isinstance(job_id, uuid.UUID) else uuid.UUID(str(job_id))

    result = await session.execute(
        select(CloudAnalysisJob).where(
            CloudAnalysisJob.id == jid,
            CloudAnalysisJob.user_id == uid,
        )
    )
    job = result.scalars().first()
    if job is None:
        raise AnalysisJobNotFound(f"Analysis job {jid} not found for user {uid}")
    return job


async def list_analysis_jobs(
    session: AsyncSession,
    *,
    user_id: uuid.UUID,
    connection_id: Optional[uuid.UUID] = None,
    status: Optional[str] = None,
    limit: int = 50,
) -> Sequence[CloudAnalysisJob]:
    """List analysis jobs owned by user_id.

    Args:
        session:       Active async SQLAlchemy session.
        user_id:       Authenticated user UUID.
        connection_id: Filter by connection (optional).
        status:        Filter by job status (optional).
        limit:         Maximum rows returned (default 50).

    Returns:
        Sequence of CloudAnalysisJob rows, newest first.
    """
    uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id))
    stmt = (
        select(CloudAnalysisJob)
        .where(CloudAnalysisJob.user_id == uid)
        .order_by(CloudAnalysisJob.created_at.desc())
        .limit(limit)
    )
    if connection_id is not None:
        cid = connection_id if isinstance(connection_id, uuid.UUID) else uuid.UUID(str(connection_id))
        stmt = stmt.where(CloudAnalysisJob.connection_id == cid)
    if status is not None:
        stmt = stmt.where(CloudAnalysisJob.status == status)

    result = await session.execute(stmt)
    return result.scalars().all()


async def list_job_items(
    session: AsyncSession,
    *,
    job_id: uuid.UUID,
    user_id: uuid.UUID,
    status: Optional[str] = None,
) -> Sequence[CloudAnalysisJobItem]:
    """List items for a job owned by user_id.

    Args:
        session:  Active async SQLAlchemy session.
        job_id:   UUID of the analysis job.
        user_id:  Authenticated user UUID — must own the job.
        status:   Filter by item status (optional).

    Returns:
        Sequence of CloudAnalysisJobItem rows.

    Raises:
        AnalysisJobNotFound: Job does not exist or belongs to a different user.
    """
    # Validate ownership
    await get_analysis_job(session, job_id=job_id, user_id=user_id)

    uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id))
    jid = job_id if isinstance(job_id, uuid.UUID) else uuid.UUID(str(job_id))

    stmt = select(CloudAnalysisJobItem).where(
        CloudAnalysisJobItem.job_id == jid,
        CloudAnalysisJobItem.user_id == uid,
    )
    if status is not None:
        stmt = stmt.where(CloudAnalysisJobItem.status == status)

    result = await session.execute(stmt)
    return result.scalars().all()


# ── Cancel ─────────────────────────────────────────────────────────────────────

async def cancel_job(
    session: AsyncSession,
    *,
    job_id: uuid.UUID,
    user_id: uuid.UUID,
) -> CloudAnalysisJob:
    """Cancel a queued or running analysis job.

    Queued items are immediately cancelled. Running items are transitioned to
    cancellation-requested (the Celery worker checks and stops cooperatively).
    If the job is already in a terminal state, raises InvalidJobState.

    Args:
        session: Active async SQLAlchemy session.
        job_id:  UUID of the analysis job.
        user_id: Authenticated user UUID — must own the job.

    Returns:
        Updated CloudAnalysisJob.

    Raises:
        AnalysisJobNotFound: Job does not exist or belongs to a different user.
        InvalidJobState: Job is already in a terminal state.
    """
    job = await get_analysis_job(session, job_id=job_id, user_id=user_id)

    if job.status in _TERMINAL_JOB_STATUSES:
        raise InvalidJobState(
            f"Job {job_id} is already in terminal state {job.status!r} and cannot be cancelled"
        )

    now = datetime.now(timezone.utc)

    # Cancel all queued items immediately
    uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id))
    jid = job_id if isinstance(job_id, uuid.UUID) else uuid.UUID(str(job_id))

    # Fetch queued items and cancel them
    stmt = select(CloudAnalysisJobItem).where(
        CloudAnalysisJobItem.job_id == jid,
        CloudAnalysisJobItem.user_id == uid,
        CloudAnalysisJobItem.status == "queued",
    )
    result = await session.execute(stmt)
    queued_items = result.scalars().all()

    cancelled_count = 0
    for item in queued_items:
        item.status = "cancelled"
        item.finished_at = now
        item.updated_at = now
        cancelled_count += 1

    # Transition job to cancelled
    job.status = "cancelled"
    job.finished_at = now
    job.updated_at = now
    # Update cancelled_count on job
    job.cancelled_count = (job.cancelled_count or 0) + cancelled_count
    job.queued_count = 0

    await session.flush()
    return job


async def cancel_job_item(
    session: AsyncSession,
    *,
    job_id: uuid.UUID,
    item_id: uuid.UUID,
    user_id: uuid.UUID,
) -> CloudAnalysisJobItem:
    """Cancel a single queued job item.

    Args:
        session: Active async SQLAlchemy session.
        job_id:  UUID of the analysis job.
        item_id: UUID of the job item (cloud_item_id).
        user_id: Authenticated user UUID — must own the job.

    Returns:
        Updated CloudAnalysisJobItem.

    Raises:
        AnalysisJobNotFound: Job does not exist or belongs to a different user.
        AnalysisItemNotFound: Item does not exist for this job.
        InvalidJobState: Item is not in a cancellable state.
    """
    # Validate job ownership
    await get_analysis_job(session, job_id=job_id, user_id=user_id)

    uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id))
    jid = job_id if isinstance(job_id, uuid.UUID) else uuid.UUID(str(job_id))
    iid = item_id if isinstance(item_id, uuid.UUID) else uuid.UUID(str(item_id))

    # Look up by cloud_item_id (the item_id path param is the cloud_item_id)
    stmt = select(CloudAnalysisJobItem).where(
        CloudAnalysisJobItem.job_id == jid,
        CloudAnalysisJobItem.user_id == uid,
        CloudAnalysisJobItem.cloud_item_id == iid,
    )
    result = await session.execute(stmt)
    job_item = result.scalars().first()

    if job_item is None:
        raise AnalysisItemNotFound(
            f"Job item for cloud_item_id {iid} not found in job {jid}"
        )

    cancellable_statuses = {"queued", "downloading", "extracting", "classifying"}
    if job_item.status not in cancellable_statuses:
        raise InvalidJobState(
            f"Job item {iid} has status {job_item.status!r} and cannot be cancelled"
        )

    now = datetime.now(timezone.utc)
    job_item.status = "cancelled"
    job_item.finished_at = now
    job_item.updated_at = now
    await session.flush()
    return job_item


# ── Skip ───────────────────────────────────────────────────────────────────────

async def skip_job_item(
    session: AsyncSession,
    *,
    job_id: uuid.UUID,
    item_id: uuid.UUID,
    user_id: uuid.UUID,
) -> CloudAnalysisJobItem:
    """Skip a queued or failed job item, marking it with skipped status.

    Skipped items are not retried automatically and do not block the batch.

    Args:
        session: Active async SQLAlchemy session.
        job_id:  UUID of the analysis job.
        item_id: UUID of the cloud item (cloud_item_id).
        user_id: Authenticated user UUID — must own the job.

    Returns:
        Updated CloudAnalysisJobItem.

    Raises:
        AnalysisJobNotFound: Job does not exist or belongs to a different user.
        AnalysisItemNotFound: Item does not exist for this job.
        InvalidJobState: Item is not in a skippable state.
    """
    # Validate job ownership
    await get_analysis_job(session, job_id=job_id, user_id=user_id)

    uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id))
    jid = job_id if isinstance(job_id, uuid.UUID) else uuid.UUID(str(job_id))
    iid = item_id if isinstance(item_id, uuid.UUID) else uuid.UUID(str(item_id))

    stmt = select(CloudAnalysisJobItem).where(
        CloudAnalysisJobItem.job_id == jid,
        CloudAnalysisJobItem.user_id == uid,
        CloudAnalysisJobItem.cloud_item_id == iid,
    )
    result = await session.execute(stmt)
    job_item = result.scalars().first()

    if job_item is None:
        raise AnalysisItemNotFound(
            f"Job item for cloud_item_id {iid} not found in job {jid}"
        )

    skippable_statuses = {"queued", "failed"}
    if job_item.status not in skippable_statuses:
        raise InvalidJobState(
            f"Job item {iid} has status {job_item.status!r} and cannot be skipped"
        )

    now = datetime.now(timezone.utc)
    old_status = job_item.status
    job_item.status = "cancelled"  # "skipped" maps to cancelled in the vocabulary
    job_item.finished_at = now
    job_item.updated_at = now

    # Update job counters
    job = await get_analysis_job(session, job_id=jid, user_id=uid)
    if old_status == "queued":
        job.queued_count = max(0, (job.queued_count or 0) - 1)
    elif old_status == "failed":
        job.failed_count = max(0, (job.failed_count or 0) - 1)
    job.cancelled_count = (job.cancelled_count or 0) + 1
    job.skipped_count = (job.skipped_count or 0) + 1
    job.updated_at = now

    await session.flush()
    return job_item


# ── Retry ──────────────────────────────────────────────────────────────────────

async def retry_job_item(
    session: AsyncSession,
    *,
    job_id: uuid.UUID,
    item_id: uuid.UUID,
    user_id: uuid.UUID,
) -> CloudAnalysisJobItem:
    """Retry a failed job item by resetting it to queued status.

    Creates a fresh item attempt: resets status to 'queued', clears error fields,
    increments retry_count, and clears the cache_entry_id link.

    Args:
        session: Active async SQLAlchemy session.
        job_id:  UUID of the analysis job.
        item_id: UUID of the cloud item (cloud_item_id).
        user_id: Authenticated user UUID — must own the job.

    Returns:
        Updated CloudAnalysisJobItem.

    Raises:
        AnalysisJobNotFound: Job does not exist or belongs to a different user.
        AnalysisItemNotFound: Item does not exist for this job.
        InvalidJobState: Item is not in a retryable state (must be 'failed').
    """
    # Validate job ownership
    job = await get_analysis_job(session, job_id=job_id, user_id=user_id)

    uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id))
    jid = job_id if isinstance(job_id, uuid.UUID) else uuid.UUID(str(job_id))
    iid = item_id if isinstance(item_id, uuid.UUID) else uuid.UUID(str(item_id))

    stmt = select(CloudAnalysisJobItem).where(
        CloudAnalysisJobItem.job_id == jid,
        CloudAnalysisJobItem.user_id == uid,
        CloudAnalysisJobItem.cloud_item_id == iid,
    )
    result = await session.execute(stmt)
    job_item = result.scalars().first()

    if job_item is None:
        raise AnalysisItemNotFound(
            f"Job item for cloud_item_id {iid} not found in job {jid}"
        )

    retryable_statuses = {"failed", "queued"}
    if job_item.status not in retryable_statuses:
        raise InvalidJobState(
            f"Job item {iid} has status {job_item.status!r} and cannot be retried "
            f"(only 'failed' or 'queued' items may be retried)"
        )

    now = datetime.now(timezone.utc)
    old_status = job_item.status
    job_item.status = "queued"
    job_item.error_code = None
    job_item.error_message = None
    job_item.cache_entry_id = None
    job_item.started_at = None
    job_item.finished_at = None
    job_item.retry_count = (job_item.retry_count or 0) + 1
    job_item.updated_at = now

    # Update job counters
    if old_status == "failed":
        job.failed_count = max(0, (job.failed_count or 0) - 1)
        job.queued_count = (job.queued_count or 0) + 1
    # If job was in a terminal state, reopen it
    if job.status in ("completed", "failed"):
        job.status = "queued"
        job.finished_at = None
    job.updated_at = now

    await session.flush()
    return job_item