From 52acd5634b06a6247c0a33782d2a8701e2c0fdc3 Mon Sep 17 00:00:00 2001 From: curo1305 Date: Fri, 26 Jun 2026 21:59:50 +0200 Subject: [PATCH] feat(14.1-02): force re-analyze flag + single-item retry-job creation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 | 89 ++++++++++++++++++++++++++++++ backend/services/cloud_analysis.py | 85 +++++++++++++++++++++++++++- 2 files changed, 171 insertions(+), 3 deletions(-) diff --git a/backend/api/cloud/analysis.py b/backend/api/cloud/analysis.py index aeda92c..3728cf2 100644 --- a/backend/api/cloud/analysis.py +++ b/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, + ) diff --git a/backend/services/cloud_analysis.py b/backend/services/cloud_analysis.py index 38ace68..bfa4423 100644 --- a/backend/services/cloud_analysis.py +++ b/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