feat(14-05): add Celery cloud analysis task with retry harness (Task 2)

- Implement tasks/cloud_analysis_tasks.py: sync Celery bridge to async processing
- Broker payload contains IDs only — credentials decrypted in worker (T-14-10)
- _TransientError and _ClassificationRetry sentinels for bounded retry (max=3)
- Auth errors are terminal (not retried); MaxRetriesExceeded writes failure status
- celery_app.py: add cloud_analysis_tasks route and import registration
- Add 6 contract tests: task registration, broker payload shape, queue routing,
  bounded retries, sentinel pattern, connection-not-found graceful drop

backend/celery_app.py

@@ -36,6 +36,7 @@ celery_app.conf.task_routes = {
     "tasks.email_tasks.*": {"queue": "email"},
     "tasks.audit_tasks.*": {"queue": "documents"},
     "tasks.cloud_tasks.*": {"queue": "documents"},
+    "tasks.cloud_analysis_tasks.*": {"queue": "documents"},
 }
 
 # Celery beat schedule:
@@ -56,6 +57,7 @@ celery_app.conf.timezone = "UTC"
 # Explicitly import task modules — autodiscover_tasks(["tasks"]) looks for
 # tasks.tasks (appends ".tasks") which doesn't exist in our structure.
 import tasks.audit_tasks  # noqa: F401, E402
+import tasks.cloud_analysis_tasks  # noqa: F401, E402
 import tasks.cloud_tasks  # noqa: F401, E402
 import tasks.document_tasks  # noqa: F401, E402
 import tasks.email_tasks  # noqa: F401, E402

backend/tasks/cloud_analysis_tasks.py

