feat(14-04): add analysis API routes and typed schemas

- POST /analysis/connections/{id}/estimate: scope estimate without bytes (T-14-04)
- POST /analysis/connections/{id}/jobs: enqueue job, returns job_id (ANALYZE-01)
- GET /analysis/jobs, GET /analysis/jobs/{id}: job status list and detail
- POST /analysis/jobs/{id}/cancel: batch cancel (queued items immediate; running cooperative)
- POST /analysis/jobs/{id}/items/{item_id}/cancel|skip|retry: per-item controls (ANALYZE-05)
- AnalysisEstimateRequest/Out, AnalysisEnqueueRequest/Out, AnalysisJobOut: typed schemas
- AnalysisJobOut: both simple (waiting/working/done/skipped/failed) and per-stage counts
- All routes use get_regular_user — admin blocked (T-14-01)
- Response schemas exclude credentials_enc, object_key, raw provider URLs (T-14-02)
- 14/14 test_cloud_analysis_api.py pass; 7 new analysis security tests pass (33/33 total)

backend/api/cloud/analysis.py

@@ -5,25 +5,518 @@ Provides the /analysis sub-router tree under /api/cloud.  All routes are
 owner-scoped and blocked for admin accounts (get_regular_user enforced at
 each leaf router).
 
-Currently registered routes:
+Route family (Plan 03 cache + Plan 04 jobs/controls):
-  GET  /analysis/cache          — cache usage/settings (cloud/cache.py)
+  GET   /analysis/cache                                   — cache usage/settings
-  PATCH /analysis/cache/settings — update settings    (cloud/cache.py)
+  PATCH /analysis/cache/settings                          — update settings
+  POST  /analysis/connections/{id}/estimate               — scope estimate (no bytes)
+  POST  /analysis/connections/{id}/jobs                   — enqueue analysis job
+  GET   /analysis/jobs                                    — list jobs
+  GET   /analysis/jobs/{job_id}                           — job status
+  POST  /analysis/jobs/{job_id}/cancel                    — cancel batch
+  POST  /analysis/jobs/{job_id}/items/{item_id}/cancel    — cancel single item
+  POST  /analysis/jobs/{job_id}/items/{item_id}/skip      — skip single item
+  POST  /analysis/jobs/{job_id}/items/{item_id}/retry     — retry failed item
 
