feat(job): create job definition only after successful upload

.env.example

@@ -10,6 +10,7 @@
 # RAPIDATA_cacheShards=128
 # RAPIDATA_batchSize=1000
 # RAPIDATA_batchPollInterval=0.5
+# RAPIDATA_failureTolerance=0.0
 
 # --- Logging ---
 # RAPIDATA_level=WARNING

docs/error_handling.md

@@ -2,38 +2,53 @@
 
 ## Introduction
 
-When creating job definitions or orders with the Rapidata SDK, datapoints may fail to upload due to various reasons such as missing files, invalid formats, or network issues. Understanding how to handle these failures is essential for building robust integrations.
+When creating job definitions with the Rapidata SDK, datapoints may fail to upload for various reasons such as missing files, invalid formats, or network issues. Understanding how to handle these failures is essential for building robust integrations.
 
-When one or more datapoints fail to upload, the SDK raises a `FailedUploadException`. This exception provides detailed information about what went wrong and gives you several recovery options:
+Job creation is **atomic by default**: if any datapoint fails to upload, the SDK does **not** create the job definition — so you never end up with an incomplete definition — and raises a `FailedUploadException`. You then fix the problem and call `exception.retry()`, which re-uploads only the failed datapoints into the **same** dataset and finishes creating the definition. Retrying never creates a duplicate dataset or definition.
 
