feat(job): require a capability below 10 responses per datapoint

docs/error_handling.md

@@ -200,6 +200,14 @@ except FailedUploadException as e:
 
 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.
 
+## Fewer than 10 responses per datapoint
+
+Job definitions collect at least 10 responses per datapoint. Asking for fewer, in a `create_*_job_definition` call or by setting `responses_per_datapoint` on an existing definition, raises `LowResponseCountError` before anything is uploaded:
+
+```
+Requested 5 responses per datapoint (minimum 10). At least 10 responses per datapoint are required. To collect fewer, email info@rapidata.ai with your use case and we will enable it for your organization.
+```
+
 ## Too many open files
 
 If uploads fail with `OSError: [Errno 24] Too many open files`, the process has hit its file-descriptor limit — the upload cache, worker pool, and HTTP connections all consume descriptors. When the SDK detects this it appends a hint to the `FailedUploadException` message pointing at the relevant knobs.

docs/job_definition_parameters.md

@@ -101,10 +101,12 @@ datapoints=[
 
 The minimum number of responses to collect for each datapoint. The actual number may slightly exceed this due to concurrent labelers.
 
+The minimum is `10`. Fewer raises `LowResponseCountError` before any datapoint is uploaded. If your use case needs fewer, email [info@rapidata.ai](mailto:info@rapidata.ai) with it and we will enable it for your organization. Ranking jobs are not affected: their `responses_per_comparison` can stay below 10.
+
 **Best Practices:**
 
 - Use 15-25 for ambiguous or subjective tasks
-- Use 5-10 for clear-cut decisions
+- Use 10 for clear-cut decisions, with `confidence_threshold` or `quorum_threshold` to stop early once the answer is clear
 
 ```python
 responses_per_datapoint=15

src/rapidata/__init__.py

@@ -95,6 +95,7 @@
     ContextManager,
     FailedUploadException,
     FailedUpload,
+    LowResponseCountError,
     RevisionConflictError,
     SampleUpload,
     rapidata_config,

src/rapidata/_skill/SKILL.md

@@ -98,7 +98,7 @@ The file is just one transport. To move the token over any transport (key-value
 
 Follow these for every job definition and audience example. Design each task so a labeler can answer it in about 10 seconds, 15 at most: labelers also answer validation tasks in the same session, mostly on phones, and within that time they read the instruction and the context and look at the media. A task they can't take in at a glance gets guessed. If a task needs longer, contact Rapidata (info@rapidata.ai) before launching it. The limits the SDK enforces are caps, not targets.
 
-1. **Keep `responses_per_datapoint` at 10 or more.** 10 is the default; never lower it to cut cost. With 3 responses, a single labeler who misread the task can flip the result. To spend less, label fewer datapoints, or add `confidence_threshold` / `quorum_threshold` when the task has one correct answer.
+1. **Keep `responses_per_datapoint` at 10 or more.** 10 is the default and the minimum; never lower it to cut cost. With 3 responses, a single labeler who misread the task can flip the result. To spend less, label fewer datapoints, or add `confidence_threshold` / `quorum_threshold` when the task has one correct answer. Below 10, `create_*_job_definition` and the `job_def.responses_per_datapoint` setter raise `LowResponseCountError` before any upload, unless the organization was enabled for it. Don't work around it: tell the user to email info@rapidata.ai with their use case. Ranking's `responses_per_comparison` is not affected.
 2. **Ask one short question about one criterion.** Aim for under 80 characters, like the instructions in this guide. To judge several aspects (motion, on-screen text, sound, …), create one job definition per aspect.
    - ✅ `"Which video matches the description better?"`
    - ❌ `"Which video is better overall? Consider the motion, timing, camera, style, on-screen text and sound."`

src/rapidata/_skill/reference.md

@@ -12,7 +12,7 @@ The new job-definition API exposes **classification**, **comparison**, **locate*
 | `instruction` | str | Task description shown to labelers (max 250 characters — longer raises `ValueError`) |
 | `datapoints` | list | Data to label (URLs or local paths) |
 | `data_type` | `"media"` \| `"text"` | `"media"` (default, covers image/video/audio) or `"text"`. **Text assets are NOT translated** unless `TranslateTextAssetsSetting` is added (source assumed English) — otherwise labelers see them verbatim |
-| `responses_per_datapoint` | int | Responses per item (default 10; never set it below 10 — see **Task design rules** in the main guide) |
+| `responses_per_datapoint` | int | Responses per item (default and minimum 10; fewer raises `LowResponseCountError` before any upload unless the organization was enabled for it via info@rapidata.ai — see **Task design rules** in the main guide) |
 | `contexts` | list[str] \| None | Text context per datapoint: one sentence under 150 characters with only what the instruction asks about (see **Task design rules** in the main guide). Max 400 characters each; contexts over the limit are always shortened against the instruction before upload — set `rapidata_config.upload.contextShortening = True` to shorten every context, or use `client.context` to shorten manually |
 | `media_contexts` | list[list[str]] \| None | Reference images per datapoint; each entry is a list of at most 10 image URLs/paths (one inner list per datapoint). Shown in a collapsible drawer unless `PinContextMediaSetting` is added (only when there is enough space — check the preview); for a compare of two image edits, `ReferenceOverlaySetting` lays a single image context over both options instead |
 | `confidence_threshold` | float \| None | Confidence-based early stopping threshold (0-1); cannot combine with `quorum_threshold` |

src/rapidata/rapidata_client/__init__.py

@@ -95,6 +95,11 @@
     DeviceFilter,
     DeviceType,
 )
-from .exceptions import FailedUploadException, FailedUpload, RevisionConflictError
+from .exceptions import (
+    FailedUploadException,
+    FailedUpload,
+    LowResponseCountError,
+    RevisionConflictError,
+)
 from .config import rapidata_config, logger, managed_print
 from .config.upload_config import CompressionConfig

src/rapidata/rapidata_client/exceptions/__init__.py

@@ -2,6 +2,7 @@
 from .asset_warning import AssetWarning
 from .failed_upload import FailedUpload
 from .failed_upload_exception import FailedUploadException
+from .low_response_count_error import LowResponseCountError
 from .rapidata_error import RapidataError
 from .revision_conflict_error import RevisionConflictError
 
@@ -10,6 +11,7 @@
     "AssetWarning",
     "FailedUpload",
     "FailedUploadException",
+    "LowResponseCountError",
     "RapidataError",
     "RevisionConflictError",
 ]

src/rapidata/rapidata_client/exceptions/low_response_count_error.py

@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+MIN_RESPONSES_PER_DATAPOINT = 10
+LOW_RESPONSE_COUNT_MESSAGE = (
+    "At least 10 responses per datapoint are required. To collect fewer, email "
+    "info@rapidata.ai with your use case and we will enable it for your organization."
+)
+
+
+class LowResponseCountError(ValueError):
+    """Raised before any upload when a job asks for fewer than 10 responses per
+    datapoint and the organization has not been enabled for it.
+
+    Attributes:
+        requested (int): The responses per datapoint that were asked for.
+        minimum (int): The minimum without the capability (10).
+    """
+
+    def __init__(self, requested: int):
+        self.requested = requested
+        self.minimum = MIN_RESPONSES_PER_DATAPOINT
+        super().__init__(
+            f"Requested {requested} responses per datapoint (minimum {self.minimum}). "
+            f"{LOW_RESPONSE_COUNT_MESSAGE}"
+        )

src/rapidata/rapidata_client/exceptions/rapidata_error.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 from typing import Optional, Any
-import json
 
 
 class RapidataError(Exception):
@@ -44,6 +43,20 @@ def _trace_id_from_details(details: Any) -> str | None:
                 return raw
         return None
 
+    @staticmethod
+    def _format_errors(errors: Any) -> list[str]:
+        """One ``field: message`` line per validation message."""
+        if isinstance(errors, dict):
+            lines = []
+            for field, messages in errors.items():
+                if not isinstance(messages, list):
+                    messages = [messages]
+                lines.extend(f"{field}: {m}" if field else str(m) for m in messages)
+            return lines
+        if isinstance(errors, list):
+            return [str(e) for e in errors]
+        return [str(errors)]
+
     def get_reason(self) -> str:
         """Get a concise reason string suitable for grouping and display.
 
@@ -76,10 +89,7 @@ def __str__(self):
             error_parts.append(f"{self.message or 'Unknown error'}")
 
         if errors:
-            if isinstance(errors, dict):
-                error_parts.append(f"Reasons: {json.dumps({'errors': errors})}")
-            else:
-                error_parts.append(f"Reasons: {errors}")
+            error_parts.extend(self._format_errors(errors))
 
         if self.trace_id:
             error_parts.append(f"Trace Id: {self.trace_id}")

src/rapidata/rapidata_client/job/_response_minimum.py

@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+import json
+import weakref
+from typing import TYPE_CHECKING, Any
+
+from rapidata.rapidata_client.config import logger
+from rapidata.rapidata_client.job._revision_body import _as_dict
+from rapidata.rapidata_client.exceptions.low_response_count_error import (
+    MIN_RESPONSES_PER_DATAPOINT,
+    LowResponseCountError,
+)
+
+if TYPE_CHECKING:
+    from rapidata.api_client.models.i_order_workflow_input import IOrderWorkflowInput
+    from rapidata.service.openapi_service import OpenAPIService
+
+_LOW_RESPONSE_COUNT_CAPABILITY = "LowResponseCount"
+
+# Only grants are cached: an organization enabled mid-session works without a new client.
+_granted: weakref.WeakKeyDictionary[OpenAPIService, bool] = weakref.WeakKeyDictionary()
+
+
+def ensure_response_count_allowed(
+    openapi_service: OpenAPIService,
+    workflow: IOrderWorkflowInput,
+    responses: int,
+) -> None:
+    """Raise LowResponseCountError before anything is uploaded.
+
+    Only simple workflows are gated, as on the backend; ranking is not.
+    """
+    if responses >= MIN_RESPONSES_PER_DATAPOINT:
+        return
+    if _as_dict(workflow).get("_t") != "SimpleWorkflow":
+        return
+    if _low_response_count_allowed(openapi_service) is False:
+        raise LowResponseCountError(responses)
+
+
+def _low_response_count_allowed(openapi_service: OpenAPIService) -> bool | None:
+    """None when the lookup failed, which leaves the decision to the backend."""
+    if _granted.get(openapi_service):
+        return True
+
+    from rapidata.rapidata_client.api.rapidata_api_client import (
+        suppress_rapidata_error_logging,
+    )
+
+    try:
+        with suppress_rapidata_error_logging():
+            response = openapi_service.api_client.call_api(
+                "GET",
+                f"https://auth.{openapi_service.environment}/connect/userinfo",
+                _request_timeout=5,
+            )
+        if response.status != 200:
+            raise RuntimeError(f"userinfo returned {response.status}")
+        userinfo: dict[str, Any] = json.loads(response.read().decode("utf-8"))
+    except Exception as e:
+        logger.debug("Capability lookup failed, the backend decides: %s", e)
+        return None
+
+    allowed = _LOW_RESPONSE_COUNT_CAPABILITY in _claim_values(
+        userinfo, "capabilities"
+    ) or "admin" in _claim_values(userinfo, "platform_relations")
+    if allowed:
+        _granted[openapi_service] = True
+    return allowed
+
+
+def _claim_values(userinfo: dict[str, Any], claim: str) -> list[str]:
+    # A claim with a single value comes back as a bare string.
+    value = userinfo.get(claim)
+    if isinstance(value, str):
+        return [value]
+    return value if isinstance(value, list) else []

src/rapidata/rapidata_client/job/rapidata_job_definition.py

@@ -29,6 +29,9 @@
     _with_instruction,
     _with_responses,
 )
+from rapidata.rapidata_client.job._response_minimum import (
+    ensure_response_count_allowed,
+)
 import webbrowser
 import urllib.parse
 from colorama import Fore
@@ -214,7 +217,10 @@ def answer_options(self, value: list[str]) -> None:
 
     @property
     def responses_per_datapoint(self) -> int:
-        """The maximum number of responses collected per datapoint."""
+        """The maximum number of responses collected per datapoint.
+
+        Setting fewer than 10 raises LowResponseCountError unless your organization was enabled for it.
+        """
         return self._effective(
             "responses_per_datapoint", _get_responses(self._snapshot.referee)
         )
@@ -224,6 +230,9 @@ def responses_per_datapoint(self, value: int) -> None:
         if isinstance(value, bool) or not isinstance(value, int) or value < 1:
             raise ValueError("responses_per_datapoint must be a positive integer")
         _with_responses(self._snapshot.referee, value)
+        ensure_response_count_allowed(
+            self._openapi_service, self._snapshot.workflow, value
+        )
         self._pending["responses_per_datapoint"] = value
 
     @property