feat: define public generation invocation contract

packages/data-designer-config/src/data_designer/config/__init__.py

@@ -32,6 +32,11 @@
     from data_designer.config.config_builder import DataDesignerConfigBuilder  # noqa: F401
     from data_designer.config.custom_column import custom_column_generator  # noqa: F401
     from data_designer.config.data_designer_config import DataDesignerConfig  # noqa: F401
+    from data_designer.config.errors import (  # noqa: F401
+        InvalidConfigError,
+        InvalidFileFormatError,
+        InvalidFilePathError,
+    )
     from data_designer.config.mcp import (  # noqa: F401
         LocalStdioMCPProvider,
         MCPProvider,
@@ -160,6 +165,10 @@
     "custom_column_generator": (f"{_MOD_BASE}.custom_column", "custom_column_generator"),
     # data_designer_config
     "DataDesignerConfig": (f"{_MOD_BASE}.data_designer_config", "DataDesignerConfig"),
+    # errors
+    "InvalidConfigError": (f"{_MOD_BASE}.errors", "InvalidConfigError"),
+    "InvalidFileFormatError": (f"{_MOD_BASE}.errors", "InvalidFileFormatError"),
+    "InvalidFilePathError": (f"{_MOD_BASE}.errors", "InvalidFilePathError"),
     # mcp
     "LocalStdioMCPProvider": (_MOD_MCP, "LocalStdioMCPProvider"),
     "MCPProvider": (_MOD_MCP, "MCPProvider"),

packages/data-designer-config/src/data_designer/config/config_builder.py

@@ -8,6 +8,7 @@
 import logging
 from pathlib import Path
 
+import yaml
 from pydantic import model_validator
 from pygments import highlight
 from pygments.formatters import HtmlFormatter
@@ -24,7 +25,13 @@
 )
 from data_designer.config.data_designer_config import DataDesignerConfig
 from data_designer.config.default_model_settings import get_default_model_configs
-from data_designer.config.errors import BuilderConfigurationError, BuilderSerializationError, InvalidColumnTypeError
+from data_designer.config.errors import (
+    BuilderConfigurationError,
+    BuilderSerializationError,
+    InvalidColumnTypeError,
+    InvalidFileFormatError,
+    InvalidFilePathError,
+)
 from data_designer.config.exportable_config import ExportableConfigBase
 from data_designer.config.mcp import ToolConfig
 from data_designer.config.models import ModelConfig, load_model_configs
@@ -111,11 +118,19 @@ def from_config(cls, config: dict | str | Path | BuilderConfig) -> Self:
         Raises:
             ValueError: If the config format is invalid.
             ValidationError: If the builder config loaded from the config is invalid.
