diff --git a/docs/examples/ga/nsga2/nsga2_python.ipynb b/docs/examples/ga/nsga2/nsga2_python.ipynb
index 08d5ee2ac..83aeafdfb 100644
--- a/docs/examples/ga/nsga2/nsga2_python.ipynb
+++ b/docs/examples/ga/nsga2/nsga2_python.ipynb
@@ -14,7 +14,6 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "import json\n",
     "import logging\n",
     "import matplotlib.pyplot as plt\n",
     "import os\n",
@@ -28,7 +27,7 @@
     "    SimulatedBinaryCrossover,\n",
     ")\n",
     "from xopt.resources.test_functions.zdt import construct_zdt\n",
-    "from xopt import Xopt, Evaluator, VOCS"
+    "from xopt import Xopt, Evaluator"
    ]
   },
   {
@@ -202,8 +201,8 @@
     "\n",
     "The output files are the following.\n",
     " - `data.csv`: All data evaluated during the optimization\n",
-    " - `vocs.txt`: The VOCS object so that the objectives, constraints, decision variables are retained alongside the data\n",
     " - `populations.csv`: Each population is written here with a column `xopt_generation` to distinguish which generation the row belongs to\n",
+    " - `vocs.txt`: The VOCS object so that the objectives, constraints, and decision variables are retained alongside the data\n",
     " - `checkpoints`: This generator periodically saves its full state to timestamped files in this directory\n",
     " - `log.txt`: Log output from the generator is recorded to this file\n",
     "\n",
@@ -288,20 +287,6 @@
     "df.head()"
    ]
   },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "# Read the VOCS object back in. This can be used for data analysis / restarting optimizations\n",
-    "with open(os.path.join(my_xopt.generator.output_dir, \"vocs.txt\")) as f:\n",
-    "    vocs_from_file = VOCS(**json.load(f))\n",
-    "\n",
-    "# Show the objectives\n",
-    "vocs_from_file.objectives"
-   ]
-  },
   {
    "cell_type": "code",
    "execution_count": null,
@@ -357,7 +342,7 @@
  ],
  "metadata": {
   "kernelspec": {
-   "display_name": "Python 3 (ipykernel)",
+   "display_name": "xopt-dev",
    "language": "python",
    "name": "python3"
   },
@@ -371,7 +356,7 @@
    "name": "python",
    "nbconvert_exporter": "python",
    "pygments_lexer": "ipython3",
-   "version": "3.12.10"
+   "version": "3.12.0"
   }
  },
  "nbformat": 4,
