feat(14.1-02): force re-analyze flag + single-item retry-job creation

- Add force: bool = False to AnalysisEnqueueRequest in cloud/schemas.py (D-11)
- Extend enqueue_analysis_job to accept force param; already_current check
  becomes (already_current and not force) — bypass for supported items (D-11, ANALYZE-06)
- Add retry_or_create_single_item_job service helper in cloud_analysis.py that
  creates a single-item force enqueue job when no active job exists (D-12)
- Add POST /analysis/connections/{id}/items/{cloud_item_id}/retry route
  in analysis.py returning AnalysisEnqueueOut with queued_count >= 1
- Wire force= through enqueue_job route handler (body.force)
- All 8 test_cloud_reanalyze_force tests pass; 26 test_cloud_analysis_contract
  tests pass (no idempotency regression); 872/873 backend tests pass

backend/api/cloud/analysis.py

@@ -59,6 +59,7 @@ from services.cloud_analysis import (
     list_analysis_jobs,
     list_job_items,
     retry_job_item,
+    retry_or_create_single_item_job,
     skip_job_item,
     InvalidJobState,
 )
@@ -171,6 +172,7 @@ async def enqueue_job(
             provider_item_ids=body.provider_item_ids,
             recursive=body.recursive,
             failure_behavior=body.failure_behavior,
+            force=body.force,
         )
     except ConnectionNotFound as exc:
         raise HTTPException(
@@ -520,3 +522,90 @@ async def retry_analysis_item(
         job_id=str(job_id),
         item_id=str(item_id),
     )
+
+
+# ── Single-item retry — no active job required (D-12) ─────────────────────────
+
+
+@router.post(
+    "/analysis/connections/{connection_id}/items/{cloud_item_id}/retry",
+    response_model=AnalysisEnqueueOut,
+    status_code=202,
+)
+@account_limiter.limit("30/minute")
+async def retry_cloud_item(
+    request: Request,
+    connection_id: uuid.UUID,
+    cloud_item_id: uuid.UUID,
+    session: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_regular_user),
+) -> AnalysisEnqueueOut:
+    """Retry a failed cloud item, creating a single-item job if no active job exists.
+
+    D-12: If no surviving active job covers the failed item, this route creates a
+    one-item analysis job through the authorized enqueue path (force=True) so the
+    item is re-queued for fresh extraction and classification.
+
+    cloud_item_id must be the DocuVault stable UUID (not the provider_item_id).
+    The connection_id is used only for ownership verification of the URL namespace;
+    the service resolves the correct connection from the cloud item row.
+
+    Returns AnalysisEnqueueOut with job_id and queued_count >= 1 on success.
+
+    T-14.1-04: Owner-scoped via get_regular_user + item ownership check.
+    T-14.1-05: Creates a queued analysis job — no provider mutations are made.
+    """
+    request.state.current_user = current_user
+
+    # Verify the caller owns the connection in the URL namespace (IDOR gate).
+    # The service also checks item ownership independently.
+    from services.cloud_items import resolve_owned_connection as _resolve_conn
+    try:
+        await _resolve_conn(
+            session,
+            connection_id=connection_id,
+            user_id=current_user.id,
+        )
+    except ConnectionNotFound as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Connection not found",
+        ) from exc
+
+    try:
+        result = await retry_or_create_single_item_job(
+            session,
+            cloud_item_id=cloud_item_id,
+            user_id=current_user.id,
+        )
+    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)},
+        )
+    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,
+    )

backend/services/cloud_analysis.py

