feat(14-06): route preview/download bytes through durable byte cache

- Add hydrate_and_cache_bytes to services/cloud_cache.py (Plan 06 cache lifecycle)
  - Cache-hit path: pin existing entry, read from MinIO, release pin in finally
  - Cache-miss path: fetch from provider, store in MinIO, create/reactivate entry, pin during response, release pin in finally
  - CacheQuotaExceeded and MinIO failures are non-fatal (bytes returned from provider path)
  - evict_lru_entries run after pin release to stay within user cache limit
- Update preview_cloud_file and download_cloud_file in operations.py to call hydrate_and_cache_bytes
  - Version key computed from CloudItem metadata (no provider call required for key resolution)
  - Shared _make_minio_cache_wrapper helper eliminates duplicate adapter code
  - Falls through to direct provider fetch when no CloudItem row found
- Add _make_minio_cache_wrapper helper to operations.py (DRY, shared by preview and download)
- Add 5 new tests to test_cloud_cache.py: cache-miss fetches provider, cache-hit skips provider, pin released on hit, pin released on miss, user isolation
- Add 3 new security tests to test_cloud_security.py: preview response excludes object_key, download response excludes object_key, foreign-user IDOR via download blocked

backend/api/cloud/operations.py

@@ -1,5 +1,5 @@
 """
-Phase 13 cloud content and mutation endpoints.
+Phase 13 cloud content and mutation endpoints (Phase 14 Plan 06: byte cache integration).
 
 Owner-scoped routes for opening, previewing, downloading, renaming, moving,
 deleting, creating folders, and uploading cloud files. All routes are
@@ -14,6 +14,16 @@ Security invariants (T-13-14):
   - Cross-connection moves are rejected before the adapter is called (D-08).
   - Self-destination moves are rejected before the adapter is called (D-09).
   - No health probe triggered by folder navigation (D-13).
+  - cache entries (object_key) are never returned in any response (T-14-02).
+
+Phase 14 Plan 06 byte cache integration:
+  - preview_cloud_file and download_cloud_file route provider bytes through
+    hydrate_and_cache_bytes — cache hits avoid a provider round-trip.
+  - The cache pin lifecycle protects active preview/download entries from LRU
+    eviction during the response stream.
+  - cache_limit_bytes is read from UserAnalysisSettings (default 512 MB).
+  - All byte storage is behind the MinIO private bucket — object keys are
+    never exposed.
 
 Route family:
   GET  /connections/{id}/items/{item_id}/open    — D-02 authorized access URL
@@ -132,6 +142,44 @@ async def _resolve_and_get_adapter(
     return conn, adapter
 
 
+def _make_minio_cache_wrapper(backend):
+    """Return a lightweight MinIO wrapper for use with hydrate_and_cache_bytes.
+
+    The wrapper adapts the MinIOBackend instance to the interface expected by
+    hydrate_and_cache_bytes: async get_object(key), put_object(key, data, content_type),
+    and delete_object(key).  This avoids duplicating the wrapper class at every
+    call site (Plan 06, DRY principle).
+    """
+    import io as _io
+    import asyncio as _asyncio
+
+    class _MinIOCacheWrapper:
+        def __init__(self, b):
+            self._b = b
+
+        async def get_object(self, key: str) -> bytes:
+            return await self._b.get_object(key)
+
+        async def put_object(self, key: str, data: bytes, content_type: str = None) -> None:
+            buf = _io.BytesIO(data)
+            await _asyncio.to_thread(
+                self._b._client.put_object,
+                self._b._bucket,
+                key,
+                buf,
+                length=len(data),
+                content_type=content_type or "application/octet-stream",
+            )
+
+        async def delete_object(self, key: str) -> None:
+            try:
+                await self._b.delete_object(key)
+            except Exception:
+                pass
+
+    return _MinIOCacheWrapper(backend)
+
+
 def _mutation_error_json(result: dict, http_status: int) -> JSONResponse:
     """Return a typed JSON error response for a failed mutation.
 
