Add Reprocess to retry failed IC reviews without the create wizard.

Successful reviews are kept; skipped ICs such as a DeepSeek 400 are run again. Reprocess all re-reviews every IC while still using the library cache.

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/services/pipeline.py

@@ -1644,15 +1644,17 @@ async def run_pipeline(
     the project is left in ``paused_insufficient_credits`` with a
     checkpoint so it can be resumed later.
 
-    When ``resume=True``, prior completed review refs and spent credits are
-    restored from the project's ``pause_checkpoint`` so completed work is
-    skipped on the next pass.
+    When ``resume=True``, prior completed review refs are restored so
+    already-reviewed ICs are skipped (paused credit resume, or user
+    reprocess of failed reviews).
 
     When ``free=True`` (admin-initiated rerun), every call runs through
     ``ApiLogger(free=True)`` so ``credits_charged`` is zeroed, the credit
     gate is bypassed, and ``meta.total_cost_usd`` is preserved rather than
-    incremented.  The raw Anthropic cost is still captured in log entries.
+    incremented.      The raw Anthropic cost is still captured in log entries.
     """
+    ctx: PipelineContext | None = None
+    api_logger: ApiLogger | None = None
     try:
         meta = proj_svc.get_project(storage, user_id, project_id)
         if not meta:
@@ -1826,37 +1828,59 @@ async def run_pipeline(
         # asyncio.CancelledError can also arrive during local-dev
         # subprocess shutdown (SIGTERM). Both are handled the same way.
         try:
+            extra: dict = {
+                "pipeline_state": {"error": "Pipeline cancelled by user"},
+                "cancel_requested": False,
+            }
+            if ctx is not None:
+                extra["completed_review_refs"] = sorted(
+                    ctx.completed_review_refs, key=natural_sort_key,
+                )
+                extra["skipped_components"] = (
+                    [s.to_dict() for s in ctx.skipped] or None
+                )
             proj_svc.transition_status(
                 storage, user_id, project_id,
                 from_status={proj_svc.STATUS_RUNNING, proj_svc.STATUS_QUEUED},
                 to_status=proj_svc.STATUS_CANCELLED,
-                pipeline_state={"error": "Pipeline cancelled by user"},
-                cancel_requested=False,
+                **extra,
             )
         except proj_svc.StatusConflict:
             pass
         broker.publish(project_id, "pipeline_cancelled", {"error": "Pipeline cancelled by user"})
         # Last-mile flush so partial billing is captured.
         try:
-            api_logger.flush(storage, user_id, project_id)  # type: ignore[has-type]
+            if api_logger is not None:
+                api_logger.flush(storage, user_id, project_id)
         except Exception:
             pass
 
     except Exception as e:
         logger.exception("Pipeline run crashed for project %s", project_id)
         try:
+            extra = {
+                "pipeline_state": {"error": str(e)},
+                "cancel_requested": False,
+            }
+            if ctx is not None:
+                extra["completed_review_refs"] = sorted(
+                    ctx.completed_review_refs, key=natural_sort_key,
+                )
+                extra["skipped_components"] = (
+                    [s.to_dict() for s in ctx.skipped] or None
+                )
             proj_svc.transition_status(
                 storage, user_id, project_id,
                 from_status={proj_svc.STATUS_RUNNING, proj_svc.STATUS_QUEUED},
                 to_status=proj_svc.STATUS_ERROR,
-                pipeline_state={"error": str(e)},
-                cancel_requested=False,
+                **extra,
             )
         except proj_svc.StatusConflict:
             pass
         broker.publish(project_id, "pipeline_error", {"error": str(e)})
         try:
-            api_logger.flush(storage, user_id, project_id)  # type: ignore[has-type]
+            if api_logger is not None:
+                api_logger.flush(storage, user_id, project_id)
         except Exception:
             pass
 

backend/services/projects.py

@@ -129,6 +129,37 @@ class ProjectMeta(BaseModel):
     cancel_requested: bool = False
 
 
+def completed_review_refs_for_retry(
+    storage: StorageBackend, user_id: str, project_id: str,
+) -> list[str]:
+    """ICs that already finished review and should be skipped on reprocess.
+
+    Drops refs that failed (skipped_components / report.review_errors) so
+    those ICs are tried again.
+    """
+    meta = get_project(storage, user_id, project_id)
+    if not meta:
+        return []
+    failed: set[str] = set()
+    for item in meta.skipped_components or []:
+        stage = (item.get("stage") or "")
+        ident = (item.get("identifier") or "").strip()
+        if ident and stage in ("validation", "review"):
+            failed.add(ident)
+    report_key = f"{_project_prefix(user_id, project_id)}/report.json"
+    if storage.exists(report_key):
+        try:
+            report = storage.read_json(report_key)
+        except Exception:
+            report = {}
+        for ref in (report.get("review_errors") or {}):
+            if ref:
+                failed.add(str(ref))
+    from backend.pinscopex.utils import natural_sort_key
+    kept = [r for r in (meta.completed_review_refs or []) if r and r not in failed]
+    return sorted(kept, key=natural_sort_key)
+
+
 def _project_prefix(user_id: str, project_id: str) -> str:
     return f"users/{user_id}/projects/{project_id}"