-Future routes (Phase 14 plans 04-09):
+Security invariants (T-14-01, T-14-02, T-14-03):
-  POST  /analysis/connections/{id}/estimate
+  - All routes use get_regular_user — admin accounts blocked.
-  POST  /analysis/connections/{id}/jobs
+  - All routes are owner-scoped: connection_id and job_id resolved against user.
-  GET   /analysis/jobs/{id}
+  - Response schemas never include credentials_enc, object_key, or provider URLs.
-  POST  /analysis/jobs/{id}/cancel
+  - No provider mutation methods are called in this module.
-  POST  /analysis/jobs/{id}/items/{item_id}/skip
 
 Module is importable as api.cloud.analysis for tests and callers that verify
 router registration before endpoint implementation is complete.
 """
 from __future__ import annotations
 
-from fastapi import APIRouter
+import uuid
+from typing import List, Optional
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
+from fastapi.responses import JSONResponse
+from sqlalchemy.ext.asyncio import AsyncSession
 
 from api.cloud.cache import router as cache_router
+from api.cloud.schemas import (
+    AnalysisControlOut,
+    AnalysisEnqueueOut,
+    AnalysisEnqueueRequest,
+    AnalysisEstimateOut,
+    AnalysisEstimateRequest,
+    AnalysisJobItemOut,
+    AnalysisJobOut,
+)
+from db.models import User
+from deps.auth import get_regular_user
+from deps.db import get_db
+from services.cloud_analysis import (
+    AnalysisItemNotFound,
+    AnalysisJobNotFound,
+    cancel_job,
+    cancel_job_item,
+    enqueue_analysis_job,
+    estimate_scope,
+    get_analysis_job,
+    list_analysis_jobs,
+    list_job_items,
+    retry_job_item,
+    skip_job_item,
+    InvalidJobState,
+)
+from services.cloud_items import ConnectionNotFound
+from services.rate_limiting import account_limiter
+
 
 router = APIRouter()
 router.include_router(cache_router)
+
+# ── Estimate ───────────────────────────────────────────────────────────────────
+
+
+@router.post(
+    "/analysis/connections/{connection_id}/estimate",
+    response_model=AnalysisEstimateOut,
+)
+@account_limiter.limit("60/minute")
+async def estimate_analysis_scope(
+    request: Request,
+    connection_id: uuid.UUID,
+    body: AnalysisEstimateRequest,
+    session: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_regular_user),
+) -> AnalysisEstimateOut:
+    """Estimate analysis scope without downloading provider bytes.
+
+    Returns counts and byte totals for the requested scope.  No bytes are
+    fetched from the provider.  No provider mutations are made.
+
+    T-14-04: bytes never downloaded during estimate.
+    T-14-03: no provider mutations.
+    T-14-01: connection must be owned by current_user.
+    """
+    request.state.current_user = current_user
+
+    try:
+        result = await estimate_scope(
+            session,
+            user_id=current_user.id,
+            connection_id=connection_id,
+            scope=body.scope,
+            provider_item_ids=body.provider_item_ids,
+            recursive=body.recursive,
+        )
+    except ConnectionNotFound as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=str(exc),
+        ) from exc
+    except ValueError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=str(exc),
+        ) from exc
+
+    return AnalysisEstimateOut(
+        supported_count=result.supported_count,
+        unsupported_count=result.unsupported_count,
+        total_provider_bytes=result.total_provider_bytes,
+        recursive=result.recursive,
+        is_partial=result.is_partial,
+        scope_kind=result.scope_kind,
+    )
+
+
+# ── Enqueue ────────────────────────────────────────────────────────────────────
+
+
+@router.post(
+    "/analysis/connections/{connection_id}/jobs",
+    response_model=AnalysisEnqueueOut,
+    status_code=202,
+)
+@account_limiter.limit("30/minute")
+async def enqueue_job(
+    request: Request,
+    connection_id: uuid.UUID,
+    body: AnalysisEnqueueRequest,
+    session: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_regular_user),
+) -> AnalysisEnqueueOut:
+    """Enqueue an analysis job for the requested scope.
+
+    Creates a CloudAnalysisJob and populates job items.  Already-current items
+    are skipped without downloading bytes.  Returns a stable job_id for polling.
+
+    Accepts failure_behavior override for this job (D-11).
+
+    T-14-01: connection must be owned by current_user.
+    T-14-03: no provider mutations.
+    T-14-04: no bytes downloaded for already-current checks.
+    """
+    request.state.current_user = current_user
+
+    # Validate failure_behavior before calling the service
+    valid_fb = {"pause_batch", "continue_item"}
+    if body.failure_behavior not in valid_fb:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=f"Invalid failure_behavior {body.failure_behavior!r}. Valid values: {sorted(valid_fb)}",
+        )
+
+    try:
+        result = await enqueue_analysis_job(
+            session,
+            user_id=current_user.id,
+            connection_id=connection_id,
+            scope=body.scope,
+            provider_item_ids=body.provider_item_ids,
+            recursive=body.recursive,
+            failure_behavior=body.failure_behavior,
+        )
+    except ConnectionNotFound as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=str(exc),
+        ) from exc
+    except ValueError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=str(exc),
+        ) from exc
+
+    await session.commit()
+
+    return AnalysisEnqueueOut(
+        job_id=str(result.job_id),
+        status="queued",
+        total_count=result.total_count,
+        queued_count=result.queued_count,
+        already_current_count=result.already_current_count,
+        unsupported_count=result.unsupported_count,
+    )
+
+
+# ── Job status / list ──────────────────────────────────────────────────────────
+
+
+def _build_job_out(job, *, detail: bool = False) -> AnalysisJobOut:
+    """Build a credential-free AnalysisJobOut from a CloudAnalysisJob row.
+
+    Simple mode aggregates internal statuses into human-friendly labels:
+      waiting = queued
+      working = downloading + extracting + classifying
+      done    = indexed + already_current
+      skipped = cancelled + unsupported
+      failed  = failed
+
+    Detailed mode includes per-stage counts.
+    """
+    # Simple aggregate labels
+    waiting_count = job.queued_count or 0
+    working_count = (
+        (job.downloading_count or 0)
+        + (job.extracting_count or 0)
+        + (job.classifying_count or 0)
+    )
+    done_count = (job.indexed_count or 0) + (job.already_current_count or 0)
+    skipped_count = (job.cancelled_count or 0) + (job.unsupported_count or 0)
+    failed_count = job.failed_count or 0
+
+    out = AnalysisJobOut(
+        job_id=str(job.id),
+        connection_id=str(job.connection_id),
+        scope_kind=job.scope_kind,
+        status=job.status,
+        failure_behavior=job.failure_behavior,
+        recursive=job.recursive,
+        total_count=job.total_count or 0,
+        waiting_count=waiting_count,
+        working_count=working_count,
+        done_count=done_count,
+        skipped_count=skipped_count,
+        failed_count=failed_count,
+        # Per-stage counts always included (0 when not in detailed mode)
+        queued_count=job.queued_count or 0,
+        downloading_count=job.downloading_count or 0,
+        extracting_count=job.extracting_count or 0,
+        classifying_count=job.classifying_count or 0,
+        indexed_count=job.indexed_count or 0,
+        already_current_count=job.already_current_count or 0,
+        cancelled_count=job.cancelled_count or 0,
+        unsupported_count=job.unsupported_count or 0,
+        created_at=job.created_at,
+        started_at=job.started_at,
+        finished_at=job.finished_at,
+    )
+
+    return out
+
+
+@router.get(
+    "/analysis/jobs",
+    response_model=List[AnalysisJobOut],
+)
+@account_limiter.limit("120/minute")
+async def list_jobs(
+    request: Request,
+    connection_id: Optional[uuid.UUID] = Query(default=None),
+    job_status: Optional[str] = Query(default=None, alias="status"),
+    detail: bool = Query(default=False),
+    session: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_regular_user),
+) -> List[AnalysisJobOut]:
+    """List analysis jobs owned by the current user.
+
+    Optional filters: connection_id, status.
+    Use ?detail=true for per-stage counts (ANALYZE-04).
+    """
+    request.state.current_user = current_user
+
+    jobs = await list_analysis_jobs(
+        session,
+        user_id=current_user.id,
+        connection_id=connection_id,
+        status=job_status,
+    )
+    return [_build_job_out(j, detail=detail) for j in jobs]
+
+
+@router.get(
+    "/analysis/jobs/{job_id}",
+    response_model=AnalysisJobOut,
+)
+@account_limiter.limit("120/minute")
+async def get_job_status(
+    request: Request,
+    job_id: uuid.UUID,
+    detail: bool = Query(default=False),
+    session: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_regular_user),
+) -> AnalysisJobOut:
+    """Return job status for a single analysis job.
+
+    Use ?detail=true for per-stage item counts (ANALYZE-04).
+    Response never includes credentials, object_key, or raw provider data.
+    """
+    request.state.current_user = current_user
+
+    try:
+        job = await get_analysis_job(session, job_id=job_id, user_id=current_user.id)
+    except AnalysisJobNotFound as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=str(exc),
+        ) from exc
+
+    return _build_job_out(job, detail=detail)
+
+
+# ── Cancel job ─────────────────────────────────────────────────────────────────
+
+
+@router.post(
+    "/analysis/jobs/{job_id}/cancel",
+    response_model=AnalysisControlOut,
+)
+@account_limiter.limit("30/minute")
+async def cancel_analysis_job(
+    request: Request,
+    job_id: uuid.UUID,
+    session: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_regular_user),
+) -> AnalysisControlOut:
+    """Cancel a queued or running analysis batch.
+
+    Queued items transition to cancelled immediately.
+    Running items are marked for cooperative cancellation.
+
+    Returns a typed result — not HTTPException — so kind/reason appear at
+    response top level (Phase 13 mutation response style).
+    """
+    request.state.current_user = current_user
+
+    try:
+        await cancel_job(session, job_id=job_id, user_id=current_user.id)
+    except AnalysisJobNotFound as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=str(exc),
+        ) from exc
+    except InvalidJobState:
+        # Already terminal — still return 200 with typed reason
+        return AnalysisControlOut(
+            kind="cancelled",
+            reason="already_terminal",
+            job_id=str(job_id),
+        )
+
+    await session.commit()
+
+    return AnalysisControlOut(
+        kind="cancelled",
+        reason="batch_cancelled",
+        job_id=str(job_id),
+    )
+
+
+# ── Cancel item ────────────────────────────────────────────────────────────────
+
+
+@router.post(
+    "/analysis/jobs/{job_id}/items/{item_id}/cancel",
+    response_model=AnalysisControlOut,
+)
+@account_limiter.limit("60/minute")
+async def cancel_analysis_item(
+    request: Request,
+    job_id: uuid.UUID,
+    item_id: uuid.UUID,
+    session: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_regular_user),
+) -> AnalysisControlOut:
+    """Cancel a single queued or in-flight job item.
+
+    item_id is the cloud_item_id (DocuVault stable UUID).
+    """
+    request.state.current_user = current_user
+
+    try:
+        await cancel_job_item(
+            session,
+            job_id=job_id,
+            item_id=item_id,
+            user_id=current_user.id,
+        )
+    except AnalysisJobNotFound as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=str(exc),
+        ) from exc
+    except AnalysisItemNotFound as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=str(exc),
+        ) from exc
+    except InvalidJobState as exc:
+        return JSONResponse(
+            status_code=status.HTTP_409_CONFLICT,
+            content={"kind": "invalid_state", "reason": str(exc)},
+        )
+
+    await session.commit()
+
+    return AnalysisControlOut(
+        kind="cancelled",
+        reason="item_cancelled",
+        job_id=str(job_id),
+        item_id=str(item_id),
+    )
+
+
+# ── Skip item ──────────────────────────────────────────────────────────────────
+
+
+@router.post(
+    "/analysis/jobs/{job_id}/items/{item_id}/skip",
+    response_model=AnalysisControlOut,
+)
+@account_limiter.limit("60/minute")
+async def skip_analysis_item(
+    request: Request,
+    job_id: uuid.UUID,
+    item_id: uuid.UUID,
+    session: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_regular_user),
+) -> AnalysisControlOut:
+    """Skip a queued or failed job item.
+
+    Marks the item as cancelled (skip is a user-initiated cancel variant).
+    item_id is the cloud_item_id (DocuVault stable UUID).
+    """
+    request.state.current_user = current_user
+
+    try:
+        await skip_job_item(
+            session,
+            job_id=job_id,
+            item_id=item_id,
+            user_id=current_user.id,
+        )
+    except AnalysisJobNotFound as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=str(exc),
+        ) from exc
+    except AnalysisItemNotFound as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=str(exc),
+        ) from exc
+    except InvalidJobState as exc:
+        return JSONResponse(
+            status_code=status.HTTP_409_CONFLICT,
+            content={"kind": "invalid_state", "reason": str(exc)},
+        )
+
+    await session.commit()
+
+    return AnalysisControlOut(
+        kind="skipped",
+        reason="item_skipped",
+        job_id=str(job_id),
+        item_id=str(item_id),
+    )
+
+
+# ── Retry item ─────────────────────────────────────────────────────────────────
+
+
+@router.post(
+    "/analysis/jobs/{job_id}/items/{item_id}/retry",
+    response_model=AnalysisControlOut,
+)
+@account_limiter.limit("30/minute")
+async def retry_analysis_item(
+    request: Request,
+    job_id: uuid.UUID,
+    item_id: uuid.UUID,
+    session: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_regular_user),
+) -> AnalysisControlOut:
+    """Retry a failed job item from the beginning.
+
+    Resets the item to queued status, increments retry_count, clears error fields.
+    item_id is the cloud_item_id (DocuVault stable UUID).
+    """
+    request.state.current_user = current_user
+
+    try:
+        await retry_job_item(
+            session,
+            job_id=job_id,
+            item_id=item_id,
+            user_id=current_user.id,
+        )
+    except AnalysisJobNotFound as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=str(exc),
+        ) from exc
+    except AnalysisItemNotFound as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=str(exc),
+        ) from exc
+    except InvalidJobState as exc:
+        return JSONResponse(
+            status_code=status.HTTP_409_CONFLICT,
+            content={"kind": "invalid_state", "reason": str(exc)},
+        )
+
+    await session.commit()
+
+    return AnalysisControlOut(
+        kind="retried",
+        reason="item_requeued",
+        job_id=str(job_id),
+        item_id=str(item_id),
+    )

backend/api/cloud/schemas.py

@@ -255,3 +255,143 @@ class CacheSettingsUpdateRequest(BaseModel):
         ge=1,
         description="Preferred byte ceiling for the local byte cache",
     )
+
+
+# ── Phase 14 analysis request schemas ─────────────────────────────────────────
+
+class AnalysisEstimateRequest(BaseModel):
+    """Request body for POST /analysis/connections/{id}/estimate.
+
+    scope: "file" | "selection" | "folder" | "connection"
+    provider_item_ids: Required for file/selection/folder scope.  Omitted for connection scope.
+    recursive: Expand folder subtree recursively (folder scope only; connection always recursive).
+    """
+
+    scope: str = Field(..., description="file | selection | folder | connection")
+    provider_item_ids: Optional[List[str]] = Field(
+        default=None,
+        description="Provider item IDs for file/selection/folder scope",
+    )
+    recursive: bool = Field(
+        default=False,
+        description="Expand folder children recursively (folder scope)",
+    )
+
+
+class AnalysisEnqueueRequest(BaseModel):
+    """Request body for POST /analysis/connections/{id}/jobs.
+
+    scope: "file" | "selection" | "folder" | "connection"
+    provider_item_ids: Required for file/selection/folder scope.
+    recursive: Expand folder subtree recursively.
+    failure_behavior: "pause_batch" (default) | "continue_item" (D-11).
+    """
+
+    scope: str = Field(..., description="file | selection | folder | connection")
+    provider_item_ids: Optional[List[str]] = Field(default=None)
+    recursive: bool = Field(default=False)
+    failure_behavior: str = Field(
+        default="pause_batch",
+        description="pause_batch | continue_item",
+    )
+
+
+# ── Phase 14 analysis response schemas ────────────────────────────────────────
+
+class AnalysisEstimateOut(BaseModel):
+    """Estimate response — no credentials, bytes, or object_key (T-14-02).
+
+    supported_count:      Files that can be analysed.
+    unsupported_count:    Items that cannot be analysed (unsupported type, folder).
+    total_provider_bytes: Sum of provider_size across supported items.
+    recursive:            Whether the estimate was recursive.
+    is_partial:           True when metadata expansion was incomplete.
+    scope_kind:           Echo of the requested scope.
+    """
+
+    supported_count: int = 0
+    unsupported_count: int = 0
+    total_provider_bytes: int = 0
+    recursive: bool = False
+    is_partial: bool = False
+    scope_kind: str
+
+
+class AnalysisJobOut(BaseModel):
+    """Job status response — no credentials or object_key (T-14-02).
+
+    Exposes both simple and detailed aggregate counts.  Caller requests detail
+    via ?detail=true query parameter.
+
+    Simple (default):  waiting_count, working_count, done_count,
+                       skipped_count, failed_count, total_count.
+    Detailed:          queued_count, downloading_count, extracting_count,
+                       classifying_count, indexed_count, already_current_count,
+                       cancelled_count, failed_count, unsupported_count.
+    """
+
+    job_id: str
+    connection_id: str
+    scope_kind: str
+    status: str
+    failure_behavior: str
+    recursive: bool = False
+    total_count: int = 0
+
+    # Simple labels (always present)
+    waiting_count: int = 0
+    working_count: int = 0
+    done_count: int = 0
+    skipped_count: int = 0
+    failed_count: int = 0
+
+    # Detailed labels (always present — 0 when detail=false, per-stage values when detail=true)
+    queued_count: int = 0
+    downloading_count: int = 0
+    extracting_count: int = 0
+    classifying_count: int = 0
+    indexed_count: int = 0
+    already_current_count: int = 0
+    cancelled_count: int = 0
+    unsupported_count: int = 0
+
+    created_at: Optional[datetime] = None
+    started_at: Optional[datetime] = None
+    finished_at: Optional[datetime] = None
+
+
+class AnalysisEnqueueOut(BaseModel):
+    """Enqueue response — returns a stable job_id for status polling (ANALYZE-01).
+
+    No credentials or object_key in response (T-14-02).
+    """
+
+    job_id: str
+    status: str
+    total_count: int = 0
+    queued_count: int = 0
+    already_current_count: int = 0
+    unsupported_count: int = 0
+
+
+class AnalysisJobItemOut(BaseModel):
+    """Per-item job status — no credentials, object_key, or raw provider data (T-14-02)."""
+
+    id: str
+    cloud_item_id: str
+    provider_item_id: str
+    status: str
+    error_code: Optional[str] = None
+    error_message: Optional[str] = None
+    retry_count: int = 0
+    created_at: Optional[datetime] = None
+    finished_at: Optional[datetime] = None
+
+
+class AnalysisControlOut(BaseModel):
+    """Generic control result for cancel/skip/retry operations."""
+
+    kind: str        # "cancelled" | "skipped" | "retried"
+    reason: Optional[str] = None
+    job_id: Optional[str] = None
+    item_id: Optional[str] = None

backend/tests/test_cloud_security.py

@@ -704,3 +704,205 @@ async def test_cache_settings_limit_above_tier_cap_returns_422(async_client, db_
     assert resp.status_code == 422, (
         f"Cache limit above tier cap must return 422, got {resp.status_code}"
     )
+
+
+# ── Phase 14: Analysis job API security (T-14-01, T-14-02, T-14-03) ────────────
+
+
+async def _create_analysis_connection(session, user_id, provider="google_drive"):
+    """Helper: create a cloud connection for analysis security tests."""
+    from db.models import CloudConnection
+    from storage.cloud_utils import encrypt_credentials
+    from config import settings as app_settings
+
+    master_key = app_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="Security Test Connection",
+        credentials_enc=creds_enc,
+        status="ACTIVE",
+    )
+    session.add(conn)
+    await session.commit()
+    return conn
+
+
+async def _create_analysis_item(session, user_id, connection_id, name="doc.pdf"):
+    """Helper: create a cloud item for analysis security tests."""
+    from db.models import CloudItem
+
+    item = CloudItem(
+        id=_uuid.uuid4(),
+        user_id=user_id,
+        connection_id=connection_id,
+        provider_item_id=f"sec-pitem-{_uuid.uuid4().hex[:8]}",
+        name=name,
+        kind="file",
+        content_type="application/pdf",
+        provider_size=102400,
+        etag="etag-sec-v1",
+        analysis_status="pending",
+    )
+    session.add(item)
+    await session.commit()
+    return item
+
+
+async def test_analysis_estimate_admin_blocked(async_client, db_session):
+    """T-14-01: Admin cannot access analysis estimate endpoint."""
+    admin_ctx = await _create_user_and_token(db_session, role="admin")
+    user_ctx = await _create_user_and_token(db_session, role="user")
+    conn = await _create_analysis_connection(db_session, user_ctx["user"].id)
+
+    resp = await async_client.post(
+        f"/api/cloud/analysis/connections/{conn.id}/estimate",
+        json={"scope": "connection", "recursive": True},
+        headers=admin_ctx["headers"],
+    )
+
+    assert resp.status_code in (403, 404), (
+        f"Admin must not access analysis estimate — got {resp.status_code}"
+    )
+
+
+async def test_analysis_enqueue_admin_blocked(async_client, db_session):
+    """T-14-01: Admin cannot enqueue analysis jobs."""
+    admin_ctx = await _create_user_and_token(db_session, role="admin")
+    user_ctx = await _create_user_and_token(db_session, role="user")
+    conn = await _create_analysis_connection(db_session, user_ctx["user"].id)
+
+    resp = await async_client.post(
+        f"/api/cloud/analysis/connections/{conn.id}/jobs",
+        json={"scope": "connection", "recursive": True},
+        headers=admin_ctx["headers"],
+    )
+
+    assert resp.status_code in (403, 404), (
+        f"Admin must not enqueue analysis jobs — got {resp.status_code}"
+    )
+
+
+async def test_analysis_estimate_unauthenticated_blocked(async_client, db_session):
+    """Analysis estimate requires authentication."""
+    conn_id = str(_uuid.uuid4())
+
+    resp = await async_client.post(
+        f"/api/cloud/analysis/connections/{conn_id}/estimate",
+        json={"scope": "connection", "recursive": True},
+    )
+
+    assert resp.status_code in (401, 403), (
+        f"Unauthenticated analysis estimate must be rejected — got {resp.status_code}"
+    )
+
+
+async def test_analysis_enqueue_response_excludes_credentials(async_client, db_session):
+    """T-14-02: Enqueue response must not contain credentials or object_key."""
+    from unittest.mock import patch as _patch
+
+    user_ctx = await _create_user_and_token(db_session, role="user")
+    conn = await _create_analysis_connection(db_session, user_ctx["user"].id)
+    item = await _create_analysis_item(db_session, user_ctx["user"].id, conn.id)
+
+    with _patch("services.cloud_analysis.get_adapter", return_value=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=user_ctx["headers"],
+        )
+
+    assert resp.status_code in (200, 202), (
+        f"Enqueue must succeed — got {resp.status_code}: {resp.text}"
+    )
+    body_text = resp.text
+
+    # T-14-02: No credential or internal storage fields in response
+    for field in ("credentials_enc", "access_token", "refresh_token", "object_key"):
+        assert field not in body_text, (
+            f"T-14-02: Sensitive field '{field}' must not appear in enqueue response"
+        )
+
+
+async def test_analysis_job_status_excludes_credentials(async_client, db_session):
+    """T-14-02: Job status response must not contain credentials or object_key."""
+    from unittest.mock import patch as _patch
+
+    user_ctx = await _create_user_and_token(db_session, role="user")
+    conn = await _create_analysis_connection(db_session, user_ctx["user"].id)
+    item = await _create_analysis_item(db_session, user_ctx["user"].id, conn.id)
+
+    with _patch("services.cloud_analysis.get_adapter", return_value=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=user_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=user_ctx["headers"],
+    )
+    assert status_resp.status_code == 200
+    body_text = status_resp.text
+
+    for field in ("credentials_enc", "access_token", "refresh_token", "object_key"):
+        assert field not in body_text, (
+            f"T-14-02: Sensitive field '{field}' must not appear in job status response"
+        )
+
+
+async def test_foreign_user_cannot_access_analysis_estimate(async_client, db_session):
+    """T-14-01: Foreign user cannot estimate analysis scope for another user's connection."""
+    owner_ctx = await _create_user_and_token(db_session, role="user")
+    foreign_ctx = await _create_user_and_token(db_session, role="user")
+    owner_conn = await _create_analysis_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"],
+    )
+
+    assert resp.status_code in (403, 404), (
+        f"Foreign user must not estimate owner's connection — got {resp.status_code}"
+    )
+
+
+async def test_foreign_user_cannot_read_analysis_job_status(async_client, db_session):
+    """T-14-01: Foreign user cannot read job status for another user's job."""
+    from unittest.mock import patch as _patch
+
+    owner_ctx = await _create_user_and_token(db_session, role="user")
+    foreign_ctx = await _create_user_and_token(db_session, role="user")
+    owner_conn = await _create_analysis_connection(db_session, owner_ctx["user"].id)
+    owner_item = await _create_analysis_item(db_session, owner_ctx["user"].id, owner_conn.id)
+
+    with _patch("services.cloud_analysis.get_adapter", return_value=MagicMock()):
+        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"],
+        )
+
+    assert enqueue_resp.status_code in (200, 202)
+    job_id = enqueue_resp.json().get("job_id")
+
+    resp = await async_client.get(
+        f"/api/cloud/analysis/jobs/{job_id}",
+        headers=foreign_ctx["headers"],
+    )
+
+    assert resp.status_code in (403, 404), (
+        f"Foreign user must not read owner's job — got {resp.status_code}"
+    )