diff --git a/backend/tests/test_cloud_analysis_api.py b/backend/tests/test_cloud_analysis_api.py
new file mode 100644
index 0000000..839a1af
--- /dev/null
+++ b/backend/tests/test_cloud_analysis_api.py
@@ -0,0 +1,450 @@
+"""
+Phase 14 Plan 01 — RED API integration tests for cloud analysis endpoints.
+
+Requirements: ANALYZE-01, ANALYZE-04, ANALYZE-05, CACHE-05
+
+Threat coverage:
+  T-14-01 — IDOR: all analysis and job-status endpoints are owner-scoped
+  T-14-02 — No credentials/object keys in any API response
+  T-14-03 — No provider mutations triggered by analysis routes
+
+These tests use the async_client fixture (real FastAPI + SQLite in-memory).
+They MUST FAIL due to missing route registrations until Phase 14 implementation
+adds the /api/cloud/analysis/* router.
+"""
+from __future__ import annotations
+
+import uuid
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+pytestmark = pytest.mark.asyncio
+
+from tests.conftest import _TEST_USER_AGENT
+
+
+# ─── Helpers ─────────────────────────────────────────────────────────────────
+
+async def _create_user_and_token(session, role: str = "user"):
+    from db.models import User, Quota
+    from services.auth import hash_password, create_access_token
+
+    user_id = uuid.uuid4()
+    user = User(
+        id=user_id,
+        handle=f"api_user_{user_id.hex[:8]}",
+        email=f"api_{user_id.hex[:8]}@example.com",
+        password_hash=hash_password("Testpassword123!"),
+        role=role,
+        is_active=True,
+        password_must_change=False,
+    )
+    quota = Quota(user_id=user_id, limit_bytes=104857600, used_bytes=0)
+    session.add(user)
+    session.add(quota)
+    await session.commit()
+    await session.refresh(user)
+
+    token = create_access_token(str(user_id), role, user_agent=_TEST_USER_AGENT)
+    return {
+        "user": user,
+        "token": token,
+        "headers": {
+            "Authorization": f"Bearer {token}",
+            "User-Agent": _TEST_USER_AGENT,
+        },
+    }
+
+
+async def _create_cloud_connection(session, user_id, provider="google_drive", status="ACTIVE"):
+    from db.models import CloudConnection
+    from storage.cloud_utils import encrypt_credentials
+    from config import settings
+
+    master_key = settings.cloud_creds_key.encode()
+    creds_enc = encrypt_credentials(
+        master_key,
+        str(user_id),
+        {"access_token": "tok", "refresh_token": "ref"},
+    )
+    conn = CloudConnection(
+        id=uuid.uuid4(),
+        user_id=user_id,
+        provider=provider,
+        display_name="API Test Connection",
+        credentials_enc=creds_enc,
+        status=status,
+    )
+    session.add(conn)
+    await session.commit()
+    return conn
+
+
+async def _create_cloud_item(session, user_id, connection_id, *, name="doc.pdf",
+                              kind="file", etag="etag-v1", analysis_status="pending"):
+    from db.models import CloudItem
+
+    item = CloudItem(
+        id=uuid.uuid4(),
+        user_id=user_id,
+        connection_id=connection_id,
+        provider_item_id=f"pitem-{uuid.uuid4().hex[:8]}",
+        name=name,
+        kind=kind,
+        content_type="application/pdf",
+        provider_size=204800,
+        etag=etag,
+        analysis_status=analysis_status,
+    )
+    session.add(item)
+    await session.commit()
+    return item
+
+
+# ─── Route registration check ─────────────────────────────────────────────────
+
+async def test_analysis_estimate_route_exists(async_client, db_session):
+    """Phase 14 router must be registered — estimate route returns non-404/405.
+
+    Until the analysis router is added to main.py, this returns 404. That is
+    the expected RED failure.
+    """
+    ctx = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, ctx["user"].id)
+
+    resp = await async_client.post(
+        f"/api/cloud/analysis/connections/{conn.id}/estimate",
+        json={"scope": "connection", "recursive": True},
+        headers=ctx["headers"],
+    )
+
+    # Phase 14 RED: route not yet registered, must return 404
+    # After implementation: route must exist and return 200
+    assert resp.status_code != 405, (
+        "Method not allowed suggests the router is registered but the route pattern is wrong"
+    )
+    # This assertion FAILS (404) until the router is wired in — expected RED behavior
+    assert resp.status_code == 200, (
+        f"Analysis estimate route not yet registered (Phase 14 RED — expected). "
+        f"Got {resp.status_code}"
+    )
+
+
+async def test_analysis_jobs_route_exists(async_client, db_session):
+    """Phase 14 router must be registered — job enqueue route returns non-404/405."""
+    ctx = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, ctx["user"].id)
+
+    resp = await async_client.post(
+        f"/api/cloud/analysis/connections/{conn.id}/jobs",
+        json={"scope": "connection", "recursive": True},
+        headers=ctx["headers"],
+    )
+
+    assert resp.status_code != 405
+    # RED: expected to fail until implementation
+    assert resp.status_code in (200, 202), (
+        f"Analysis job enqueue route not yet registered (Phase 14 RED). Got {resp.status_code}"
+    )
+
+
+async def test_job_status_route_exists(async_client, db_session):
+    """Phase 14: GET /api/cloud/analysis/jobs/{id} route must exist."""
+    ctx = await _create_user_and_token(db_session)
+
+    fake_job_id = str(uuid.uuid4())
+    resp = await async_client.get(
+        f"/api/cloud/analysis/jobs/{fake_job_id}",
+        headers=ctx["headers"],
+    )
+
+    # RED: returns 404 for route until implementation; then 404 for unknown job is ok
+    # After implementation: must return either 200 (job found) or 404 (not found) — not 405
+    assert resp.status_code != 405, (
+        "Method not allowed suggests the job-status route pattern is wrong"
+    )
+
+
+async def test_job_cancel_route_exists(async_client, db_session):
+    """Phase 14: POST /api/cloud/analysis/jobs/{id}/cancel route must exist."""
+    ctx = await _create_user_and_token(db_session)
+
+    fake_job_id = str(uuid.uuid4())
+    resp = await async_client.post(
+        f"/api/cloud/analysis/jobs/{fake_job_id}/cancel",
+        headers=ctx["headers"],
+    )
+
+    assert resp.status_code != 405, "Method not allowed on cancel route"
+
+
+# ─── ANALYZE-04: Progress detail labels ──────────────────────────────────────
+
+async def test_job_status_includes_simplified_labels_by_default(async_client, db_session):
+    """ANALYZE-04/D-06: Job status defaults to simplified progress labels.
+
+    Simplified: waiting, working, done, skipped, failed (not raw internal states).
+    """
+    ctx = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, ctx["user"].id)
+    item = await _create_cloud_item(db_session, ctx["user"].id, conn.id)
+
+    with patch("services.cloud_analysis.get_adapter", new_callable=MagicMock):
+        enqueue_resp = await async_client.post(
+            f"/api/cloud/analysis/connections/{conn.id}/jobs",
+            json={"scope": "file", "provider_item_ids": [item.provider_item_id]},
+            headers=ctx["headers"],
+        )
+
+    assert enqueue_resp.status_code in (200, 202)
+    job_id = enqueue_resp.json().get("job_id")
+
+    status_resp = await async_client.get(
+        f"/api/cloud/analysis/jobs/{job_id}",
+        headers=ctx["headers"],
+    )
+    assert status_resp.status_code == 200
+    body = status_resp.json()
+
+    # Default response must include aggregate counts using simplified vocabulary
+    simplified_states = {
+        "waiting_count", "working_count", "done_count", "skipped_count", "failed_count",
+        # or equivalently: queued_count, running_count, indexed_count, ...
+        "total_count",
+    }
+    # At least one of these must be present
+    assert any(k in body for k in simplified_states | {"status", "total_count"}), (
+        f"Job status must include aggregate counts. Got keys: {list(body.keys())}"
+    )
+
+
+async def test_job_status_detailed_mode_includes_per_stage_counts(async_client, db_session):
+    """ANALYZE-04/D-06: With detailed setting, job status includes per-stage counts.
+
+    Detailed: queued, downloading, extracting, classifying, indexed, cancelled, failed.
+    """
+    ctx = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, ctx["user"].id)
+    item = await _create_cloud_item(db_session, ctx["user"].id, conn.id)
+
+    with patch("services.cloud_analysis.get_adapter", new_callable=MagicMock):
+        enqueue_resp = await async_client.post(
+            f"/api/cloud/analysis/connections/{conn.id}/jobs",
+            json={"scope": "file", "provider_item_ids": [item.provider_item_id]},
+            headers=ctx["headers"],
+        )
+
+    assert enqueue_resp.status_code in (200, 202)
+    job_id = enqueue_resp.json().get("job_id")
+
+    # Request detailed progress (via query param or header per implementation design)
+    status_resp = await async_client.get(
+        f"/api/cloud/analysis/jobs/{job_id}?detail=true",
+        headers=ctx["headers"],
+    )
+    assert status_resp.status_code == 200
+    body = status_resp.json()
+
+    # Detailed mode must expose per-stage counts
+    detailed_states = {
+        "queued_count", "downloading_count", "extracting_count",
+        "classifying_count", "indexed_count", "cancelled_count", "failed_count",
+    }
+    assert any(k in body for k in detailed_states), (
+        f"Detailed mode must include per-stage counts. Got keys: {list(body.keys())}"
+    )
+
+
+# ─── ANALYZE-05: Controls — cancel job items ─────────────────────────────────
+
+async def test_skip_item_transitions_it_to_skipped(async_client, db_session):
+    """ANALYZE-05: Skip control on a queued item transitions it to skipped status."""
+    ctx = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, ctx["user"].id)
+    item = await _create_cloud_item(db_session, ctx["user"].id, conn.id)
+
+    with patch("services.cloud_analysis.get_adapter", new_callable=MagicMock):
+        enqueue_resp = await async_client.post(
+            f"/api/cloud/analysis/connections/{conn.id}/jobs",
+            json={"scope": "file", "provider_item_ids": [item.provider_item_id]},
+            headers=ctx["headers"],
+        )
+
+    job_id = enqueue_resp.json().get("job_id")
+
+    resp = await async_client.post(
+        f"/api/cloud/analysis/jobs/{job_id}/items/{item.id}/skip",
+        headers=ctx["headers"],
+    )
+    assert resp.status_code in (200, 202, 204)
+
+
+async def test_unauthenticated_request_blocked(async_client, db_session):
+    """Security: unauthenticated requests to analysis endpoints return 401."""
+    conn_id = str(uuid.uuid4())
+
+    resp = await async_client.post(
+        f"/api/cloud/analysis/connections/{conn_id}/estimate",
+        json={"scope": "connection", "recursive": True},
+        # No Authorization header
+    )
+
+    # Analysis routes must require authentication (401 or 403 depending on middleware)
+    # 404 is acceptable during RED phase when route is not yet registered
+    assert resp.status_code in (401, 403, 404), (
+        f"Unauthenticated request to analysis route must be blocked — got {resp.status_code}"
+    )
+
+
+# ─── Response schema shape ────────────────────────────────────────────────────
+
+async def test_estimate_response_schema_shape(async_client, db_session):
+    """ANALYZE-03/D-03: Estimate response includes required fields.
+
+    D-03 fields: supported_count, total_provider_bytes, unsupported_count,
+    recursive (for folder/connection scope).
+    """
+    ctx = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, ctx["user"].id)
+    item = await _create_cloud_item(db_session, ctx["user"].id, conn.id)
+
+    with patch("services.cloud_analysis.get_adapter", new_callable=MagicMock):
+        resp = await async_client.post(
+            f"/api/cloud/analysis/connections/{conn.id}/estimate",
+            json={"scope": "file", "provider_item_ids": [item.provider_item_id]},
+            headers=ctx["headers"],
+        )
+
+    # RED: will fail until routes are registered
+    assert resp.status_code == 200, (
+        f"Estimate route not yet registered (Phase 14 RED). Got {resp.status_code}"
+    )
+
+    body = resp.json()
+    required_fields = {"supported_count", "unsupported_count", "total_provider_bytes"}
+    missing = required_fields - set(body.keys())
+    assert not missing, f"Estimate response missing required fields: {missing}"
+
+    # No sensitive fields in response
+    sensitive = {"credentials_enc", "object_key", "access_token", "refresh_token"}
+    leaked = sensitive & set(body.keys())
+    assert not leaked, f"Estimate response contains sensitive fields: {leaked}"
+
+
+async def test_enqueue_response_includes_job_id(async_client, db_session):
+    """ANALYZE-01: Enqueue response must include a stable job_id for status polling."""
+    ctx = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, ctx["user"].id)
+    item = await _create_cloud_item(db_session, ctx["user"].id, conn.id)
+
+    with patch("services.cloud_analysis.get_adapter", new_callable=MagicMock):
+        resp = await async_client.post(
+            f"/api/cloud/analysis/connections/{conn.id}/jobs",
+            json={"scope": "file", "provider_item_ids": [item.provider_item_id]},
+            headers=ctx["headers"],
+        )
+
+    assert resp.status_code in (200, 202), (
+        f"Enqueue route not yet registered (Phase 14 RED). Got {resp.status_code}"
+    )
+    body = resp.json()
+    assert "job_id" in body, "Enqueue response must include job_id"
+    # job_id must be a valid UUID string
+    try:
+        uuid.UUID(body["job_id"])
+    except (ValueError, AttributeError):
+        pytest.fail(f"job_id must be a valid UUID — got: {body.get('job_id')!r}")
+
+
+async def test_enqueue_response_excludes_credentials(async_client, db_session):
+    """T-14-02: Enqueue response must not contain credentials or raw object keys."""
+    ctx = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, ctx["user"].id)
+    item = await _create_cloud_item(db_session, ctx["user"].id, conn.id)
+
+    with patch("services.cloud_analysis.get_adapter", new_callable=MagicMock):
+        resp = await async_client.post(
+            f"/api/cloud/analysis/connections/{conn.id}/jobs",
+            json={"scope": "file", "provider_item_ids": [item.provider_item_id]},
+            headers=ctx["headers"],
+        )
+
+    if resp.status_code in (200, 202):
+        body_text = resp.text
+        sensitive_fields = [
+            "credentials_enc", "access_token", "refresh_token",
+            "client_secret", "object_key",
+        ]
+        for field in sensitive_fields:
+            assert field not in body_text, (
+                f"Enqueue response contains sensitive field '{field}' (T-14-02)"
+            )
+
+
+# ─── Scan quota seam ─────────────────────────────────────────────────────────
+
+async def test_scan_quota_seam_module_exists():
+    """ANALYZE-13/D-13: Phase 14 must expose a scan quota seam distinct from storage quota.
+
+    This test checks that the scan quota helper exists as a callable with the
+    correct signature, even if tier limits are not enforced in Phase 14.
+    """
+    from services.cloud_analysis import check_scan_quota  # Phase 14 — not yet implemented
+
+    import inspect
+    sig = inspect.signature(check_scan_quota)
+    params = list(sig.parameters.keys())
+
+    # Must accept session and user_id at minimum
+    assert "session" in params or "user_id" in params, (
+        "check_scan_quota must accept session and/or user_id"
+    )
+
+
+async def test_failure_behavior_setting_accepted_by_enqueue(async_client, db_session):
+    """ANALYZE-05/D-11: Enqueue must accept failure_behavior field.
+
+    Accepted values: 'pause_batch' (default) and 'continue_item'.
+    """
+    ctx = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, ctx["user"].id)
+    item = await _create_cloud_item(db_session, ctx["user"].id, conn.id)
+
+    with patch("services.cloud_analysis.get_adapter", new_callable=MagicMock):
+        resp = await async_client.post(
+            f"/api/cloud/analysis/connections/{conn.id}/jobs",
+            json={
+                "scope": "file",
+                "provider_item_ids": [item.provider_item_id],
+                "failure_behavior": "continue_item",
+            },
+            headers=ctx["headers"],
+        )
+
+    # Must accept 'continue_item' without 422/400 validation error
+    assert resp.status_code not in (400, 422), (
+        f"Enqueue must accept failure_behavior='continue_item'. Got {resp.status_code}: {resp.text}"
+    )
+
+
+async def test_invalid_failure_behavior_rejected(async_client, db_session):
+    """ANALYZE-05: Unknown failure_behavior values must be rejected."""
+    ctx = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, ctx["user"].id)
+
+    resp = await async_client.post(
+        f"/api/cloud/analysis/connections/{conn.id}/jobs",
+        json={
+            "scope": "file",
+            "provider_item_ids": [],
+            "failure_behavior": "explode_everything",  # Invalid
+        },
+        headers=ctx["headers"],
+    )
+
+    # Route may 404 (RED phase) or 422 (validation error after implementation)
+    # Must NOT be 200/202 with an invalid value accepted
+    assert resp.status_code not in (200, 202), (
+        f"Invalid failure_behavior must not be accepted — got {resp.status_code}"
+    )
diff --git a/backend/tests/test_cloud_analysis_contract.py b/backend/tests/test_cloud_analysis_contract.py
new file mode 100644
index 0000000..6451c53
--- /dev/null
+++ b/backend/tests/test_cloud_analysis_contract.py
@@ -0,0 +1,660 @@
+"""
+Phase 14 Plan 01 — RED contract tests for cloud analysis scope, estimates, and
+idempotency (no bytes downloaded during estimate or already-current checks).
+
+Requirements: ANALYZE-01, ANALYZE-02, ANALYZE-03, ANALYZE-06, ANALYZE-07
+
+Threat coverage:
+  T-14-01 — IDOR: foreign user cannot estimate/enqueue another user's connection
+  T-14-03 — Hidden provider mutation: fake adapter mutation counters stay zero
+  T-14-04 — Hidden byte download: estimate and already-current checks never call
+             provider byte fetch
+
+All tests import non-existent Phase 14 modules and must FAIL due to
+ImportError or missing endpoints. They define the implementation contract.
+"""
+from __future__ import annotations
+
+import uuid
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+pytestmark = pytest.mark.asyncio
+
+from tests.conftest import _TEST_USER_AGENT
+
+
+# ─── Minimal fake adapter ────────────────────────────────────────────────────
+
+class FakeCloudAdapter:
+    """Counts calls to mutation and byte-download methods.
+
+    Zero counts on all methods after analysis operations confirms
+    T-14-03 (no mutation) and T-14-04 (no byte download).
+    """
+
+    def __init__(self):
+        self.upload_calls = 0
+        self.delete_calls = 0
+        self.rename_calls = 0
+        self.move_calls = 0
+        self.create_folder_calls = 0
+        self.get_object_calls = 0       # byte download
+        self.put_object_calls = 0       # byte upload
+
+    # Mutations — must stay zero during analysis
+    async def upload(self, *a, **kw):
+        self.upload_calls += 1
+        raise AssertionError("upload must not be called during analysis")
+
+    async def delete(self, *a, **kw):
+        self.delete_calls += 1
+        raise AssertionError("delete must not be called during analysis")
+
+    async def rename(self, *a, **kw):
+        self.rename_calls += 1
+        raise AssertionError("rename must not be called during analysis")
+
+    async def move(self, *a, **kw):
+        self.move_calls += 1
+        raise AssertionError("move must not be called during analysis")
+
+    async def create_folder(self, *a, **kw):
+        self.create_folder_calls += 1
+        raise AssertionError("create_folder must not be called during analysis")
+
+    # Byte downloads — must stay zero during estimate and already-current checks
+    async def get_object(self, *a, **kw):
+        self.get_object_calls += 1
+        raise AssertionError("get_object must not be called during estimate or already-current check")
+
+    async def put_object(self, *a, **kw):
+        self.put_object_calls += 1
+        raise AssertionError("put_object must not be called during analysis path")
+
+    @property
+    def mutation_call_count(self):
+        return (
+            self.upload_calls + self.delete_calls + self.rename_calls +
+            self.move_calls + self.create_folder_calls
+        )
+
+    @property
+    def byte_call_count(self):
+        return self.get_object_calls + self.put_object_calls
+
+
+# ─── Shared test helpers ─────────────────────────────────────────────────────
+
+async def _create_user_and_token(session, role: str = "user"):
+    from db.models import User, Quota
+    from services.auth import hash_password, create_access_token
+
+    user_id = uuid.uuid4()
+    user = User(
+        id=user_id,
+        handle=f"ana_user_{user_id.hex[:8]}",
+        email=f"ana_{user_id.hex[:8]}@example.com",
+        password_hash=hash_password("Testpassword123!"),
+        role=role,
+        is_active=True,
+        password_must_change=False,
+    )
+    quota = Quota(user_id=user_id, limit_bytes=104857600, used_bytes=0)
+    session.add(user)
+    session.add(quota)
+    await session.commit()
+    await session.refresh(user)
+
+    token = create_access_token(str(user_id), role, user_agent=_TEST_USER_AGENT)
+    return {
+        "user": user,
+        "token": token,
+        "headers": {
+            "Authorization": f"Bearer {token}",
+            "User-Agent": _TEST_USER_AGENT,
+        },
+    }
+
+
+async def _create_cloud_connection(session, user_id, provider="google_drive", status="ACTIVE"):
+    from db.models import CloudConnection
+    from storage.cloud_utils import encrypt_credentials
+    from config import settings
+
+    master_key = settings.cloud_creds_key.encode()
+    creds_enc = encrypt_credentials(
+        master_key,
+        str(user_id),
+        {"access_token": "tok", "refresh_token": "ref"},
+    )
+    conn = CloudConnection(
+        id=uuid.uuid4(),
+        user_id=user_id,
+        provider=provider,
+        display_name="Test Connection",
+        credentials_enc=creds_enc,
+        status=status,
+    )
+    session.add(conn)
+    await session.commit()
+    return conn
+
+
+async def _create_cloud_item(session, user_id, connection_id, *, name="doc.pdf",
+                              kind="file", etag="etag-v1", analysis_status="pending"):
+    from db.models import CloudItem
+
+    item = CloudItem(
+        id=uuid.uuid4(),
+        user_id=user_id,
+        connection_id=connection_id,
+        provider_item_id=f"pitem-{uuid.uuid4().hex[:8]}",
+        name=name,
+        kind=kind,
+        content_type="application/pdf",
+        provider_size=102400,
+        etag=etag,
+        analysis_status=analysis_status,
+    )
+    session.add(item)
+    await session.commit()
+    return item
+
+
+# ─── ANALYZE-01: Single file analysis estimate ───────────────────────────────
+
+async def test_estimate_single_file_returns_supported_count_and_bytes(async_client, db_session):
+    """ANALYZE-01: Single-file estimate returns count=1 and provider_bytes without downloading.
+
+    T-14-04: no bytes fetched during estimate.
+    """
+    # Requires Phase 14 endpoint: POST /api/cloud/analysis/connections/{id}/estimate
+    from api.cloud.analysis import router as analysis_router  # Phase 14 — not yet implemented
+
+    ctx = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, ctx["user"].id)
+    item = await _create_cloud_item(db_session, ctx["user"].id, conn.id)
+
+    adapter = FakeCloudAdapter()
+    with patch("services.cloud_analysis.get_adapter", return_value=adapter):
+        resp = await async_client.post(
+            f"/api/cloud/analysis/connections/{conn.id}/estimate",
+            json={"scope": "file", "provider_item_ids": [item.provider_item_id]},
+            headers=ctx["headers"],
+        )
+
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["supported_count"] == 1
+    assert body["unsupported_count"] == 0
+    assert "total_provider_bytes" in body
+    assert body["total_provider_bytes"] == item.provider_size
+
+    # T-14-04: no byte downloads during estimate
+    assert adapter.byte_call_count == 0, "Estimate must not download provider bytes"
+    # T-14-03: no mutations during estimate
+    assert adapter.mutation_call_count == 0, "Estimate must not call any provider mutation"
+
+
+# ─── ANALYZE-02: Selection estimate ─────────────────────────────────────────
+
+async def test_estimate_selection_aggregates_multiple_files(async_client, db_session):
+    """ANALYZE-02: Multi-file selection estimate aggregates counts and bytes.
+
+    T-14-04: bytes not fetched during selection estimate.
+    """
+    from api.cloud.analysis import router as analysis_router  # Phase 14
+
+    ctx = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, ctx["user"].id)
+    item1 = await _create_cloud_item(db_session, ctx["user"].id, conn.id, name="a.pdf")
+    item2 = await _create_cloud_item(db_session, ctx["user"].id, conn.id, name="b.docx")
+
+    adapter = FakeCloudAdapter()
+    with patch("services.cloud_analysis.get_adapter", return_value=adapter):
+        resp = await async_client.post(
+            f"/api/cloud/analysis/connections/{conn.id}/estimate",
+            json={
+                "scope": "selection",
+                "provider_item_ids": [item1.provider_item_id, item2.provider_item_id],
+            },
+            headers=ctx["headers"],
+        )
+
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["supported_count"] == 2
+    # total bytes = sum of provider sizes
+    assert body["total_provider_bytes"] == (item1.provider_size or 0) + (item2.provider_size or 0)
+
+    assert adapter.byte_call_count == 0
+    assert adapter.mutation_call_count == 0
+
+
+async def test_estimate_includes_unsupported_item_count(async_client, db_session):
+    """ANALYZE-02: Estimate reports unsupported items without blocking supportable ones.
+
+    Unsupported items are reported but do not cause estimate to fail (D-23).
+    """
+    from api.cloud.analysis import router as analysis_router  # Phase 14
+
+    ctx = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, ctx["user"].id)
+    pdf_item = await _create_cloud_item(db_session, ctx["user"].id, conn.id, name="report.pdf")
+    # An item with an unsupported content type (e.g. .exe) should count as unsupported
+    exe_item = await _create_cloud_item(db_session, ctx["user"].id, conn.id, name="setup.exe")
+
+    adapter = FakeCloudAdapter()
+    with patch("services.cloud_analysis.get_adapter", return_value=adapter):
+        resp = await async_client.post(
+            f"/api/cloud/analysis/connections/{conn.id}/estimate",
+            json={
+                "scope": "selection",
+                "provider_item_ids": [pdf_item.provider_item_id, exe_item.provider_item_id],
+            },
+            headers=ctx["headers"],
+        )
+
+    assert resp.status_code == 200
+    body = resp.json()
+    # At least one supported (pdf), at least one unsupported (exe)
+    assert body["supported_count"] >= 1
+    assert body["unsupported_count"] >= 1
+    assert adapter.byte_call_count == 0
+
+
+# ─── ANALYZE-03: Whole-connection estimate ───────────────────────────────────
+
+async def test_whole_connection_estimate_always_required(async_client, db_session):
+    """ANALYZE-03: Whole-connection scope always produces an estimate response before enqueue.
+
+    The estimate endpoint must be called before enqueue for connection-scope (D-02).
+    T-14-04: bytes not fetched.
+    """
+    from api.cloud.analysis import router as analysis_router  # Phase 14
+
+    ctx = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, ctx["user"].id)
+    await _create_cloud_item(db_session, ctx["user"].id, conn.id, name="file1.pdf")
+
+    adapter = FakeCloudAdapter()
+    with patch("services.cloud_analysis.get_adapter", return_value=adapter):
+        resp = await async_client.post(
+            f"/api/cloud/analysis/connections/{conn.id}/estimate",
+            json={"scope": "connection", "recursive": True},
+            headers=ctx["headers"],
+        )
+
+    assert resp.status_code == 200
+    body = resp.json()
+    # Whole-connection always includes a recursive flag in the response
+    assert "recursive" in body or body.get("scope_kind") == "connection"
+    assert "supported_count" in body
+    assert adapter.byte_call_count == 0
+
+
+async def test_folder_estimate_asks_about_recursion(async_client, db_session):
+    """ANALYZE-03/D-04: Folder-scope estimate accepts recursive flag and reports it back."""
+    from api.cloud.analysis import router as analysis_router  # Phase 14
+
+    ctx = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, ctx["user"].id)
+    folder_item = await _create_cloud_item(db_session, ctx["user"].id, conn.id,
+                                            name="Reports", kind="folder")
+
+    adapter = FakeCloudAdapter()
+    with patch("services.cloud_analysis.get_adapter", return_value=adapter):
+        resp = await async_client.post(
+            f"/api/cloud/analysis/connections/{conn.id}/estimate",
+            json={
+                "scope": "folder",
+                "provider_item_ids": [folder_item.provider_item_id],
+                "recursive": True,
+            },
+            headers=ctx["headers"],
+        )
+
+    assert resp.status_code == 200
+    body = resp.json()
+    assert "recursive" in body
+    assert adapter.byte_call_count == 0
+
+
+# ─── ANALYZE-06: Duplicate / already-current skip ────────────────────────────
+
+async def test_already_current_item_skipped_without_byte_fetch(async_client, db_session):
+    """ANALYZE-06: Item with unchanged etag is skipped without downloading provider bytes.
+
+    T-14-04: get_object must not be called for already-current items.
+    """
+    from api.cloud.analysis import router as analysis_router  # Phase 14
+
+    ctx = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, ctx["user"].id)
+    # Item already indexed with same etag
+    item = await _create_cloud_item(
+        db_session, ctx["user"].id, conn.id,
+        etag="etag-stable-v1", analysis_status="indexed"
+    )
+
+    adapter = FakeCloudAdapter()
+    with patch("services.cloud_analysis.get_adapter", return_value=adapter):
+        resp = await async_client.post(
+            f"/api/cloud/analysis/connections/{conn.id}/jobs",
+            json={
+                "scope": "file",
+                "provider_item_ids": [item.provider_item_id],
+            },
+            headers=ctx["headers"],
+        )
+
+    # Enqueue succeeds (202 Accepted or 200 with job status)
+    assert resp.status_code in (200, 202)
+    body = resp.json()
+
+    # Item reported as already_current — not re-downloaded
+    # The job should report 0 queued items (or the item appears as already_current)
+    job_id = body.get("job_id")
+    assert job_id is not None
+
+    # Fetch job status to verify already_current
+    status_resp = await async_client.get(
+        f"/api/cloud/analysis/jobs/{job_id}",
+        headers=ctx["headers"],
+    )
+    assert status_resp.status_code == 200
+    status_body = status_resp.json()
+
+    # Item should not be in "queued" state — it's already current
+    assert status_body.get("already_current_count", 0) >= 1
+    assert status_body.get("queued_count", 0) == 0
+
+    # Critical: no bytes were fetched
+    assert adapter.byte_call_count == 0, "already-current check must not download provider bytes"
+    # Critical: no mutations occurred
+    assert adapter.mutation_call_count == 0
+
+
+async def test_changed_etag_triggers_reanalysis(async_client, db_session):
+    """ANALYZE-06: Item with changed etag is queued for reanalysis (not skipped)."""
+    from api.cloud.analysis import router as analysis_router  # Phase 14
+
+    ctx = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, ctx["user"].id)
+    # Item previously indexed but etag has changed
+    item = await _create_cloud_item(
+        db_session, ctx["user"].id, conn.id,
+        etag="etag-OLD", analysis_status="indexed"
+    )
+
+    adapter = FakeCloudAdapter()
+    # Provider now reports a different etag
+    with patch("services.cloud_analysis.get_adapter", return_value=adapter):
+        with patch("services.cloud_analysis.resolve_provider_metadata",
+                   new_callable=AsyncMock,
+                   return_value={"etag": "etag-NEW", "size": 102400}):
+            resp = await async_client.post(
+                f"/api/cloud/analysis/connections/{conn.id}/jobs",
+                json={"scope": "file", "provider_item_ids": [item.provider_item_id]},
+                headers=ctx["headers"],
+            )
+
+    assert resp.status_code in (200, 202)
+    body = resp.json()
+    job_id = body.get("job_id")
+    assert job_id is not None
+
+    status_resp = await async_client.get(
+        f"/api/cloud/analysis/jobs/{job_id}",
+        headers=ctx["headers"],
+    )
+    assert status_resp.status_code == 200
+    status_body = status_resp.json()
+    # Item is queued for re-analysis (not marked already_current)
+    assert status_body.get("queued_count", 0) >= 1
+
+
+# ─── ANALYZE-07: Provider file not mutated ───────────────────────────────────
+
+async def test_analysis_never_mutates_provider_file(async_client, db_session):
+    """ANALYZE-07: Analysis job completion never calls any provider mutation method.
+
+    T-14-03: mutation counters must remain zero for the full analysis lifecycle.
+    """
+    from api.cloud.analysis import router as analysis_router  # Phase 14
+
+    ctx = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, ctx["user"].id)
+    item = await _create_cloud_item(db_session, ctx["user"].id, conn.id)
+
+    adapter = FakeCloudAdapter()
+    with patch("services.cloud_analysis.get_adapter", return_value=adapter):
+        resp = await async_client.post(
+            f"/api/cloud/analysis/connections/{conn.id}/jobs",
+            json={"scope": "file", "provider_item_ids": [item.provider_item_id]},
+            headers=ctx["headers"],
+        )
+
+    assert resp.status_code in (200, 202)
+    # No mutations at any point in analysis
+    assert adapter.mutation_call_count == 0, (
+        f"Analysis called provider mutations: upload={adapter.upload_calls}, "
+        f"delete={adapter.delete_calls}, rename={adapter.rename_calls}, "
+        f"move={adapter.move_calls}, create_folder={adapter.create_folder_calls}"
+    )
+
+
+# ─── T-14-01: IDOR — foreign user cannot estimate/enqueue ───────────────────
+
+async def test_foreign_user_cannot_estimate_another_users_connection(async_client, db_session):
+    """T-14-01: Estimate request against another user's connection returns 403/404."""
+    from api.cloud.analysis import router as analysis_router  # Phase 14
+
+    owner_ctx = await _create_user_and_token(db_session)
+    foreign_ctx = await _create_user_and_token(db_session)
+    owner_conn = await _create_cloud_connection(db_session, owner_ctx["user"].id)
+
+    resp = await async_client.post(
+        f"/api/cloud/analysis/connections/{owner_conn.id}/estimate",
+        json={"scope": "connection", "recursive": True},
+        headers=foreign_ctx["headers"],  # Foreign user's token
+    )
+
+    assert resp.status_code in (403, 404), (
+        f"Foreign user estimate must be rejected — got {resp.status_code}"
+    )
+
+
+async def test_foreign_user_cannot_enqueue_another_users_connection(async_client, db_session):
+    """T-14-01: Enqueue request against another user's connection is rejected."""
+    from api.cloud.analysis import router as analysis_router  # Phase 14
+
+    owner_ctx = await _create_user_and_token(db_session)
+    foreign_ctx = await _create_user_and_token(db_session)
+    owner_conn = await _create_cloud_connection(db_session, owner_ctx["user"].id)
+    owner_item = await _create_cloud_item(db_session, owner_ctx["user"].id, owner_conn.id)
+
+    resp = await async_client.post(
+        f"/api/cloud/analysis/connections/{owner_conn.id}/jobs",
+        json={"scope": "file", "provider_item_ids": [owner_item.provider_item_id]},
+        headers=foreign_ctx["headers"],
+    )
+
+    assert resp.status_code in (403, 404), (
+        f"Foreign user enqueue must be rejected — got {resp.status_code}"
+    )
+
+
+async def test_admin_cannot_enqueue_another_users_analysis_job(async_client, db_session):
+    """T-14-01: Admin role cannot access cloud analysis routes (same as browse restriction)."""
+    from api.cloud.analysis import router as analysis_router  # Phase 14
+
+    user_ctx = await _create_user_and_token(db_session, role="user")
+    admin_ctx = await _create_user_and_token(db_session, role="admin")
+    user_conn = await _create_cloud_connection(db_session, user_ctx["user"].id)
+    user_item = await _create_cloud_item(db_session, user_ctx["user"].id, user_conn.id)
+
+    resp = await async_client.post(
+        f"/api/cloud/analysis/connections/{user_conn.id}/jobs",
+        json={"scope": "file", "provider_item_ids": [user_item.provider_item_id]},
+        headers=admin_ctx["headers"],
+    )
+
+    assert resp.status_code in (403, 404), (
+        f"Admin must not access cloud analysis — got {resp.status_code}"
+    )
+
+
+# ─── Job status and controls ─────────────────────────────────────────────────
+
+async def test_job_status_response_excludes_credentials_and_object_keys(async_client, db_session):
+    """T-14-02: Job status response never includes credentials or raw object keys."""
+    from api.cloud.analysis import router as analysis_router  # Phase 14
+
+    ctx = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, ctx["user"].id)
+    item = await _create_cloud_item(db_session, ctx["user"].id, conn.id)
+
+    adapter = FakeCloudAdapter()
+    with patch("services.cloud_analysis.get_adapter", return_value=adapter):
+        enqueue_resp = await async_client.post(
+            f"/api/cloud/analysis/connections/{conn.id}/jobs",
+            json={"scope": "file", "provider_item_ids": [item.provider_item_id]},
+            headers=ctx["headers"],
+        )
+
+    assert enqueue_resp.status_code in (200, 202)
+    job_id = enqueue_resp.json().get("job_id")
+    assert job_id is not None
+
+    status_resp = await async_client.get(
+        f"/api/cloud/analysis/jobs/{job_id}",
+        headers=ctx["headers"],
+    )
+    assert status_resp.status_code == 200
+    body_text = status_resp.text
+
+    # T-14-02: No credential fields or raw object keys in response
+    forbidden_fields = [
+        "credentials_enc", "access_token", "refresh_token",
+        "client_secret", "object_key",
+    ]
+    for field in forbidden_fields:
+        assert field not in body_text, (
+            f"Sensitive field '{field}' must not appear in job status response"
+        )
+
+
+async def test_foreign_user_cannot_read_job_status(async_client, db_session):
+    """T-14-01: Foreign user cannot read job status for another user's job."""
+    from api.cloud.analysis import router as analysis_router  # Phase 14
+
+    owner_ctx = await _create_user_and_token(db_session)
+    foreign_ctx = await _create_user_and_token(db_session)
+    owner_conn = await _create_cloud_connection(db_session, owner_ctx["user"].id)
+    owner_item = await _create_cloud_item(db_session, owner_ctx["user"].id, owner_conn.id)
+
+    adapter = FakeCloudAdapter()
+    with patch("services.cloud_analysis.get_adapter", return_value=adapter):
+        enqueue_resp = await async_client.post(
+            f"/api/cloud/analysis/connections/{owner_conn.id}/jobs",
+            json={"scope": "file", "provider_item_ids": [owner_item.provider_item_id]},
+            headers=owner_ctx["headers"],
+        )
+
+    job_id = enqueue_resp.json().get("job_id")
+
+    # Foreign user tries to read the job
+    resp = await async_client.get(
+        f"/api/cloud/analysis/jobs/{job_id}",
+        headers=foreign_ctx["headers"],
+    )
+    assert resp.status_code in (403, 404)
+
+
+async def test_foreign_user_cannot_cancel_another_users_job(async_client, db_session):
+    """T-14-01: Foreign user cannot cancel a job owned by another user."""
+    from api.cloud.analysis import router as analysis_router  # Phase 14
+
+    owner_ctx = await _create_user_and_token(db_session)
+    foreign_ctx = await _create_user_and_token(db_session)
+    owner_conn = await _create_cloud_connection(db_session, owner_ctx["user"].id)
+    owner_item = await _create_cloud_item(db_session, owner_ctx["user"].id, owner_conn.id)
+
+    adapter = FakeCloudAdapter()
+    with patch("services.cloud_analysis.get_adapter", return_value=adapter):
+        enqueue_resp = await async_client.post(
+            f"/api/cloud/analysis/connections/{owner_conn.id}/jobs",
+            json={"scope": "file", "provider_item_ids": [owner_item.provider_item_id]},
+            headers=owner_ctx["headers"],
+        )
+
+    job_id = enqueue_resp.json().get("job_id")
+
+    resp = await async_client.post(
+        f"/api/cloud/analysis/jobs/{job_id}/cancel",
+        headers=foreign_ctx["headers"],
+    )
+    assert resp.status_code in (403, 404)
+
+
+# ─── Cancel and retry state transitions ──────────────────────────────────────
+
+async def test_cancel_job_transitions_to_cancelled_status(async_client, db_session):
+    """ANALYZE-05: Cancel job transitions the job status to 'cancelled'."""
+    from api.cloud.analysis import router as analysis_router  # Phase 14
+
+    ctx = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, ctx["user"].id)
+    item = await _create_cloud_item(db_session, ctx["user"].id, conn.id)
+
+    adapter = FakeCloudAdapter()
+    with patch("services.cloud_analysis.get_adapter", return_value=adapter):
+        enqueue_resp = await async_client.post(
+            f"/api/cloud/analysis/connections/{conn.id}/jobs",
+            json={"scope": "file", "provider_item_ids": [item.provider_item_id]},
+            headers=ctx["headers"],
+        )
+
+    job_id = enqueue_resp.json().get("job_id")
+
+    cancel_resp = await async_client.post(
+        f"/api/cloud/analysis/jobs/{job_id}/cancel",
+        headers=ctx["headers"],
+    )
+    assert cancel_resp.status_code in (200, 202, 204)
+
+    status_resp = await async_client.get(
+        f"/api/cloud/analysis/jobs/{job_id}",
+        headers=ctx["headers"],
+    )
+    assert status_resp.json().get("status") in ("cancelled", "cancelling")
+
+
+async def test_retry_failed_item_requeues_it(async_client, db_session):
+    """ANALYZE-05: Retry on a failed item transitions it back to queued status."""
+    from api.cloud.analysis import router as analysis_router  # Phase 14
+
+    ctx = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, ctx["user"].id)
+    item = await _create_cloud_item(db_session, ctx["user"].id, conn.id)
+
+    adapter = FakeCloudAdapter()
+    with patch("services.cloud_analysis.get_adapter", return_value=adapter):
+        enqueue_resp = await async_client.post(
+            f"/api/cloud/analysis/connections/{conn.id}/jobs",
+            json={"scope": "file", "provider_item_ids": [item.provider_item_id]},
+            headers=ctx["headers"],
+        )
+
+    job_id = enqueue_resp.json().get("job_id")
+
+    # Simulate item in failed state then retry
+    resp = await async_client.post(
+        f"/api/cloud/analysis/jobs/{job_id}/items/{item.id}/retry",
+        headers=ctx["headers"],
+    )
+    # Must return a valid status that indicates retry was accepted
+    assert resp.status_code in (200, 202, 204)
diff --git a/backend/tests/test_cloud_cache.py b/backend/tests/test_cloud_cache.py
new file mode 100644
index 0000000..a6adf4b
--- /dev/null
+++ b/backend/tests/test_cloud_cache.py
@@ -0,0 +1,495 @@
+"""
+Phase 14 Plan 01 — RED contract tests for the byte cache service.
+
+Requirements: CACHE-03, CACHE-04, CACHE-05
+
+Threat coverage:
+  T-14-01 — IDOR: cache metadata queries are always owner-scoped
+  T-14-02 — Object-key secrecy: object_key never appears in API responses
+  T-14-05 — Cache quota corruption: quota must be atomically incremented on
+             cache retain and decremented on eviction; a safe in-memory
+             simulation is used because SQLite cannot run the atomic UPDATE.
+
+All tests import non-existent Phase 14 modules and must FAIL due to
+ImportError until implementation is complete.
+"""
+from __future__ import annotations
+
+import uuid
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+pytestmark = pytest.mark.asyncio
+
+from tests.conftest import _TEST_USER_AGENT
+
+
+# ─── Helpers ─────────────────────────────────────────────────────────────────
+
+async def _create_user_and_token(session, role: str = "user"):
+    from db.models import User, Quota
+    from services.auth import hash_password, create_access_token
+
+    user_id = uuid.uuid4()
+    user = User(
+        id=user_id,
+        handle=f"cache_user_{user_id.hex[:8]}",
+        email=f"cache_{user_id.hex[:8]}@example.com",
+        password_hash=hash_password("Testpassword123!"),
+        role=role,
+        is_active=True,
+        password_must_change=False,
+    )
+    quota = Quota(user_id=user_id, limit_bytes=104857600, used_bytes=0)
+    session.add(user)
+    session.add(quota)
+    await session.commit()
+    await session.refresh(user)
+
+    token = create_access_token(str(user_id), role, user_agent=_TEST_USER_AGENT)
+    return {
+        "user": user,
+        "token": token,
+        "headers": {
+            "Authorization": f"Bearer {token}",
+            "User-Agent": _TEST_USER_AGENT,
+        },
+    }
+
+
+async def _create_cloud_connection(session, user_id, provider="google_drive", status="ACTIVE"):
+    from db.models import CloudConnection
+    from storage.cloud_utils import encrypt_credentials
+    from config import settings
+
+    master_key = settings.cloud_creds_key.encode()
+    creds_enc = encrypt_credentials(
+        master_key,
+        str(user_id),
+        {"access_token": "tok", "refresh_token": "ref"},
+    )
+    conn = CloudConnection(
+        id=uuid.uuid4(),
+        user_id=user_id,
+        provider=provider,
+        display_name="Cache Test Connection",
+        credentials_enc=creds_enc,
+        status=status,
+    )
+    session.add(conn)
+    await session.commit()
+    return conn
+
+
+async def _create_cloud_item(session, user_id, connection_id, *, name="doc.pdf",
+                              kind="file", etag="etag-v1"):
+    from db.models import CloudItem
+
+    item = CloudItem(
+        id=uuid.uuid4(),
+        user_id=user_id,
+        connection_id=connection_id,
+        provider_item_id=f"pitem-{uuid.uuid4().hex[:8]}",
+        name=name,
+        kind=kind,
+        content_type="application/pdf",
+        provider_size=102400,
+        etag=etag,
+        analysis_status="pending",
+    )
+    session.add(item)
+    await session.commit()
+    return item
+
+
+# ─── CACHE-03: Cache bytes only during open/preview/analysis ─────────────────
+
+async def test_cache_entry_created_during_analysis_job(db_session):
+    """CACHE-03: A cache entry is created when bytes are hydrated for analysis.
+
+    Imports CloudByteCacheEntry (Phase 14 schema addition).
+    """
+    from services.cloud_cache import create_cache_entry  # Phase 14
+
+    user_id = uuid.uuid4()
+    conn_id = uuid.uuid4()
+    item_id = uuid.uuid4()
+
+    entry = await create_cache_entry(
+        session=db_session,
+        user_id=user_id,
+        connection_id=conn_id,
+        cloud_item_id=item_id,
+        provider_item_id="pitem-001",
+        version_key="etag-v1",
+        object_key=f"cache/{user_id}/{uuid.uuid4()}.pdf",
+        content_type="application/pdf",
+        size_bytes=102400,
+    )
+
+    assert entry is not None
+    assert entry.user_id == user_id
+    assert entry.cloud_item_id == item_id
+    assert entry.size_bytes == 102400
+    assert entry.pin_count == 0
+    assert entry.active_job_count == 0
+
+
+async def test_browse_without_analysis_does_not_create_cache_entry(db_session):
+    """CACHE-03: Listing a folder never creates a cache entry (no byte download).
+
+    Verifies that reconcile_cloud_listing (existing Phase 12/13 code path) leaves
+    cloud_byte_cache_entries empty.
+    """
+    from db.models import CloudConnection, CloudItem
+    from sqlalchemy import select
+
+    # Try to import Phase 14 cache model
+    try:
+        from db.models import CloudByteCacheEntry  # Phase 14 — not yet in models
+    except ImportError:
+        pytest.fail(
+            "CloudByteCacheEntry model must be added to db/models.py in Phase 14"
+        )
+
+    user_id = uuid.uuid4()
+    # After any browse/reconcile operation, zero cache entries should exist
+    result = await db_session.execute(
+        select(CloudByteCacheEntry).where(CloudByteCacheEntry.user_id == user_id)
+    )
+    rows = result.scalars().all()
+    assert len(rows) == 0, "Browse operations must not create cache entries"
+
+
+# ─── CACHE-04: LRU eviction and cache limits ─────────────────────────────────
+
+async def test_eviction_does_not_remove_pinned_entry(db_session):
+    """CACHE-04/D-16: Eviction must not evict entries with active jobs or pins.
+
+    An entry with pin_count > 0 must survive LRU eviction.
+    """
+    from services.cloud_cache import create_cache_entry, evict_lru_entries  # Phase 14
+
+    user_id = uuid.uuid4()
+    conn_id = uuid.uuid4()
+
+    # Create a pinned cache entry (pin_count=1 simulates active analysis/preview)
+    pinned = await create_cache_entry(
+        session=db_session,
+        user_id=user_id,
+        connection_id=conn_id,
+        cloud_item_id=uuid.uuid4(),
+        provider_item_id="pinned-item",
+        version_key="etag-pinned",
+        object_key=f"cache/{user_id}/{uuid.uuid4()}.pdf",
+        content_type="application/pdf",
+        size_bytes=10 * 1024 * 1024,  # 10 MB
+        pin_count=1,  # pinned — must survive eviction
+    )
+
+    # Create an unpinned entry to confirm eviction can happen at all
+    evictable = await create_cache_entry(
+        session=db_session,
+        user_id=user_id,
+        connection_id=conn_id,
+        cloud_item_id=uuid.uuid4(),
+        provider_item_id="evictable-item",
+        version_key="etag-evict",
+        object_key=f"cache/{user_id}/{uuid.uuid4()}.pdf",
+        content_type="application/pdf",
+        size_bytes=5 * 1024 * 1024,  # 5 MB
+        pin_count=0,
+    )
+
+    # Run eviction with a tight byte limit that forces eviction of unpinned entries
+    evicted_ids = await evict_lru_entries(
+        session=db_session,
+        user_id=user_id,
+        max_cache_bytes=1,  # 1 byte limit forces eviction
+    )
+
+    # The pinned entry must NOT be evicted
+    assert str(pinned.id) not in [str(e) for e in evicted_ids], (
+        "Eviction must preserve entries with pin_count > 0 (D-16)"
+    )
+
+
+async def test_eviction_removes_oldest_unpinned_entry_first(db_session):
+    """CACHE-04/D-14: LRU eviction removes least-recently-accessed unpinned entry first."""
+    from services.cloud_cache import create_cache_entry, evict_lru_entries  # Phase 14
+    from datetime import datetime, timezone, timedelta
+
+    user_id = uuid.uuid4()
+    conn_id = uuid.uuid4()
+
+    older_entry = await create_cache_entry(
+        session=db_session,
+        user_id=user_id,
+        connection_id=conn_id,
+        cloud_item_id=uuid.uuid4(),
+        provider_item_id="old-item",
+        version_key="etag-old",
+        object_key=f"cache/{user_id}/{uuid.uuid4()}.pdf",
+        content_type="application/pdf",
+        size_bytes=5 * 1024 * 1024,
+        pin_count=0,
+        # Simulate last accessed 2 hours ago
+        last_accessed_at=datetime.now(timezone.utc) - timedelta(hours=2),
+    )
+
+    newer_entry = await create_cache_entry(
+        session=db_session,
+        user_id=user_id,
+        connection_id=conn_id,
+        cloud_item_id=uuid.uuid4(),
+        provider_item_id="new-item",
+        version_key="etag-new",
+        object_key=f"cache/{user_id}/{uuid.uuid4()}.pdf",
+        content_type="application/pdf",
+        size_bytes=5 * 1024 * 1024,
+        pin_count=0,
+        # Simulate last accessed 1 minute ago
+        last_accessed_at=datetime.now(timezone.utc) - timedelta(minutes=1),
+    )
+
+    # Evict enough to remove exactly one entry (limit below combined size)
+    evicted_ids = await evict_lru_entries(
+        session=db_session,
+        user_id=user_id,
+        max_cache_bytes=6 * 1024 * 1024,  # fits one entry; must evict the older one
+    )
+
+    # The older entry should be evicted first (LRU order)
+    assert str(older_entry.id) in [str(e) for e in evicted_ids], (
+        "LRU eviction must remove least-recently-accessed entry first"
+    )
+    # The newer entry should survive
+    assert str(newer_entry.id) not in [str(e) for e in evicted_ids]
+
+
+# ─── CACHE-04/T-14-05: Quota accounting ─────────────────────────────────────
+
+async def test_cache_retain_increments_quota(db_session):
+    """T-14-05: Creating a cache entry must atomically increment used_bytes in quotas.
+
+    The increment requirement is tested via the service layer; the actual atomic
+    SQL UPDATE…RETURNING runs during implementation against real PostgreSQL.
+    In SQLite we verify the function signature exists and accepts the right arguments.
+    """
+    from services.cloud_cache import increment_quota_for_cache  # Phase 14
+
+    # Verify the function exists with the expected signature
+    import inspect
+    sig = inspect.signature(increment_quota_for_cache)
+    params = list(sig.parameters.keys())
+
+    # Must accept session, user_id, and size_bytes
+    assert "session" in params, "increment_quota_for_cache must accept session"
+    assert "user_id" in params, "increment_quota_for_cache must accept user_id"
+    assert "size_bytes" in params, "increment_quota_for_cache must accept size_bytes"
+
+
+async def test_cache_eviction_decrements_quota(db_session):
+    """T-14-05: Evicting a cache entry must decrement used_bytes in quotas.
+
+    Verifies the decrement function signature exists (full atomic SQL in PostgreSQL).
+    """
+    from services.cloud_cache import decrement_quota_for_cache  # Phase 14
+
+    import inspect
+    sig = inspect.signature(decrement_quota_for_cache)
+    params = list(sig.parameters.keys())
+
+    assert "session" in params, "decrement_quota_for_cache must accept session"
+    assert "user_id" in params, "decrement_quota_for_cache must accept user_id"
+    assert "size_bytes" in params, "decrement_quota_for_cache must accept size_bytes"
+
+
+# ─── CACHE-05: Owner isolation ───────────────────────────────────────────────
+
+async def test_cache_entry_query_is_owner_scoped(db_session):
+    """CACHE-05/T-14-01: Cache metadata queries are scoped to owner.
+
+    Foreign user must not see another user's cache entries.
+    """
+    from services.cloud_cache import create_cache_entry, list_cache_entries  # Phase 14
+
+    owner_id = uuid.uuid4()
+    foreign_id = uuid.uuid4()
+    conn_id = uuid.uuid4()
+
+    # Create entry for owner
+    await create_cache_entry(
+        session=db_session,
+        user_id=owner_id,
+        connection_id=conn_id,
+        cloud_item_id=uuid.uuid4(),
+        provider_item_id="owner-item",
+        version_key="etag-v1",
+        object_key=f"cache/{owner_id}/{uuid.uuid4()}.pdf",
+        content_type="application/pdf",
+        size_bytes=1024,
+    )
+
+    # Foreign user should see zero entries
+    foreign_entries = await list_cache_entries(session=db_session, user_id=foreign_id)
+    assert len(foreign_entries) == 0, (
+        "Cache entries must be invisible to non-owner users (CACHE-05)"
+    )
+
+    # Owner should see their own entries
+    owner_entries = await list_cache_entries(session=db_session, user_id=owner_id)
+    assert len(owner_entries) == 1
+
+
+async def test_cache_api_excludes_object_key_from_response(async_client, db_session):
+    """T-14-02: Cache settings/status API response never includes the MinIO object key.
+
+    Object keys are internal and must not appear in any API response.
+    """
+    from api.cloud.analysis import router as analysis_router  # Phase 14
+
+    ctx = await _create_user_and_token(db_session)
+
+    # GET cache settings/status endpoint
+    resp = await async_client.get(
+        "/api/cloud/analysis/cache",
+        headers=ctx["headers"],
+    )
+
+    # May return 200 (empty state) or 404/422 if the endpoint exists
+    assert resp.status_code != 500, "Cache status endpoint must not 500"
+
+    if resp.status_code == 200:
+        body_text = resp.text
+        # T-14-02: object_key must not appear in any response
+        assert "object_key" not in body_text, (
+            "MinIO object_key must not appear in cache API responses (T-14-02)"
+        )
+
+
+async def test_admin_cannot_read_user_cache_status(async_client, db_session):
+    """T-14-01/CACHE-05: Admin cannot access user's cache metadata via the API."""
+    from api.cloud.analysis import router as analysis_router  # Phase 14
+
+    user_ctx = await _create_user_and_token(db_session, role="user")
+    admin_ctx = await _create_user_and_token(db_session, role="admin")
+
+    # Admin tries to read the user's cache
+    resp = await async_client.get(
+        "/api/cloud/analysis/cache",
+        headers=admin_ctx["headers"],
+    )
+
+    # Admin is blocked from cloud analysis routes entirely (same as browse/mutations)
+    assert resp.status_code in (403, 404), (
+        f"Admin must not read cache metadata — got {resp.status_code}"
+    )
+
+
+# ─── Version-key helper ───────────────────────────────────────────────────────
+
+async def test_version_key_prefers_version_over_etag():
+    """ANALYZE-06/D-19: Version key computation prefers 'version' field over etag.
+
+    Precedence: version > etag > metadata fingerprint > content hash.
+    Content hash must never be a pre-download requirement.
+    """
+    from services.cloud_cache import compute_version_key  # Phase 14
+
+    # version takes priority
+    key_with_version = compute_version_key(
+        provider_item_id="item-001",
+        version="v3",
+        etag="etag-stale",
+        size=1024,
+        modified_at="2026-06-01T00:00:00Z",
+        content_type="application/pdf",
+    )
+    key_from_etag_only = compute_version_key(
+        provider_item_id="item-001",
+        version=None,
+        etag="etag-stable",
+        size=1024,
+        modified_at="2026-06-01T00:00:00Z",
+        content_type="application/pdf",
+    )
+    key_from_fingerprint = compute_version_key(
+        provider_item_id="item-001",
+        version=None,
+        etag=None,
+        size=1024,
+        modified_at="2026-06-01T00:00:00Z",
+        content_type="application/pdf",
+    )
+
+    # Keys derived from different inputs must differ
+    assert key_with_version != key_from_etag_only, (
+        "version-derived key must differ from etag-only key"
+    )
+    assert key_from_etag_only != key_from_fingerprint, (
+        "etag-derived key must differ from metadata-fingerprint key"
+    )
+
+    # Same inputs must produce stable keys (deterministic)
+    key_repeat = compute_version_key(
+        provider_item_id="item-001",
+        version="v3",
+        etag="etag-stale",
+        size=1024,
+        modified_at="2026-06-01T00:00:00Z",
+        content_type="application/pdf",
+    )
+    assert key_with_version == key_repeat, "compute_version_key must be deterministic"
+
+
+async def test_version_key_fallback_to_fingerprint_without_etag():
+    """ANALYZE-06/D-19: Version key falls back to metadata fingerprint when etag absent."""
+    from services.cloud_cache import compute_version_key  # Phase 14
+
+    key = compute_version_key(
+        provider_item_id="item-fingerprint",
+        version=None,
+        etag=None,
+        size=2048,
+        modified_at="2026-06-15T10:00:00Z",
+        content_type="text/plain",
+    )
+
+    assert key is not None
+    assert isinstance(key, str)
+    assert len(key) > 0
+
+    # Same metadata must produce same fingerprint key
+    key_repeat = compute_version_key(
+        provider_item_id="item-fingerprint",
+        version=None,
+        etag=None,
+        size=2048,
+        modified_at="2026-06-15T10:00:00Z",
+        content_type="text/plain",
+    )
+    assert key == key_repeat
+
+
+async def test_content_hash_not_required_during_version_key_computation():
+    """ANALYZE-06/D-20: Content hash is not computed during version key resolution.
+
+    Passing content_hash=None must not cause a byte-fetch fallback.
+    """
+    from services.cloud_cache import compute_version_key  # Phase 14
+
+    # Providing no content_hash still produces a valid key via metadata fingerprint
+    key = compute_version_key(
+        provider_item_id="item-no-hash",
+        version=None,
+        etag=None,
+        size=512,
+        modified_at="2026-06-20T12:00:00Z",
+        content_type="application/pdf",
+        content_hash=None,  # Must be accepted without triggering any download
+    )
+
+    assert key is not None
+    assert isinstance(key, str)