diff --git a/src/cloudai/configurator/__init__.py b/src/cloudai/configurator/__init__.py
index a88432c41..3965bc93b 100644
--- a/src/cloudai/configurator/__init__.py
+++ b/src/cloudai/configurator/__init__.py
@@ -16,9 +16,10 @@
 
 from .base_agent import BaseAgent
 from .base_gym import BaseGym
-from .cloudai_gym import CloudAIGymEnv, TrajectoryEntry
+from .cloudai_gym import CloudAIGymEnv
 from .grid_search import GridSearchAgent
 from .gymnasium_adapter import GymnasiumAdapter
+from .trajectory import Trajectory
 
 __all__ = [
     "BaseAgent",
@@ -26,5 +27,5 @@
     "CloudAIGymEnv",
     "GridSearchAgent",
     "GymnasiumAdapter",
-    "TrajectoryEntry",
+    "Trajectory",
 ]
diff --git a/src/cloudai/configurator/cloudai_gym.py b/src/cloudai/configurator/cloudai_gym.py
index 21b3f82af..cdafd5ce5 100644
--- a/src/cloudai/configurator/cloudai_gym.py
+++ b/src/cloudai/configurator/cloudai_gym.py
@@ -15,29 +15,20 @@
 # limitations under the License.
 
 import copy
-import csv
-import dataclasses
 import logging
 from pathlib import Path
-from typing import Any, Dict, Optional, Tuple
+from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, cast
 
 from cloudai.core import METRIC_ERROR, BaseRunner, Registry, TestRun
 from cloudai.util.lazy_imports import lazy
 
 from .base_agent import RewardOverrides
 from .base_gym import BaseGym
-from .env_params import EnvParams, ObsLeafDescriptor, write_env_params
+from .env_params import EnvParams, ObsLeafDescriptor
+from .trajectory import Trajectory
 
-
-@dataclasses.dataclass(frozen=True)
-class TrajectoryEntry:
-    """Represents a trajectory entry."""
-
-    step: int
-    action: dict[str, Any]
-    reward: float
-    observation: list
-    env_params: dict[str, Any] = dataclasses.field(default_factory=dict)
+if TYPE_CHECKING:
+    import pandas as pd
 
 
 class CloudAIGymEnv(BaseGym):
@@ -47,7 +38,12 @@ class CloudAIGymEnv(BaseGym):
     Uses the TestRun object and actual runner methods to execute jobs.
     """
 
-    def __init__(self, test_run: TestRun, runner: BaseRunner, rewards: RewardOverrides):
+    def __init__(
+        self,
+        test_run: TestRun,
+        runner: BaseRunner,
+        rewards: RewardOverrides,
+    ):
         """
         Initialize the Gym environment using the TestRun object.
 