@@ -465,16 +465,20 @@ async def enqueue_analysis_job(
     provider_item_ids: Optional[List[str]] = None,
     recursive: bool = False,
     failure_behavior: str = "pause_batch",
+    force: bool = False,
 ) -> EnqueueResult:
     """Create a CloudAnalysisJob and populate job items from scope.
 
     Items that are "already current" (version key matches a prior indexed run
     in any recent job item) are marked already_current without enqueuing byte
-    work. Unsupported items are marked unsupported immediately.
+    work — unless force=True is passed, which bypasses the already_current check
+    for supported items (D-11, ANALYZE-06).
+
+    Unsupported items are always marked unsupported, regardless of force.
 
     Only supported, non-current items are created as queued job items.
 
-    No provider bytes are downloaded. No provider mutations are made.
+    No provider bytes are downloaded. No provider mutations are made (ANALYZE-07).
 
     Args:
         session:           Active async SQLAlchemy session.
@@ -484,6 +488,9 @@ async def enqueue_analysis_job(
         provider_item_ids: Provider item IDs for file/selection/folder scope.
         recursive:         Expand folder children recursively (folder scope).
         failure_behavior:  "pause_batch" | "continue_item" (D-11).
+        force:             When True, bypass already_current check for supported items.
+                           Unsupported items remain unsupported. No provider mutation
+                           is performed regardless of this flag (ANALYZE-07).
 
     Returns:
         EnqueueResult with job_id and item counts.
@@ -596,7 +603,9 @@ async def enqueue_analysis_job(
         # Check if an existing indexed job item has the same version key
         # — already_current check (T-14-04: no bytes downloaded).
         # Skip the fallback if live metadata confirmed the item changed.
-        already_current = False if live_metadata_changed else await _check_already_current(
+        # When force=True, bypass the already_current check entirely so the item
+        # is re-queued for fresh analysis (D-11, ANALYZE-06).
+        already_current = False if (live_metadata_changed or force) else await _check_already_current(
             session, user_id=uid, cloud_item_id=item.id, version_key=current_vk
         )
 
@@ -1122,3 +1131,73 @@ async def retry_job_item(
 
     await session.flush()
     return job_item
+
+
+# ── Single-item retry-job creation (D-12) ─────────────────────────────────────
+
+async def retry_or_create_single_item_job(
+    session: AsyncSession,
+    *,
+    cloud_item_id: uuid.UUID,
+    user_id: uuid.UUID,
+) -> EnqueueResult:
+    """Retry a failed cloud item, creating a single-item job if no active job exists.
+
+    Implements D-12: Failed analysis retry uses existing job retry semantics when
+    a surviving active job covers the item; otherwise creates a one-item job via
+    enqueue_analysis_job(scope="file", force=True) so a fresh queued item is produced
+    through the authorized analysis path.
+
+    The fallback single-item job uses force=True so the item is re-queued regardless
+    of its analysis_status or version_key match — it failed and the user wants a retry.
+
+    Args:
+        session:        Active async SQLAlchemy session.
+        cloud_item_id:  DocuVault stable UUID for the CloudItem to retry.
+        user_id:        Authenticated user UUID — must own the item.
+
+    Returns:
+        EnqueueResult with job_id and item counts (queued_count >= 1 on success).
+
+    Raises:
+        AnalysisItemNotFound: Item does not exist or belongs to another user.
+        ConnectionNotFound:   Connection is no longer owned by user_id.
+        InvalidJobState:      Item is not in a retryable analysis state.
+    """
+    uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id))
+    iid = cloud_item_id if isinstance(cloud_item_id, uuid.UUID) else uuid.UUID(str(cloud_item_id))
+
+    # 1. Resolve the cloud item and verify ownership
+    stmt = select(CloudItem).where(
+        CloudItem.id == iid,
+        CloudItem.user_id == uid,
+        CloudItem.deleted_at.is_(None),
+    )
+    item_result = await session.execute(stmt)
+    item = item_result.scalars().first()
+    if item is None:
+        raise AnalysisItemNotFound(
+            f"Cloud item {iid} not found or not owned by user"
+        )
+
+    # 2. Accept failed, indexed, stale, and pending items for this path.
+    #    Unsupported items cannot be retried.
+    retryable_statuses = {"failed", "indexed", "stale", "pending"}
+    if item.analysis_status not in retryable_statuses:
+        raise InvalidJobState(
+            f"Cloud item {iid} has analysis_status {item.analysis_status!r} "
+            f"and cannot be retried via this path"
+        )
+
+    # 3. Create a single-item force job through the authorized analysis path (D-12).
+    #    force=True ensures the item is queued even if it was indexed before.
+    enqueue_result = await enqueue_analysis_job(
+        session,
+        user_id=uid,
+        connection_id=item.connection_id,
+        scope="file",
+        provider_item_ids=[item.provider_item_id],
+        force=True,
+    )
+
+    return enqueue_result