diff --git a/xopt/generators/checkpoints.py b/xopt/generators/checkpoints.py
new file mode 100644
index 000000000..c189e8366
--- /dev/null
+++ b/xopt/generators/checkpoints.py
@@ -0,0 +1,123 @@
+from datetime import datetime
+from pydantic import BaseModel, Field, model_validator
+import json
+import os
+
+from ..vocs import VOCS
+
+
+class CheckpointMixin(BaseModel):
+    """
+    Mix-in class adding checkpoint saving and loading to a generator.
+
+    Checkpoints are written to a caller-supplied directory. The VOCS object is
+    serialized into the checkpoint itself. Legacy checkpoints which predate this
+    instead carry the VOCS object in a "vocs.txt" file one level above the
+    directory holding the checkpoints and are still supported when loading.
+
+    The host class must provide a ``vocs`` attribute and a ``to_json`` method.
+    Writing of the checkpoint file is the responsibility of the concrete class
+    as it will be implementation dependent.
+
+    Parameters
+    ----------
+    checkpoint_file : str, optional
+        Path to checkpoint file to load from. If provided, the generator will be
+        initialized from the checkpoint state. User-specified parameters will
+        override checkpoint values.
+    """
+
+    checkpoint_file: str | None = Field(
+        None, description="Path to checkpoint file to load from", exclude=True
+    )
+
+    @staticmethod
+    def _load_checkpoint_data(fname: str) -> dict:
+        """
+        Internal function to load generator data from checkpoint file as well as VOCS object.
+
+        Parameters
+        ----------
+        fname : str
+            Path to the checkpoint file
+
+        Returns
+        -------
+        dict
+            Dictionary containing VOCS and checkpoint data
+        """
+        # Load the checkpoint
+        with open(fname) as f:
+            checkpoint_data = json.load(f)
+
+        if "vocs" in checkpoint_data:
+            return checkpoint_data
+
+        # Legacy checkpoints w/o VOCS
+        vocs_fname = os.path.join(os.path.dirname(fname), "../vocs.txt")
+        if not os.path.exists(vocs_fname):
+            raise ValueError(
+                f'Checkpoint "{fname}" does not contain a VOCS object and no '
+                f'VOCS file was found at "{vocs_fname}".'
+            )
+
+        with open(vocs_fname) as f:
+            vocs = VOCS(**json.load(f))
+
+        return {"vocs": vocs, **checkpoint_data}
+
+    @model_validator(mode="before")
+    @classmethod
+    def load_from_checkpoint(cls, values):
+        """
+        Load from checkpoint file if checkpoint_file is provided.
+        """
+        # Case when a checkpoint file has been supplied
+        if isinstance(values, dict) and "checkpoint_file" in values:
+            checkpoint_file = values.pop("checkpoint_file")
+            if checkpoint_file is not None:
+                # Load checkpoint data
+                checkpoint_data = cls._load_checkpoint_data(checkpoint_file)
+
+                # Merge with user data precedence
+                merged_data = {**checkpoint_data, **values}
+                return merged_data
+
+        # No checkpoint
+        return values
+
+    def _save_checkpoint(self, path: str | os.PathLike) -> str:
+        """
+        Write a checkpoint of the generator state to disk.
+
+        Parameters
+        ----------
+        path : str or os.PathLike
+            Directory into which the checkpoint file is written. Created if it
+            does not already exist.
+
+        Returns
+        -------
+        str
+            Path to the checkpoint file which was written.
+        """
+        # Set up the output directory
+        os.makedirs(path, exist_ok=True)
+
+        # Create a base filename
+        base_checkpoint_filename = datetime.now().strftime("%Y%m%d_%H%M%S")
+        checkpoint_path = os.path.join(path, f"{base_checkpoint_filename}_1.txt")
+
+        # Check if file exists and increment counter until we find a free filename
+        counter = 2
+        while os.path.exists(checkpoint_path):
+            checkpoint_path = os.path.join(
+                path, f"{base_checkpoint_filename}_{counter}.txt"
+            )
+            counter += 1
+
+        # Now we have a unique filename
+        with open(checkpoint_path, "w") as f:
+            f.write(self.to_json())
+
+        return checkpoint_path
diff --git a/xopt/generators/ga/base.py b/xopt/generators/ga/base.py
new file mode 100644
index 000000000..8c96b2950
--- /dev/null
+++ b/xopt/generators/ga/base.py
@@ -0,0 +1,186 @@
+from pydantic import Field, field_validator
+import logging
+import os
+import pandas as pd
+import time
+
+from ..checkpoints import CheckpointMixin
+from ..deduplicated import DeduplicatedGeneratorBase
+
+POPULATION_METADATA_COLUMNS = [
+    "xopt_generation",
+    "xopt_candidate_idx",
+    "xopt_runtime",
+    "xopt_error",
+]
+
+
+class GAGeneratorBase(CheckpointMixin, DeduplicatedGeneratorBase):
+    """
+    Base class for genetic algorithm generators which write output and checkpoints.
+
+    Handles the output directory, log file, and periodic checkpointing on behalf of
+    subclasses. Subclasses call `end_generation` once each time a generation is
+    completed and everything else is taken care of.
+
+    Nothing is written to disk until the generator is used, so building or
+    deserializing one never touches the filesystem.
+
+    Parameters
+    ----------
+    output_dir : str or os.PathLike, optional
+        Directory to save algorithm state and population history, or None to write
+        nothing. Stored as a string, unexpanded; environment variables and "~" are
+        expanded when the path is used. If the directory already contains data, a
+        number is appended to avoid overwriting it.
+    checkpoint_freq : int, default=1
+        Frequency (in generations) at which checkpoints are saved. Set to -1 to
+        disable checkpointing.
+    log_level : int
+        Level of log messages written to "log.txt".
+
+    Attributes
+    ----------
+    expanded_output_dir : str or None
+        `output_dir` with environment variables and "~" expanded. All file writes go
+        here.
+    """
+
+    output_dir: str | None = None
+    checkpoint_freq: int = Field(
+        1,
+        description="How often (in generations) to save checkpoints (set to -1 to disable)",
+    )
+    log_level: int = Field(
+        logging.INFO, description="Log message level output to log.txt"
+    )
+    _output_prepared: bool = (
+        False  # Whether the output directory has been resolved and created
+    )
+
+    @field_validator("output_dir", mode="before")
+    @classmethod
+    def validate_output_dir(cls, value):
+        """Accept any os.PathLike, storing it as a string."""
+        if isinstance(value, os.PathLike):
+            return os.fspath(value)
+        return value
+
+    @property
+    def expanded_output_dir(self) -> str | None:
+        """Output directory with environment variables and "~" expanded."""
+        if self.output_dir is None:
+            return None
+        return os.path.expanduser(os.path.expandvars(self.output_dir))
+
+    def model_post_init(self, context):
+        # Get a unique logger per object. Naming it after the concrete class keeps
+        # records propagating through that class's module logger.
+        self._logger = logging.getLogger(
+            f"{type(self).__module__}.{type(self).__name__}.{id(self)}"
+        )
+        self._logger.setLevel(self.log_level)
+
+    def _prepare_output(self) -> None:
+        """
+        Resolve and create the output directory and begin logging to file.
+
+        Repeated calls do nothing. If the requested directory already holds data, a
+        number is appended and `output_dir` is updated to the path actually used.
+        """
+        if (self.output_dir is None) or self._output_prepared:
+            return
+
+        # Check if directory exists and do collision avoidance. Resolve into a local
+        # so the field is only assigned once, since assignment revalidates the model.
+        # Suffixes are applied to the unexpanded path, but tested against the expanded one.
+        requested = self.output_dir
+        counter = 2
+        output_dir = requested
+        expanded = self.expanded_output_dir
+        while os.path.exists(expanded) and os.listdir(expanded):
+            output_dir = f"{requested}_{counter}"
+            expanded = os.path.expanduser(os.path.expandvars(output_dir))
+            counter += 1
+        if output_dir != requested:
+            self._logger.info(
+                f'detected existing output_dir "{requested}" and corrected '
+                f'to "{output_dir}" to avoid overwriting'
+            )
+        self.output_dir = output_dir
+
+        # We are now setup
+        os.makedirs(self.expanded_output_dir, exist_ok=True)
+        self._output_prepared = True
+
+        # Set up file logging
+        log_file_path = os.path.join(self.expanded_output_dir, "log.txt")
+        file_handler = logging.FileHandler(log_file_path, mode="w")
+        file_handler.setLevel(self.log_level)
+        file_handler.setFormatter(
+            logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
+        )
+        self._logger.addHandler(file_handler)
+        self._logger.info(f"routing log output to file: {log_file_path}")
+
+        # Record the problem definition alongside the data
+        # Note: this is necessary to include in output for users running analysis on the results
+        # ie to plot Pareto front, you need to know the names and direction of the objectives
+        with open(os.path.join(self.expanded_output_dir, "vocs.txt"), "w") as f:
+            f.write(self.vocs.model_dump_json())
+
+    def end_generation(self, generation_index: int, population: list[dict]) -> None:
+        """
+        Record a completed generation, writing output and checkpoints as configured.
+
+        Parameters
+        ----------
+        generation_index : int
+            Index of the generation which was just completed.
+        population : list of dict
+            The individuals making up the completed population.
+        """
+        self._prepare_output()
+        if self.output_dir is None:
+            return
+        output_dir = self.expanded_output_dir
+        save_start_t = time.perf_counter()
+
+        # Save all Xopt data
+        self.data.to_csv(os.path.join(output_dir, "data.csv"), index=False)
+
+        # Construct the DataFrame for this population
+        pop_df = pd.DataFrame(population)
+        pop_df["xopt_generation"] = generation_index
+
+        # Normalize the columns in the DataFrame
+        # Avoid schema changing part way through optimization so we can write CSV in append mode
+        pop_df = pop_df.reindex(
+            columns=self.vocs.all_names + POPULATION_METADATA_COLUMNS
+        )
+
+        # Write population DataFrame to file
+        csv_path = os.path.join(output_dir, "populations.csv")
+        pop_df.to_csv(
+            csv_path, index=False, mode="a", header=not os.path.isfile(csv_path)
+        )
+        self._logger.debug(
+            f'saved optimization data to "{output_dir}" '
+            f"in {1000 * (time.perf_counter() - save_start_t):.2f}ms"
+        )
+
+        # Save a checkpoint if one is due
+        if self.checkpoint_freq > 0 and (generation_index % self.checkpoint_freq == 0):
+            checkpoint_path = self._save_checkpoint(
+                os.path.join(output_dir, "checkpoints")
+            )
+            self._logger.debug(f'saved checkpoint file "{checkpoint_path}"')
+
+    def close_log_file(self):
+        """
+        Closes out the log file (if used)
+        """
+        for handler in list(self._logger.handlers):
+            if isinstance(handler, logging.FileHandler):
+                handler.close()
+            self._logger.removeHandler(handler)
diff --git a/xopt/generators/ga/nsga2.py b/xopt/generators/ga/nsga2.py
index ba716cccb..5e35092e5 100644
--- a/xopt/generators/ga/nsga2.py
+++ b/xopt/generators/ga/nsga2.py
@@ -1,11 +1,7 @@
-from datetime import datetime
 from itertools import chain
 from pydantic import Field, Discriminator, model_validator
 from typing import Annotated
