fix: validate workflow rerun inputs

Author: andreatgretel

fern/versions/latest/pages/concepts/workflow-chaining.mdx

@@ -129,7 +129,7 @@ results = workflow.run(
 )
 ```
 
-If the reviewed data replaces a stage's selected output in place, use `rerun_from="expanded"` to force that stage and its descendants to rebuild from the current boundary output.
+If the reviewed data replaces a stage's selected output in place, run with `resume=ResumeMode.IF_POSSIBLE` and `rerun_from="expanded"` to rebuild that stage and its descendants from the current boundary output.
 
 ## Current limits
 

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

@@ -258,6 +258,10 @@ def run(
         """
         if not self._stages:
             raise DataDesignerWorkflowError(f"Workflow {self.name!r} has no stages.")
+        if rerun_from is not None and resume == ResumeMode.NEVER:
+            raise DataDesignerWorkflowError(
+                "rerun_from requires resume to be ResumeMode.IF_POSSIBLE or ResumeMode.ALWAYS."
+            )
 
         stage_indices = _stage_indices_by_name(self._stages)
         target_stage_names = _normalize_stage_names(targets, stage_indices, "target")
@@ -537,6 +541,13 @@ def _validate_stage_output_overrides(
         raise DataDesignerWorkflowError(
             f"Stage output override(s) must be ancestors of a target stage: {', '.join(non_ancestors)}."
         )
+    for name in sorted(stage_output_overrides):
+        override_path = _stage_output_override(name, stage_output_overrides)
+        if override_path is not None:
+            try:
+                _count_parquet_records(override_path)
+            except DataDesignerWorkflowError as exc:
+                raise DataDesignerWorkflowError(f"Invalid stage output override for stage {name!r}: {exc}") from exc
 
 
 def _stage_output_override(

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

@@ -493,6 +493,24 @@ def test_composite_workflow_rerun_from_forces_stage_and_descendants(
     assert [call.kwargs["resume"] for call in create_mock.call_args_list] == [ResumeMode.NEVER, ResumeMode.NEVER]
 
 
+def test_composite_workflow_rerun_from_requires_resume(
+    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="rerun-from-no-resume")
+    workflow.add_stage("base", _category_builder(stub_model_configs), num_records=2)
+    workflow.add_stage("copy", _copy_builder(stub_model_configs))
+
+    with pytest.raises(DataDesignerWorkflowError, match="rerun_from requires resume"):
+        workflow.run(rerun_from="copy")
+
+    assert create_mock.call_count == 0
+
+
 def test_composite_workflow_stage_output_override_seeds_descendants(
     tmp_path: Path,
     stub_model_providers: list[ModelProvider],
@@ -532,6 +550,24 @@ def test_composite_workflow_stage_output_override_seeds_descendants(
     assert metadata["stages"][1]["seed_path"] == str(approved_path.resolve())
 
 
+def test_composite_workflow_stage_output_override_path_must_exist(
+    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="missing-override")
+    workflow.add_stage("base", _category_builder(stub_model_configs), num_records=2)
+    workflow.add_stage("copy", _copy_builder(stub_model_configs))
+
+    with pytest.raises(DataDesignerWorkflowError, match="Invalid stage output override"):
+        workflow.run(targets="copy", stage_output_overrides={"base": stub_artifact_path / "missing.parquet"})
+
+    assert create_mock.call_count == 0
+
+
 def test_composite_workflow_resume_if_possible_skips_completed_stages(
     stub_artifact_path: Path,
     stub_model_providers: list[ModelProvider],