@@ -0,0 +1,344 @@
+"""
+Celery tasks for cloud analysis processing — Phase 14 Plan 05.
+
+Provides the sync Celery bridge to the async processing service.
+
+Task: process_cloud_analysis_item
+  - Broker payload: job_id (str), item_id (str), user_id (str), connection_id (str),
+    cloud_item_id (str) — IDs only, no credentials, bytes, or filenames (T-14-10).
+  - Credentials are decrypted inside the worker after ownership revalidation.
+  - Supports bounded transient provider/classifier retry (max_retries=3).
+  - Terminal auth failures do NOT retry — they write a controlled failure status.
+  - MaxRetriesExceededError writes final "failed" status to DB.
+  - Cooperative cancellation is checked between all major processing stages by the
+    processing service.
+
+Retry harness (same pattern as cloud_tasks.py / document_tasks.py):
+  CRITICAL: self.retry() MUST be raised from the outer sync task, NOT inside
+  asyncio.run(). _TransientError and _ClassificationRetry are sentinels raised by
+  _run_processing() to signal retryable failures that escape asyncio.run().
+
+Aggregate job counter updates:
+  - Handled by services.cloud_analysis_processing._set_item_status and
+    _update_job_completion (the processing service owns all counter transitions).
+  - The Celery task only records the final outcome returned by process_job_item.
+
+Queue routing:
+  - Added to celery_app.py task_routes under "tasks.cloud_analysis_tasks.*" → "documents".
+
+Security:
+  - T-14-10: Broker payload contains no credentials, provider URLs, filenames, or bytes.
+  - Owner revalidation inside the worker — no trust in the broker message alone.
+"""
+from __future__ import annotations
+
+import asyncio
+import random
+
+from celery.exceptions import MaxRetriesExceededError
+
+from celery_app import celery_app
+
+
+# ── Retry sentinels ───────────────────────────────────────────────────────────
+
+class _TransientError(Exception):
+    """Retryable transient provider or network failure.
+
+    Raised by _run_processing() to escape asyncio.run() so self.retry()
+    can be called in the outer sync Celery task layer.
+    """
+
+
+class _ClassificationRetry(Exception):
+    """Retryable AI classification failure.
+
+    Separate sentinel from _TransientError so the caller can apply appropriate
+    backoff and log the classification failure path distinctly.
+    """
+
+
+# ── Async processing worker ───────────────────────────────────────────────────
+
+async def _run_processing(
+    job_id: str,
+    item_id: str,
+    user_id: str,
+    connection_id: str,
+    cloud_item_id: str,
+) -> dict:
+    """Async body of process_cloud_analysis_item.
+
+    Opens its own AsyncSession (never shared with other tasks/requests).
+    Decrypts credentials inside the worker — never put them in the broker message.
+
+    Returns a result dict with at least: {"status": "indexed"|"failed"|"cancelled"|"stale"}.
+
+    Raises:
+        _TransientError:       Provider/network failure that should be retried.
+        _ClassificationRetry:  AI classification failure that should be retried.
+    """
+    import uuid as _uuid
+
+    from db.session import AsyncSessionLocal
+    from services.cloud_items import resolve_owned_connection, ConnectionNotFound
+    from storage.cloud_backend_factory import build_cloud_resource_adapter
+    from storage.cloud_utils import decrypt_credentials
+    from storage import get_storage_backend
+    from config import settings
+
+    master_key = settings.cloud_creds_key.encode()
+
+    async with AsyncSessionLocal() as session:
+        # Revalidate ownership — worker never trusts broker payload alone (T-14-10)
+        try:
+            conn = await resolve_owned_connection(
+                session,
+                connection_id=connection_id,
+                user_id=user_id,
+            )
+        except ConnectionNotFound:
+            # Connection deleted or user changed — silently drop, do not retry
+            return {"status": "cancelled", "reason": "connection_not_found"}
+
+        # Decrypt credentials inside the worker (T-14-10)
+        try:
+            credentials = decrypt_credentials(master_key, user_id, conn.credentials_enc)
+        except Exception as exc:
+            # Decryption failure is terminal — write failure status
+            await _write_item_failure(
+                session,
+                job_id=job_id,
+                item_id=item_id,
+                user_id=user_id,
+                error_code="credential_error",
+                error_message="Credential decryption failed. Re-connect the account.",
+            )
+            return {"status": "failed", "reason": "credential_error"}
+
+        # Build read-only provider adapter
+        try:
+            adapter = build_cloud_resource_adapter(conn.provider, credentials)
+        except Exception as exc:
+            await _write_item_failure(
+                session,
+                job_id=job_id,
+                item_id=item_id,
+                user_id=user_id,
+                error_code="adapter_error",
+                error_message="Failed to build provider adapter.",
+            )
+            return {"status": "failed", "reason": "adapter_error"}
+
+        # Build MinIO client for cache byte storage
+        minio_client = _build_minio_client()
+
+        # Delegate to the processing service
+        from services.cloud_analysis_processing import (
+            process_job_item,
+            ProcessingResult,
+            _ClassificationError,
+            OwnerValidationError,
+        )
+
+        try:
+            result: ProcessingResult = await process_job_item(
+                session,
+                job_id=_uuid.UUID(job_id),
+                item_id=_uuid.UUID(item_id),
+                user_id=_uuid.UUID(user_id),
+                connection_id=_uuid.UUID(connection_id),
+                cloud_item_id=_uuid.UUID(cloud_item_id),
+                minio_client=minio_client,
+                provider_adapter=adapter,
+            )
+            await session.commit()
+            return {
+                "status": result.status,
+                "error_code": result.error_code,
+                "cache_entry_id": str(result.cache_entry_id) if result.cache_entry_id else None,
+            }
+
+        except _ClassificationError as exc:
+            # Re-raise as our sentinel so the outer sync layer can self.retry()
+            raise _ClassificationRetry(str(exc)) from exc
+
+        except Exception as exc:
+            err_str = str(exc).lower()
+            # Detect auth/scope errors — terminal, do not retry
+            if any(kw in err_str for kw in (
+                "invalid_grant", "unauthorized", "401", "403", "scope", "revoked",
+                "credential_error", "auth_error",
+            )):
+                await _write_item_failure(
+                    session,
+                    job_id=job_id,
+                    item_id=item_id,
+                    user_id=user_id,
+                    error_code="auth_error",
+                    error_message="Authentication failed. Re-connect the account.",
+                )
+                await session.commit()
+                return {"status": "failed", "reason": "auth_error"}
+            # Transient failure — signal outer layer to retry
+            raise _TransientError(f"Transient error during processing: {exc}") from exc
+
+
+def _build_minio_client():
+    """Build a MinIO-compatible storage client for cache byte operations.
+
+    Returns an object with async get_object(key), put_object(key, data, content_type),
+    and delete_object(key) methods backed by the project's MinIO storage backend.
+    """
+    from storage import get_storage_backend
+
+    class _MinIOClientWrapper:
+        def __init__(self, backend):
+            self._backend = backend
+
+        async def get_object(self, key: str) -> bytes:
+            return await self._backend.get_object(key)
+
+        async def put_object(self, key: str, data: bytes, content_type: str = None) -> None:
+            await self._backend.put_object(key, data, content_type=content_type or "application/octet-stream")
+
+        async def delete_object(self, key: str) -> None:
+            try:
+                await self._backend.delete_object(key)
+            except Exception:
+                pass
+
+    return _MinIOClientWrapper(get_storage_backend())
+
+
+async def _write_item_failure(
+    session,
+    *,
+    job_id: str,
+    item_id: str,
+    user_id: str,
+    error_code: str,
+    error_message: str,
+) -> None:
+    """Write a failure status to a CloudAnalysisJobItem without leaking exceptions.
+
+    Used for terminal errors (credential, adapter) where the processing service
+    could not run its normal status-transition path.
+    """
+    import uuid as _uuid
+    from sqlalchemy import select
+    from db.models import CloudAnalysisJob, CloudAnalysisJobItem
+
+    try:
+        uid = _uuid.UUID(user_id)
+        jid = _uuid.UUID(job_id)
+        iid = _uuid.UUID(item_id)
+
+        job = (await session.execute(
+            select(CloudAnalysisJob).where(
+                CloudAnalysisJob.id == jid,
+                CloudAnalysisJob.user_id == uid,
+            )
+        )).scalars().first()
+
+        job_item = (await session.execute(
+            select(CloudAnalysisJobItem).where(
+                CloudAnalysisJobItem.id == iid,
+                CloudAnalysisJobItem.job_id == jid,
+            )
+        )).scalars().first()
+
+        if job_item is not None and job is not None:
+            from services.cloud_analysis_processing import _set_item_status, _update_job_completion
+            old_status = job_item.status
+            await _set_item_status(
+                session,
+                job_item=job_item,
+                job=job,
+                new_status="failed",
+                old_status=old_status,
+                error_code=error_code,
+                error_message=error_message,
+            )
+            await _update_job_completion(session, job=job)
+    except Exception:
+        pass  # Best-effort — do not mask the original failure
+
+
+async def _write_max_retry_failure(
+    job_id: str, item_id: str, user_id: str
+) -> None:
+    """Write final 'failed' status after all Celery retries are exhausted."""
+    from db.session import AsyncSessionLocal
+
+    async with AsyncSessionLocal() as session:
+        await _write_item_failure(
+            session,
+            job_id=job_id,
+            item_id=item_id,
+            user_id=user_id,
+            error_code="max_retries_exceeded",
+            error_message="Processing failed after 3 retry attempts. Use per-item retry to requeue.",
+        )
+        await session.commit()
+
+
+# ── Celery task ───────────────────────────────────────────────────────────────
+
+@celery_app.task(
+    bind=True,
+    max_retries=3,
+    name="tasks.cloud_analysis_tasks.process_cloud_analysis_item",
+    serializer="json",
+    acks_late=True,
+    reject_on_worker_lost=True,
+)
+def process_cloud_analysis_item(
+    self,
+    job_id: str,
+    item_id: str,
+    user_id: str,
+    connection_id: str,
+    cloud_item_id: str,
+) -> dict:
+    """Sync Celery entry-point for processing one cloud analysis job item.
+
+    Broker payload contains IDs only — no credentials, provider URLs, filenames,
+    or bytes (T-14-10). Credentials are decrypted inside the worker.
+
+    Retry harness:
+      - Bounded exponential backoff: 30s / 90s / 270s (± 10s jitter).
+      - _TransientError and _ClassificationRetry sentinels escape asyncio.run()
+        and trigger self.retry() here in the sync Celery layer.
+      - MaxRetriesExceededError writes final failure status to DB.
+      - Terminal auth failures are NOT retried.
+
+    Args:
+        job_id:         CloudAnalysisJob UUID string.
+        item_id:        CloudAnalysisJobItem UUID string.
+        user_id:        Owner User UUID string.
+        connection_id:  CloudConnection UUID string.
+        cloud_item_id:  CloudItem UUID string.
+
+    Returns:
+        Result dict: {"status": "indexed"|"failed"|"cancelled"|"stale", ...}
+    """
+    try:
+        return asyncio.run(
+            _run_processing(job_id, item_id, user_id, connection_id, cloud_item_id)
+        )
+    except (_TransientError, _ClassificationRetry) as exc:
+        try:
+            # Exponential backoff with jitter: 30s, 90s, 270s (± 10s)
+            jitter = random.randint(-10, 10)
+            countdown = (30 * (3 ** self.request.retries)) + jitter
+            raise self.retry(exc=exc, countdown=countdown)
+        except MaxRetriesExceededError:
+            # All retries exhausted — write final failure status
+            asyncio.run(_write_max_retry_failure(job_id, item_id, user_id))
+            return {
+                "status": "failed",
+                "reason": "max_retries_exceeded",
+                "job_id": job_id,
+                "item_id": item_id,
+            }

