feat(batch-processing-cli): add --max-image-failure-rate flag (#2360)

* feat(batch-processing-cli): add --max-image-failure-rate flag

Exposes the per-shard image-failure tolerance knob through the CLI.
Adds the flag to `process-images-with-workflow` and `restart-job`,
threads it through `WorkflowProcessingJobV1` to the wire as
`maxImageFailureRate`, and appends it to the restart override
payload parallel to `maxParallelTasks`.

Pure pass-through plumbing; the threshold is enforced inside the
workflows-data-processor worker (shipped separately in the
batch-processing-services repo). The server is the source of truth
for `[0.0, 1.0]` validation — no client-side range check is added,
matching the existing behavior of other numeric override flags in
this command.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* add openspec to gitignore

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Co-authored-by: Paweł Pęczek <146137186+PawelPeczek-Roboflow@users.noreply.github.com>

Author: 3 people

.gitignore

@@ -214,4 +214,7 @@ inference_models/tests/e2e_platform_tests/assets/
 inference_testing
 
 # rerun.io recordings
-*.rrd
+*.rrd
+
+openspec*/
+opsx/

inference_cli/lib/roboflow_cloud/batch_processing/api_operations.py

@@ -401,6 +401,7 @@ def trigger_job_with_workflows_images_processing(
     api_key: str,
     inference_backend: Optional[InferenceBackend] = None,
     job_name: Optional[str] = None,
+    max_image_failure_rate: Optional[float] = None,
 ) -> str:
     workspace = get_workspace(api_key=api_key)
     compute_configuration = ComputeConfigurationV2(
@@ -431,6 +432,7 @@ def trigger_job_with_workflows_images_processing(
         compute_configuration=compute_configuration,
         processing_timeout_seconds=max_runtime_seconds,
         max_parallel_tasks=max_parallel_tasks,
+        max_image_failure_rate=max_image_failure_rate,
         processing_specification=processing_specification,
         notifications_url=notifications_url,
         inference_backend=inference_backend,
@@ -755,6 +757,7 @@ def restart_batch_job(
     workers_per_machine: Optional[int] = None,
     max_runtime_seconds: Optional[float] = None,
     max_parallel_tasks: Optional[int] = None,
+    max_image_failure_rate: Optional[float] = None,
 ) -> None:
     workspace = get_workspace(api_key=api_key)
     response = send_restart_job_request(
@@ -765,6 +768,7 @@ def restart_batch_job(
         workers_per_machine=workers_per_machine,
         max_runtime_seconds=max_runtime_seconds,
         max_parallel_tasks=max_parallel_tasks,
+        max_image_failure_rate=max_image_failure_rate,
     )
     console = Console()
     console.print(JSON.from_data(response, indent=2))
@@ -784,6 +788,7 @@ def send_restart_job_request(
     workers_per_machine: Optional[int] = None,
     max_runtime_seconds: Optional[float] = None,
     max_parallel_tasks: Optional[int] = None,
+    max_image_failure_rate: Optional[float] = None,
 ) -> dict:
     payload = {
         "type": "parameters-override-v1",
@@ -799,6 +804,8 @@ def send_restart_job_request(
         payload["maxParallelTasks"] = max_parallel_tasks
     if max_runtime_seconds is not None:
         payload["processingTimeoutSeconds"] = max_runtime_seconds
+    if max_image_failure_rate is not None:
+        payload["maxImageFailureRate"] = max_image_failure_rate
     params = {"api_key": api_key}
     try:
         response = requests.post(

inference_cli/lib/roboflow_cloud/batch_processing/core.py

@@ -189,6 +189,19 @@ def process_images_with_workflow(
             "--max-parallel-tasks", help="Max number of concurrent processing tasks"
         ),
     ] = None,
+    max_image_failure_rate: Annotated[
+        Optional[float],
+        typer.Option(
+            "--max-image-failure-rate",
+            help=(
+                "Maximum fraction of images per shard that may fail before the shard "
+                "is aborted (0.0-1.0). Default: 10 absolute failures per shard "
+                "(~3.9% on a full 256-image shard, higher effective tolerance on "
+                "partial last shards). No client-side validation is performed; "
+                "out-of-range values are rejected by the server with HTTP 400."
+            ),
+        ),
+    ] = None,
     aggregation_format: Annotated[
         Optional[AggregationFormat],
         typer.Option("--aggregation-format", help="Format of results aggregation"),
@@ -261,6 +274,7 @@ def process_images_with_workflow(
             api_key=api_key,
             inference_backend=inference_backend,
             job_name=job_name,
+            max_image_failure_rate=max_image_failure_rate,
         )
         print(f"Triggered job with ID: {job_id}")
     except KeyboardInterrupt:
@@ -591,6 +605,19 @@ def restart_job(
             "--max-parallel-tasks", help="Max number of concurrent processing tasks"
         ),
     ] = None,
+    max_image_failure_rate: Annotated[
+        Optional[float],
+        typer.Option(
+            "--max-image-failure-rate",
+            help=(
+                "Maximum fraction of images per shard that may fail before the shard "
+                "is aborted (0.0-1.0). Default: 10 absolute failures per shard "
+                "(~3.9% on a full 256-image shard, higher effective tolerance on "
+                "partial last shards). No client-side validation is performed; "
+                "out-of-range values are rejected by the server with HTTP 400."
+            ),
+        ),
+    ] = None,
     debug_mode: Annotated[
         bool,
         typer.Option(
@@ -610,6 +637,7 @@ def restart_job(
             workers_per_machine=workers_per_machine,
             max_runtime_seconds=max_runtime_seconds,
             max_parallel_tasks=max_parallel_tasks,
+            max_image_failure_rate=max_image_failure_rate,
         )
     except KeyboardInterrupt:
         print("Command interrupted.")

inference_cli/lib/roboflow_cloud/batch_processing/entities.py

@@ -144,6 +144,9 @@ class WorkflowProcessingJobV1(BaseModel):
     max_parallel_tasks: Optional[int] = Field(
         serialization_alias="maxParallelTasks", default=None
     )
+    max_image_failure_rate: Optional[float] = Field(
+        serialization_alias="maxImageFailureRate", default=None
+    )
     processing_specification: WorkflowsProcessingSpecificationV1 = Field(
         serialization_alias="processingSpecification"
     )