-import json
-import logging
 import numpy as np
-import os
 import pandas as pd
 import time
 import warnings
@@ -14,8 +10,8 @@
 from ...errors import DataError
 from ...generator import StateOwner
 from ...vocs import VOCS
-from ..deduplicated import DeduplicatedGeneratorBase
 from ..utils import fast_dominated_argsort
+from .base import GAGeneratorBase
 from .operators import (
     PolynomialMutation,
     DummyMutation,
@@ -312,7 +308,7 @@ def generate_candidates_from_population(
 ########################################################################################################################
 
 
-class NSGA2Generator(DeduplicatedGeneratorBase, StateOwner):
+class NSGA2Generator(GAGeneratorBase, StateOwner):
     """
     Non-dominated Sorting Genetic Algorithm II (NSGA-II) generator.  Implements the NSGA-II algorithm
     for multi-objective optimization as described in [1]. This generator accomdates user selected mutation
@@ -331,7 +327,7 @@ class NSGA2Generator(DeduplicatedGeneratorBase, StateOwner):
         Operator used to perform crossover between parent solutions.
     mutation_operator : PolynomialMutation or DummyMutation, default=PolynomialMutation()
         Operator used to perform mutation on offspring solutions.
-    output_dir : str, optional
+    output_dir : str or os.PathLike, optional
         Directory to save algorithm state and population history.
     checkpoint_freq : int, default=1
         Frequency (in generations) at which to save checkpoints.
@@ -373,11 +369,6 @@ class NSGA2Generator(DeduplicatedGeneratorBase, StateOwner):
     supports_constraints: bool = True
     supports_single_objective: bool = True
 
-    # Checkpoint loading
-    checkpoint_file: str | None = Field(
-        None, description="Path to checkpoint file to load from", exclude=True
-    )
-
     population_size: int = Field(50, description="Population size")
     crossover_operator: Annotated[
         (
@@ -392,20 +383,6 @@ class NSGA2Generator(DeduplicatedGeneratorBase, StateOwner):
         Discriminator("name"),
     ] = PolynomialMutation()
 
-    # Output options
-    output_dir: str | None = None
-    checkpoint_freq: int = Field(
-        1,
-        description="How often (in generations) to save checkpoints (set to -1 to disable)",
-    )
-    log_level: int = Field(
-        logging.INFO, description="Log message level output to log.txt"
-    )
-    _output_dir_setup: bool = (
-        False  # Used in initializing the directory. PLEASE DO NOT CHANGE
-    )
-    _logger: logging.Logger | None = None
-
     # Metadata
     fevals: int = Field(
         0,
@@ -431,63 +408,6 @@ class NSGA2Generator(DeduplicatedGeneratorBase, StateOwner):
     pop: list[dict] = Field(default=[])
     child: list[dict] = Field(default=[])
 
-    def model_post_init(self, context):
-        # Get a unique logger per object
-        self._logger = logging.getLogger(f"{__name__}.NSGA2Generator.{id(self)}")
-        self._logger.setLevel(self.log_level)
-
-    @staticmethod
-    def _load_checkpoint_data(fname: str) -> dict:
-        """
-        Internal function to load generator data from checkpoint file as well as VOCS object.
-
-        Parameters
-        ----------
-        fname : str
-            Path to the checkpoint file
-
-        Returns
-        -------
-        dict
-            Dictionary containing VOCS and checkpoint data
-        """
-        # Load the VOCS object
-        vocs_fname = os.path.join(os.path.dirname(fname), "../vocs.txt")
-        if not os.path.exists(vocs_fname):
-            raise ValueError(
-                f'Could not load VOCS file at "{vocs_fname}". Complete NSGA2Generator '
-                "output directory is required for loading from checkpoint."
-            )
-
-        with open(vocs_fname) as f:
-            vocs = VOCS(**json.load(f))
-
-        # Load the checkpoint
-        with open(fname) as f:
-            checkpoint_data = json.load(f)
-
-        return {"vocs": vocs, **checkpoint_data}
-
-    @model_validator(mode="before")
-    @classmethod
-    def load_from_checkpoint(cls, values):
-        """
-        Load from checkpoint file if checkpoint_file is provided.
-        """
-        # Case when a checkpoint file has been supplied
-        if isinstance(values, dict) and "checkpoint_file" in values:
-            checkpoint_file = values.pop("checkpoint_file")
-            if checkpoint_file is not None:
-                # Load checkpoint data
-                checkpoint_data = cls._load_checkpoint_data(checkpoint_file)
-
-                # Merge with user data precedence
-                merged_data = {**checkpoint_data, **values}
-                return merged_data
-
-        # No checkpoint
-        return values
-
     @model_validator(mode="after")
     def vocs_compatible(self):
         """
@@ -542,7 +462,7 @@ def data_in_bounds(self, data: dict) -> bool:
         )
 
     def _generate(self, n_candidates: int) -> list[dict]:
-        self.ensure_output_dir_setup()
+        self._prepare_output()
         start_t = time.perf_counter()
 
         # If we have a population create children, otherwise generate randomly sampled points
@@ -583,7 +503,7 @@ def _generate(self, n_candidates: int) -> list[dict]:
         return candidates
 
     def add_data(self, new_data: pd.DataFrame):
-        self.ensure_output_dir_setup()
+        self._prepare_output()
 
         # Validate data is at least compatible with selection / genetic operators
         vocs_names = (
@@ -644,75 +564,8 @@ def add_data(self, new_data: pd.DataFrame):
             self.child = self.child[self.population_size :]
             self.n_generations += 1
 
-            # Save the history file
-            if self.output_dir is not None:
-                save_start_t = time.perf_counter()
-
-                # Save all Xopt data
-                self.data.to_csv(os.path.join(self.output_dir, "data.csv"), index=False)
-                with open(os.path.join(self.output_dir, "vocs.txt"), "w") as f:
-                    json.dump(self.vocs.dict(), f)
-
-                # Construct the DataFrame for this population
-                pop_df = pd.DataFrame(self.pop)
-                pop_df["xopt_generation"] = self.n_generations
-
-                # Normalize the columns in the DataFrame
-                # Avoid schema changing part way through optimization so we can write CSV in append mode
-                columns = self.vocs.all_names + [
-                    "xopt_generation",
-                    "xopt_candidate_idx",
-                    "xopt_runtime",
-                    "xopt_error",
-                ]
-                pop_df = pop_df.reindex(columns=columns)
-
-                # Write population DataFrame to file
-                csv_path = os.path.join(self.output_dir, "populations.csv")
-                pop_df.to_csv(
-                    csv_path, index=False, mode="a", header=not os.path.isfile(csv_path)
-                )
-
-                # Log some things
-                self._logger.debug(
-                    f'saved optimization data to "{self.output_dir}" '
-                    f"in {1000 * (time.perf_counter() - save_start_t):.2f}ms"
-                )
-
-                if self.checkpoint_freq > 0 and (
-                    self.n_generations % self.checkpoint_freq == 0
-                ):
-                    self._save_checkpoint()
-
-    def _save_checkpoint(self):
-        # Confirm we are ready to save checkpoint
-        if self.output_dir is None:
-            raise ValueError("Cannot save checkpoint without an output directory")
-        self.ensure_output_dir_setup()
-
-        # Create a base filename
-        os.makedirs(os.path.join(self.output_dir, "checkpoints"), exist_ok=True)
-        base_checkpoint_filename = datetime.now().strftime("%Y%m%d_%H%M%S")
-        checkpoint_path = os.path.join(
-            self.output_dir,
-            "checkpoints",
-            f"{base_checkpoint_filename}_1.txt",
-        )
-
-        # Check if file exists and increment counter until we find a free filename
-        counter = 2
-        while os.path.exists(checkpoint_path):
-            checkpoint_path = os.path.join(
-                self.output_dir,
-                "checkpoints",
-                f"{base_checkpoint_filename}_{counter}.txt",
-            )
-            counter += 1
-
-        # Now we have a unique filename
-        with open(checkpoint_path, "w") as f:
-            f.write(self.to_json())
-        self._logger.debug(f'saved checkpoint file "{checkpoint_path}"')
+            # Write output files and save a checkpoint if one is due
+            self.end_generation(self.n_generations, self.pop)
 
     def set_data(self, data):
         self.data = data
@@ -728,51 +581,3 @@ def __repr__(self) -> str:
 
     def __str__(self) -> str:
         return self.__repr__()
-
-    def ensure_output_dir_setup(self):
-        if (self.output_dir is None) or self._output_dir_setup:
-            return
-
-        # Check if directory exists and do collision avoidance
-        counter = 2
-        output_dir_dedup = self.output_dir
-        while os.path.exists(output_dir_dedup) and os.listdir(output_dir_dedup):
-            output_dir_dedup = f"{self.output_dir}_{counter}"
-            counter += 1
-        self._logger.info(
-            f'detected existing output_dir "{self.output_dir}" and corrected '
-            f'to "{output_dir_dedup}" to avoid overwriting'
-        )
-        self.output_dir = output_dir_dedup
-
-        # We are now setup
-        self._output_dir_setup = True
-
-        # Setup the directory
-        os.makedirs(self.output_dir, exist_ok=True)
-
-        # Set up file logging
-        log_file_path = os.path.join(self.output_dir, "log.txt")
-        file_handler = logging.FileHandler(log_file_path, mode="w")
-        file_handler.setLevel(self.log_level)
-
-        # Use the same format as the default logger
-        formatter = logging.Formatter(
-            "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
-        )
-        file_handler.setFormatter(formatter)
-
-        # Add the file handler to the logger
-        self._logger.addHandler(file_handler)
-        self._logger.info(f"routing log output to file: {log_file_path}")
-
-    def close_log_file(self):
-        """
-        Closes out the log file (if used)
-        """
-        if self.output_dir is not None and self._output_dir_setup:
-            # Remove all handlers from the logger
-            for handler in list(self._logger.handlers):
-                if isinstance(handler, logging.FileHandler):
-                    handler.close()
-                self._logger.removeHandler(handler)
diff --git a/xopt/tests/generators/ga/test_base.py b/xopt/tests/generators/ga/test_base.py
new file mode 100644
index 000000000..557c49de6
--- /dev/null
+++ b/xopt/tests/generators/ga/test_base.py
@@ -0,0 +1,245 @@
+import json
+import logging
+import os
+
+import pandas as pd
+import pytest
+
+from xopt.generators.ga.base import GAGeneratorBase
+from xopt.resources.test_functions.tnk import tnk_vocs
+
+
+class OutputTestGenerator(GAGeneratorBase):
+    """Minimal concrete generator for exercising the base class output behavior."""
+
+    name = "ga_base_test"
+    supports_single_objective: bool = True
+    supports_multi_objective: bool = True
+    supports_constraints: bool = True
+
+    def _generate(self, n_candidates: int) -> list[dict]:
+        return []
+
+
+class RecordingHandler(logging.Handler):
+    """Captures messages so propagation to the module logger can be checked."""
+
+    def __init__(self):
+        super().__init__()
+        self.messages = []
+
+    def emit(self, record):
+        self.messages.append(record.getMessage())
+
+
+@pytest.fixture
+def module_logger():
+    """Handler on the logger records propagate to, named after the concrete class."""
+    logger = logging.getLogger(OutputTestGenerator.__module__)
+    handler = RecordingHandler()
+    logger.addHandler(handler)
+    logger.setLevel(logging.DEBUG)
+    yield handler
+    logger.removeHandler(handler)
+
+
+def make_generator(output_dir, **kwargs):
+    return OutputTestGenerator(
+        vocs=tnk_vocs,
+        output_dir=None if output_dir is None else str(output_dir),
+        log_level=logging.DEBUG,
+        **kwargs,
+    )
+
+
+def make_population(
+    size: int, generation: int, extra: dict | None = None
+) -> list[dict]:
+    """Build a population of individuals carrying all VOCS and metadata columns."""
+    population = []
+    for idx in range(size):
+        individual = {
+            "x1": 0.1 * idx,
+            "x2": 0.2 * idx,
+            "y1": 1.0 * idx,
+            "y2": 2.0 * idx,
+            "c1": -1.0,
+            "c2": -1.0,
+            "xopt_candidate_idx": generation * size + idx,
+            "xopt_runtime": 0.1,
+            "xopt_error": False,
+        }
+        if extra is not None:
+            individual.update(extra)
+        population.append(individual)
+    return population
+
+
+def make_data(n_rows: int) -> pd.DataFrame:
+    return pd.DataFrame(
+        {
+            "x1": [0.1 * i for i in range(n_rows)],
+            "x2": [0.2 * i for i in range(n_rows)],
+            "xopt_candidate_idx": list(range(n_rows)),
+            "xopt_parent_generation": [0] * n_rows,
+        }
+    )
+
+
+def run_generation(generator, index, size=4, extra=None, n_data=None):
+    """Feed the generator a completed generation."""
+    generator.data = make_data(n_data if n_data is not None else index * size)
+    generator.end_generation(index, make_population(size, index - 1, extra))
+
+
+def test_construction_touches_nothing(tmp_path):
+    # Generators are built and deserialized freely, so neither may create anything
+    requested = tmp_path / "run"
+    generator = make_generator(requested)
+    OutputTestGenerator.model_validate(json.loads(generator.to_json()))
+
+    assert generator.output_dir == str(requested)
+    assert os.listdir(tmp_path) == []
+
+
+def test_prepare_output_creates_directory_and_is_idempotent(tmp_path):
+    requested = tmp_path / "run"
+    generator = make_generator(requested)
+
+    generator._prepare_output()
+    assert generator.output_dir == str(requested)
+    assert os.path.isdir(requested)
+
+    generator._prepare_output()
+    assert generator.output_dir == str(requested)
+    generator.close_log_file()
+
+
+def test_existing_empty_directory_is_not_renamed(tmp_path, module_logger):
+    # A directory which exists but holds nothing is reused as-is. Tests which hand
+    # the generator a TemporaryDirectory depend on this.
+    requested = tmp_path / "run"
+    os.makedirs(requested)
+
+    generator = make_generator(requested)
+    generator._prepare_output()
+
+    assert generator.output_dir == str(requested)
+    assert not any("corrected" in m for m in module_logger.messages)
+    generator.close_log_file()
+
+
+def test_non_empty_directory_is_renamed(tmp_path, module_logger):
+    requested = tmp_path / "run"
+    os.makedirs(requested)
+    (requested / "data.csv").write_text("existing\n")
+
+    first = make_generator(requested)
+    first._prepare_output()
+    assert first.output_dir == f"{requested}_2"
+    assert os.path.isdir(f"{requested}_2")
+    assert any("corrected" in m for m in module_logger.messages)
+
+    # The original directory is left untouched
+    assert (requested / "data.csv").read_text() == "existing\n"
+    first.close_log_file()
+
+    # A second collision steps to the next suffix
+    (tmp_path / "run_2" / "data.csv").write_text("existing\n")
+    second = make_generator(requested)
+    second._prepare_output()
+    assert second.output_dir == f"{requested}_3"
+    second.close_log_file()
+
+
+def test_end_generation_writes_both_files(tmp_path):
+    generator = make_generator(tmp_path / "run")
+    run_generation(generator, 1, n_data=8)
+
+    assert len(pd.read_csv(os.path.join(generator.output_dir, "data.csv"))) == 8
+
+    pop_df = pd.read_csv(os.path.join(generator.output_dir, "populations.csv"))
+    assert len(pop_df) == 4
+    assert (pop_df["xopt_generation"] == 1).all()
+    assert list(pop_df.columns) == tnk_vocs.all_names + [
+        "xopt_generation",
+        "xopt_candidate_idx",
+        "xopt_runtime",
+        "xopt_error",
+    ]
+    generator.close_log_file()
+
+
+def test_data_overwritten_while_populations_accumulate(tmp_path):
+    generator = make_generator(tmp_path / "run")
+    run_generation(generator, 1, n_data=4)
+    run_generation(generator, 2, n_data=8)
+
+    # data.csv is a full overwrite, so it reflects only the latest generation
+    assert len(pd.read_csv(os.path.join(generator.output_dir, "data.csv"))) == 8
+
+    # populations.csv is appended and carries exactly one header line
+    population_path = os.path.join(generator.output_dir, "populations.csv")
+    pop_df = pd.read_csv(population_path)
+    assert len(pop_df) == 8
+    assert sorted(pop_df["xopt_generation"].unique()) == [1, 2]
+    with open(population_path) as f:
+        assert sum(1 for line in f if line.startswith("x1,")) == 1
+    generator.close_log_file()
+
+
+def test_end_generation_normalizes_changing_schema(tmp_path):
+    generator = make_generator(tmp_path / "run")
+    run_generation(generator, 1)
+
+    # A later generation gaining an extra key must not shift the appended columns
+    run_generation(generator, 2, extra={"obs1": 3.0})
+
+    # ... nor may one missing a metadata key
+    generator.data = make_data(12)
+    sparse = make_population(4, 2)
+    for individual in sparse:
+        del individual["xopt_runtime"]
+    generator.end_generation(3, sparse)
+
+    pop_df = pd.read_csv(os.path.join(generator.output_dir, "populations.csv"))
+    assert len(pop_df) == 12
+    assert "obs1" not in pop_df.columns
+    assert pop_df[pop_df["xopt_generation"] == 3]["xopt_runtime"].isna().all()
+    assert pop_df[pop_df["xopt_generation"] == 2]["xopt_runtime"].notna().all()
+    generator.close_log_file()
+
+
+@pytest.mark.parametrize("checkpoint_freq, expected", [(1, 4), (2, 2), (-1, 0)])
+def test_checkpoint_frequency(tmp_path, checkpoint_freq, expected):
+    generator = make_generator(tmp_path / "run", checkpoint_freq=checkpoint_freq)
+    for index in range(1, 5):
+        run_generation(generator, index)
+
+    checkpoint_dir = os.path.join(generator.output_dir, "checkpoints")
+    written = len(os.listdir(checkpoint_dir)) if os.path.isdir(checkpoint_dir) else 0
+    assert written == expected
+    generator.close_log_file()
+
+
+def test_no_output_dir(tmp_path, module_logger):
+    generator = make_generator(None)
+    run_generation(generator, 1)
+
+    # Nothing written, but the generator still logs to the module logger
+    assert os.listdir(tmp_path) == []
+    generator._logger.info("still logging")
+    assert "still logging" in module_logger.messages
+
+
+def test_log_file_receives_records_and_closes(tmp_path, module_logger):
+    generator = make_generator(tmp_path / "run")
+    generator._prepare_output()
+    generator._logger.info("after prepare")
+
+    assert "after prepare" in module_logger.messages
+    generator.close_log_file()
+
+    with open(os.path.join(generator.output_dir, "log.txt")) as f:
+        assert "after prepare" in f.read()
+    assert not generator._logger.handlers
diff --git a/xopt/tests/generators/ga/test_nsga2.py b/xopt/tests/generators/ga/test_nsga2.py
index 635d93dc2..8cd753f4f 100644
--- a/xopt/tests/generators/ga/test_nsga2.py
+++ b/xopt/tests/generators/ga/test_nsga2.py
@@ -76,7 +76,6 @@ def test_nsga2_output_data():
         # Verify that the data files are created
         assert os.path.exists(os.path.join(output_dir, "data.csv"))
         assert os.path.exists(os.path.join(output_dir, "populations.csv"))
-        assert os.path.exists(os.path.join(output_dir, "vocs.txt"))
         assert os.path.exists(os.path.join(output_dir, "log.txt"))
 
         # Read the data file and check its contents
@@ -110,11 +109,6 @@ def test_nsga2_output_data():
         # Check that the populations file contains the expected columns
         assert "xopt_generation" in pop_df.columns
 
-        # Check that the VOCS file contains valid JSON
-        with open(os.path.join(output_dir, "vocs.txt"), "r") as f:
-            vocs_dict = json.load(f)
-            VOCS(**vocs_dict)
-
         # Verify that the log file exists and has content
         with open(os.path.join(output_dir, "log.txt"), "r") as f:
             log_content = f.read()
@@ -224,7 +218,7 @@ def nsga2_optimization_with_checkpoint():
         )
 
         # Hack to avoid log error on windows: "The process cannot access the file because it is being used by another process"
-        generator.ensure_output_dir_setup()
+        generator._prepare_output()
         generator.close_log_file()
 
         # Run a few optimization steps
@@ -894,7 +888,7 @@ def test_nsga2_output_inhomogenous_data():
         )
 
         # Hack to avoid log error on windows: "The process cannot access the file because it is being used by another process"