backend/tests/test_cloud_analysis_contract.py

@@ -1067,3 +1067,175 @@ async def test_processing_never_calls_provider_mutation_methods(db_session):
     )
     # Should succeed or fail gracefully (not from a mutation)
     assert proc_result.status in ("indexed", "failed", "stale", "cancelled")
+
+
+# ─── Plan 05 Task 2: Celery task broker payload and retry contract ────────────
+
+def test_celery_task_registered_in_app():
+    """ANALYZE-04: process_cloud_analysis_item task is registered in the Celery app.
+
+    Verifies that the task module is imported and the task name is in the registry.
+    """
+    from celery_app import celery_app
+    import tasks.cloud_analysis_tasks  # noqa: F401 — side effect: registers the task
+
+    task_names = list(celery_app.tasks.keys())
+    assert "tasks.cloud_analysis_tasks.process_cloud_analysis_item" in task_names, (
+        "process_cloud_analysis_item must be registered in the Celery task registry"
+    )
+
+
+def test_celery_broker_payload_contains_ids_only():
+    """T-14-10: Celery task signature accepts only ID strings — no credentials/bytes.
+
+    Inspects the task function signature to verify it accepts only string ID
+    parameters (job_id, item_id, user_id, connection_id, cloud_item_id).
+    """
+    import inspect
+    from tasks.cloud_analysis_tasks import process_cloud_analysis_item
+
+    sig = inspect.signature(process_cloud_analysis_item)
+    params = list(sig.parameters.keys())
+
+    # Required ID-only parameters
+    required = ["job_id", "item_id", "user_id", "connection_id", "cloud_item_id"]
+    for param in required:
+        assert param in params, (
+            f"Celery task must accept '{param}' ID parameter — broker payload "
+            "must contain IDs only (T-14-10)"
+        )
+
+    # Forbidden parameters that would indicate credential leakage
+    forbidden = ["credentials", "credentials_enc", "access_token", "file_bytes",
+                 "provider_url", "filename", "object_key"]
+    for param in forbidden:
+        assert param not in params, (
+            f"Celery task must NOT accept '{param}' — credentials and bytes "
+            "must not appear in broker payload (T-14-10)"
+        )
+
+
+def test_celery_task_routes_to_documents_queue():
+    """ANALYZE-04: cloud_analysis_tasks routes to the 'documents' queue."""
+    from celery_app import celery_app
+
+    routes = celery_app.conf.task_routes or {}
+    assert "tasks.cloud_analysis_tasks.*" in routes, (
+        "cloud_analysis_tasks must have a task route entry"
+    )
+    route = routes["tasks.cloud_analysis_tasks.*"]
+    assert route.get("queue") == "documents", (
+        "cloud_analysis_tasks must route to the 'documents' queue"
+    )
+
+
+def test_celery_task_max_retries_is_bounded():
+    """ANALYZE-04: Celery task has bounded max_retries (not unlimited).
+
+    Unlimited retries could cause runaway retry storms on persistent failures.
+    """
+    from tasks.cloud_analysis_tasks import process_cloud_analysis_item
+
+    # Access the underlying task object's max_retries attribute
+    task_obj = process_cloud_analysis_item
+    assert hasattr(task_obj, "max_retries"), "Task must have max_retries set"
+    assert task_obj.max_retries is not None, "max_retries must not be None (unlimited)"
+    assert 1 <= task_obj.max_retries <= 10, (
+        f"max_retries must be a reasonable bounded value, got {task_obj.max_retries}"
+    )
+
+
+def test_celery_task_retry_sentinel_escapes_asyncio_run():
+    """ANALYZE-04: _TransientError and _ClassificationRetry sentinels escape asyncio.run().
+
+    The sentinel pattern requires that exceptions raised inside asyncio.run()
+    propagate out to the sync Celery task layer where self.retry() can be called.
+    This test verifies that the sentinels are plain Exception subclasses (not
+    coroutines or awaitables), so they can propagate through asyncio.run() normally.
+    """
+    from tasks.cloud_analysis_tasks import _TransientError, _ClassificationRetry
+
+    # The sentinels must be importable and be plain Exception subclasses
+    assert issubclass(_TransientError, Exception), (
+        "_TransientError must be a plain Exception subclass to escape asyncio.run()"
+    )
+    assert issubclass(_ClassificationRetry, Exception), (
+        "_ClassificationRetry must be a plain Exception subclass to escape asyncio.run()"
+    )
+
+    # Verify they are NOT derived from asyncio-specific types that would be swallowed
+    import asyncio
+    assert not issubclass(_TransientError, asyncio.CancelledError), (
+        "_TransientError must not derive from CancelledError (would be swallowed)"
+    )
+    assert not issubclass(_ClassificationRetry, asyncio.CancelledError), (
+        "_ClassificationRetry must not derive from CancelledError (would be swallowed)"
+    )
+
+    # Verify they can be raised and caught normally (synchronous check)
+    caught_transient = False
+    try:
+        raise _TransientError("test")
+    except _TransientError:
+        caught_transient = True
+    assert caught_transient, "_TransientError must be catchable as Exception"
+
+    caught_classification = False
+    try:
+        raise _ClassificationRetry("test")
+    except _ClassificationRetry:
+        caught_classification = True
+    assert caught_classification, "_ClassificationRetry must be catchable as Exception"
+
+
+async def test_run_processing_returns_cancelled_when_connection_deleted():
+    """ANALYZE-04: _run_processing returns cancelled/skipped when connection no longer exists.
+
+    The worker revalidates ownership at the start — if the connection is deleted
+    between enqueue and execution, the task silently drops (no retry).
+
+    Uses mocked AsyncSessionLocal to avoid requiring a live PostgreSQL connection
+    in the test environment.
+    """
+    from tasks.cloud_analysis_tasks import _run_processing
+    from unittest.mock import patch, AsyncMock, MagicMock
+    from services.cloud_items import ConnectionNotFound
+
+    import uuid
+
+    # Mock the AsyncSessionLocal so the task uses an in-memory session
+    mock_session = AsyncMock()
+    mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+    mock_session.__aexit__ = AsyncMock(return_value=False)
+
+    # Mock resolve_owned_connection to raise ConnectionNotFound (deleted connection)
+    with patch("tasks.cloud_analysis_tasks._run_processing") as patched_run:
+        # Verify the function exists and is importable
+        from tasks.cloud_analysis_tasks import _run_processing as real_run
+        assert callable(real_run)
+
+    # Test the connection-not-found path by patching at the service layer
+    with patch("services.cloud_items.resolve_owned_connection",
+               new_callable=AsyncMock,
+               side_effect=ConnectionNotFound("not found")):
+        with patch("db.session.AsyncSessionLocal") as mock_session_local:
+            mock_ctx = AsyncMock()
+            mock_ctx.__aenter__ = AsyncMock(return_value=mock_ctx)
+            mock_ctx.__aexit__ = AsyncMock(return_value=False)
+            mock_session_local.return_value = mock_ctx
+
+            result = await _run_processing(
+                job_id=str(uuid.uuid4()),
+                item_id=str(uuid.uuid4()),
+                user_id=str(uuid.uuid4()),
+                connection_id=str(uuid.uuid4()),  # Does not exist in DB
+                cloud_item_id=str(uuid.uuid4()),
+            )
+
+    # Worker must return a graceful status for deleted connection
+    assert result.get("status") in ("cancelled", "failed"), (
+        f"Deleted connection must return cancelled/failed status, got: {result}"
+    )
+    assert result.get("reason") == "connection_not_found", (
+        f"Expected reason='connection_not_found', got: {result.get('reason')}"
+    )