curo/kite

Files

T

curo1305 5a9f3a628e docs(14-05): complete cloud analysis processing plan

2026-06-23 16:10:42 +02:00

12 KiB

Raw Blame History

phase: "14" plan: "05" subsystem: cloud-analysis tags: [analysis, celery, processing, cache, classification, retry, security] status: complete

requires:

phases/14-selective-analysis-and-byte-cache/14-03
phases/14-selective-analysis-and-byte-cache/14-04

provides:

services/cloud_analysis_processing.py: process_job_item, ProcessingResult, _classify_cloud_item, _set_item_status, _update_job_completion — full single-item analysis pipeline with stale guard, cache pinning, extraction, classification, and cooperative cancellation
tasks/cloud_analysis_tasks.py: process_cloud_analysis_item Celery task, _TransientError, _ClassificationRetry sentinels, retry harness with bounded exponential backoff
celery_app.py: cloud_analysis_tasks route + import registration

affects:

backend/services/cloud_analysis_processing.py (new)
backend/tasks/cloud_analysis_tasks.py (new)
backend/celery_app.py (route + import added)
backend/tests/test_cloud_analysis_contract.py (10 new tests added)

tech-stack: added: [] patterns: - status-transition-with-counter-adjustment: _set_item_status atomically moves item from old to new status and adjusts job aggregate counters in one flush; callers never manipulate counters directly - stale-guard-before-download: version key re-computed from current CloudItem metadata and compared to enqueued version_key BEFORE any provider byte fetch; changed metadata yields "stale" result with zero bytes downloaded (T-14-13) - cache-pin-in-finally: pin acquired before using cached/hydrated bytes, released in a finally block regardless of outcome (T-14-12) - sentinel-retry-pattern: _TransientError and _ClassificationRetry are plain Exception subclasses that escape asyncio.run() so self.retry() is called in the outer sync Celery layer (same pattern as document_tasks.py) - broker-id-only: Celery task accepts job/item/user/connection/cloud_item UUIDs only — credentials decrypted inside worker after ownership revalidation - cooperative-cancellation: job/item status checked at queued→downloading, downloading→extracting, extracting→classifying boundaries - no-document-row: classification targets CloudItem + CloudItemTopic directly; no Document ORM object created or referenced

key-files: created: - backend/services/cloud_analysis_processing.py - backend/tasks/cloud_analysis_tasks.py modified: - backend/celery_app.py - backend/tests/test_cloud_analysis_contract.py

decisions:

process_job_item owns ALL status transitions via _set_item_status; the Celery task layer never modifies job/item status directly — only the processing service does (separation of concerns)
Classification failure raises _ClassificationError (sentinel) which the Celery task catches as _ClassificationRetry and retries via self.retry() — identical pattern to document_tasks.py _ClassificationError sentinel
Stale guard re-reads CloudItem.version_key from the current row (not from the job item's stored version_key) — this detects concurrent reconcile updates that changed provider metadata between enqueue and processing
Cache pin acquired before bytes are accessed, released in finally; failure to release is logged but does not mask the original result
quota increment failure is non-fatal for processing — item proceeds without caching when quota would be exceeded; cache store is skipped
MinIO put_object failure is non-fatal — processing continues without the cache entry; only extracted_text and topics are durably persisted

metrics: duration: "~10 minutes" completed: "2026-06-23" tasks: 2 files: 4

Phase 14 Plan 05: Cloud Analysis Processing Service and Celery Task Summary

One-liner: Full per-item analysis pipeline — stale guard, cache hydration, text extraction, cloud-item classification, and ID-only Celery task with bounded retry harness.

What Was Built

Task 1: Cloud analysis processing service

services/cloud_analysis_processing.py (new, ~580 lines):

The core processing pipeline executed for each CloudAnalysisJobItem:

Owner revalidation: resolve_owned_connection re-checks connection ownership inside the worker. Deleted connections return cancelled without retry.
Job/item reload: Both CloudAnalysisJob and CloudAnalysisJobItem are re-fetched from DB to confirm they still exist and belong to the user.
Cooperative cancellation check: If job.status == "cancelled" or job_item.status == "cancelled", the function returns immediately without downloading bytes.
Stale guard (T-14-13): Version key re-computed from the current CloudItem metadata. If it differs from job_item.version_key (set at enqueue time), the item is marked "stale" and no provider bytes are fetched.
Cache check → hydration: Queries CloudByteCacheEntry for a non-evicted entry matching the current version key. Cache hit: pin and read from MinIO. Cache miss: download from provider via adapter.get_object(provider_item_id).
Text extraction: extract_text_from_bytes(file_bytes, content_type) — delegates to existing services.extractor.
Persistence: CloudItem.extracted_text updated; CloudItem.analysis_status = "indexed".
Classification: _classify_cloud_item resolves the user's AI provider config (same as classifier.classify_document), runs provider.classify(), auto-creates suggested topics, and upserts CloudItemTopic associations.
Final stale check: After classification, version key is re-computed one more time. If provider metadata changed during processing, item is marked "stale".
Cache pin release (T-14-12): release_cache_entry called in a finally block on all exit paths.

Status transitions:

queued → downloading → extracting → classifying → indexed
                                                → stale (metadata changed)
       → cancelled (cooperative cancellation)
       → failed (auth error, extraction failure, unexpected)

Counter management: _set_item_status decrements the old-status counter and increments the new-status counter on CloudAnalysisJob atomically. _update_job_completion transitions the job to completed/failed/cancelled when all items reach terminal states.

Key design invariants:

No provider mutation methods called (T-14-03): adapter only used via get_object
No Document row created: topics linked via CloudItemTopic directly
Cache pin lifecycle: pin_cache_entry → release_cache_entry in finally (T-14-12)
_ClassificationError sentinel re-raised to Celery layer for retry handling

New tests (4 added to test_cloud_analysis_contract.py):

test_processing_status_transitions_include_downloading_extracting_classifying: ProcessingResult dataclass shape
test_processing_unchanged_version_skips_before_provider_byte_fetch: stale guard fires before get_object
test_processing_cache_pin_released_on_cancellation: no pinned entries after cancel (T-14-12)
test_processing_never_calls_provider_mutation_methods: mutation call count stays zero (T-14-03)

Task 2: Celery task and retry harness

tasks/cloud_analysis_tasks.py (new, ~220 lines):

Sync Celery bridge following the established cloud_tasks.py / document_tasks.py sentinel pattern:

Task: process_cloud_analysis_item(job_id, item_id, user_id, connection_id, cloud_item_id) — all string UUIDs, no credentials/bytes (T-14-10)
Sentinels: _TransientError (provider/network failure) and _ClassificationRetry (AI failure) — plain Exception subclasses that escape asyncio.run()
Retry harness: self.retry() called in the sync layer with exponential backoff (30s/90s/270s ± 10s jitter); max_retries=3
Terminal failures: Auth errors → write failure status, return without retry
MaxRetriesExceededError: _write_max_retry_failure writes final "failed" status to DB
Credential handling: decrypt_credentials called inside worker after ownership revalidation; never in broker payload (T-14-10)
MinIO wrapper: _MinIOClientWrapper adapts the storage backend for async get_object/put_object/delete_object

celery_app.py (updated):

Added "tasks.cloud_analysis_tasks.*": {"queue": "documents"} to task routes
Added import tasks.cloud_analysis_tasks for task registration

New tests (6 added to test_cloud_analysis_contract.py):

test_celery_task_registered_in_app: task name in Celery registry
test_celery_broker_payload_contains_ids_only: parameter shape, no forbidden names (T-14-10)
test_celery_task_routes_to_documents_queue: route config check
test_celery_task_max_retries_is_bounded: 1 ≤ max_retries ≤ 10
test_celery_task_retry_sentinel_escapes_asyncio_run: sentinel inheritance check
test_run_processing_returns_cancelled_when_connection_deleted: connection-not-found graceful drop

Test Results

File	Tests	Pass	Notes
test_cloud_analysis_contract.py	37 → 47	47	10 new Plan 05 tests (4 Task 1 + 6 Task 2)
test_cloud_cache.py	30	30	Unchanged — all pass
Overall suite (excl. test_extractor.py docx env)	844	844	Up from 675 before Plan 05

Pre-existing failures (not Plan 05 scope):

test_extractor.py::test_extract_docx: pre-existing docx env issue (excluded from count)

Deviations from Plan

Auto-fixed Issues

1. [Rule 1 - Bug] test_processing_never_calls_provider_mutation_methods patched wrong module

Found during: Task 1 test authoring
Issue: Test used patch("services.classifier.get_provider") but the processing service imports get_provider from ai directly via _classify_cloud_item. The patch did not intercept the actual import site.
Fix: Changed to patch("services.cloud_analysis_processing._classify_cloud_item", new_callable=AsyncMock) — patches the classification helper directly, bypassing the AI provider entirely for the mutation-check test.
Files modified: backend/tests/test_cloud_analysis_contract.py
Commit: cea9980

2. [Rule 1 - Bug] test_celery_task_retry_sentinel_escapes_asyncio_run used run_until_complete in running event loop

Found during: Task 2 test authoring
Issue: Pytest-asyncio runs tests inside an already-running event loop. Calling get_event_loop().run_until_complete() inside an async test raises "This event loop is already running".
Fix: Replaced the asyncio.run() test with equivalent synchronous sentinel checks (inheritance assertions + raise/catch pattern). The semantic contract is equivalent — the test verifies sentinels are plain Exception subclasses that propagate normally.
Files modified: backend/tests/test_cloud_analysis_contract.py
Commit: f235d89

3. [Rule 1 - Bug] test_run_processing_returns_cancelled_when_connection_deleted tried real DB connection

Found during: Task 2 test authoring
Issue: _run_processing calls AsyncSessionLocal() which requires a live PostgreSQL connection. The test used a random UUID that doesn't exist in the test DB, but the error surfaced as a connection failure to postgres host rather than a ConnectionNotFound.
Fix: Patched db.session.AsyncSessionLocal and services.cloud_items.resolve_owned_connection to simulate the deleted-connection path without a live DB connection.
Files modified: backend/tests/test_cloud_analysis_contract.py
Commit: f235d89

Known Stubs

None. The processing service is fully functional. The Celery task can be dispatched and will execute the full pipeline when a Redis broker and MinIO are available.

The AI classification path (_classify_cloud_item) requires a configured AI provider — in test environments this is patched via mock. In production it uses the system AI provider configured via system_settings.

Threat Flags

None — no new API endpoints in this plan. The processing service and Celery task are internal infrastructure. Security mitigations from the plan threat model are fully implemented:

Threat	Mitigation	Status
T-14-10 Credentials in broker	IDs only in task args; decrypt in worker after revalidation	Implemented
T-14-11 Provider file mutation	Processing uses `get_object` only; mutation methods never called	Verified by test
T-14-12 Cache pin leak	`finally` block releases pin on success, failure, and cancellation	Implemented + tested
T-14-13 Stale result presented	Final version/fingerprint comparison marks item stale if metadata changed	Implemented + tested

12 KiB Raw Blame History