Implement Celery-backed cloud analysis processing: hydrate bytes on demand, retain/reuse cache entries, extract text, classify topics onto `CloudItem`, update per-item and aggregate states, and honor cancellation/retry semantics.
Scope fence: do not build frontend controls in this plan.
@.planning/phases/14-selective-analysis-and-byte-cache/14-PATTERNS.md
@backend/tasks/cloud_tasks.py
@backend/tasks/document_tasks.py
@backend/services/extractor.py
@backend/services/classifier.py
@backend/storage/cloud_backend_factory.py
@backend/services/cloud_byte_cache.py
Task 1: Add processing service for one job item
backend/services/cloud_analysis_processing.py, backend/services/cloud_byte_cache.py, backend/tests/test_cloud_analysis_contract.py, backend/tests/test_cloud_cache.py
- backend/services/extractor.py
- backend/services/classifier.py
- backend/services/cloud_byte_cache.py
- backend/services/cloud_analysis_versioning.py
Implement a service that processes a single job item by revalidating owner/connection/item, checking cancellation, checking version-key currentness, reusing or hydrating cache bytes, extracting text, persisting `CloudItem.extracted_text`, classifying topics through existing classifier logic adapted for cloud items, updating status to indexed/failed/cancelled/stale, and releasing cache pins. Compute content hash only while bytes are already available. Never create a local `Document` row and never call provider mutation methods.
- Status transitions include downloading, extracting, classifying, indexed, cancelled, failed, stale.
- Unchanged version skips processing before provider byte fetch.
- Provider byte fetch occurs only for actionable analysis/open/preview work.
- Cache pins are released on success, failure, and cancellation.
cd backend && pytest -q tests/test_cloud_analysis_contract.py tests/test_cloud_cache.py -k "processing or idempotent or cancellation"
Task 2: Add Celery task and retry harness
backend/tasks/cloud_analysis_tasks.py, backend/celery_app.py, backend/tests/test_cloud_analysis_contract.py
- backend/tasks/cloud_tasks.py
- backend/tasks/document_tasks.py
- backend/celery_app.py
Add sync Celery tasks that bridge to async processing with the existing sentinel retry pattern. Task payloads include job id/item id only. Decrypt credentials inside the worker after owner revalidation. Support bounded transient provider/classifier retry, terminal auth failures, max-retry final failure status, and cooperative cancellation checks between stages.
- Celery broker payload contains no credentials, provider URLs, filenames, or bytes.
- `self.retry()` is raised only in the outer sync task layer.
- Max retry writes controlled failure status.
- Aggregate job counters update after item transitions.
cd backend && pytest -q tests/test_cloud_analysis_contract.py -k "celery or retry or broker"
<threat_model>
Threat ID
Threat
Mitigation
T-14-10
Credentials in broker payload
IDs only in Celery task args; decrypt in worker
T-14-11
Provider file mutation
Processing service uses content-read adapter path only and tests fake mutation counters