@@ -304,14 +352,68 @@ async def preview_cloud_file(
     if not PreviewSupport.is_supported(content_type):
         return _unsupported_preview_response(connection_id, item_id, "unsupported_format")
 
-    # Step 4: Fetch bytes via the adapter (supported binary format confirmed above)
+    # Step 4: Resolve the adapter (requires credential decryption).
     try:
         conn, adapter = await _resolve_and_get_adapter(session, connection_id, current_user.id)
-        file_bytes = await adapter.get_object(item_id)
     except HTTPException:
-        # Re-raise 401 (credential failure) and 404 (ownership race) — never mask as
-        # unsupported_preview. The IDOR guard in Step 1 already passed, but a 401
-        # from credential decryption should surface as an auth error to the client.
+        raise
+    except Exception:
+        return _unsupported_preview_response(connection_id, item_id, "provider_error")
+
+    # Step 5: Fetch bytes via the byte cache (PATTERNS.md lifecycle, Plan 06).
+    # hydrate_and_cache_bytes: cache hit → read from MinIO without a provider call;
+    # cache miss → fetch from provider, store in MinIO, pin for duration of response.
+    try:
+        from services.cloud_cache import hydrate_and_cache_bytes
+        from services.cloud_analysis_versioning import compute_version_key
+        from services.cloud_analysis_settings import (
+            get_or_create_user_analysis_settings,
+            DEFAULT_CACHE_LIMIT_BYTES,
+        )
+        from storage import get_storage_backend
+
+        # Compute the version key from CloudItem metadata (no provider call required).
+        version_key = compute_version_key(
+            provider_item_id=item_id,
+            version=cloud_item.version if cloud_item else None,
+            etag=cloud_item.etag if cloud_item else None,
+            size=cloud_item.provider_size if cloud_item else None,
+            modified_at=str(cloud_item.modified_at) if (cloud_item and cloud_item.modified_at) else None,
+            content_type=content_type,
+        ) if cloud_item else None
+
+        # Resolve user's cache byte limit.
+        try:
+            user_settings = await get_or_create_user_analysis_settings(
+                session, user_id=current_user.id
+            )
+            cache_limit = user_settings.cloud_cache_limit_bytes
+        except Exception:
+            cache_limit = DEFAULT_CACHE_LIMIT_BYTES
+
+        if cloud_item and version_key:
+            # MinIO client wrapper for cache operations.
+            _backend = get_storage_backend()
+
+            minio_wrapper = _make_minio_cache_wrapper(_backend)
+
+            file_bytes, _ = await hydrate_and_cache_bytes(
+                session,
+                user_id=current_user.id,
+                connection_id=connection_id,
+                cloud_item_id=cloud_item.id,
+                provider_item_id=item_id,
+                version_key=version_key,
+                content_type=content_type,
+                fetch_fn=lambda: adapter.get_object(item_id),
+                minio_client=minio_wrapper,
+                cache_limit_bytes=cache_limit,
+            )
+        else:
+            # No CloudItem row or no version key — fetch directly from provider.
+            file_bytes = await adapter.get_object(item_id)
+
+    except HTTPException:
         raise
     except Exception:
         return _unsupported_preview_response(connection_id, item_id, "provider_error")
@@ -364,10 +466,59 @@ async def download_cloud_file(
     filename = cloud_item.name if cloud_item else item_id
     content_type = (cloud_item.content_type if cloud_item else None) or "application/octet-stream"
 
+    # Fetch bytes via the byte cache (Plan 06, PATTERNS.md cache lifecycle).
+    # hydrate_and_cache_bytes: cache hit → read from MinIO without a provider call;
+    # cache miss → fetch from provider, store in MinIO, pin for duration of response.
     try:
-        file_bytes = await adapter.get_object(item_id)
+        from services.cloud_cache import hydrate_and_cache_bytes
+        from services.cloud_analysis_versioning import compute_version_key
+        from services.cloud_analysis_settings import (
+            get_or_create_user_analysis_settings,
+            DEFAULT_CACHE_LIMIT_BYTES,
+        )
+        from storage import get_storage_backend
+
+        version_key = compute_version_key(
+            provider_item_id=item_id,
+            version=cloud_item.version if cloud_item else None,
+            etag=cloud_item.etag if cloud_item else None,
+            size=cloud_item.provider_size if cloud_item else None,
+            modified_at=str(cloud_item.modified_at) if (cloud_item and cloud_item.modified_at) else None,
+            content_type=content_type if content_type != "application/octet-stream" else None,
+        ) if cloud_item else None
+
+        try:
+            user_settings = await get_or_create_user_analysis_settings(
+                session, user_id=current_user.id
+            )
+            cache_limit = user_settings.cloud_cache_limit_bytes
+        except Exception:
+            cache_limit = DEFAULT_CACHE_LIMIT_BYTES
+
+        if cloud_item and version_key:
+            _backend = get_storage_backend()
+            minio_wrapper = _make_minio_cache_wrapper(_backend)
+
+            file_bytes, _ = await hydrate_and_cache_bytes(
+                session,
+                user_id=current_user.id,
+                connection_id=connection_id,
+                cloud_item_id=cloud_item.id,
+                provider_item_id=item_id,
+                version_key=version_key,
+                content_type=content_type if content_type != "application/octet-stream" else None,
+                fetch_fn=lambda: adapter.get_object(item_id),
+                minio_client=minio_wrapper,
+                cache_limit_bytes=cache_limit,
+            )
+        else:
+            # No CloudItem metadata — fetch directly from provider.
+            file_bytes = await adapter.get_object(item_id)
+
+    except HTTPException:
+        raise
     except Exception as exc:
-        result_dict = adapter._normalize_error(exc)
+        result_dict = getattr(adapter, "_normalize_error", lambda e: {})(exc)
         if result_dict.get("kind") == MUT_KIND_REAUTH:
             raise HTTPException(
                 status_code=status.HTTP_401_UNAUTHORIZED,