+            InvalidFilePathError: If a local config path cannot be read.
+            InvalidFileFormatError: If a local or inline config contains malformed YAML or JSON.
         """
         if isinstance(config, BuilderConfig):
             builder_config = config
         else:
-            json_config = json.loads(serialize_data(smart_load_yaml(config)))
+            try:
+                loaded_config = smart_load_yaml(config)
+            except OSError as e:
+                raise InvalidFilePathError(f"Failed to load builder config: {e}") from e
+            except yaml.YAMLError as e:
+                raise InvalidFileFormatError(f"Failed to parse builder config: {e}") from e
+            json_config = json.loads(serialize_data(loaded_config))
             # Normalize shorthand DataDesignerConfig into full BuilderConfig
             if "columns" in json_config and "data_designer" not in json_config:
                 json_config = {"data_designer": json_config}

packages/data-designer-config/src/data_designer/config/utils/io_helpers.py

@@ -190,7 +190,7 @@ def _smart_load_yaml_internal(yaml_in: str | Path | dict, *, from_url: bool) ->
         with open(yaml_in) as file:
             yaml_out = yaml.safe_load(file)
     elif isinstance(yaml_in, str):
-        if not from_url and yaml_in.endswith((".yaml", ".yml")) and not os.path.isfile(yaml_in):
+        if not from_url and yaml_in.endswith((".yaml", ".yml", ".json")) and not os.path.isfile(yaml_in):
             raise FileNotFoundError(f"File not found: {yaml_in}")
         else:
             yaml_out = yaml.safe_load(yaml_in)

packages/data-designer-config/tests/config/test_public_contract.py

@@ -0,0 +1,38 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+import data_designer.config.errors as config_errors
+from data_designer.config import (
+    DataDesignerConfigBuilder,
+    InvalidConfigError,
+    InvalidFileFormatError,
+    InvalidFilePathError,
+)
+
+
+def test_invalid_config_error_is_publicly_exported() -> None:
+    assert InvalidConfigError is config_errors.InvalidConfigError
+
+
+def test_builder_file_errors_are_publicly_exported() -> None:
+    assert InvalidFileFormatError is config_errors.InvalidFileFormatError
+    assert InvalidFilePathError is config_errors.InvalidFilePathError
+
+
+@pytest.mark.parametrize("filename", ["missing.yaml", "missing.json"])
+def test_from_config_normalizes_missing_file_error(tmp_path: Path, filename: str) -> None:
+    with pytest.raises(InvalidFilePathError) as exc_info:
+        DataDesignerConfigBuilder.from_config(str(tmp_path / filename))
+
+    assert isinstance(exc_info.value.__cause__, FileNotFoundError)
+
+
+def test_from_config_normalizes_malformed_config_error() -> None:
+    with pytest.raises(InvalidFileFormatError):
+        DataDesignerConfigBuilder.from_config("data_designer: [")

packages/data-designer/src/data_designer/interface/__init__.py

@@ -8,6 +8,7 @@
 
 if TYPE_CHECKING:
     from data_designer.config.run_config import ResumeMode  # noqa: F401
+    from data_designer.engine.dataset_builders.errors import ArtifactStorageError  # noqa: F401
     from data_designer.interface.composite_workflow import (  # noqa: F401
         CompositeWorkflow,
         CompositeWorkflowResults,
@@ -24,6 +25,7 @@
     from data_designer.interface.results import DatasetCreationResults  # noqa: F401
 
 _LAZY_IMPORTS: dict[str, tuple[str, str]] = {
+    "ArtifactStorageError": ("data_designer.engine.dataset_builders.errors", "ArtifactStorageError"),
     "CompositeWorkflow": ("data_designer.interface.composite_workflow", "CompositeWorkflow"),
     "CompositeWorkflowResults": ("data_designer.interface.composite_workflow", "CompositeWorkflowResults"),
     "DataDesigner": ("data_designer.interface.data_designer", "DataDesigner"),

packages/data-designer/src/data_designer/interface/composite_workflow.py

@@ -369,6 +369,9 @@ def run(
                     stage=stage,
                     stage_dir_name=stage_dir_name,
                     stage_builder=stage_builder,
+                    requested_num_records=stage_metadata.get(
+                        "num_records_actual" if stage.output_processors else "num_records_requested"
+                    ),
                 )
                 stage_results[stage.name] = output_result
                 stage_output_paths[stage.name] = output_seed_path
@@ -671,6 +674,7 @@ def _stage_result_from_metadata(
     stage: _WorkflowStage,
     stage_dir_name: str,
     stage_builder: DataDesignerConfigBuilder,
+    requested_num_records: int | None,
 ) -> DatasetCreationResults:
     main_storage = ArtifactStorage(artifact_path=workflow_path, dataset_name=stage_dir_name, resume=ResumeMode.ALWAYS)
     result_storage = main_storage
@@ -691,6 +695,7 @@ def _stage_result_from_metadata(
         analysis=_load_stage_analysis(result_storage),
         config_builder=result_builder,
         dataset_metadata=DatasetMetadata(),
+        requested_num_records=requested_num_records,
     )
 
 

packages/data-designer/src/data_designer/interface/data_designer.py

@@ -266,10 +266,12 @@ def create(
 
         Returns:
             DatasetCreationResults object with methods for loading the generated dataset,
-            analysis results, and displaying sample records for inspection.
+            analysis results, displaying sample records, and inspecting completion,
+            artifact, and resume metadata.
 
         Raises:
             DataDesignerGenerationError: If an error occurs during dataset generation.
+            DataDesignerEarlyShutdownError: If early shutdown produces no records.
             DataDesignerProfilingError: If an error occurs during dataset profiling.
         """
         logger.info("🎨 Creating Data Designer dataset")