-        generator.ensure_output_dir_setup()
+        generator._prepare_output()
         generator.close_log_file()
 
         # Run a few optimization steps
diff --git a/xopt/tests/generators/test_checkpoints.py b/xopt/tests/generators/test_checkpoints.py
new file mode 100644
index 000000000..380343207
--- /dev/null
+++ b/xopt/tests/generators/test_checkpoints.py
@@ -0,0 +1,122 @@
+from copy import deepcopy
+from datetime import datetime
+import json
+import os
+
+import pytest
+
+from xopt.generators.checkpoints import CheckpointMixin
+from xopt.generators.random import RandomGenerator
+from xopt.resources.testing import TEST_VOCS_BASE
+from xopt.vocs import VOCS
+
+
+class CheckpointingRandomGenerator(CheckpointMixin, RandomGenerator):
+    """Minimal host for the checkpoint mixin with a field to round trip."""
+
+    counter: int = 0
+
+
+def parse_checkpoint_filename(filename: str) -> tuple[datetime, int]:
+    """Split a checkpoint filename into its timestamp and deduplication index."""
+    base, index = filename.rsplit("_", 1)
+    return datetime.strptime(base, "%Y%m%d_%H%M%S"), int(index.split(".")[0])
+
+
+def make_legacy_checkpoint(checkpoint_path: str) -> None:
+    """Rewrite a checkpoint into the legacy layout with VOCS held in "vocs.txt"."""
+    with open(checkpoint_path) as f:
+        checkpoint_data = json.load(f)
+    vocs = checkpoint_data.pop("vocs")
+
+    with open(checkpoint_path, "w") as f:
+        json.dump(checkpoint_data, f)
+    legacy_dir = os.path.dirname(os.path.dirname(checkpoint_path))
+    with open(os.path.join(legacy_dir, "vocs.txt"), "w") as f:
+        json.dump(vocs, f)
+
+
+def test_save_checkpoint_layout(tmp_path):
+    generator = CheckpointingRandomGenerator(vocs=deepcopy(TEST_VOCS_BASE))
+    checkpoint_path = generator._save_checkpoint(tmp_path / "checkpoints")
+
+    # The checkpoint is written directly into the supplied directory
+    assert os.path.dirname(checkpoint_path) == str(tmp_path / "checkpoints")
+    assert os.listdir(tmp_path / "checkpoints") == [os.path.basename(checkpoint_path)]
+    assert os.listdir(tmp_path) == ["checkpoints"]
+
+    # Filename follows the timestamp plus deduplication index scheme
+    _, index = parse_checkpoint_filename(os.path.basename(checkpoint_path))
+    assert index == 1
+
+    # The checkpoint holds valid JSON and carries the VOCS object itself
+    with open(checkpoint_path) as f:
+        checkpoint_data = json.load(f)
+    assert "counter" in checkpoint_data
+    assert VOCS(**checkpoint_data["vocs"]) == generator.vocs
+
+
+def test_checkpoint_round_trip(tmp_path):
+    generator = CheckpointingRandomGenerator(vocs=deepcopy(TEST_VOCS_BASE), counter=17)
+    checkpoint_path = generator._save_checkpoint(tmp_path / "checkpoints")
+
+    # VOCS comes from the checkpoint itself, not from the user
+    reloaded = CheckpointingRandomGenerator(checkpoint_file=checkpoint_path)
+    assert reloaded.counter == 17
+    assert reloaded.vocs == generator.vocs
+
+    # The path used to load is not carried into the reloaded generator's state
+    assert reloaded.checkpoint_file is None
+    assert "checkpoint_file" not in json.loads(reloaded.to_json())
+
+
+def test_checkpoint_user_values_take_precedence(tmp_path):
+    generator = CheckpointingRandomGenerator(vocs=deepcopy(TEST_VOCS_BASE), counter=17)
+    checkpoint_path = generator._save_checkpoint(tmp_path / "checkpoints")
+
+    reloaded = CheckpointingRandomGenerator(checkpoint_file=checkpoint_path, counter=99)
+    assert reloaded.counter == 99
+
+
+def test_save_checkpoint_avoids_overwriting(tmp_path):
+    generator = CheckpointingRandomGenerator(vocs=deepcopy(TEST_VOCS_BASE))
+    first = generator._save_checkpoint(tmp_path / "checkpoints")
+    second = generator._save_checkpoint(tmp_path / "checkpoints")
+
+    assert first != second
+    assert len(os.listdir(tmp_path / "checkpoints")) == 2
+
+
+def test_load_legacy_checkpoint(tmp_path):
+    generator = CheckpointingRandomGenerator(vocs=deepcopy(TEST_VOCS_BASE), counter=17)
+    checkpoint_path = generator._save_checkpoint(tmp_path / "checkpoints")
+    make_legacy_checkpoint(checkpoint_path)
+
+    # VOCS is recovered from "vocs.txt" since the checkpoint does not carry it
+    reloaded = CheckpointingRandomGenerator(checkpoint_file=checkpoint_path)
+    assert reloaded.counter == 17
+    assert reloaded.vocs == generator.vocs
+
+
+def test_load_legacy_checkpoint_missing_vocs_file(tmp_path):
+    generator = CheckpointingRandomGenerator(vocs=deepcopy(TEST_VOCS_BASE))
+    checkpoint_path = generator._save_checkpoint(tmp_path / "checkpoints")
+    make_legacy_checkpoint(checkpoint_path)
+    os.remove(tmp_path / "vocs.txt")
+
+    with pytest.raises(ValueError, match="does not contain a VOCS object"):
+        CheckpointingRandomGenerator(checkpoint_file=checkpoint_path)
+
+
+def test_embedded_vocs_preferred_over_legacy_file(tmp_path):
+    generator = CheckpointingRandomGenerator(vocs=deepcopy(TEST_VOCS_BASE))
+    checkpoint_path = generator._save_checkpoint(tmp_path / "checkpoints")
+
+    # A stale legacy file beside a modern checkpoint must be ignored
+    stale_vocs = deepcopy(TEST_VOCS_BASE)
+    stale_vocs.variables.pop(next(iter(stale_vocs.variables)))
+    with open(tmp_path / "vocs.txt", "w") as f:
+        json.dump(stale_vocs.model_dump(), f)
+
+    reloaded = CheckpointingRandomGenerator(checkpoint_file=checkpoint_path)
+    assert reloaded.vocs == generator.vocs