fix(builder): handle resume when metadata.json missing (interrupted before first batch)

When a run is interrupted before any row group or batch completes, metadata.json
is never written. Previously resume=True would raise DatasetGenerationError in
this case. Now build() detects the missing file, logs an info message, clears
any leftover partial results and falls back to a clean fresh run.

This is the common scenario for small datasets (fewer records than buffer_size)
where all records fit in a single row group.

Author: przemekboruta

packages/data-designer-engine/src/data_designer/engine/dataset_builders/dataset_builder.py

@@ -185,6 +185,17 @@ def build(
         start_time = time.perf_counter()
         buffer_size = self._resource_provider.run_config.buffer_size
 
+        if resume and not self.artifact_storage.metadata_file_path.exists():
+            # No metadata.json means the previous run was interrupted before any batch (sync) or
+            # row group (async) completed.  Nothing to resume — discard any leftover partial
+            # results and start fresh.
+            logger.info(
+                "▶️ No metadata.json found — the previous run was interrupted before any batch "
+                "completed. Starting generation from the beginning."
+            )
+            self.artifact_storage.clear_partial_results()
+            resume = False
+
         generated = True
         if DATA_DESIGNER_ASYNC_ENGINE:
             self._validate_async_compatibility()

packages/data-designer-engine/tests/engine/dataset_builders/test_dataset_builder.py

@@ -968,20 +968,27 @@ def _make_resume_builder(stub_resource_provider, stub_test_config_builder, tmp_p
     )
 
 
-def test_build_resume_raises_without_metadata(stub_resource_provider, stub_test_config_builder, tmp_path):
-    """resume=True when only the folder exists (no metadata.json) raises DatasetGenerationError.
+def test_build_resume_starts_fresh_without_metadata(stub_resource_provider, stub_test_config_builder, tmp_path, caplog):
+    """resume=True when only the folder exists (no metadata.json) logs an info message and starts fresh.
 
     This covers the case where a run was interrupted before any batch completed — the
     folder was created by _write_builder_config but metadata.json was never written.
+    Previously this raised DatasetGenerationError; now it silently restarts from batch 0.
     """
     # Pre-create the folder with content so resolved_dataset_name(resume=True) returns "dataset"
     dataset_dir = tmp_path / "dataset"
     dataset_dir.mkdir()
     (dataset_dir / "builder_config.json").write_text("{}")  # non-empty, no metadata
 
     builder = _make_resume_builder(stub_resource_provider, stub_test_config_builder, tmp_path)
-    with pytest.raises(DatasetGenerationError, match="metadata.json not found"):
-        builder.build(num_records=4, resume=True)
+    with caplog.at_level(logging.INFO):
+        with patch.object(builder, "_run_model_health_check_if_needed"):
+            with patch.object(builder, "_run_batch"):
+                with patch.object(builder.batch_manager, "finish"):
+                    # resume=False is set internally; build dispatches to the normal (non-resume) path
+                    builder.build(num_records=4, resume=True)
+
+    assert any("interrupted before any batch completed" in record.message for record in caplog.records)
 
 
 def test_build_resume_raises_on_num_records_mismatch(stub_resource_provider, stub_test_config_builder, tmp_path):
@@ -1132,18 +1139,31 @@ def test_build_async_resume_logs_warning_when_already_complete(
     assert any("already complete" in record.message for record in caplog.records)
 
 
-def test_build_async_resume_raises_without_metadata(stub_resource_provider, stub_test_config_builder, tmp_path):
-    """Async resume raises DatasetGenerationError when metadata.json is missing."""
+def test_build_async_resume_starts_fresh_without_metadata(
+    stub_resource_provider, stub_test_config_builder, tmp_path, caplog
+):
+    """Async resume with no metadata.json logs an info message and starts fresh.
+
+    Previously this raised DatasetGenerationError; now it silently restarts from row group 0.
+    The log is emitted in build() before dispatching to _build_async, so mocking _build_async
+    does not suppress the message.
+    """
     dataset_dir = tmp_path / "dataset"
     dataset_dir.mkdir()
     (dataset_dir / "builder_config.json").write_text("{}")
 
     builder = _make_resume_builder(stub_resource_provider, stub_test_config_builder, tmp_path)
 
-    with patch.object(builder_mod, "DATA_DESIGNER_ASYNC_ENGINE", True):
-        with patch.object(builder, "_run_model_health_check_if_needed"):
-            with pytest.raises(DatasetGenerationError, match="metadata.json not found"):
-                builder.build(num_records=4, resume=True)
+    with caplog.at_level(logging.INFO):
+        with patch.object(builder_mod, "DATA_DESIGNER_ASYNC_ENGINE", True):
+            with patch.object(builder, "_run_model_health_check_if_needed"):
+                with patch.object(builder, "_build_async", return_value=True) as mock_async:
+                    builder.build(num_records=4, resume=True)
+
+    # _build_async is called with resume=False because the no-metadata path resets the flag
+    _, kwargs = mock_async.call_args
+    assert kwargs.get("resume") is False
+    assert any("interrupted before any batch completed" in record.message for record in caplog.records)
 
 
 def test_build_async_resume_already_complete_does_not_run_after_generation_processors(

packages/data-designer/tests/cli/commands/test_create_command.py

@@ -18,7 +18,9 @@ def test_create_command_delegates_to_controller(mock_ctrl_cls: MagicMock) -> Non
     mock_ctrl = MagicMock()
     mock_ctrl_cls.return_value = mock_ctrl
 
-    create_command(config_source="config.yaml", num_records=10, dataset_name="dataset", artifact_path=None, output_format=None)
+    create_command(
+        config_source="config.yaml", num_records=10, dataset_name="dataset", artifact_path=None, output_format=None
+    )
 
     mock_ctrl_cls.assert_called_once()
     mock_ctrl.run_create.assert_called_once_with(
@@ -59,7 +61,9 @@ def test_create_command_default_artifact_path_is_none(mock_ctrl_cls: MagicMock)
     mock_ctrl = MagicMock()
     mock_ctrl_cls.return_value = mock_ctrl
 
-    create_command(config_source="config.yaml", num_records=5, dataset_name="ds", artifact_path=None, output_format=None)
+    create_command(
+        config_source="config.yaml", num_records=5, dataset_name="ds", artifact_path=None, output_format=None
+    )
 
     mock_ctrl.run_create.assert_called_once_with(
         config_source="config.yaml",