@@ -378,6 +380,10 @@ def create(
             config_builder=config_builder,
             dataset_metadata=dataset_metadata,
             task_traces=task_traces,
+            requested_num_records=num_records,
+            early_shutdown=builder.early_shutdown,
+            requested_resume_mode=resume,
+            effective_resume_mode=builder.artifact_storage.resume,
         )
 
     async def acreate(

packages/data-designer/src/data_designer/interface/results.py

@@ -11,6 +11,7 @@
 from data_designer.config.config_builder import DataDesignerConfigBuilder
 from data_designer.config.dataset_metadata import DatasetMetadata
 from data_designer.config.errors import InvalidFileFormatError
+from data_designer.config.run_config import ResumeMode
 from data_designer.config.seed_source_dataframe import DataFrameSeedSource
 from data_designer.config.utils.visualization import WithRecordSamplerMixin
 from data_designer.engine.dataset_builders.errors import ArtifactStorageError
@@ -50,6 +51,10 @@ def __init__(
         config_builder: DataDesignerConfigBuilder,
         dataset_metadata: DatasetMetadata,
         task_traces: list[TaskTrace] | None = None,
+        requested_num_records: int | None = None,
+        early_shutdown: bool | None = None,
+        requested_resume_mode: ResumeMode | None = None,
+        effective_resume_mode: ResumeMode | None = None,
     ):
         """Creates a new instance with results based on a dataset creation run.
 
@@ -62,12 +67,40 @@ def __init__(
                 Resume note: only contains traces for the current invocation; traces
                 from earlier ``create()`` calls that this run resumed are not
                 retained.
+            requested_num_records: Number of records requested for this invocation.
+            early_shutdown: Whether generation stopped at the early-shutdown gate.
+                ``None`` when no generation invocation produced this result object.
+            requested_resume_mode: Resume mode requested for this invocation, or
+                ``None`` when no generation invocation produced this result object.
+            effective_resume_mode: Resume mode selected after compatibility checks,
+                or ``None`` when no generation invocation produced this result object.
         """
         self.artifact_storage = artifact_storage
         self._analysis = analysis
         self._config_builder = config_builder
         self.dataset_metadata = dataset_metadata
         self.task_traces: list[TaskTrace] = task_traces or []
+        self.requested_num_records = requested_num_records
+        self.early_shutdown = early_shutdown
+        self.requested_resume_mode = requested_resume_mode
+        self.effective_resume_mode = effective_resume_mode
+
+    @property
+    def dataset_path(self) -> Path:
+        """Return the resolved dataset directory for this result."""
+        return self.artifact_storage.base_dataset_path
+
+    @property
+    def actual_num_records(self) -> int:
+        """Return the total number of records in the result dataset."""
+        return self.count_records()
+
+    @property
+    def is_partial(self) -> bool | None:
+        """Return whether the result contains fewer records than requested."""
+        if self.requested_num_records is None:
+            return None
+        return self.actual_num_records < self.requested_num_records
 
     def load_analysis(self) -> DatasetProfilerResults:
         """Load the profiling analysis results for the generated dataset.

packages/data-designer/tests/interface/test_composite_workflow.py

@@ -590,6 +590,39 @@ def test_composite_workflow_resume_if_possible_skips_completed_stages(
     assert create_mock.call_count == 0
     assert results.count_records() == 3
     assert results.load_dataset()["category"].tolist() == ["alpha", "alpha", "alpha"]
+    base_result = results["base"]
+    assert isinstance(base_result, DatasetCreationResults)
+    assert base_result.requested_num_records == 3
+    assert base_result.actual_num_records == 3
+    assert base_result.is_partial is False
+    assert base_result.early_shutdown is None
+    assert base_result.requested_resume_mode is None
+    assert base_result.effective_resume_mode is None
+
+
+def test_composite_workflow_reused_stage_reports_partial_result(
+    stub_artifact_path: Path,
+    stub_model_providers: list[ModelProvider],
+    stub_model_configs: list[ModelConfig],
+    stub_dataset_profiler_results,
+) -> None:
+    data_designer = _data_designer(stub_artifact_path, stub_model_providers)
+    create_mock = _patch_create(data_designer, stub_dataset_profiler_results)
+    workflow = data_designer.compose_workflow(name="resume-partial")
+    workflow.add_stage("base", _category_builder(stub_model_configs), num_records=3)
+    first = workflow.run()
+    batch_path = first["base"].artifact_storage.final_dataset_path / "batch_00000.parquet"
+    lazy.pd.read_parquet(batch_path).head(2).to_parquet(batch_path, index=False)
+    create_mock.reset_mock()
+
+    resumed = data_designer.compose_workflow(name="resume-partial")
+    resumed.add_stage("base", _category_builder(stub_model_configs), num_records=3)
+    results = resumed.run(resume=ResumeMode.IF_POSSIBLE)
+
+    assert create_mock.call_count == 0
+    assert results["base"].requested_num_records == 3
+    assert results["base"].actual_num_records == 2
+    assert results["base"].is_partial is True
 
 
 def test_composite_workflow_resume_if_possible_skips_stage_with_output_processors(