@@ -328,6 +328,7 @@ def build(
         if self._post_generation_processed_resume_result(resume, num_records) is not None:
             return self.artifact_storage.final_dataset_path
 
+        self.artifact_storage.create_dataset_directory(resume=resume)
         self._write_builder_config()
 
         # Set media storage mode based on parameters
 
@@ -11,6 +11,7 @@
 from functools import cached_property
 from pathlib import Path
 from typing import TYPE_CHECKING
+from uuid import uuid4
 
 from pydantic import BaseModel, ConfigDict, PrivateAttr, field_validator, model_validator
 
@@ -165,6 +166,30 @@ def refresh_media_storage_path(self) -> None:
         self._media_storage.base_path = self.base_dataset_path
         self._media_storage.images_dir = self.base_dataset_path / images_subdir
 
+    def create_dataset_directory(self, *, resume: ResumeMode) -> None:
+        """Exclusively claim a fresh dataset directory, or reuse it when resuming."""
+        if resume != ResumeMode.NEVER:
+            self.mkdir_if_needed(self.base_dataset_path)
+            return
+
+        dataset_path = self.base_dataset_path
+        while True:
+            try:
+                dataset_path.mkdir(exist_ok=False)
+            except FileExistsError:
+                dataset_path = self.artifact_path / (
+                    f"{self.dataset_name}_{datetime.now().strftime('%m-%d-%Y_%H%M%S')}_{uuid4().hex[:8]}"
+                )
+            except OSError as exc:
+                raise ArtifactStorageError(f"Cannot create dataset directory {str(dataset_path)!r}: {exc}") from exc
+            else:
+                break
+
+        if dataset_path != self.base_dataset_path:
+            logger.info(f"📂 Dataset directory already claimed. This run will be saved to {str(dataset_path)!r}.")
+        self.__dict__["resolved_dataset_name"] = dataset_path.name
+        self.refresh_media_storage_path()
+
     @staticmethod
     def mkdir_if_needed(path: Path | str) -> Path:
         """Create the directory if it does not exist."""
 
@@ -4,8 +4,12 @@
 from __future__ import annotations
 
 import json
+import multiprocessing as mp
 from datetime import datetime
-from unittest.mock import patch
+from multiprocessing.queues import Queue
+from multiprocessing.synchronize import Barrier
+from pathlib import Path
+from unittest.mock import MagicMock, patch
 
 import pytest
 from pyarrow import ArrowNotImplementedError
@@ -502,3 +506,92 @@ def test_clear_partial_results_is_noop_when_no_partial_folder(tmp_path):
     storage = ArtifactStorage(artifact_path=tmp_path)
     assert not storage.partial_results_path.exists()
     storage.clear_partial_results()  # must not raise
+
+
+@pytest.mark.parametrize("existing", [False, True], ids=["fresh", "occupied"])
+@patch("data_designer.engine.storage.artifact_storage.datetime")
+def test_create_dataset_directory_is_exclusive(mock_datetime: MagicMock, tmp_path: Path, existing: bool) -> None:
+    mock_datetime.now.return_value = datetime(2025, 1, 1, 12, 3, 4)
+    if existing:
+        (tmp_path / "dataset").mkdir()
+        (tmp_path / "dataset" / "sentinel").write_text("original")
+    storages = [ArtifactStorage(artifact_path=tmp_path) for _ in range(3)]
+
+    for storage in storages:
+        storage.create_dataset_directory(resume=ResumeMode.NEVER)
+        assert storage.base_dataset_path.is_dir()
+        assert storage.media_storage.base_path == storage.base_dataset_path
+        assert storage.media_storage.images_dir == storage.base_dataset_path / "images"
+
+    assert len({storage.base_dataset_path for storage in storages}) == 3
+    assert len(list(tmp_path.iterdir())) == 3 + existing
+    if existing:
+        assert (tmp_path / "dataset" / "sentinel").read_text() == "original"
+
+
+def test_create_dataset_directory_does_not_reuse_empty_directory(tmp_path: Path) -> None:
+    storage = ArtifactStorage(artifact_path=tmp_path)
+    (tmp_path / "dataset").mkdir()
+
+    storage.create_dataset_directory(resume=ResumeMode.NEVER)
+
+    assert storage.base_dataset_path != tmp_path / "dataset"
+    assert storage.base_dataset_path.is_dir()
+    assert list((tmp_path / "dataset").iterdir()) == []
+
+
+@pytest.mark.parametrize("resume", [ResumeMode.ALWAYS, ResumeMode.IF_POSSIBLE])
+def test_create_dataset_directory_reuses_resumed_directory(tmp_path: Path, resume: ResumeMode) -> None:
+    dataset_path = tmp_path / "dataset"
+    dataset_path.mkdir()
+    (dataset_path / "sentinel").write_text("original")
+    storage = ArtifactStorage(artifact_path=tmp_path, resume=resume)
+
+    storage.create_dataset_directory(resume=resume)
+
+    assert storage.base_dataset_path == dataset_path
+    assert list(tmp_path.iterdir()) == [dataset_path]
+    assert (dataset_path / "sentinel").read_text() == "original"
+
+
+def test_create_dataset_directory_reports_filesystem_error(tmp_path: Path) -> None:
+    storage = ArtifactStorage(artifact_path=tmp_path)
+    with patch.object(Path, "mkdir", side_effect=PermissionError("permission denied")):
+        with pytest.raises(ArtifactStorageError, match="Cannot create dataset directory"):
+            storage.create_dataset_directory(resume=ResumeMode.NEVER)
+
+
+def _claim_dataset_directory(artifact_path: Path, barrier: Barrier, output: Queue) -> None:
+    with patch("data_designer.engine.storage.artifact_storage.datetime") as clock:
+        clock.now.return_value = datetime(2025, 1, 1, 12, 3, 4)
+        storage = ArtifactStorage(artifact_path=artifact_path)
+        barrier.wait(timeout=30)
+        storage.create_dataset_directory(resume=ResumeMode.NEVER)
+        output.put(str(storage.base_dataset_path))
+
+
+@pytest.mark.parametrize("existing", [False, True], ids=["fresh", "occupied"])
+def test_create_dataset_directory_is_exclusive_across_processes(tmp_path: Path, existing: bool) -> None:
+    if existing:
+        (tmp_path / "dataset").mkdir()
+        (tmp_path / "dataset" / "sentinel").write_text("original")
+    context = mp.get_context("spawn")
+    barrier = context.Barrier(2)
+    output = context.Queue()
+    processes = [context.Process(target=_claim_dataset_directory, args=(tmp_path, barrier, output)) for _ in range(2)]
+    try:
+        for process in processes:
+            process.start()
+        for process in processes:
+            process.join(timeout=45)
+            assert process.exitcode == 0
+        paths = [Path(output.get(timeout=5)) for _ in processes]
+        assert len(set(paths)) == 2
+        assert all(path.is_dir() for path in paths)
+    finally:
+        for process in processes:
+            if process.is_alive():
+                process.terminate()
+            process.join(timeout=5)
+        output.close()
+        output.join_thread()
@@ -235,11 +235,12 @@ def create(
                 configuration (columns, constraints, seed data, etc.).
             num_records: Number of records to generate.
             dataset_name: Name of the dataset. This name will be used as the dataset
-                folder name in the artifact path directory. If a non-empty directory with the
+                folder name in the artifact path directory. If a directory with the
                 same name already exists, dataset will be saved to a new directory with
                 a datetime stamp. For example, if the dataset name is "awesome_dataset" and a directory
                 with the same name already exists, the dataset will be saved to a new directory
-                with the name "awesome_dataset_2025-01-01_12-00-00".
+                with the name "awesome_dataset_01-01-2025_120000". A random suffix may be added
+                to avoid collisions.
             resume: Controls how interrupted runs are handled.
 
                 - ``ResumeMode.NEVER`` (default): always start a fresh generation run.
 
@@ -27,7 +27,7 @@
 from data_designer.config.errors import InvalidConfigError
 from data_designer.config.models import ChatCompletionInferenceParams, ModelConfig, ModelProvider
 from data_designer.config.processors import DropColumnsProcessorConfig
-from data_designer.config.run_config import JinjaRenderingEngine, RequestAdmissionTuningConfig, RunConfig
+from data_designer.config.run_config import JinjaRenderingEngine, RequestAdmissionTuningConfig, ResumeMode, RunConfig
 from data_designer.config.sampler_constraints import ScalarInequalityConstraint
 from data_designer.config.sampler_params import (
     CategorySamplerParams,
@@ -2468,3 +2468,39 @@ def test_create_raises_error_when_all_trace_files_are_skipped(
 
     with pytest.raises(DataDesignerGenerationError, match="did not produce any rows"):
         data_designer.create(builder, num_records=1, dataset_name="invalid-trace-seed")
+
+
+@pytest.mark.parametrize("resume", [ResumeMode.NEVER, ResumeMode.IF_POSSIBLE])
+@patch("data_designer.engine.storage.artifact_storage.datetime")
+def test_create_same_timestamp_keeps_runs_isolated(
+    mock_datetime: MagicMock, tmp_path: Path, stub_model_providers: list[ModelProvider], resume: ResumeMode
+) -> None:
+    mock_datetime.now.return_value = datetime(2025, 1, 1, 12, 3, 4)
+    designer = DataDesigner(artifact_path=tmp_path, model_providers=stub_model_providers, auto_configure_logging=False)
+    designer.set_run_config(RunConfig(buffer_size=10, display_tui=False))
+    results = []
+    for tag, num_records in [("A", 20), ("B", 10), ("C", 10)]:
+        builder = DataDesignerConfigBuilder(model_configs=[])
+        builder.add_column(
+            SamplerColumnConfig(name="tag", sampler_type="category", params=CategorySamplerParams(values=[tag]))
+        )
+        result = designer.create(builder, num_records=num_records, resume=resume)
+        results.append((tag, num_records, result))
+
+    assert len({result.artifact_storage.base_dataset_path for _, _, result in results}) == 3
+    assert len(list(tmp_path.iterdir())) == 3
+    for tag, num_records, result in results:
+        assert result.load_dataset()["tag"].tolist() == [tag] * num_records
+        assert result.artifact_storage.media_storage.base_path == result.artifact_storage.base_dataset_path
+
+
+def test_preview_does_not_claim_dataset_directory(tmp_path: Path, stub_model_providers: list[ModelProvider]) -> None:
+    builder = DataDesignerConfigBuilder(model_configs=[])
+    builder.add_column(
+        SamplerColumnConfig(name="tag", sampler_type="category", params=CategorySamplerParams(values=["A"]))
+    )
+    designer = DataDesigner(artifact_path=tmp_path, model_providers=stub_model_providers, auto_configure_logging=False)
+    designer.set_run_config(RunConfig(display_tui=False))
+
+    assert designer.preview(builder, num_records=1).dataset["tag"].tolist() == ["A"]
+    assert list(tmp_path.iterdir()) == []