@@ -62,15 +58,10 @@ def __init__(self, test_run: TestRun, runner: BaseRunner, rewards: RewardOverrid
         self.rewards = rewards
         self.max_steps = test_run.test.agent_steps
         self.reward_function = Registry().get_reward_function(test_run.test.agent_reward_function)
-        self.trajectory: dict[int, list[TrajectoryEntry]] = {}
         self.params: EnvParams | None = EnvParams.from_test(test_run.test)
+        self.trajectory = Trajectory(iteration_dir=self.iteration_dir)
         super().__init__()
 
-    @property
-    def env_params_record_path(self) -> Path:
-        """``env.csv`` lives alongside ``trajectory.csv`` so a plain ``merge`` joins them."""
-        return self.iteration_dir / "env.csv"
-
     @property
     def upcoming_trial(self) -> int:
         """
@@ -104,7 +95,7 @@ def define_observation_space(self) -> list:
     def reset(
         self,
         seed: Optional[int] = None,
-        options: Optional[dict[str, Any]] = None,  # noqa: Vulture
+        options: Optional[dict[str, Any]] = None,
     ) -> Tuple[list, dict[str, Any]]:
         """
         Reset the environment and reinitialize the TestRun.
@@ -118,6 +109,7 @@ def reset(
                 - observation (list): Initial observation.
                 - info (dict): Additional info for debugging.
         """
+        del options
         if seed is not None:
             lazy.np.random.seed(seed)
         self.test_run.current_iteration = 0
@@ -152,59 +144,50 @@ def step(self, action: Any) -> Tuple[list, float, bool, dict]:
         if cached_result is not None:
             logging.info(
                 "Retrieved cached result from trajectory with reward %s (from step %s). Skipping execution.",
-                cached_result.reward,
-                cached_result.step,
+                cached_result["reward"],
+                cached_result["step"],
             )
-            self.write_trajectory(
-                TrajectoryEntry(
-                    step=self.test_run.step,
-                    action=action,
-                    reward=cached_result.reward,
-                    observation=cached_result.observation,
-                    env_params=sampled_env_params,
-                )
-            )
-
-            return cached_result.observation, cached_result.reward, False, info
-
-        if not self.test_run.test.constraint_check(self.test_run, self.runner.system):
-            logging.info("Constraint check failed. Skipping step.")
-            return [-1.0], self.rewards.constraint_failure, True, info
-
-        new_tr = copy.deepcopy(self.test_run)
-        new_tr.output_path = self.runner.get_job_output_path(new_tr)
-        self.runner.test_scenario.test_runs = [new_tr]
-
-        self.runner.shutting_down = False
-        self.runner.jobs.clear()
-        self.runner.testrun_to_job_map.clear()
-
-        try:
-            self.runner.run()
-        except Exception as e:
-            logging.error(f"Error running step {self.test_run.step}: {e}")
-
-        if self.runner.test_scenario.test_runs and self.runner.test_scenario.test_runs[0].output_path.exists():
-            self.test_run = self.runner.test_scenario.test_runs[0]
+            observation = {
+                metric: cached_result[f"observation.{metric}"] for metric in self.test_run.test.agent_metrics
+            }
+            reward = cast(float, cached_result["reward"])
         else:
-            self.test_run = copy.deepcopy(self.original_test_run)
-            self.test_run.step = new_tr.step
-            self.test_run.output_path = new_tr.output_path
-
-        metrics = self.get_observation(action)
-        reward = self.compute_reward(metrics)
-
-        self.write_trajectory(
-            TrajectoryEntry(
-                step=self.test_run.step,
-                action=action,
-                reward=reward,
-                observation=metrics,
-                env_params=sampled_env_params,
-            )
+            if not self.test_run.test.constraint_check(self.test_run, self.runner.system):
+                logging.info("Constraint check failed. Skipping step.")
+                return [-1.0], self.rewards.constraint_failure, True, info
+
+            new_tr = copy.deepcopy(self.test_run)
+            new_tr.output_path = self.runner.get_job_output_path(new_tr)
+            self.runner.test_scenario.test_runs = [new_tr]
+
+            self.runner.shutting_down = False
+            self.runner.jobs.clear()
+            self.runner.testrun_to_job_map.clear()
+
+            try:
+                self.runner.run()
+            except Exception as e:
+                logging.error(f"Error running step {self.test_run.step}: {e}")
+
+            if self.runner.test_scenario.test_runs and self.runner.test_scenario.test_runs[0].output_path.exists():
+                self.test_run = self.runner.test_scenario.test_runs[0]
+            else:
+                self.test_run = copy.deepcopy(self.original_test_run)
+                self.test_run.step = new_tr.step
+                self.test_run.output_path = new_tr.output_path
+
+            observation = self.get_observation()
+            reward = self.compute_reward(observation)
+
+        self.trajectory.append(
+            step=self.test_run.step,
+            action=action,
+            reward=reward,
+            observation=observation,
+            env_params=sampled_env_params,
         )
 
-        return metrics, reward, False, info
+        return list(observation.values()), reward, False, info
 
     def render(self, mode: str = "human"):
         """
@@ -225,38 +208,35 @@ def seed(self, seed: Optional[int] = None):
         if seed is not None:
             lazy.np.random.seed(seed)
 
-    def compute_reward(self, observation: list) -> float:
+    def compute_reward(self, observation: dict[str, Any]) -> float:
         """
         Compute a reward based on the TestRun result.
 
         Args:
-            observation (list): The observation list containing the average value.
+            observation: Metric values keyed by configured metric name.
 
         Returns:
             float: Reward value.
         """
-        return self.reward_function(observation)
+        return self.reward_function(list(observation.values()))
 
-    def get_observation(self, action: Any) -> list:
+    def get_observation(self) -> dict[str, Any]:
         """
         Get the observation from the TestRun object.
 
-        Args:
-            action (Any): Action taken by the agent.
-
         Returns:
-            list: The observation.
+            Metric values keyed by configured metric name.
         """
         all_metrics = self.test_run.test.agent_metrics
         if not all_metrics:
             raise ValueError("No agent metrics defined for the test run")
 
-        observation = []
+        observation: dict[str, Any] = {}
         for metric in all_metrics:
             v = self.test_run.get_metric_value(self.runner.system, metric)
             if v is METRIC_ERROR:
                 v = self.rewards.metric_failure
-            observation.append(v)
+            observation[metric] = v
         return observation
 
     def structured_observation_descriptors(self) -> Optional[Dict[str, ObsLeafDescriptor]]:
@@ -277,83 +257,15 @@ def encode_env_params(self, env_params: dict[str, Any]) -> Dict[str, Any]:
         """
         return self.params.encode(env_params) if self.params is not None else {}
 
-    def write_trajectory(self, entry: TrajectoryEntry):
-        """
-        Append the entry to the in-memory cache and trajectory.csv (plus env.csv when declared).
-
-        ``trajectory.csv`` and the ``env.csv`` projection are sunk from the same
-        ``TrajectoryEntry`` here, so a trial that never produces an entry (e.g. a
-        constraint failure returns before this call) lands in neither file and the
-        two stay 1:1 step-aligned.
-        """
-        self.current_trajectory.append(entry)
-
-        file_exists = self.trajectory_file_path.exists()
-        logging.debug(f"Writing trajectory into {self.trajectory_file_path} (exists: {file_exists})")
-        self.trajectory_file_path.parent.mkdir(parents=True, exist_ok=True)
-
-        with open(self.trajectory_file_path, mode="a", newline="") as file:
-            writer = csv.writer(file)
-            if not file_exists:
-                writer.writerow(["step", "action", "reward", "observation"])
-            writer.writerow([entry.step, entry.action, entry.reward, entry.observation])
-
-        write_env_params(self.env_params_record_path, entry.step, entry.env_params)
-
     @property
     def iteration_dir(self) -> Path:
-        """Per-iteration output dir; trajectory.csv and env.csv both live here, step-aligned."""
+        """Per-iteration output directory containing the trajectory output."""
         return self.runner.scenario_root / self.test_run.name / f"{self.test_run.current_iteration}"
 
     @property
     def trajectory_file_path(self) -> Path:
-        return self.iteration_dir / "trajectory.csv"
-
-    @property
-    def current_trajectory(self) -> list[TrajectoryEntry]:
-        return self.trajectory.setdefault(self.test_run.current_iteration, [])
-
-    def get_cached_trajectory_result(self, action: Any, env_params: dict[str, Any]) -> TrajectoryEntry | None:
-        """
-        Return a cached entry only when the full trial identity matches.
-
-        Trial identity is ``(action, env_params)``: env-randomized parameters
-        change the workload's behaviour, so a trial repeating the same action
-        under a different ``env_params`` sample must miss and re-run. Empty
-        env_params on both sides is the back-compat path for workloads that
-        do not declare any ``[env_params.*]`` block. The sample is passed in (a
-        per-trial local owned by ``step``), exactly like ``action``.
-        """
-        for entry in self.current_trajectory:
-            action_match = self._values_match_exact(entry.action, action)
-            env_params_match = self._values_match_exact(entry.env_params, env_params)
-            if action_match and env_params_match:
-                return entry
+        return self.trajectory.output_path
 
-        return None
-
-    @classmethod
-    def _values_match_exact(cls, left: Any, right: Any) -> bool:
-        if type(left) is not type(right):
-            return False
-
-        elif isinstance(left, dict):
-            left_keys = set(left.keys())
-            right_keys = set(right.keys())
-            if left_keys != right_keys:
-                return False
-
-            return all(cls._values_match_exact(left[key], right[key]) for key in left_keys)
-
-        elif isinstance(left, (list, tuple)):
-            if len(left) != len(right):
-                return False
-
-            for left_item, right_item in zip(left, right, strict=True):
-                if not cls._values_match_exact(left_item, right_item):
-                    return False
-
-            return True
-
-        else:
-            return left == right
+    def get_cached_trajectory_result(self, action: Any, env_params: dict[str, Any]) -> "pd.Series | None":
+        """Return the first trajectory row matching the action and environment parameters."""
+        return self.trajectory.find(action=action, env_params=env_params)
diff --git a/src/cloudai/configurator/env_params.py b/src/cloudai/configurator/env_params.py
index a824d6205..af53b8a49 100644
--- a/src/cloudai/configurator/env_params.py
+++ b/src/cloudai/configurator/env_params.py
@@ -27,11 +27,9 @@
 
 from __future__ import annotations
 
-import csv
 import dataclasses
 import math
 import random
-from pathlib import Path
 from typing import TYPE_CHECKING, Any, Dict, List, Literal, Optional, Protocol, runtime_checkable
 
 from pydantic import BaseModel, ConfigDict, Field, model_validator
@@ -239,30 +237,6 @@ def observation_descriptors(self) -> Dict[str, ObsLeafDescriptor]:
         return {name: param.observation_descriptor() for name, param in self.params.items()}
 
 
-def write_env_params(path: Path, step: int, sample: Dict[str, Any]) -> None:
-    """
-    Append one trial's env_params sample to a step-aligned CSV.
-
-    The CSV mirrors how ``trajectory.csv`` serialises its ``action`` column
-    (one row per env.step(), sample dict stringified in a single cell) so the
-    two files align 1:1 on ``step`` and a plain ``merge`` joins them.
-
-    Empty samples are skipped, so a run without env_params writes nothing and
-    callers can sink every trial unconditionally.
-    """
-    if step < 1:
-        raise ValueError(f"step must be a positive trial index (cloudai DSE is 1-based); got {step}")
-    if not sample:
-        return
-    new_file = not path.exists()
-    path.parent.mkdir(parents=True, exist_ok=True)
-    with path.open("a", newline="") as f:
-        writer = csv.writer(f)
-        if new_file:
-            writer.writerow(("step", "env"))
-        writer.writerow([step, sample])
-
-
 def validate_domain_randomization_active(test_scenario: "TestScenario") -> None:
     """
     Reject prepped configs that declare domain randomization no agent will run.
diff --git a/src/cloudai/configurator/trajectory.py b/src/cloudai/configurator/trajectory.py
new file mode 100644
index 000000000..85543b5d9
--- /dev/null
+++ b/src/cloudai/configurator/trajectory.py
@@ -0,0 +1,237 @@
+# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES
+# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Pandas-native trajectory storage with flat, namespaced columns."""
+
+from __future__ import annotations
+
+import csv
+import logging
+from collections.abc import Mapping, Sequence
+from copy import deepcopy
+from numbers import Integral
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+from cloudai.util.lazy_imports import lazy
+
+if TYPE_CHECKING:
+    import pandas as pd
+
+
+class Trajectory:
+    """An ordered DataFrame of DSE steps persisted as core and metadata CSVs."""
+
+    file_name = "trajectory.csv"
+    metadata_file_name = "metadata.csv"
+    _core_fields = ("step", "action", "reward", "observation")
+    _core_domains = frozenset(_core_fields)
+
+    def __init__(
+        self,
+        *,
+        iteration_dir: Path,
+        dataframe: pd.DataFrame | None = None,
+    ) -> None:
+        self._iteration_dir = iteration_dir
+        self._dataframe = lazy.pd.DataFrame() if dataframe is None else _copy_dataframe(dataframe)
+        self._validate_dataframe()
+        self._dataframe = self._dataframe.astype(object)
+        logging.debug(
+            "Initializing Trajectory: entries=%s, columns=%s.",
+            len(self),
+            list(self._dataframe.columns),
+        )
+
+    def __len__(self) -> int:
+        """Return the number of trajectory rows."""
+        return len(self._dataframe)
+
+    @property
+    def dataframe(self) -> pd.DataFrame:
+        """Return a copy of the trajectory DataFrame for analysis."""
+        return _copy_dataframe(self._dataframe)
+
+    @property
+    def output_path(self) -> Path:
+        """Return the trajectory CSV path."""
+        return self._iteration_dir / self.file_name
+
+    @property
+    def metadata_path(self) -> Path:
+        """Return the trajectory metadata CSV path."""
+        return self._iteration_dir / self.metadata_file_name
+
+    def append(
+        self,
+        *,
+        step: int,
+        action: object,
+        reward: object,
+        observation: object,
+        **values: object,
+    ) -> pd.Series:
+        """Flatten, persist, and store one trajectory row."""
+        self._validate_step(step)
+        if len(self) and step <= self._dataframe.iloc[-1]["step"]:
+            raise ValueError(
+                f"trajectory steps must increase: last step is {self._dataframe.iloc[-1]['step']}, got {step}"
+            )
+
+        record: dict[str, object] = {"step": step}
+        domains = {"action": action, "reward": reward, "observation": observation, **values}
+        for domain, value in domains.items():
+            _flatten_value(record, domain, value)
+
+        fields = tuple(record)
+        expected_fields = tuple(self._dataframe.columns)
+        if expected_fields and fields != expected_fields:
+            raise ValueError(f"trajectory record fields changed: expected {expected_fields}, got {fields}")
+
+        row = lazy.pd.Series(record, dtype=object)
+        self._persist(row, action=action, reward=reward, observation=observation)
+
+        row_frame = row.to_frame().T.astype(object)
+        self._dataframe = lazy.pd.concat([self._dataframe, row_frame], ignore_index=True).astype(object)
+        logging.debug("Appended trajectory row for step %s (total rows: %s).", step, len(self))
+        return _copy_series(row)
+
+    def find(self, **values: object) -> pd.Series | None:
+        """Return a copy of the first row matching all supplied domain values."""
+        criteria: dict[str, object] = {}
+        for domain, value in values.items():
+            _flatten_value(criteria, domain, value)
+
+        if any(field not in self._dataframe.columns for field in criteria):
+            return None
+
+        for _, row in self._dataframe.iterrows():
+            if all(_values_match_exact(row[field], value) for field, value in criteria.items()):
+                logging.debug("Found matching trajectory row at step %s for %s.", row["step"], values)
+                return _copy_series(row)
+        logging.debug("No matching trajectory row found for %s.", values)
+        return None
+
+    def _validate_dataframe(self) -> None:
+        if not self._dataframe.columns.is_unique:
+            raise ValueError("trajectory dataframe columns must be unique")
+        non_string_columns = [column for column in self._dataframe.columns if not isinstance(column, str)]
+        if non_string_columns:
+            raise TypeError(f"trajectory dataframe columns must be strings: {non_string_columns}")
+        if self._dataframe.empty and len(self._dataframe.columns) == 0:
+            return
+        if "step" not in self._dataframe.columns:
+            raise ValueError("trajectory dataframe must contain a step column")
+
+        previous_step: int | None = None
+        for step in self._dataframe["step"]:
+            self._validate_step(step)
+            if previous_step is not None and step <= previous_step:
+                raise ValueError(f"trajectory steps must increase: last step is {previous_step}, got {step}")
+            previous_step = int(step)
+
+    @staticmethod
+    def _validate_step(step: object) -> None:
+        if isinstance(step, bool) or not isinstance(step, Integral) or step < 1:
+            raise ValueError(f"trajectory step must be a positive integer; got {step}")
+
+    def _persist(self, row: pd.Series, *, action: object, reward: object, observation: object) -> None:
+        core_row = lazy.pd.Series(
+            {
+                "step": row["step"],
+                "action": action,
+                "reward": reward,
+                "observation": list(observation.values()) if isinstance(observation, Mapping) else observation,
+            },
+            dtype=object,
+        )
+        metadata_fields = tuple(
+            field for field in row.index if field == "step" or field.split(".", maxsplit=1)[0] not in self._core_domains
+        )
+        metadata_row = row[list(metadata_fields)] if len(metadata_fields) > 1 else None
+
+        self.output_path.parent.mkdir(parents=True, exist_ok=True)
+        core_header = _validate_csv_header(self.output_path, self._core_fields)
+        metadata_header = (
+            _validate_csv_header(self.metadata_path, metadata_fields) if metadata_row is not None else False
+        )
+
+        core_row.to_frame().T.to_csv(self.output_path, mode="a", header=core_header, index=False)
+        logging.debug("Wrote trajectory row to %s.", self.output_path)
+        if metadata_row is not None:
+            metadata_row.to_frame().T.to_csv(
+                self.metadata_path,
+                mode="a",
+                header=metadata_header,
+                index=False,
+            )
+            logging.debug("Wrote trajectory metadata row to %s.", self.metadata_path)
+
+
+def _validate_csv_header(path: Path, fields: tuple[str, ...]) -> bool:
+    """Validate an existing CSV header and return whether a header must be written."""
+    write_header = not path.exists() or path.stat().st_size == 0
+    if write_header:
+        return True
+    with path.open(newline="") as file:
+        existing_fields = tuple(next(csv.reader(file), ()))
+    if existing_fields != fields:
+        raise ValueError(f"trajectory file fields do not match: expected {fields}, got {existing_fields}")
+    return False
+
+
+def _flatten_value(record: dict[str, object], key: str, value: object) -> None:
+    """Flatten mappings into dot-separated columns while preserving leaf values."""
+    if isinstance(value, Mapping):
+        for child_key, child_value in value.items():
+            if not isinstance(child_key, str):
+                raise TypeError(f"trajectory mapping keys must be strings: {child_key}")
+            _flatten_value(record, f"{key}.{child_key}", child_value)
+        return
+    if key in record:
+        raise ValueError(f"trajectory values produce duplicate column: {key}")
+    record[key] = deepcopy(value)
+
+
+def _copy_dataframe(dataframe: pd.DataFrame) -> pd.DataFrame:
+    """Copy a DataFrame and recursively snapshot object cells."""
+    copied = dataframe.copy(deep=True).astype(object)
+    for row_index in range(dataframe.shape[0]):
+        for column_index in range(dataframe.shape[1]):
+            copied.iat[row_index, column_index] = deepcopy(dataframe.iat[row_index, column_index])
+    return copied
+
+
+def _copy_series(series: pd.Series) -> pd.Series:
+    """Copy a Series and recursively snapshot object cells."""
+    copied = series.copy(deep=True).astype(object)
+    for index in range(len(series)):
+        copied.iat[index] = deepcopy(series.iat[index])
+    return copied
+
+
+def _values_match_exact(left: Any, right: Any) -> bool:
+    if type(left) is not type(right):
+        return False
+    if isinstance(left, Mapping):
+        if set(left) != set(right):
+            return False
+        return all(_values_match_exact(left[key], right[key]) for key in left)
+    if isinstance(left, Sequence) and not isinstance(left, (str, bytes)):
+        return len(left) == len(right) and all(
+            _values_match_exact(left_item, right_item) for left_item, right_item in zip(left, right, strict=True)
+        )
+    return bool(left == right)
diff --git a/src/cloudai/models/workload.py b/src/cloudai/models/workload.py
index 72416afaa..975237540 100644
--- a/src/cloudai/models/workload.py
+++ b/src/cloudai/models/workload.py
@@ -127,7 +127,7 @@ class TestDefinition(BaseModel, ABC):
         description=(
             "Environment parameters sampled by the env per trial. Sibling to "
             "cmd_args; not part of the agent's action space. CloudAIGymEnv samples, "
-            "persists to env.csv, and includes them in the trajectory cache key."
+            "persists them in the trajectory output, and includes them in the trajectory cache key."
         ),
     )
 
diff --git a/src/cloudai/report_generator/dse_report.py b/src/cloudai/report_generator/dse_report.py
index efb85956e..ddd0b8165 100644
--- a/src/cloudai/report_generator/dse_report.py
+++ b/src/cloudai/report_generator/dse_report.py
@@ -128,6 +128,14 @@ def _safe_literal_eval(raw: Any, default: Any) -> Any:
     return default
 
 
+def load_trajectory_dataframe(iteration_dir: Path) -> tuple[Path, Any] | None:
+    """Load a trajectory CSV as a DataFrame."""
+    trajectory_file = iteration_dir / "trajectory.csv"
+    if not trajectory_file.is_file():
+        return None
+    return trajectory_file, lazy.pd.read_csv(trajectory_file)
+
+
 def _format_scalar(value: Any) -> str:
     if isinstance(value, float):
         return f"{value:.4f}".rstrip("0").rstrip(".")
@@ -239,12 +247,12 @@ def _build_trajectory_steps(
     test_case: TestRun,
     test_runs: list[TestRun],
 ) -> list[TrajectoryStep] | None:
-    trajectory_file = iteration_dir / "trajectory.csv"
-    if not trajectory_file.is_file():
-        logging.warning(f"No trajectory file found for {test_case.name} at {trajectory_file}")
+    loaded_trajectory = load_trajectory_dataframe(iteration_dir)
+    if loaded_trajectory is None:
+        logging.warning(f"No trajectory file found for {test_case.name} in {iteration_dir}")
         return None
 
-    df = lazy.pd.read_csv(trajectory_file)
+    trajectory_file, df = loaded_trajectory
     if df.empty:
         logging.warning(f"No trajectory data found for {test_case.name} at {trajectory_file}")
         return None
diff --git a/src/cloudai/reporter.py b/src/cloudai/reporter.py
index a897015c3..a97ef2fc3 100644
--- a/src/cloudai/reporter.py
+++ b/src/cloudai/reporter.py
@@ -27,9 +27,8 @@
 from rich.console import Console
 from rich.table import Table
 
-from cloudai.report_generator.dse_report import build_dse_summaries
+from cloudai.report_generator.dse_report import build_dse_summaries, load_trajectory_dataframe
 from cloudai.report_generator.util import load_system_metadata
-from cloudai.util.lazy_imports import lazy
 
 from .core import CommandGenStrategy, Reporter, TestRun, case_name
 from .models.scenario import TestRunDetails
@@ -182,12 +181,12 @@ def report_best_dse_config(self):
                 continue
 
             tr_root = self.results_root / tr.name / f"{tr.current_iteration}"
-            trajectory_file = tr_root / "trajectory.csv"
-            if not trajectory_file.is_file():
-                logging.warning("No trajectory file found for %s at %s", tr.name, trajectory_file)
+            loaded_trajectory = load_trajectory_dataframe(tr_root)
+            if loaded_trajectory is None:
+                logging.warning("No trajectory file found for %s in %s", tr.name, tr_root)
                 continue
 
-            df = lazy.pd.read_csv(trajectory_file)
+            _, df = loaded_trajectory
             best_step = df.loc[df["reward"].idxmax()]["step"]
             best_step_details = tr_root / f"{best_step}" / CommandGenStrategy.TEST_RUN_DUMP_FILE_NAME
             if not best_step_details.is_file():
diff --git a/src/cloudai/systems/slurm/single_sbatch_runner.py b/src/cloudai/systems/slurm/single_sbatch_runner.py
index 8a3e769d9..decb1c0ea 100644
--- a/src/cloudai/systems/slurm/single_sbatch_runner.py
+++ b/src/cloudai/systems/slurm/single_sbatch_runner.py
@@ -21,7 +21,8 @@
 from pathlib import Path
 from typing import Generator, Optional, cast
 
-from cloudai.configurator import CloudAIGymEnv, TrajectoryEntry
+from cloudai.configurator import CloudAIGymEnv
+from cloudai.configurator.env_params import EnvParams
 from cloudai.core import BaseJob, JobIdRetrievalError, Registry, System, TestRun, TestScenario
 from cloudai.util import CommandShell, format_time_limit, parse_time_limit
 
@@ -125,9 +126,11 @@ def get_single_tr_block(self, tr: TestRun) -> str:
         return srun_cmd
 
     def unroll_dse(self, tr: TestRun) -> Generator[TestRun, None, None]:
-        for idx, combination in enumerate(tr.all_combinations):
-            next_tr = tr.apply_params_set(combination)
-            next_tr.step = idx + 1
+        params = EnvParams.from_test(tr.test)
+        for idx, combination in enumerate(tr.all_combinations, start=1):
+            sampled_env_params = params.sample(idx) if params is not None else {}
+            next_tr = tr.apply_params_set(combination, env_params=sampled_env_params)
+            next_tr.step = idx
             next_tr.output_path = self.get_job_output_path(next_tr)
 
             if next_tr.test.constraint_check(next_tr, self.system):
@@ -205,27 +208,29 @@ def handle_dse(self):
             if not tr.is_dse_job:
                 continue
 
+            agent_class = registry.get_agent(tr.test.agent)
+            agent_config_data = tr.test.agent_config or {}
+            agent_config = agent_class.get_config_class()(**agent_config_data)
+            gym = CloudAIGymEnv(tr, self, rewards=agent_config.rewards)
+
             for idx, combination in enumerate(tr.all_combinations, start=1):
-                next_tr = tr.apply_params_set(combination)
+                sampled_env_params = gym.params.sample(idx) if gym.params is not None else {}
+                next_tr = tr.apply_params_set(combination, env_params=sampled_env_params)
                 next_tr.step = idx
                 next_tr.output_path = self.get_job_output_path(next_tr)
 
-                rewards = None
-                agent_class = registry.get_agent(next_tr.test.agent)
-                agent_config_data = next_tr.test.agent_config or {}
-                agent_config = agent_class.get_config_class()(**agent_config_data)
-                rewards = agent_config.rewards
+                if not next_tr.test.constraint_check(next_tr, self.system):
+                    continue
 
-                gym = CloudAIGymEnv(next_tr, self, rewards=rewards)
-                observation = gym.get_observation({})
+                gym.test_run = next_tr
+                observation = gym.get_observation()
                 reward = gym.compute_reward(observation)
-                gym.write_trajectory(
-                    TrajectoryEntry(
-                        step=idx,
-                        action=combination,
-                        reward=reward,
-                        observation=observation,
-                    )
+                gym.trajectory.append(
+                    step=idx,
+                    action=combination,
+                    reward=reward,
+                    observation=observation,
+                    env_params=sampled_env_params,
                 )
 
     def completed_test_runs(self, job: BaseJob) -> list[TestRun]:
diff --git a/tests/test_cloudaigym.py b/tests/test_cloudaigym.py
index dfc7ea7ae..c1d8c3153 100644
--- a/tests/test_cloudaigym.py
+++ b/tests/test_cloudaigym.py
@@ -18,9 +18,14 @@
 from typing import cast
 from unittest.mock import MagicMock, PropertyMock, patch
 
+import pandas as pd
 import pytest
 
-from cloudai.configurator import CloudAIGymEnv, GridSearchAgent, TrajectoryEntry
+from cloudai.configurator import (
+    CloudAIGymEnv,
+    GridSearchAgent,
+    Trajectory,
+)
 from cloudai.configurator.env_params import EnvParamSpec, ObsLeafDescriptor
 from cloudai.core import BaseRunner, RewardOverrides, Runner, TestRun, TestScenario
 from cloudai.systems.slurm import SlurmSystem
@@ -37,6 +42,20 @@
 from tests.test_env_params import EnvVarCmdArgs, EnvVarTestDefinition
 
 
+def _trajectory_row(
+    step: int,
+    action: dict[str, object],
+    reward: float,
+    observation: list[float] | list[int],
+) -> dict[str, object]:
+    return {
+        "step": step,
+        **{f"action.{key}": value for key, value in action.items()},
+        "reward": reward,
+        "observation.default": observation[0],
+    }
+
+
 @pytest.fixture
 def nemorun() -> NeMoRunTestDefinition:
     return NeMoRunTestDefinition(
@@ -130,7 +149,8 @@ def test_compute_reward(reward_function, test_cases, base_tr: TestRun):
     env = CloudAIGymEnv(test_run=base_tr, runner=MagicMock(), rewards=RewardOverrides())
 
     for input_value, expected_reward in test_cases:
-        reward = env.compute_reward(input_value)
+        observation = {f"metric_{index}": value for index, value in enumerate(input_value)}
+        reward = env.compute_reward(observation)
         assert reward == expected_reward
 
 
@@ -170,7 +190,9 @@ def test_tr_output_path(setup_env: tuple[TestRun, BaseRunner]):
         pytest.param(RewardOverrides(constraint_failure=-2.5), -2.5, id="custom_penalty"),
     ],
 )
-def test_constraint_failure(nemorun: NeMoRunTestDefinition, rewards: RewardOverrides, expected_reward: float):
+def test_constraint_failure(
+    nemorun: NeMoRunTestDefinition, tmp_path: Path, rewards: RewardOverrides, expected_reward: float
+):
     tdef = nemorun.model_copy(deep=True)
     tdef.cmd_args.data.global_batch_size = 8
     tdef.agent_metrics = ["default"]
@@ -182,6 +204,7 @@ def test_constraint_failure(nemorun: NeMoRunTestDefinition, rewards: RewardOverr
         reports={NeMoRunReportGenerationStrategy},
     )
     runner = MagicMock(spec=BaseRunner)
+    runner.scenario_root = tmp_path / "scenario"
     runner.system = MagicMock()
     env = CloudAIGymEnv(test_run=test_run, runner=runner, rewards=rewards)
 
@@ -353,31 +376,25 @@ def test_apply_params_set__preserves_installables_state(setup_env: tuple[TestRun
 
 
 @pytest.mark.parametrize(
-    ("trajectory", "current_iteration", "action", "expected_step"),
+    ("entries", "action", "expected_step"),
     [
-        ({}, 0, {"x": 1}, None),
-        ({0: [TrajectoryEntry(1, {"x": 1}, 1, [1])]}, 0, {"x": 1}, 1),
-        ({0: [TrajectoryEntry(1, {"x": 1.0}, 1, [1])]}, 0, {"x": 1}, None),
+        ([], {"x": 1}, None),
+        ([_trajectory_row(1, {"x": 1}, 1, [1])], {"x": 1}, 1),
+        ([_trajectory_row(1, {"x": 1.0}, 1, [1])], {"x": 1}, None),
         (
-            {
-                0: [
-                    TrajectoryEntry(1, {"x": 1.0}, 1, [1]),
-                    TrajectoryEntry(2, {"x": 1}, 1, [1]),
-                ]
-            },
-            0,
+            [
+                _trajectory_row(1, {"x": 1.0}, 1, [1]),
+                _trajectory_row(2, {"x": 1}, 1, [1]),
+            ],
             {"x": 1},
             2,
         ),
-        ({0: [TrajectoryEntry(1, {"x": 1}, 1, [1])]}, 1, {"x": 1}, None),
-        ({1: [TrajectoryEntry(3, {"x": 1}, 1, [1])]}, 1, {"x": 1}, 3),
     ],
 )
 def test_get_cached_trajectory_result(
     base_tr: TestRun,
     tmp_path: Path,
-    trajectory: dict[int, list[TrajectoryEntry]],
-    current_iteration: int,
+    entries: list[dict[str, object]],
     action: dict[str, object],
     expected_step: int | None,
 ) -> None:
@@ -390,18 +407,17 @@ def test_get_cached_trajectory_result(
     runner.get_job_output_path.return_value = tmp_path / "scenario" / base_tr.name / "0" / "7"
 
     env = CloudAIGymEnv(test_run=base_tr, runner=runner, rewards=RewardOverrides())
-    env.test_run.current_iteration = current_iteration
-    env.trajectory = trajectory
+    env.trajectory = Trajectory(iteration_dir=env.iteration_dir, dataframe=pd.DataFrame(entries, dtype=object))
 
     actual = env.get_cached_trajectory_result(action, {})
     if actual is None:
         assert expected_step is None
     else:
-        assert actual.step == expected_step
+        assert actual["step"] == expected_step
 
 
-def test_cached_step_appends_trajectory_row(nemorun: NeMoRunTestDefinition, tmp_path: Path) -> None:
-    """Cache hits must still append a row to trajectory.csv so the visible step list matches agent_steps."""
+def test_cached_step_appends_trajectory_record(nemorun: NeMoRunTestDefinition, tmp_path: Path) -> None:
+    """Cache hits must still append a record so the visible step list matches agent_steps."""
     tdef = nemorun.model_copy(deep=True)
     tdef.cmd_args.data.global_batch_size = 8
     tdef.agent_metrics = ["default"]
@@ -420,7 +436,7 @@ def test_cached_step_appends_trajectory_row(nemorun: NeMoRunTestDefinition, tmp_
     env = CloudAIGymEnv(test_run=test_run, runner=runner, rewards=RewardOverrides())
     cached_action = {"trainer.max_steps": 1000}
     env.test_run.current_iteration = 0
-    env.trajectory = {0: [TrajectoryEntry(step=1, action=cached_action, reward=0.42, observation=[0.84])]}
+    env.trajectory.append(step=1, action=cached_action, reward=0.42, observation={"default": 0.84})
 
     env.test_run.step = 4
     obs, reward, done, _info = env.step(cached_action)
@@ -429,18 +445,19 @@ def test_cached_step_appends_trajectory_row(nemorun: NeMoRunTestDefinition, tmp_
     assert reward == 0.42
     assert obs == [0.84]
     assert done is False
-    rows = env.trajectory[0]
+    rows = env.trajectory.dataframe
     assert len(rows) == 2
-    assert rows[-1].step == 5, (
+    assert rows.iloc[-1]["step"] == 5, (
         "CloudAIGymEnv.step() advances test_run.step before recording the trajectory row; "
         "the cached row must be tagged with the advanced trial index, not the pre-step value."
     )
-    assert rows[-1].reward == 0.42
-    assert rows[-1].action == cached_action
+    assert rows.iloc[-1]["reward"] == 0.42
+    assert rows.iloc[-1]["action.trainer.max_steps"] == cached_action["trainer.max_steps"]
 
-    csv_path = env.trajectory_file_path
-    assert csv_path.exists()
-    contents = csv_path.read_text().strip().splitlines()
+    trajectory_path = env.trajectory_file_path
+    assert trajectory_path.exists()
+    assert trajectory_path.name == "trajectory.csv"
+    contents = trajectory_path.read_text().strip().splitlines()
     assert contents[0] == "step,action,reward,observation"
     assert contents[-1].startswith("5,")
 
@@ -448,10 +465,16 @@ def test_cached_step_appends_trajectory_row(nemorun: NeMoRunTestDefinition, tmp_
 def _seed_cached_entry_with_env_params(
     env: CloudAIGymEnv, action: dict[str, object], env_params: dict[str, object]
 ) -> None:
-    """Seed env.trajectory with one entry carrying the given env_params."""
-    entry = TrajectoryEntry(step=1, action=action, reward=0.5, observation=[100.0], env_params=env_params)
-    env.test_run.current_iteration = 0
-    env.trajectory = {0: [entry]}
+    """Seed an environment-parameter-aware trajectory with one entry."""
+    trajectory = Trajectory(iteration_dir=env.iteration_dir)
+    trajectory.append(
+        step=1,
+        action=action,
+        reward=0.5,
+        observation={"default": 100.0},
+        env_params=dict(env_params),
+    )
+    env.trajectory = trajectory
 
 
 def test_cache_miss_when_env_params_differ(base_tr: TestRun, tmp_path: Path) -> None:
@@ -506,7 +529,10 @@ def test_cache_hit_when_neither_has_env_params(base_tr: TestRun, tmp_path: Path)
 
     env = CloudAIGymEnv(test_run=base_tr, runner=runner, rewards=RewardOverrides())
     env.test_run.current_iteration = 0
-    env.trajectory = {0: [TrajectoryEntry(step=1, action={"x": 10}, reward=0.5, observation=[100.0])]}
+    env.trajectory = Trajectory(
+        iteration_dir=env.iteration_dir,
+        dataframe=pd.DataFrame([_trajectory_row(1, {"x": 10}, 0.5, [100.0])], dtype=object),
+    )
     # Note: neither the cached entry nor the trial carries env_params -> existing behavior.
 
     result = env.get_cached_trajectory_result({"x": 10}, {})
@@ -519,7 +545,7 @@ def test_step_reruns_workload_when_env_params_change(tmp_path: Path) -> None:
 
     Counterpart to test_cache_miss_when_env_params_differ but exercising the
     full step() flow: increment_step -> sample env_params -> apply_params_set ->
-    cache lookup -> runner.run() -> write_trajectory. With seed 42 the sampler
+    cache lookup -> runner.run() -> trajectory append. With seed 42 the sampler
     draws ball_speed=3 then ball_speed=1 on the two consecutive trials, so the
     cache key differs and the workload must re-run both times.
     """
@@ -552,9 +578,9 @@ def test_step_reruns_workload_when_env_params_change(tmp_path: Path) -> None:
 
     env = CloudAIGymEnv(test_run=test_run, runner=runner, rewards=RewardOverrides())
     action = {"paddle_width": 4}
-    fake_obs = iter([[100.0], [50.0]])
+    fake_obs = iter([{"default": 100.0}, {"default": 50.0}])
 
-    with patch.object(env, "get_observation", side_effect=lambda _action: next(fake_obs)):
+    with patch.object(env, "get_observation", side_effect=lambda: next(fake_obs)):
         env.test_run.step = 0
         *_, info1 = env.step(action)  # samples ball_speed=3
         *_, info2 = env.step(action)  # samples ball_speed=1
@@ -568,13 +594,8 @@ def test_step_reruns_workload_when_env_params_change(tmp_path: Path) -> None:
     )
 
 
-def test_env_csv_is_step_aligned_with_trajectory(tmp_path: Path) -> None:
-    """env.csv must have exactly one row per env.step() call, with steps aligned 1:1 to trajectory.csv.
-
-    This pins the corpus-friendly contract: a downstream consumer can
-    ``pd.merge(traj, env, on="step")`` without losing rows on either side,
-    independent of whether the trial hit the trajectory cache.
-    """
+def test_env_params_are_recorded_in_trajectory_metadata(tmp_path: Path) -> None:
+    """Every recorded trial includes its sampled environment in trajectory metadata."""
     tdef = EnvVarTestDefinition(
         name="dr",
         description="dr",
@@ -603,35 +624,23 @@ def test_env_csv_is_step_aligned_with_trajectory(tmp_path: Path) -> None:
 
     env = CloudAIGymEnv(test_run=test_run, runner=runner, rewards=RewardOverrides())
     action_a, action_b = {"paddle_width": 4}, {"paddle_width": 8}
-    fake_obs = iter([[100.0], [50.0], [25.0]])
+    fake_obs = iter([{"default": 100.0}, {"default": 50.0}, {"default": 25.0}])
 
-    with patch.object(env, "get_observation", side_effect=lambda _action: next(fake_obs)):
+    with patch.object(env, "get_observation", side_effect=lambda: next(fake_obs)):
         env.test_run.step = 0
         for action in (action_a, action_b, action_a):
             env.step(action)
 
-    env_csv = env.env_params_record_path
-    traj_csv = env.trajectory_file_path
-    assert env_csv.exists(), "env.csv must be written when env_params is declared"
-
-    env_steps = [int(line.split(",", 1)[0]) for line in env_csv.read_text().strip().splitlines()[1:]]
-    traj_steps = [int(line.split(",", 1)[0]) for line in traj_csv.read_text().strip().splitlines()[1:]]
-    assert env_steps == traj_steps == [1, 2, 3], (
-        f"step columns must align 1:1 across env.csv ({env_steps}) and trajectory.csv ({traj_steps})"
-    )
+    trajectory = pd.read_csv(env.trajectory_file_path)
+    metadata = pd.read_csv(env.trajectory.metadata_path)
+    assert list(trajectory.columns) == ["step", "action", "reward", "observation"]
+    assert list(metadata.columns) == ["step", "env_params.ball_speed"]
+    assert trajectory["step"].tolist() == metadata["step"].tolist() == [1, 2, 3]
+    assert metadata["env_params.ball_speed"].notna().all()
 
 
-def test_env_csv_step_alignment_holds_on_constraint_failure(tmp_path: Path) -> None:
-    """A constraint failure must not desync env.csv from trajectory.csv.
-
-    Runs three steps where the middle one fails ``constraint_check`` and the
-    other two succeed. ``env.csv`` is sunk inside ``write_trajectory`` from the
-    same ``TrajectoryEntry``, which is never reached on the early-return
-    constraint-failure path - so the failed step lands in neither file. The
-    corpus-friendly contract (``pd.merge(traj, env, on="step")`` loses no rows)
-    therefore holds via shared absence: both files record exactly the surviving
-    steps, aligned 1:1.
-    """
+def test_constraint_failure_omits_the_complete_trajectory_row(tmp_path: Path) -> None:
+    """A constraint failure records no trajectory components."""
     tdef = EnvVarTestDefinition(
         name="dr",
         description="dr",
@@ -662,40 +671,35 @@ def test_env_csv_step_alignment_holds_on_constraint_failure(tmp_path: Path) -> N
 
     # Step 2 fails the constraint; steps 1 and 3 survive. get_observation is only
     # reached on the surviving steps, so it yields exactly two values.
-    fake_obs = iter([[100.0], [25.0]])
+    fake_obs = iter([{"default": 100.0}, {"default": 25.0}])
     with (
         patch.object(EnvVarTestDefinition, "constraint_check", side_effect=[True, False, True]),
-        patch.object(env, "get_observation", side_effect=lambda _action: next(fake_obs)),
+        patch.object(env, "get_observation", side_effect=lambda: next(fake_obs)),
     ):
         env.test_run.step = 0
         for action in ({"paddle_width": 4}, {"paddle_width": 6}, {"paddle_width": 8}):
             env.step(action)
 
-    env_csv = env.env_params_record_path
-    traj_csv = env.trajectory_file_path
-
-    assert env_csv.exists(), "surviving steps declare env_params -> env.csv must exist"
-    env_steps = [int(line.split(",", 1)[0]) for line in env_csv.read_text().strip().splitlines()[1:]]
+    trajectory_path = env.trajectory_file_path
+    metadata_path = env.trajectory.metadata_path
     traj_steps = (
-        [int(line.split(",", 1)[0]) for line in traj_csv.read_text().strip().splitlines()[1:]]
-        if traj_csv.exists()
+        [int(line.split(",", 1)[0]) for line in trajectory_path.read_text().strip().splitlines()[1:]]
+        if trajectory_path.exists()
         else []
     )
-    assert env_steps == traj_steps == [1, 3], (
-        f"the constraint-failed step (2) must appear in neither file; env.csv ({env_steps}) "
-        f"and trajectory.csv ({traj_steps}) must stay 1:1 aligned on the surviving steps"
+    metadata_steps = (
+        [int(line.split(",", 1)[0]) for line in metadata_path.read_text().strip().splitlines()[1:]]
+        if metadata_path.exists()
+        else []
     )
+    assert traj_steps == metadata_steps == [1, 3]
 
 
-def test_step_cache_hit_with_declared_env_params_still_writes_env_csv(tmp_path: Path) -> None:
-    """End-to-end: cache HIT under observer-driven env_params still records env.csv.
+def test_step_cache_hit_with_declared_env_params_records_complete_trajectory_row(tmp_path: Path) -> None:
+    """End-to-end: a cache hit records its environment in the trajectory output.
 
-    A cache hit still calls ``write_trajectory``, which sinks the trajectory row
-    and the matching env.csv row from the same entry - keeping the two files
-    step-aligned even when the workload itself is short-circuited.
-    Asserts: (a) the workload is NOT re-run (cache short-circuit), (b)
-    env.csv gains a row, (c) trajectory.csv gains a row carrying the
-    sampled env_params.
+    A cache hit still appends a complete row even though workload execution is
+    short-circuited.
     """
     import random as _random
 
@@ -726,13 +730,17 @@ def test_step_cache_hit_with_declared_env_params_still_writes_env_csv(tmp_path:
     env = CloudAIGymEnv(test_run=test_run, runner=runner, rewards=RewardOverrides())
     assert env.params is not None, "TestDefinition.env_params declared -> EnvParams must be built"
 
-    expected_sample = {"ball_speed": _random.Random("42:ball_speed:1").choice([1, 2, 3])}
+    expected_sample = {"ball_speed": _random.Random("42:ball_speed:2").choice([1, 2, 3])}  # noqa: S311, RUF100
     action = {"paddle_width": 4}
     env.test_run.current_iteration = 0
-    env.trajectory = {
-        0: [TrajectoryEntry(step=0, action=action, reward=0.42, observation=[0.84], env_params=expected_sample)]
-    }
-    env.test_run.step = 0
+    env.trajectory.append(
+        step=1,
+        action=action,
+        reward=0.42,
+        observation={"default": 0.84},
+        env_params=expected_sample,
+    )
+    env.test_run.step = 1
 
     with patch.object(env, "get_observation", side_effect=AssertionError("cache miss path must not run")):
         obs, reward, _done, info = env.step(action)
@@ -744,14 +752,14 @@ def test_step_cache_hit_with_declared_env_params_still_writes_env_csv(tmp_path:
         "the per-trial regime behind this observation is reported on info['env_params']"
     )
 
-    env_csv = env.env_params_record_path
-    assert env_csv.exists(), "cache HIT must NOT skip the observer; env.csv must record the trial"
-    env_rows = env_csv.read_text().strip().splitlines()
-    assert env_rows[0] == "step,env"
-    assert env_rows[1].startswith("1,"), f"expected step 1 row in env.csv, got {env_rows[1]!r}"
+    trajectory = pd.read_csv(env.trajectory_file_path)
+    metadata = pd.read_csv(env.trajectory.metadata_path)
+    assert list(trajectory.columns) == ["step", "action", "reward", "observation"]
+    assert trajectory.iloc[-1]["step"] == 2
+    assert metadata.iloc[-1]["env_params.ball_speed"] == expected_sample["ball_speed"]
 
-    traj_rows = env.trajectory[0]
-    assert len(traj_rows) == 2 and traj_rows[-1].env_params == expected_sample, (
+    traj_rows = env.trajectory.dataframe
+    assert len(traj_rows) == 2 and traj_rows.iloc[-1]["env_params.ball_speed"] == expected_sample["ball_speed"], (
         "cache-hit trajectory entry must record the per-trial env_params sample"
     )
 
@@ -794,7 +802,7 @@ def test_step_overlays_env_params_onto_cmd_args(tmp_path: Path) -> None:
     env = CloudAIGymEnv(test_run=test_run, runner=runner, rewards=RewardOverrides())
     expected = _random.Random("42:ball_speed:1").choice([1, 2])
 
-    with patch.object(env, "get_observation", side_effect=lambda _action: [1.0]):
+    with patch.object(env, "get_observation", side_effect=lambda: {"default": 1.0}):
         env.test_run.step = 0
         env.step({"paddle_width": 4})
 
@@ -826,8 +834,10 @@ def test_param_space_excludes_env_params_keys(setup_env: tuple[TestRun, BaseRunn
     )
 
 
-def test_no_env_csv_when_env_params_not_declared(nemorun: NeMoRunTestDefinition, tmp_path: Path) -> None:
-    """Workloads without [env_params.*] pay zero overhead: no observer, no env.csv."""
+def test_csv_trajectory_has_no_env_params_column_when_not_declared(
+    nemorun: NeMoRunTestDefinition, tmp_path: Path
+) -> None:
+    """Workloads without env_params retain the base trajectory schema."""
     tdef = nemorun.model_copy(deep=True)
     tdef.cmd_args.data.global_batch_size = 8
     test_run = TestRun(
@@ -842,10 +852,16 @@ def test_no_env_csv_when_env_params_not_declared(nemorun: NeMoRunTestDefinition,
     runner.scenario_root = tmp_path / "scenario"
     runner.system = MagicMock()
 
-    env = CloudAIGymEnv(test_run=test_run, runner=runner, rewards=RewardOverrides())
+    env = CloudAIGymEnv(
+        test_run=test_run,
+        runner=runner,
+        rewards=RewardOverrides(),
+    )
 
     assert env.params is None, "no env_params declared -> no EnvParams object"
-    assert not env.env_params_record_path.exists()
+    env.trajectory.append(step=1, action={}, reward=1.0, observation={"default": 1.0})
+    assert env.trajectory_file_path.read_text().splitlines()[0] == "step,action,reward,observation"
+    assert not env.trajectory.metadata_path.exists()
 
 
 def _dr_env(tmp_path: Path, candidates: list, *, seed: int = 42) -> CloudAIGymEnv:
@@ -883,6 +899,7 @@ def test_metrics_only_env_opts_out(self, nemorun: NeMoRunTestDefinition, tmp_pat
         tdef.cmd_args.data.global_batch_size = 8
         test_run = TestRun(name="plain_tr", test=tdef, num_nodes=1, nodes=[])
         runner = MagicMock(spec=BaseRunner)
+        runner.scenario_root = tmp_path / "scenario"
         runner.system = MagicMock()
         env = CloudAIGymEnv(test_run=test_run, runner=runner, rewards=RewardOverrides())
 
diff --git a/tests/test_env_params.py b/tests/test_env_params.py
index 9eb11d2ac..14acb54ba 100644
--- a/tests/test_env_params.py
+++ b/tests/test_env_params.py
@@ -33,7 +33,6 @@
 
 import dataclasses
 import random
-from pathlib import Path
 from typing import Any, List, Union
 
 import pytest
@@ -45,7 +44,6 @@
     EnvParams,
     EnvParamSpec,
     ObsLeafDescriptor,
-    write_env_params,
 )
 from cloudai.core import TestRun
 from cloudai.models.workload import CmdArgs, TestDefinition
@@ -204,27 +202,6 @@ def test_env_params_is_immutable() -> None:
         env_params.seed = 1  # pyright: ignore[reportAttributeAccessIssue]
 
 
-# --- write_env_params: unchanged persistence contract ---
-
-
-def test_csv_sink_skips_empty_samples_and_rejects_zero_step(tmp_path: Path) -> None:
-    path = tmp_path / "env.csv"
-    write_env_params(path, 1, {})  # empty -> no-op, no file
-    assert not path.exists()
-    with pytest.raises(ValueError, match="must be a positive trial index"):
-        write_env_params(path, 0, {"ball_speed": 1})
-
-
-def test_csv_sink_writes_header_then_rows(tmp_path: Path) -> None:
-    path = tmp_path / "env.csv"
-    write_env_params(path, 1, {"ball_speed": 2})
-    write_env_params(path, 2, {"ball_speed": 3})
-    contents = path.read_text().strip().splitlines()
-    assert contents[0] == "step,env"
-    assert contents[1].startswith("1,")
-    assert contents[2].startswith("2,")
-
-
 # --- EnvParams.from_test: resolves candidate lists from cmd_args, once at env formulation ---
 
 
diff --git a/tests/test_gymnasium_adapter_contract.py b/tests/test_gymnasium_adapter_contract.py
index a995fede0..54cef9630 100644
--- a/tests/test_gymnasium_adapter_contract.py
+++ b/tests/test_gymnasium_adapter_contract.py
@@ -25,7 +25,7 @@
 contextual-bandit configs (``agent_steps=1``), RLlib calls ``reset()`` before
 *every* trial. An earlier adapter rewound ``test_run.step`` on reset and
 collapsed every trial onto step 1 — silently overwriting output dirs and
-producing duplicate-step rows in trajectory.csv / env.csv.
+producing duplicate-step trajectory records.
 
 These tests pin the negative invariant: the adapter must not mutate
 ``test_run.step``. That counter is owned by ``TestRun`` and advanced
diff --git a/tests/test_single_sbatch_runner.py b/tests/test_single_sbatch_runner.py
index 91d3cdf27..6b3f029b3 100644
--- a/tests/test_single_sbatch_runner.py
+++ b/tests/test_single_sbatch_runner.py
@@ -24,6 +24,7 @@
 import pytest
 import toml
 
+from cloudai.configurator.env_params import EnvParams, EnvParamSpec
 from cloudai.core import Registry, System, TestRun, TestScenario
 from cloudai.systems.slurm import SingleSbatchRunner, SlurmJob, SlurmJobMetadata, SlurmSystem
 from cloudai.workloads.nccl_test import NCCLCmdArgs, NCCLTestDefinition
@@ -569,3 +570,29 @@ def test_trajectory_saved(dse_tr: TestRun, slurm_system: SlurmSystem) -> None:
     df = pd.read_csv(trajectory_path)
     assert df.shape[0] == len(dse_tr.all_combinations)
     assert df["step"].tolist() == list(range(1, len(dse_tr.all_combinations) + 1))
+
+
+def test_dse_env_params_are_applied_to_runs_and_trajectory(dse_tr: TestRun, slurm_system: SlurmSystem) -> None:
+    dse_tr.test.cmd_args.nthreads = [1, 2]
+    dse_tr.test.env_params = {"nthreads": EnvParamSpec()}
+    dse_tr.test.agent_config = {"random_seed": 42}
+    tc = TestScenario(name="tc", test_runs=[dse_tr])
+    runner = SingleSbatchRunner(mode="run", system=slurm_system, test_scenario=tc, output_path=slurm_system.output_path)
+    trajectory_path = runner.scenario_root / dse_tr.name / f"{dse_tr.current_iteration}" / "trajectory.csv"
+    metadata_path = trajectory_path.with_name("metadata.csv")
+    trajectory_path.unlink(missing_ok=True)
+    metadata_path.unlink(missing_ok=True)
+
+    params = EnvParams.from_test(dse_tr.test)
+    assert params is not None
+    expected_samples = [params.sample(step) for step in range(1, len(dse_tr.all_combinations) + 1)]
+
+    unrolled = list(runner.unroll_dse(dse_tr))
+    assert [tr.test.cmd_args.nthreads for tr in unrolled] == [sample["nthreads"] for sample in expected_samples]
+
+    runner.handle_dse()
+
+    trajectory = pd.read_csv(trajectory_path)
+    metadata = pd.read_csv(metadata_path)
+    assert trajectory["step"].tolist() == metadata["step"].tolist()
+    assert metadata["env_params.nthreads"].tolist() == [sample["nthreads"] for sample in expected_samples]
diff --git a/tests/test_trajectory.py b/tests/test_trajectory.py
new file mode 100644
index 000000000..8b0a3654b
--- /dev/null
+++ b/tests/test_trajectory.py
@@ -0,0 +1,341 @@
+# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES
+# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from ast import literal_eval
+from pathlib import Path
+
+import pandas as pd
+import pytest
+
+from cloudai.configurator import Trajectory
+
+
+def test_append_flattens_domains_into_dataframe_columns(tmp_path: Path) -> None:
+    trajectory = Trajectory(iteration_dir=tmp_path)
+
+    row = trajectory.append(
+        step=1,
+        action={"model": {"layers": 8}},
+        reward=0.95,
+        observation={"throughput": 120.0},
+        env_params={"network_speed": 100},
+        logging={"gpu_power_watts": 610.0},
+    )
+
+    expected = {
+        "step": 1,
+        "action.model.layers": 8,
+        "reward": 0.95,
+        "observation.throughput": 120.0,
+        "env_params.network_speed": 100,
+        "logging.gpu_power_watts": 610.0,
+    }
+    assert row.to_dict() == expected
+    assert trajectory.dataframe.to_dict(orient="records") == [expected]
+    assert all(pd.api.types.is_object_dtype(dtype) for dtype in trajectory.dataframe.dtypes)
+
+    core = pd.read_csv(trajectory.output_path)
+    assert list(core.columns) == ["step", "action", "reward", "observation"]
+    action = core.loc[0, "action"]
+    observation = core.loc[0, "observation"]
+    assert isinstance(action, str)
+    assert isinstance(observation, str)
+    assert literal_eval(action) == {"model": {"layers": 8}}
+    assert core.loc[0, "reward"] == 0.95
+    assert literal_eval(observation) == [120.0]
+
+    metadata = pd.read_csv(trajectory.metadata_path)
+    assert metadata.to_dict(orient="records") == [
+        {
+            "step": 1,
+            "env_params.network_speed": 100,
+            "logging.gpu_power_watts": 610.0,
+        }
+    ]
+
+
+def test_dataframe_and_returned_rows_are_copies(tmp_path: Path) -> None:
+    trajectory = Trajectory(iteration_dir=tmp_path)
+    samples = [1, 2]
+    row = trajectory.append(
+        step=1,
+        action={"x": 1},
+        reward=1.0,
+        observation={"metric": 2.0},
+        logging={"samples": samples},
+    )
+
+    samples.append(3)
+    row["reward"] = 99.0
+    row_samples = row["logging.samples"]
+    assert isinstance(row_samples, list)
+    row_samples.append(4)
+    dataframe = trajectory.dataframe
+    dataframe.loc[0, "reward"] = 88.0
+    dataframe_samples = dataframe.loc[0, "logging.samples"]
+    assert isinstance(dataframe_samples, list)
+    dataframe_samples.append(5)
+    found = trajectory.find(action={"x": 1})
+    assert found is not None
+    found_samples = found["logging.samples"]
+    assert isinstance(found_samples, list)
+    found_samples.append(6)
+
+    assert trajectory.dataframe.loc[0, "reward"] == 1.0
+    assert trajectory.dataframe.loc[0, "logging.samples"] == [1, 2]
+
+
+def test_find_matches_flattened_subset_and_preserves_exact_types(tmp_path: Path) -> None:
+    trajectory = Trajectory(iteration_dir=tmp_path)
+    first = trajectory.append(
+        step=1, action={"x": 1.0}, reward=1.0, observation={"metric": 1.0}, env_params={"speed": 2}
+    )
+    trajectory.append(step=2, action={"x": 1}, reward=2.0, observation={"metric": 2.0}, env_params={"speed": 2})
+
+    match = trajectory.find(action={"x": 1.0}, env_params={"speed": 2})
+
+    assert match is not None
+    assert match.to_dict() == first.to_dict()
+    assert trajectory.find(action={"x": True}, env_params={"speed": 2}) is None
+
+
+def test_find_returns_none_for_unknown_columns(tmp_path: Path) -> None:
+    trajectory = Trajectory(iteration_dir=tmp_path)
+    trajectory.append(step=1, action={"x": 1}, reward=1.0, observation={"metric": 1.0})
+
+    assert trajectory.find(env_params={"speed": 2}) is None
+
+
+def test_trajectory_rejects_invalid_or_non_increasing_steps(tmp_path: Path) -> None:
+    trajectory = Trajectory(iteration_dir=tmp_path)
+
+    with pytest.raises(ValueError, match="positive integer"):
+        trajectory.append(step=0, action={"x": 1}, reward=1.0, observation={"metric": 1.0})
+
+    trajectory.append(step=2, action={"x": 1}, reward=1.0, observation={"metric": 1.0})
+    with pytest.raises(ValueError, match="steps must increase"):
+        trajectory.append(step=2, action={"x": 2}, reward=1.0, observation={"metric": 1.0})
+
+
+def test_first_row_establishes_fixed_schema(tmp_path: Path) -> None:
+    trajectory = Trajectory(iteration_dir=tmp_path)
+    trajectory.append(step=1, action={"x": 1}, reward=1.0, observation={"metric": 1.0})
+
+    with pytest.raises(ValueError, match="record fields changed"):
+        trajectory.append(
+            step=2,
+            action={"x": 2},
+            reward=2.0,
+            observation={"metric": 2.0},
+            logging={"power": 600.0},
+        )
+
+
+def test_flattening_rejects_duplicate_columns(tmp_path: Path) -> None:
+    trajectory = Trajectory(iteration_dir=tmp_path)
+
+    with pytest.raises(ValueError, match=r"duplicate column: action\.x"):
+        trajectory.append(
+            step=1,
+            action={"x": 1},
+            reward=1.0,
+            observation={"metric": 1.0},
+            **{"action.x": 2},
+        )
+
+
+def test_flattening_rejects_non_string_mapping_keys(tmp_path: Path) -> None:
+    trajectory = Trajectory(iteration_dir=tmp_path)
+
+    with pytest.raises(TypeError, match="mapping keys must be strings"):
+        trajectory.append(step=1, action={1: "x"}, reward=1.0, observation={"metric": 1.0})
+
+
+@pytest.mark.parametrize("missing", ["action", "reward", "observation"])
+def test_append_requires_core_values(tmp_path: Path, missing: str) -> None:
+    values = {"action": {"x": 1}, "reward": 1.0, "observation": {"metric": 2.0}}
+    del values[missing]
+
+    with pytest.raises(TypeError, match=missing):
+        Trajectory(iteration_dir=tmp_path).append(step=1, **values)
+
+    assert not (tmp_path / "trajectory.csv").exists()
+
+
+def test_warm_start_dataframe_is_copied_and_not_replayed(tmp_path: Path) -> None:
+    samples = [1, 2]
+    dataframe = pd.DataFrame(
+        [
+            {
+                "step": 1,
+                "action.x": 1,
+                "reward": 1.0,
+                "observation.metric": 2.0,
+                "logging.samples": samples,
+            }
+        ],
+        dtype=object,
+    )
+    trajectory = Trajectory(dataframe=dataframe, iteration_dir=tmp_path)
+    dataframe.loc[0, "reward"] = 99.0
+    samples.append(3)
+
+    assert len(trajectory) == 1
+    assert trajectory.dataframe.loc[0, "reward"] == 1.0
+    assert trajectory.dataframe.loc[0, "logging.samples"] == [1, 2]
+    assert not (tmp_path / "trajectory.csv").exists()
+
+
+@pytest.mark.parametrize(
+    ("records", "message"),
+    [
+        ([{"action.x": 1}], "must contain a step column"),
+        ([{"step": 0}], "positive integer"),
+        ([{"step": 2}, {"step": 1}], "steps must increase"),
+    ],
+)
+def test_warm_start_dataframe_validation(tmp_path: Path, records: list[dict[str, object]], message: str) -> None:
+    with pytest.raises(ValueError, match=message):
+        Trajectory(iteration_dir=tmp_path, dataframe=pd.DataFrame(records))
+
+
+def test_warm_start_dataframe_rejects_duplicate_columns(tmp_path: Path) -> None:
+    dataframe = pd.DataFrame([[1, 2]], columns=["step", "step"])
+
+    with pytest.raises(ValueError, match="columns must be unique"):
+        Trajectory(iteration_dir=tmp_path, dataframe=dataframe)
+
+
+def test_csv_initializes_a_precreated_empty_file(tmp_path: Path) -> None:
+    path = tmp_path / "trajectory.csv"
+    path.touch()
+
+    Trajectory(iteration_dir=tmp_path).append(step=1, action={"x": 1}, reward=1.0, observation={"metric": 2.0})
+
+    assert path.read_text().splitlines() == [
+        "step,action,reward,observation",
+        "1,{'x': 1},1.0,[2.0]",
+    ]
+
+
+def test_csv_reuses_a_matching_header_without_duplication(tmp_path: Path) -> None:
+    path = tmp_path / "trajectory.csv"
+    path.write_text("step,action,reward,observation\n")
+
+    Trajectory(iteration_dir=tmp_path).append(step=1, action={"x": 1}, reward=1.0, observation={"metric": 2.0})
+
+    assert path.read_text().splitlines() == [
+        "step,action,reward,observation",
+        "1,{'x': 1},1.0,[2.0]",
+    ]
+
+
+def test_csv_rejects_an_existing_mismatched_header(tmp_path: Path) -> None:
+    path = tmp_path / "trajectory.csv"
+    path.write_text("step,reward,action,observation\n")
+
+    with pytest.raises(ValueError, match="file fields do not match"):
+        Trajectory(iteration_dir=tmp_path).append(step=1, action={"x": 1}, reward=1.0, observation={"metric": 2.0})
+
+    assert path.read_text() == "step,reward,action,observation\n"
+
+
+def test_core_only_trajectory_does_not_create_metadata_file(tmp_path: Path) -> None:
+    trajectory = Trajectory(iteration_dir=tmp_path)
+
+    trajectory.append(step=1, action={"x": 1}, reward=1.0, observation={"metric": 2.0})
+
+    assert not trajectory.metadata_path.exists()
+
+
+def test_metadata_reuses_matching_header_without_duplication(tmp_path: Path) -> None:
+    path = tmp_path / "metadata.csv"
+    path.write_text("step,env_params.speed,logging.power\n")
+    trajectory = Trajectory(iteration_dir=tmp_path)
+
+    trajectory.append(
+        step=1,
+        action={"x": 1},
+        reward=1.0,
+        observation={"metric": 2.0},
+        env_params={"speed": 100},
+        logging={"power": 600.0},
+    )
+
+    assert path.read_text().splitlines() == [
+        "step,env_params.speed,logging.power",
+        "1,100,600.0",
+    ]
+
+
+def test_metadata_initializes_a_precreated_empty_file(tmp_path: Path) -> None:
+    path = tmp_path / "metadata.csv"
+    path.touch()
+    trajectory = Trajectory(iteration_dir=tmp_path)
+
+    trajectory.append(
+        step=1,
+        action={"x": 1},
+        reward=1.0,
+        observation={"metric": 2.0},
+        env_params={"speed": 100},
+    )
+
+    assert path.read_text().splitlines() == ["step,env_params.speed", "1,100"]
+
+
+def test_metadata_rejects_an_existing_mismatched_header_before_core_write(tmp_path: Path) -> None:
+    metadata_path = tmp_path / "metadata.csv"
+    metadata_path.write_text("step,logging.power,env_params.speed\n")
+    trajectory = Trajectory(iteration_dir=tmp_path)
+
+    with pytest.raises(ValueError, match="file fields do not match"):
+        trajectory.append(
+            step=1,
+            action={"x": 1},
+            reward=1.0,
+            observation={"metric": 2.0},
+            env_params={"speed": 100},
+            logging={"power": 600.0},
+        )
+
+    assert not trajectory.output_path.exists()
+    assert len(trajectory) == 0
+
+
+def test_persistence_failure_does_not_append_to_memory(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    trajectory = Trajectory(iteration_dir=tmp_path)
+
+    def fail_write(*_args: object, **_kwargs: object) -> None:
+        raise OSError("write failed")
+
+    monkeypatch.setattr(pd.DataFrame, "to_csv", fail_write)
+
+    with pytest.raises(OSError, match="write failed"):
+        trajectory.append(step=1, action={"x": 1}, reward=1.0, observation={"metric": 2.0})
+
+    assert len(trajectory) == 0
+
+
+def test_trajectory_logs_lifecycle_and_lookup(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
+    with caplog.at_level("DEBUG"):
+        trajectory = Trajectory(iteration_dir=tmp_path)
+        trajectory.append(step=1, action={"x": 1}, reward=1.0, observation={"metric": 2.0})
+        assert trajectory.find(action={"x": 1}) is not None
+        assert trajectory.find(action={"x": 2}) is None
+
+    assert "Initializing Trajectory: entries=0, columns=[]." in caplog.messages
+    assert "Appended trajectory row for step 1 (total rows: 1)." in caplog.messages