Fix isoflop_analysis and reference to use StepSpec for dedup steps

ExecutorStep.fn receives the config object, not the output path. Using
StepSpec(...).as_executor_step() preserves the lambda(output_path) calling
convention via the round-trip short-circuit in resolve_executor_step.

Author: ravwojdyla

experiments/dedup/isoflop_analysis.py

@@ -23,6 +23,7 @@
 )
 from experiments.scaling_law_sweeps.c_adamc import create_isoflop_sweep_steps
 from marin.execution.executor import ExecutorStep, executor_main, this_output_path, versioned
+from marin.execution.step_spec import StepSpec
 
 logger = logging.getLogger(__name__)
 
@@ -61,14 +62,14 @@ def _get_vanilla_data_mixture(*, variant: str) -> LMMixtureDatasetConfig:
 
 def _get_deduped_data_mixture(*, variant: str, mode: DedupMode, max_parallelism: int = 1024) -> LMMixtureDatasetConfig:
     """Dedup fineweb-edu mixture"""
-    dedup_step = ExecutorStep(
+    dedup_step = StepSpec(
         name=f"dedup/{variant}_{mode.lower()}",
         fn=lambda op: _DEDUP_FN[mode](
             input_paths=downloads[variant],
             output_path=op,
             max_parallelism=max_parallelism,
         ),
-    )
+    ).as_executor_step()
 
     dedup_mode_to_filter_type = {
         DedupMode.EXACT_PARAGRAPH: FilterType.REMOVE_SPANS,

experiments/dedup/reference.py

@@ -4,6 +4,7 @@
 import logging
 
 from marin.execution.executor import ExecutorStep, InputName, executor_main
+from marin.execution.step_spec import StepSpec
 from marin.processing.classification.deduplication.fuzzy import dedup_fuzzy_document
 
 from experiments.pretraining_datasets.simple import downloads
@@ -12,15 +13,14 @@
 
 
 def build_dedup_step(dataset: InputName, max_parallelism: int) -> ExecutorStep:
-    return ExecutorStep(
+    return StepSpec(
         name=f"dedup_{dataset.name}",
         fn=lambda op: dedup_fuzzy_document(
             input_paths=dataset,
             output_path=op,
             max_parallelism=max_parallelism,
         ),
-        description=f"Run dedupe on {dataset.name}",
-    )
+    ).as_executor_step()
 
 
 STEPS = [