feat(annotation): calibration_max_records absolute cap + inheritance cascade

src/pragmata/api/annotation_import.py

@@ -49,10 +49,12 @@ class ImportResult:
             production dataset.
         calibration_fraction: Per-task configured fraction. May differ from
             ``realised_calibration_fraction`` on re-imports because prior
-            assignments are locked by the manifest.
+            assignments are locked by the manifest, and under a binding cap.
         realised_calibration_fraction: Per-task actual share of items routed
             to calibration this run (calibration / (calibration + production)).
             Zero when no items were assigned for that task.
+        calibration_max_items: Per-task configured absolute cap. ``None`` =
+            uncapped for that task.
         errors: Per-record validation failures (index + detail).
     """
 
@@ -62,6 +64,7 @@ class ImportResult:
     production_count: dict[Task, int] = field(default_factory=dict)
     calibration_fraction: dict[Task, float] = field(default_factory=dict)
     realised_calibration_fraction: dict[Task, float] = field(default_factory=dict)
+    calibration_max_items: dict[Task, int | None] = field(default_factory=dict)
     errors: list[RecordError] = field(default_factory=list)
 
 
@@ -75,6 +78,7 @@ def import_records(
     base_dir: str | Path | Unset = UNSET,
     config_path: str | Path | Unset = UNSET,
     calibration_fraction: float | Unset = UNSET,
+    calibration_max_items: int | None | Unset = UNSET,
     calibration_min_submitted: int | None | Unset = UNSET,
     calibration_partition_seed: int | Unset = UNSET,
     locale: Locale | Unset = UNSET,
@@ -124,6 +128,11 @@ def import_records(
             unless overridden in YAML config). Falls through to YAML config
             and the built-in default (0.1) when omitted. Set to 0.0 for
             production-only batches.
+        calibration_max_items: Deployment-level absolute cap on calibration
+            annotation items per task. ``None`` is uncapped (just the
+            fractional knob). Inherited by workspaces/tasks unless overridden
+            in YAML config. Cap unit is the annotation item: chunks for
+            retrieval, records for grounding / generation.
         calibration_min_submitted: Deployment-level overlap requirement for
             the calibration dataset. ``None`` disables calibration entirely
             (must be paired with ``calibration_fraction=0.0``). Inherits to
@@ -153,6 +162,7 @@ def import_records(
             "dataset_id": dataset_id,
             "base_dir": base_dir,
             "calibration_fraction": calibration_fraction,
+            "calibration_max_items": calibration_max_items,
             "calibration_min_submitted": calibration_min_submitted,
             "calibration_partition_seed": calibration_partition_seed,
             "locale": locale,
@@ -199,19 +209,26 @@ def import_records(
 
         # Manifest is written only after fan-out succeeds. On failure, the
         # in-memory assignments are dropped; a retry with the same corpus + seed
-        # re-derives identical buckets (deterministic per-unit hashing).
+        # re-derives identical buckets under the fraction-only path. Under a
+        # binding cap, retries are order-dependent (see
+        # ``test_cap_under_split_imports_is_order_dependent_by_design``).
         dataset_counts = fan_out_records(client, settings, partition=partition)
         write_partition_manifest(import_paths.partition_manifest, manifest)
 
+    def _per_task_summary(task: Task) -> str:
+        cap = partition.calibration_max_items[task]
+        cap_suffix = f", cap={cap}" if cap is not None else ""
+        return (
+            f"{task.value}={summary.calibration_count[task]} "
+            f"(realised={summary.realised_fraction[task]:.3f}, "
+            f"configured={summary.configured_fraction[task]:.3f}{cap_suffix})"
+        )
+
     logger.info(
         "Import complete: %d records across %d datasets. Per-task calibration: %s",
         len(raw),
         len(dataset_counts),
-        ", ".join(
-            f"{task.value}={summary.calibration_count[task]} "
-            f"(realised={summary.realised_fraction[task]:.3f}, configured={summary.configured_fraction[task]:.3f})"
-            for task in Task
-        ),
+        ", ".join(_per_task_summary(task) for task in Task),
     )
     return ImportResult(
         total_records=len(raw),
@@ -220,5 +237,6 @@ def import_records(
         production_count=summary.production_count,
         calibration_fraction=summary.configured_fraction,
         realised_calibration_fraction=summary.realised_fraction,
+        calibration_max_items=partition.calibration_max_items,
         errors=validation.errors,
     )

src/pragmata/cli/commands/annotation.py

@@ -106,11 +106,21 @@ def import_command(
         "YAML config). Falls through to YAML config and built-in default (0.1) when "
         "omitted; set to 0.0 for production-only batches.",
     ),
+    calibration_max_items: int | None = typer.Option(
+        None,
+        "--calibration-max-items",
+        help="Deployment-level absolute cap on calibration annotation items per task "
+        "(inherited by workspaces/tasks unless overridden in YAML config). Smaller of "
+        "(fraction × N_items, cap) wins. Existing assignments are never demoted. "
+        "Cap unit is the annotation item: chunks for retrieval, records for grounding "
+        "and generation. Omit to leave uncapped.",
+    ),
     no_calibration: bool = typer.Option(
         False,
         "--no-calibration",
         help="Disable calibration entirely for this batch: sets calibration_min_submitted=None "
-        "and calibration_fraction=0.0. Cannot be combined with --calibration-fraction > 0.",
+        "and calibration_fraction=0.0. Cannot be combined with --calibration-fraction > 0 "
+        "or --calibration-max-items.",
     ),
     calibration_partition_seed: int | None = typer.Option(
         None,
@@ -147,6 +157,12 @@ def import_command(
             err=True,
         )
         raise typer.Exit(code=2)
+    if no_calibration and calibration_max_items is not None:
+        typer.echo(
+            "Error: --no-calibration cannot be combined with --calibration-max-items.",
+            err=True,
+        )
+        raise typer.Exit(code=2)
 
     result = annotation.import_records(
         records,
@@ -159,6 +175,7 @@ def import_command(
         calibration_fraction=(
             0.0 if no_calibration else UNSET if calibration_fraction is None else calibration_fraction
         ),
+        calibration_max_items=UNSET if calibration_max_items is None else calibration_max_items,
         calibration_min_submitted=None if no_calibration else UNSET,
         calibration_partition_seed=UNSET if calibration_partition_seed is None else calibration_partition_seed,
         locale=parse_locale(locale) or UNSET,
@@ -169,9 +186,11 @@ def import_command(
         prod_n = result.production_count.get(task, 0)
         cfg_fraction = result.calibration_fraction.get(task, 0.0)
         realised = result.realised_calibration_fraction.get(task, 0.0)
+        cap = result.calibration_max_items.get(task)
+        cap_str = f", cap={cap}" if cap is not None else ""
         typer.echo(
             f"  {task.value}: calibration={cal_n}, production={prod_n} "
-            f"(configured={cfg_fraction:.3f}, realised={realised:.3f})"
+            f"(configured={cfg_fraction:.3f}, realised={realised:.3f}{cap_str})"
         )
     for ds, count in result.dataset_counts.items():
         typer.echo(f"  {ds}: {count}")