-- Inspect which datapoints failed and why
-- Retry the failed datapoints
-- Continue with the successfully uploaded datapoints
+If you would rather tolerate a small fraction of failures instead, raise the [failure tolerance](#failure-tolerance).
 
-This guide shows you how to handle upload failures effectively.
+## Failure tolerance
 
-## Understanding FailedUploadException
+`failure_tolerance` is the fraction of a job's datapoints (`0.0`–`1.0`) allowed to fail while still creating the definition:
 
-The `FailedUploadException` is raised during `JobDefinition` or `Order` creation when one or more datapoints cannot be uploaded. 
-**Important**: Despite the exception being raised, a `JobDefinition` or `Order` object is still created with the successfully uploaded datapoints, allowing you to continue if you catch the exception.
+- `0.0` (default) — strict. Any failed upload aborts creation; no definition is left behind and the failed datapoints can be retried.
+- `0.01` — up to 1% of datapoints may fail; the definition is created with the rest, and the failures are logged.
+- `1.0` — the definition is created regardless of how many datapoints failed, as long as at least one datapoint uploads successfully (a definition over an empty dataset is never created).
 
-### Exception Properties
+Set it globally, via the config object or the `RAPIDATA_failureTolerance` environment variable:
+
+```python
+from rapidata import rapidata_config
+rapidata_config.upload.failureTolerance = 0.01
+```
 
-The exception provides these properties to help you understand and recover from failures:
+or per call, which overrides the global default for that job:
 
 ```python
-FailedUploadException(
-    dataset: RapidataDataset, # (1)!
-    failed_uploads: list[FailedUpload], # (2)!
-    order: Optional[RapidataOrder], # (3)!
-    job_definition: Optional[JobDefinition] # (4)!
+job_def = client.job.create_classification_job_definition(
+    name="Image Classification",
+    instruction="What animal is in this image?",
+    answer_options=["Cat", "Dog", "Bird"],
+    datapoints=["cat1.jpg", "dog1.jpg", "missing.jpg"],
+    failure_tolerance=0.01,
 )
 ```
 
-1. The dataset that was being created.
-2. Basic list of failed datapoints.
-3. The order object (only present during order creation).
-4. The job definition object (only present during job definition creation).
+The ratio is measured against the whole job, so it stays meaningful across retries.
+
+## Understanding FailedUploadException
+
+The `FailedUploadException` is raised during job-definition creation when the upload stays outside the failure tolerance. When it comes from job creation, no job definition was created — recover with `exception.retry()` (see below).
+
+### Exception Properties
+
+The exception exposes these to help you understand and recover from failures:
+
+- `dataset` — the dataset that was being uploaded to. `retry()` reuses it.
+- `failed_uploads` / `detailed_failures` / `failures_by_reason` — which datapoints failed and why (see below).
+- `retry()` — re-upload the failed datapoints into `dataset` and finish creating the definition. Returns the `RapidataJobDefinition`.
+- `job_definition` — `None` for job creation (nothing was persisted). Populated only when tolerating failures on a legacy order.
 
 ### Understanding Failure Information
 
@@ -99,11 +114,11 @@ This groups all failed datapoints by their error message, making it easy to see
 
 **Datapoint Creation Failures**: After assets are successfully uploaded, datapoints are created. These failures can have different reasons depending on what went wrong (e.g., validation errors, format issues, backend constraints). Each datapoint may fail for a unique reason.
 
-## Recovery Strategies
+## Recovery
 
-### Strategy 1: Continue with Successfully Uploaded Datapoints
+### Fix and retry
 
-When a `FailedUploadException` is raised, the `JobDefinition` is still created with the successfully uploaded datapoints. You can catch the exception and continue using the created object:
+The recommended recovery path is to catch the exception, fix whatever caused the failures (for example correct the file paths), and call `exception.retry()`. This re-uploads **only** the failed datapoints into the **same** dataset and finishes creating the definition — no new dataset, no duplicate definition, and nothing to re-specify:
 
 ```python
 from rapidata import RapidataClient
@@ -116,67 +131,46 @@ try:
         name="Image Classification",
         instruction="What animal is in this image?",
         answer_options=["Cat", "Dog", "Bird"],
-        datapoints=["cat1.jpg", "dog1.jpg", "missing.jpg"]
+        datapoints=["cat1.jpg", "dog1.jpg", "missing.jpg"],
     )
 except FailedUploadException as e:
-    print(f"Warning: {len(e.failed_uploads)} datapoints failed to upload")
-
-    if len(e.failed_uploads) > len(datapoints) * 0.1: # (1)!
-        raise ValueError("Too many failures, aborting")
+    for reason, datapoints in e.failures_by_reason.items():  # (1)!
+        print(f"  {reason}: {len(datapoints)} datapoints")
 
-    job_def = e.job_definition # (2)!
+    # ...fix the failing datapoints (e.g. correct the paths on disk)...
+    job_def = e.retry()  # (2)!
 
-# The job definition holds the successful datapoints — assign it to an audience to start collecting responses.
 audience = client.audience.get_audience_by_id("global")
 job = audience.assign_job(job_def)
 ```
 
-1. Check if the failure rate is acceptable — here we abort if more than 10% failed.
-2. The job definition was still created with the successfully uploaded datapoints. You can use it normally.
-
-### Strategy 2: Retry Failed Datapoints
-
-After catching the exception, you can fix the issues (e.g., correct file paths, fix formats) and retry the failed datapoints by adding them to the dataset:
-
-```python
-from rapidata import RapidataClient
-from rapidata.rapidata_client.exceptions import FailedUploadException
+1. Inspect what failed and why before retrying.
+2. Returns the created `RapidataJobDefinition`. If some datapoints still fail beyond the tolerance, `retry()` raises `FailedUploadException` again (with the same dataset attached), so you can keep fixing and retrying.
 
-client = RapidataClient()
+!!! warning
+    Don't re-call `create_*_job_definition(...)` to retry — that starts a fresh dataset and creates a **second** definition. Use `exception.retry()` so the existing dataset is reused.
 
-try:
-    job_def = client.job.create_classification_job_definition(
-        name="Image Classification",
-        instruction="What animal is in this image?",
-        answer_options=["Cat", "Dog", "Bird"],
-        datapoints=["cat1.jpg", "dog1.jpg", "missing.jpg"]
-    )
-except FailedUploadException as e:
-    print(f"{len(e.failed_uploads)} datapoints failed:")
-    for reason, datapoints in e.failures_by_reason.items():
-        print(f"  {reason}: {len(datapoints)} datapoints")
+### Tolerate a fraction of failures
 
-    successful_retries, failed_retries = e.dataset.add_datapoints(e.failed_uploads) # (1)!
-    print(f"{len(successful_retries)} datapoints successfully added on retry")
+If a few failures are acceptable, set a [failure tolerance](#failure-tolerance). When the failed fraction is within tolerance the definition is created with the datapoints that succeeded, the failures are logged, and no exception is raised:
 
-    if failed_retries:
-        print(f"{len(failed_retries)} datapoints still failed after retry")
+```python
+job_def = client.job.create_classification_job_definition(
+    name="Image Classification",
+    instruction="What animal is in this image?",
+    answer_options=["Cat", "Dog", "Bird"],
+    datapoints=["cat1.jpg", "dog1.jpg", "missing.jpg"],
+    failure_tolerance=0.5,  # tolerate up to 50% failures for this job
+)
 ```
 
-1. Fix the underlying issues (e.g., correct file paths) before retrying. This adds the previously failed datapoints back to the dataset.
-
-### Strategy 3: Retrieve and Use After Exception (If Not Caught)
+### Manual control (advanced)
 
-If you didn't catch the exception during creation, you can still retrieve and use the job definition. It was created with the successfully uploaded datapoints and can be used through code or the app.rapidata.ai UI:
+`exception.retry()` covers the common case. If you need to drive the retry yourself — for example to substitute corrected datapoints — the failed dataset is available on the exception and you can add datapoints to it directly:
 
 ```python
-from rapidata import RapidataClient
-
-client = RapidataClient()
-
-job_def = client.job.get_job_definition_by_id(job_definition_id) # (1)!
-audience = client.audience.get_audience_by_id("global")
-audience.assign_job(job_def)
+except FailedUploadException as e:
+    successful_retries, failed_retries = e.dataset.add_datapoints(e.failed_uploads)
 ```
 
-1. Retrieve the job definition using its ID (from the exception message or the [Rapidata Dashboard](https://app.rapidata.ai)).
+Note that this only re-uploads the datapoints; it does not create the job definition. Prefer `exception.retry()` unless you specifically need the lower-level control.

src/rapidata/rapidata_client/config/upload_config.py

@@ -103,6 +103,14 @@ class UploadConfig(BaseModel):
             maximum length is automatically shortened for the order/job instruction before
             upload. When False (default), an over-long context is left unchanged and a
             warning is logged that the backend would reject it. Defaults to False.
+        failureTolerance (float): The fraction of a job's datapoints allowed to fail while
+            still creating the job definition (0.0-1.0). 0.0 (default) is strict: any failed
+            upload aborts creation so no incomplete definition is left behind, and the failed
+            datapoints can be retried into the same dataset. 1.0 creates the definition
+            regardless of how many datapoints failed, as long as at least one datapoint
+            uploads successfully (a definition over an empty dataset is never created).
+            Overridable per call via the ``failure_tolerance`` argument on
+            ``create_*_job_definition``. Defaults to 0.0.
     """
 
     model_config = ConfigDict(validate_assignment=True)
@@ -145,6 +153,10 @@ def _apply_env_vars(cls, data: Any) -> Any:
         default=False,
         description="Automatically shorten over-long datapoint contexts for the instruction before upload.",
     )
+    failureTolerance: float = Field(
+        default=0.0,
+        description="Fraction of a job's datapoints allowed to fail while still creating the definition (0.0-1.0).",
+    )
 
     @field_validator("maxWorkers")
     @classmethod
@@ -174,6 +186,13 @@ def validate_batch_size(cls, v: int) -> int:
             raise ValueError("batchSize must be at least 100")
         return v
 
+    @field_validator("failureTolerance")
+    @classmethod
+    def validate_failure_tolerance(cls, v: float) -> float:
+        if not 0.0 <= v <= 1.0:
+            raise ValueError("failureTolerance must be between 0.0 and 1.0")
+        return v
+
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
         self._migrate_cache()

src/rapidata/rapidata_client/exceptions/failed_upload_exception.py

@@ -9,6 +9,9 @@
     from rapidata.rapidata_client.job.rapidata_job_definition import (
         RapidataJobDefinition,
     )
+    from rapidata.rapidata_client.job._job_creation_state_machine import (
+        JobDefinitionCreationMachine,
+    )
     from rapidata.rapidata_client.order.rapidata_order import RapidataOrder
 
 
@@ -21,13 +24,44 @@ def __init__(
         failed_uploads: list[FailedUpload[Datapoint]],
         order: Optional[RapidataOrder] = None,
         job_definition: Optional[RapidataJobDefinition] = None,
+        machine: Optional[JobDefinitionCreationMachine] = None,
     ):
         self.dataset = dataset
         self.order = order
         self.job_definition = job_definition
         self._failed_uploads = failed_uploads
+        self._machine = machine
         super().__init__(str(self))
 
+    def retry(self) -> RapidataJobDefinition:
+        """Retry the failed datapoints and finish creating the job definition.
+
+        Re-uploads only the datapoints that failed into the **same** dataset
+        (never a new one), then creates the job definition once the upload is
+        within the configured failure tolerance. Returns the created
+        ``RapidataJobDefinition``. If some datapoints still fail beyond the
+        tolerance, this raises ``FailedUploadException`` again (with the same
+        dataset attached) so it can be caught and retried in a loop.
+
+        This is the intended recovery path for a failed job-definition
+        creation: fix whatever caused the failures (e.g. correct file paths),
+        then call ``retry()`` - no need to re-specify the datapoints or drop
+        down to ``dataset.add_datapoints`` manually.
+        """
+        if self._machine is None:
+            raise RuntimeError(
+                "retry() is only available for failed job-definition creation. "
+                "To re-upload the failed datapoints manually, use "
+                "dataset.add_datapoints(exception.failed_uploads)."
+            )
+        return self._machine.resume()
+
+    @property
+    def machine(self) -> Optional[JobDefinitionCreationMachine]:
+        """The creation state machine backing ``retry()``, when this failure
+        came from job-definition creation (``None`` for order uploads)."""
+        return self._machine
+
     @property
     def failed_uploads(self) -> list[Datapoint]:
         """
@@ -85,6 +119,13 @@ def __str__(self) -> str:
             lines.append("  ]")
 
         failed_upload_message = "\n".join(lines)
+        if self.machine is not None:
+            failed_upload_message += (
+                "\n\nNo job definition was created because the upload stayed outside the "
+                "failure tolerance. Fix the failed datapoints and retry them into the same "
+                "dataset (no new dataset is created) by catching this exception and calling: "
+                "\n\tjob_definition = exception.retry()"
+            )
         if self.order:
             failed_upload_message += f"\n\nTo run the order without the failed datapoints, call: \n\trapidata_client.order.get_order_by_id('{self.order.id}').run()"
         if self.job_definition: