diff --git a/benchmarks/test_collectors_benchmark.py b/benchmarks/test_collectors_benchmark.py
index a1119d7cd71..516cc3f8148 100644
--- a/benchmarks/test_collectors_benchmark.py
+++ b/benchmarks/test_collectors_benchmark.py
@@ -26,6 +26,7 @@
     EnvCreator,
     GymEnv,
     ParallelEnv,
+    SerialEnv,
     StepCounter,
     TransformedEnv,
 )
@@ -292,6 +293,25 @@ def single_collector_with_rb_setup_pixels():
     return ((c, rb), {})
 
 
+def flat_collector_with_rb_setup():
+    """Setup batched collection into the preferred flat replay layout."""
+    env = SerialEnv(4, functools.partial(_PayloadEnv, payload_size=4096))
+    rb = ReplayBuffer(storage=LazyTensorStorage(10000))
+    c = Collector(
+        env,
+        RandomPolicy(env.action_spec),
+        total_frames=-1,
+        frames_per_batch=128,
+        replay_buffer=rb,
+        flatten_data=True,
+    )
+    c = iter(c)
+    for i, _ in enumerate(c):
+        if i == 10:
+            break
+    return ((c, rb), {})
+
+
 def sync_collector_with_rb_setup():
     """Setup a multi-process collector whose workers write whole rollouts."""
     device = "cuda:0" if torch.cuda.device_count() else "cpu"
@@ -360,6 +380,12 @@ def test_sync_payload_with_rb(benchmark):
         collector.shutdown()
 
 
+def test_flatten_data_with_rb(benchmark):
+    """Benchmark one reshape and one extend for a batched rollout."""
+    (c, rb), _ = flat_collector_with_rb_setup()
+    benchmark(execute_collector_with_rb, c, rb)
+
+
 @pytest.mark.skipif(not torch.cuda.device_count(), reason="no rendering without cuda")
 def test_single_with_rb_pixels(benchmark):
     """Benchmark single collector with replay buffer for pixel observations."""
diff --git a/docs/source/reference/collectors_distributed.rst b/docs/source/reference/collectors_distributed.rst
index 2cb017fe47d..13eb880250c 100644
--- a/docs/source/reference/collectors_distributed.rst
+++ b/docs/source/reference/collectors_distributed.rst
@@ -96,11 +96,15 @@ topology.
 
 .. tip::
 
-  All distributed collectors support ``trajs_per_batch`` combined with
-  ``replay_buffer``.  When set, each remote worker assembles **complete
-  trajectories** and writes them to the shared buffer as flat 1-D sequences,
-  which is directly compatible with :class:`~torchrl.data.replay_buffers.SliceSampler`.
-  See :ref:`collectors_replay_trajs` for examples and best practices.
+  :class:`~torchrl.collectors.distributed.RayCollector` supports direct replay
+  writes through a Ray-backed ``replay_buffer``. Pass ``flatten_data=True`` for
+  regular fixed-frame collection, or set ``trajs_per_batch`` to have each
+  remote worker assemble complete trajectories. Both paths write flat 1-D
+  sequences compatible with
+  :class:`~torchrl.data.replay_buffers.SliceSampler`. The generic distributed
+  and RPC collectors return data to their caller instead of owning a shared
+  replay-buffer transport. See :ref:`collectors_replay_trajs` for examples and
+  best practices.
 
 .. autosummary::
     :toctree: generated/
diff --git a/docs/source/reference/collectors_replay.rst b/docs/source/reference/collectors_replay.rst
index 3231b17f3cf..f6d208d3bae 100644
--- a/docs/source/reference/collectors_replay.rst
+++ b/docs/source/reference/collectors_replay.rst
@@ -13,79 +13,56 @@ Collectors and Replay Buffers
     recommended — see :ref:`Data layout: contiguous trajectories
     <data-layout>`.
 
-Collectors and replay buffers interoperability
-----------------------------------------------
-
-In the simplest scenario where single transitions have to be sampled
-from the replay buffer, little attention has to be given to the way
-the collector is built. Flattening the data after collection will
-be a sufficient preprocessing step before populating the storage:
-
-    >>> memory = ReplayBuffer(
-    ...     storage=LazyTensorStorage(N),
-    ...     transform=lambda data: data.reshape(-1))
-    >>> for data in collector:
-    ...     memory.extend(data)
-
-If trajectory slices have to be collected, the recommended way to achieve this is to create
-a multidimensional buffer and sample using the :class:`~torchrl.data.replay_buffers.SliceSampler`
-sampler class. One must ensure that the data passed to the buffer is properly shaped, with the
-``time`` and ``batch`` dimensions clearly separated. In practice, the following configurations
-will work:
-
-    >>> # Single environment: no need for a multi-dimensional buffer
-    >>> memory = ReplayBuffer(
-    ...     storage=LazyTensorStorage(N),
-    ...     sampler=SliceSampler(num_slices=4, trajectory_key=("collector", "traj_ids"))
-    ... )
-    >>> collector = Collector(env, policy, frames_per_batch=N, total_frames=-1)
-    >>> for data in collector:
-    ...     memory.extend(data)
-    >>> # Batched environments: a multi-dim buffer is required
-    >>> memory = ReplayBuffer(
-    ...     storage=LazyTensorStorage(N, ndim=2),
-    ...     sampler=SliceSampler(num_slices=4, trajectory_key=("collector", "traj_ids"))
-    ... )
-    >>> env = ParallelEnv(4, make_env)
-    >>> collector = Collector(env, policy, frames_per_batch=N, total_frames=-1)
-    >>> for data in collector:
-    ...     memory.extend(data)
-    >>> # Synchronous process collection behaves like ParallelEnv if cat_results="stack"
-    >>> memory = ReplayBuffer(
-    ...     storage=LazyTensorStorage(N, ndim=2),
-    ...     sampler=SliceSampler(num_slices=4, trajectory_key=("collector", "traj_ids"))
-    ... )
-    >>> collector = Collector(make_env, policy,
-    ...     num_collectors=4,
-    ...     sync=True,
-    ...     frames_per_batch=N,
-    ...     total_frames=-1,
-    ...     cat_results="stack")
-    >>> for data in collector:
-    ...     memory.extend(data)
-    >>> # Process collection + parallel env: adapt ndim for both batch dimensions
-    >>> memory = ReplayBuffer(
-    ...     storage=LazyTensorStorage(N, ndim=3),
-    ...     sampler=SliceSampler(num_slices=4, trajectory_key=("collector", "traj_ids"))
-    ... )
-    >>> collector = Collector(lambda: ParallelEnv(2, make_env), policy,
-    ...     num_collectors=4,
-    ...     sync=True,
-    ...     frames_per_batch=N,
-    ...     total_frames=-1,
-    ...     cat_results="stack")
-    >>> for data in collector:
-    ...     memory.extend(data)
-
-.. important::
-
-    The ``ndim=2`` and ``ndim=3`` examples above apply to **fixed-frame
-    batches** (the default, without ``trajs_per_batch``).  When
-    ``trajs_per_batch`` is set, each trajectory is written to the buffer as a
-    **flat 1-D sequence** of variable length.  A storage with ``ndim >= 2``
-    expects a fixed second dimension that variable-length trajectories cannot
-    satisfy.  Always use the default ``ndim=1`` when combining
-    ``trajs_per_batch`` with a replay buffer.
+Preferred layout: flat 1-D storage
+----------------------------------
+
+TorchRL recommends storing replay data in a single flat, 1-D buffer
+(``ndim=1``). Collector batch dimensions describe how data was produced; they
+do not need to become replay-buffer dimensions. Trajectory boundaries remain
+available through ``("collector", "traj_ids")``, ``("next", "done")``,
+``("next", "terminated")``, and ``("next", "truncated")``, and trajectory
+slices can be reconstructed at sampling time with
+:class:`~torchrl.data.replay_buffers.SliceSampler`.
+
+When the collector writes directly to the replay buffer, set
+``flatten_data=True``. A batched rollout with shape ``[N, T]`` is then reshaped
+once to ``[N * T]`` and written with one ``extend`` call:
+
+.. code-block:: python
+
+    from torchrl.collectors import Collector
+    from torchrl.data import LazyTensorStorage, ReplayBuffer
+
+    memory = ReplayBuffer(storage=LazyTensorStorage(100_000))
+    collector = Collector(
+        env,
+        policy,
+        frames_per_batch=200,
+        total_frames=-1,
+        replay_buffer=memory,
+        flatten_data=True,
+    )
+    for _ in collector:  # yields None; transitions are written directly
+        batch = memory.sample(256)
+
+The same argument is available on :class:`~torchrl.collectors.AsyncCollector`,
+:class:`~torchrl.collectors.MultiSyncCollector`,
+:class:`~torchrl.collectors.MultiAsyncCollector`, and
+:class:`~torchrl.collectors.distributed.RayCollector`. If the caller owns the
+write instead, flatten explicitly before extending:
+
+.. code-block:: python
+
+    for data in collector:
+        memory.extend(data.reshape(-1))
+
+Higher-dimensional storage (``ndim >= 2``) remains supported for applications
+that deliberately store fixed-shape chunks. It is a specialized layout rather
+than the recommended default: the chunk dimensions become part of every replay
+item, variable-length trajectories cannot be represented directly, and shared
+multi-producer buffers require extra care around trajectory boundaries. See
+:ref:`The replay buffer ndim arg and why it doesn't multi-process well
+<data-layout-storage-ndim>` for details.
 
 .. _collectors_replay_trajs:
 
diff --git a/docs/source/reference/collectors_single.rst b/docs/source/reference/collectors_single.rst
index 0796079c692..8190b6dc2f2 100644
--- a/docs/source/reference/collectors_single.rst
+++ b/docs/source/reference/collectors_single.rst
@@ -78,6 +78,9 @@ padding) instead of being yielded.  This is the recommended pattern for
 off-policy training with :class:`~torchrl.data.replay_buffers.SliceSampler`, especially
 with multi-process collectors where fixed-frame batches can silently mix
 episodes.  See :ref:`collectors_replay_trajs` for full details and examples.
+For regular fixed-frame collection from a batched environment, pass
+``flatten_data=True`` to reshape each ``[N, T]`` rollout once and extend the
+recommended 1-D storage with ``N * T`` transitions.
 
 .. note::
     The deprecated collector aliases were removed in v0.13. Construct new
diff --git a/docs/source/reference/data_layout.rst b/docs/source/reference/data_layout.rst
index f411b9b92af..fb7a55423f6 100644
--- a/docs/source/reference/data_layout.rst
+++ b/docs/source/reference/data_layout.rst
@@ -10,6 +10,9 @@ which trajectories are concatenated end-to-end and their boundaries are
 recovered from the per-step ``is_init`` / ``("next", "done")`` /
 ``("next", "truncated")`` / ``("next", "terminated")`` markers, **not**
 from a fixed ``[B, T]`` shape with a padding mask.
+For collectors that write directly to a replay buffer, pass
+``flatten_data=True`` to reshape batched rollouts to this layout before each
+``extend`` call.
 
 Two main patterns coexist in TorchRL:
 
diff --git a/docs/source/reference/data_replaybuffers.rst b/docs/source/reference/data_replaybuffers.rst
index 8ca46081ad5..0a1ce15f697 100644
--- a/docs/source/reference/data_replaybuffers.rst
+++ b/docs/source/reference/data_replaybuffers.rst
@@ -20,6 +20,16 @@ restrictions, and expected performance trade-offs, and
 :ref:`ref_distributed_transport_layouts` for the per-operation layout
 discovery and buffer lifecycle.
 
+.. important::
+
+    TorchRL recommends a flat, 1-D replay-buffer layout (the default
+    ``ndim=1`` storage). Flatten batched collector output before insertion and
+    recover trajectory structure from per-step boundary markers with
+    :class:`~torchrl.data.replay_buffers.SliceSampler`. Higher-dimensional
+    storage is supported when fixed-shape chunks are intentionally part of each
+    replay item, but it is not the preferred general-purpose layout. See
+    :ref:`Data layout: contiguous trajectories <data-layout>`.
+
 .. code-block:: python
 
     from functools import partial
diff --git a/test/test_collectors.py b/test/test_collectors.py
index 169d04b743d..1465c2fd7e2 100644
--- a/test/test_collectors.py
+++ b/test/test_collectors.py
@@ -5683,6 +5683,114 @@ def test_ray_map_fn_and_get_distant_attr(self):
 
 
 class TestCollectorRB:
+    @pytest.mark.parametrize(
+        "collector_class",
+        [Collector, AsyncCollector, MultiSyncCollector, MultiAsyncCollector],
+    )
+    def test_flatten_data_replay_buffer(self, collector_class):
+        """Batched collector rollouts can be written to flat 1-D storage."""
+
+        def make_env():
+            return SerialEnv(2, CountingEnv)
+
+        probe = make_env()
+        policy = RandomPolicy(probe.action_spec)
+        probe.close(raise_if_closed=False)
+        rb = ReplayBuffer(storage=LazyTensorStorage(64), batch_size=4)
+        if collector_class is Collector:
+            create_env_fn = make_env()
+        elif collector_class is AsyncCollector:
+            create_env_fn = make_env
+        else:
+            create_env_fn = [make_env]
+        collector = collector_class(
+            create_env_fn,
+            policy,
+            replay_buffer=rb,
+            flatten_data=True,
+            total_frames=8,
+            frames_per_batch=8,
+        )
+        try:
+            assert all(data is None for data in collector)
+        finally:
+            collector.shutdown()
+
+        assert rb.write_count == 8
+        assert rb[:].shape == (8,)
+        traj_ids = rb["collector", "traj_ids"]
+        torch.testing.assert_close(traj_ids[:4], torch.zeros(4, dtype=torch.long))
+        torch.testing.assert_close(traj_ids[4:], torch.ones(4, dtype=torch.long))
+
+    def test_flatten_data_replay_buffer_requires_extend(self):
+        env = CountingEnv()
+        rb = ReplayBuffer(storage=LazyTensorStorage(64), batch_size=4)
+        try:
+            with pytest.raises(TypeError, match="requires extend_buffer=True"):
+                Collector(
+                    env,
+                    RandomPolicy(env.action_spec),
+                    replay_buffer=rb,
+                    flatten_data=True,
+                    extend_buffer=False,
+                    total_frames=8,
+                    frames_per_batch=8,
+                )
+        finally:
+            env.close(raise_if_closed=False)
+
+    @pytest.mark.parametrize(
+        "collector_class", [Collector, MultiSyncCollector, MultiAsyncCollector]
+    )
+    def test_flatten_data_requires_replay_buffer(self, collector_class):
+        """flatten_data without a replay buffer is an error, not a silent no-op."""
+        env = CountingEnv()
+        try:
+            policy = RandomPolicy(env.action_spec)
+            if collector_class is Collector:
+                create_env_fn = env
+            else:
+                create_env_fn = [CountingEnv]
+            with pytest.raises(TypeError, match="requires a replay buffer"):
+                collector_class(
+                    create_env_fn,
+                    policy,
+                    flatten_data=True,
+                    total_frames=8,
+                    frames_per_batch=8,
+                )
+        finally:
+            env.close(raise_if_closed=False)
+
+    def test_flatten_data_replay_buffer_postproc(self):
+        """postproc runs on the batched [N, T] rollout, before flattening."""
+        env = SerialEnv(2, CountingEnv)
+        rb = ReplayBuffer(storage=LazyTensorStorage(64), batch_size=4)
+        postproc_shapes = []
+
+        def postproc(data):
+            postproc_shapes.append(tuple(data.shape))
+            data["postproc_flag"] = torch.ones(data.shape, dtype=torch.bool)
+            return data
+
+        collector = Collector(
+            env,
+            RandomPolicy(env.action_spec),
+            replay_buffer=rb,
+            flatten_data=True,
+            postproc=postproc,
+            total_frames=8,
+            frames_per_batch=8,
+        )
+        try:
+            assert all(data is None for data in collector)
+        finally:
+            collector.shutdown()
+
+        assert postproc_shapes == [(2, 4)]
+        assert rb[:].shape == (8,)
+        assert rb["postproc_flag"].all()
+
     @pytest.mark.skipif(not _has_gym, reason="requires gym.")
     def test_collector_rb_sync(self):
         env = SerialEnv(8, lambda cp=CARTPOLE_VERSIONED(): GymEnv(cp))
diff --git a/test/test_configs.py b/test/test_configs.py
index 6dda89a8e0c..c6f11593c59 100644
--- a/test/test_configs.py
+++ b/test/test_configs.py
@@ -1194,12 +1194,62 @@ def test_collector_config(self, factory, collector):
                 assert isinstance(collector_instance, MultiSyncCollector)
             elif collector == "multi_async":
                 assert isinstance(collector_instance, MultiAsyncCollector)
+            assert collector_instance.flatten_data is False
             for _c in collector_instance:
                 # Just check that we can iterate
                 break
         finally:
             collector_instance.shutdown(timeout=10)
 
+    @pytest.mark.parametrize("collector", ["async", "multi_sync", "multi_async"])
+    @pytest.mark.skipif(not _has_gymnasium, reason="Gymnasium is not installed")
+    @pytest.mark.skipif(not _has_hydra, reason="Hydra is not installed")
+    def test_collector_config_flatten_data_requires_replay_buffer(self, collector):
+        """flatten_data is forwarded by the config: without a replay buffer the collector raises."""
+        from hydra.errors import InstantiationException
+        from hydra.utils import instantiate
+        from torchrl.trainers.algorithms.configs.collectors import (
+            AsyncCollectorConfig,
+            MultiAsyncCollectorConfig,
+            MultiSyncCollectorConfig,
+        )
+        from torchrl.trainers.algorithms.configs.envs_libs import GymEnvConfig
+        from torchrl.trainers.algorithms.configs.modules import (
+            MLPConfig,
+            TanhNormalModelConfig,
+        )
+
+        env_cfg = GymEnvConfig(env_name="Pendulum-v1")
+        policy_cfg = TanhNormalModelConfig(
+            network=MLPConfig(in_features=3, out_features=2, depth=2, num_cells=32),
+            in_keys=["observation"],
+            out_keys=["action"],
+        )
+        if collector == "async":
+            cfg = AsyncCollectorConfig(
+                create_env_fn=env_cfg,
+                policy=policy_cfg,
+                frames_per_batch=10,
+                flatten_data=True,
+            )
+        elif collector == "multi_sync":
+            cfg = MultiSyncCollectorConfig(
+                create_env_fn=[env_cfg],
+                policy=policy_cfg,
+                frames_per_batch=10,
+                flatten_data=True,
+            )
+        else:
+            cfg = MultiAsyncCollectorConfig(
+                create_env_fn=[env_cfg],
+                policy=policy_cfg,
+                frames_per_batch=10,
+                flatten_data=True,
+            )
+        # Hydra wraps the collector's TypeError in an InstantiationException.
+        with pytest.raises(InstantiationException, match="requires a replay buffer"):
+            instantiate(cfg)
+
     @pytest.mark.parametrize("factory", [True, False])
     @pytest.mark.parametrize("collector", ["async", "multi_sync", "multi_async"])
     @pytest.mark.skipif(not _has_gymnasium, reason="Gymnasium is not installed")
diff --git a/test/test_distributed.py b/test/test_distributed.py
index f6ee0802535..fdfc92ea453 100644
--- a/test/test_distributed.py
+++ b/test/test_distributed.py
@@ -48,7 +48,7 @@
     SamplerWithoutReplacement,
     TensorDictReplayBuffer,
 )
-from torchrl.envs import StepCounter, TransformedEnv
+from torchrl.envs import SerialEnv, StepCounter, TransformedEnv
 from torchrl.modules import RandomPolicy
 from torchrl.modules.inference_server import InferenceServer
 from torchrl.objectives import LossModule
@@ -105,6 +105,10 @@ def __init__(self):
         self.unexpected = nn.Parameter(torch.ones(()))
 
 
+def make_batched_counting_env():
+    return SerialEnv(2, CountingEnv)
+
+
 class ScalarRayLoss(LossModule):
     @dataclass
     class _AcceptedKeys:
@@ -1452,6 +1456,35 @@ def test_ray_owned_inference_and_replay(self):
             if replay is not None:
                 replay.shutdown()
 
+    def test_ray_replay_flatten_data_batched_env(self):
+        replay = TensorDictReplayBuffer(
+            storage=partial(LazyTensorStorage, 100),
+            batch_size=4,
+            service_backend="ray",
+            service_backend_options={"remote_config": {"num_cpus": 0}},
+            transport="auto",
+        )
+        collector = None
+        try:
+            collector = RayCollector(
+                create_env_fn=[make_batched_counting_env],
+                policy=CountingPolicy(),
+                replay_buffer=replay,
+                flatten_data=True,
+                collector_class=Collector,
+                frames_per_batch=8,
+                total_frames=16,
+                remote_configs={"num_cpus": 1, "num_gpus": 0},
+                sync=True,
+            )
+            assert all(batch is None for batch in collector)
+            assert replay.write_count == 16
+            assert replay.sample().shape == (4,)
+        finally:
+            if collector is not None:
+                collector.shutdown()
+            replay.shutdown()
+
     def test_ray_collector_pause_drains_and_resumes(self):
         replay = TensorDictReplayBuffer(
             storage=partial(LazyTensorStorage, 1000),
diff --git a/torchrl/collectors/_base.py b/torchrl/collectors/_base.py
index fc4795618cb..25809aab1ce 100644
--- a/torchrl/collectors/_base.py
+++ b/torchrl/collectors/_base.py
@@ -273,8 +273,7 @@ class BaseCollector(IterableDataset, metaclass=abc.ABCMeta):
             combined with ``replay_buffer`` is supported for
             :class:`~torchrl.collectors.MultiSyncCollector`,
             :class:`~torchrl.collectors.MultiAsyncCollector`,
-            :class:`~torchrl.collectors.distributed.RayCollector`, and
-            :class:`~torchrl.collectors.distributed.RPCCollector`.
+            and :class:`~torchrl.collectors.distributed.RayCollector`.
             Trajectory assembly is delegated to each worker's inner collector,
             which calls :meth:`_iter_by_trajectories` independently and writes
             complete trajectories to the shared replay buffer.  Both the
diff --git a/torchrl/collectors/_multi_base.py b/torchrl/collectors/_multi_base.py
index 18445148f6d..688f221a4a5 100644
--- a/torchrl/collectors/_multi_base.py
+++ b/torchrl/collectors/_multi_base.py
@@ -305,6 +305,15 @@ class MultiCollector(BaseCollector, metaclass=_MultiCollectorMeta):
             for envs without dynamic specs, ``False`` for others.
         replay_buffer (ReplayBuffer, optional): if provided, the collector will not yield tensordicts
             but populate the buffer instead. Defaults to ``None``.
+        flatten_data (bool, optional): if ``True``, flatten each worker
+            rollout before extending the replay buffer. A worker rollout with
+            shape ``[N, T]`` is therefore written as ``[N * T]`` transitions
+            to the recommended flat, 1-D replay-buffer layout. Requires
+            ``replay_buffer`` to be set (a ``TypeError`` is raised otherwise).
+            When ``trajs_per_batch`` is set this option is redundant: complete
+            trajectories are already written to the buffer as flat 1-D
+            sequences. Defaults to ``False`` for backward compatibility with
+            multidimensional replay-buffer storage.
         extend_buffer (bool, optional): if `True`, the replay buffer is extended with entire rollouts and not
             with single steps. Defaults to `True` for multiprocessed data collectors.
         trust_policy (bool, optional): if ``True``, a non-TensorDictModule policy will be trusted to be
@@ -453,6 +462,7 @@ def __init__(
         set_truncated: bool = False,
         use_buffers: bool | None = None,
         replay_buffer: ReplayBuffer | None = None,
+        flatten_data: bool = False,
         extend_buffer: bool = True,
         trust_policy: bool | None = None,
         compile_policy: bool | dict[str, Any] | None = None,
@@ -564,7 +574,7 @@ def __init__(
         self.policy = policy
         self.policy_factory = policy_factory
 
-        self._setup_multi_replay_buffer(replay_buffer, extend_buffer)
+        self._setup_multi_replay_buffer(replay_buffer, flatten_data, extend_buffer)
 
         # Set up weight receivers if provided
         if weight_recv_schemes is not None:
@@ -693,14 +703,26 @@ def _setup_env_kwargs(
     def _setup_multi_replay_buffer(
         self,
         replay_buffer: ReplayBuffer | None,
+        flatten_data: bool,
         extend_buffer: bool,
     ) -> None:
         """Set up replay buffer for multi-process collector."""
         self.local_init_rb = True
+        self.flatten_data = flatten_data
+        self.extend_buffer = extend_buffer
 
-        self._check_replay_buffer_init()
+        if self.flatten_data:
+            if replay_buffer is None:
+                raise TypeError(
+                    "flatten_data=True requires a replay buffer to be passed to the collector. "
+                    "To flatten yielded batches, reshape them at consumption time with data.reshape(-1)."
+                )
+            if not self.extend_buffer:
+                raise TypeError(
+                    "flatten_data=True requires extend_buffer=True when a replay buffer is passed."
+                )
 
-        self.extend_buffer = extend_buffer
+        self._check_replay_buffer_init()
 
         if (
             replay_buffer is not None
@@ -1052,10 +1074,12 @@ def _check_replay_buffer_init(self):
                     **self.create_env_kwargs[0]
                 ).fake_tensordict()
             fake_td = self._add_policy_outputs_to_fake_td(fake_td)
-            if getattr(self, "_worker_trajs_per_batch", None) is not None:
-                # With trajs_per_batch, workers write flat 1-D timesteps to
-                # the buffer.  Initialise the storage as 1-D so that the
-                # shapes match when real trajectories are written.
+            if (
+                getattr(self, "_worker_trajs_per_batch", None) is not None
+                or self.flatten_data
+            ):
+                # These paths write flat 1-D timesteps to the buffer. Initialise
+                # the storage as 1-D so that real writes have the same shape.
                 fake_td = fake_td.reshape(-1)[:1]
                 fake_td["collector", "traj_ids"] = torch.zeros(
                     fake_td.shape, dtype=torch.long
@@ -1395,6 +1419,7 @@ def _run_processes(self) -> None:
                     "set_truncated": self.set_truncated,
                     "use_buffers": self._use_buffers,
                     "replay_buffer": self.replay_buffer,
+                    "flatten_data": self.flatten_data,
                     "extend_buffer": self.extend_buffer,
                     "traj_pool": self._traj_pool,
                     "trust_policy": self.trust_policy,
diff --git a/torchrl/collectors/_runner.py b/torchrl/collectors/_runner.py
index a8b28904c69..49f4031ff9b 100644
--- a/torchrl/collectors/_runner.py
+++ b/torchrl/collectors/_runner.py
@@ -53,6 +53,7 @@ def _main_async_collector(
     set_truncated: bool = False,
     use_buffers: bool | None = None,
     replay_buffer: ReplayBuffer | None = None,
+    flatten_data: bool = False,
     extend_buffer: bool = True,
     traj_pool: _TrajectoryPool = None,
     trust_policy: bool = False,
@@ -100,6 +101,9 @@ def _main_async_collector(
     original_init_random_frames = (
         init_random_frames if init_random_frames is not None else 0
     )
+    # Only forward flatten_data when requested: collector_class may be a custom
+    # collector whose constructor predates (and does not accept) this kwarg.
+    flatten_kwargs = {"flatten_data": True} if flatten_data else {}
     try:
         # When trajs_per_batch is set, _iter_by_trajectories() handles RB writes
         # (with proper padding stripping for 1-D storage). Set _ignore_rb=False so
@@ -151,6 +155,7 @@ def _main_async_collector(
             pre_collect_hook=pre_collect_hook,
             post_collect_hook=post_collect_hook,
             compact_obs=compact_obs,
+            **flatten_kwargs,
         )
         # Set up weight receivers for worker process using the standard register_scheme_receiver API.
         # This properly initializes the schemes on the receiver side and stores them in _receiver_schemes.
diff --git a/torchrl/collectors/_single.py b/torchrl/collectors/_single.py
index 20f4fe692d8..103d17801fb 100644
--- a/torchrl/collectors/_single.py
+++ b/torchrl/collectors/_single.py
@@ -511,6 +511,16 @@ class Collector(BaseCollector, metaclass=_CollectorMeta):
             .. warning:: Using a replay buffer with a `postproc` or `split_trajs=True` requires
                 `extend_buffer=True`, as the whole batch needs to be observed to apply these transforms.
 
+        flatten_data (bool, optional): if ``True``, flatten all collector
+            batch dimensions before extending the replay buffer. For example, a
+            rollout with shape ``[N, T]`` is written as ``[N * T]`` transitions
+            to a flat, 1-D replay buffer instead of ``N`` items whose payload
+            shape is ``[T]``. This is the recommended replay-buffer layout in
+            TorchRL. Requires ``replay_buffer`` to be set (a ``TypeError`` is
+            raised otherwise). When ``trajs_per_batch`` is set this option is
+            redundant: complete trajectories are already written to the buffer
+            as flat 1-D sequences. Defaults to ``False`` for backward
+            compatibility with multidimensional replay-buffer storage.
         extend_buffer (bool, optional): if `True`, the replay buffer is extended with entire rollouts and not
             with single steps. Defaults to `True`.
 
@@ -708,6 +718,7 @@ def __init__(
         set_truncated: bool = False,
         use_buffers: bool | None = None,
         replay_buffer: ReplayBuffer | None = None,
+        flatten_data: bool = False,
         extend_buffer: bool = True,
         trust_policy: bool | None = None,
         compile_policy: bool | dict[str, Any] | None = None,
@@ -785,6 +796,7 @@ def __init__(
         # Set up replay buffer
         self._setup_replay_buffer(
             replay_buffer=replay_buffer,
+            flatten_data=flatten_data,
             extend_buffer=extend_buffer,
             postproc=postproc,
             split_trajs=split_trajs,
@@ -1045,6 +1057,7 @@ def _setup_policy_version_tracking(
     def _setup_replay_buffer(
         self,
         replay_buffer: ReplayBuffer | None,
+        flatten_data: bool,
         extend_buffer: bool,
         postproc: Callable | None,
         split_trajs: bool | None,
@@ -1053,9 +1066,26 @@ def _setup_replay_buffer(
     ) -> None:
         """Set up replay buffer configuration and validate compatibility."""
         self.replay_buffer = replay_buffer
+        self.flatten_data = flatten_data
         self.extend_buffer = extend_buffer
         self.local_init_rb = True
 
+        if self.flatten_data:
+            if self.replay_buffer is None:
+                raise TypeError(
+                    "flatten_data=True requires a replay buffer to be passed to the collector. "
+                    "To flatten yielded batches, reshape them at consumption time with data.reshape(-1)."
+                )
+            if not self.extend_buffer:
+                raise TypeError(
+                    "flatten_data=True requires extend_buffer=True when a replay buffer is passed."
+                )
+            if split_trajs not in (None, False):
+                raise TypeError(
+                    "flatten_data=True is incompatible with split_trajs=True because split trajectories are padded. "
+                    "Use trajs_per_batch for flat complete-trajectory writes."
+                )
+
         # Validate replay buffer compatibility
         if self.replay_buffer is not None and not self._ignore_rb:
             if postproc is not None and not self.extend_buffer:
@@ -1980,6 +2010,13 @@ def is_private(key):
                 key for key in tensordict_out.keys(True) if is_private(key)
             ]
             tensordict_out = tensordict_out.exclude(*excluded_keys, inplace=True)
+        if self.flatten_data and self.replay_buffer is not None:
+            # LLMCollector reuses the `flatten_data` attribute with different
+            # semantics: it flattens yielded rollouts inside rollout() itself
+            # (via view(-1)), so its data is already 1-D here and this reshape
+            # must remain a lazy no-op (reshape(-1) on an already-flat lazy
+            # stack preserves laziness and non-tensor data).
+            tensordict_out = tensordict_out.reshape(-1)
         return tensordict_out
 
     def _update_traj_ids(self, env_output) -> None:
@@ -2278,7 +2315,9 @@ def fake_tensordict(self) -> TensorDictBase:
         - ``set_truncated=True`` last-step ``truncated``/``done`` masking
           applied;
         - ``postproc`` / ``split_trajs`` / private-key exclusion applied,
-          mirroring :meth:`_postproc`.
+          mirroring :meth:`_postproc`;
+        - with ``flatten_data=True`` and a replay buffer, the result is
+          reshaped to the flat 1-D layout written to the buffer.
 
         Intended for storage initialization and ``torch.compile`` /
         cudagraph warmup without having to step the environment first.
diff --git a/torchrl/collectors/distributed/ray.py b/torchrl/collectors/distributed/ray.py
index fe1e8f2aeb1..df85bdcce87 100644
--- a/torchrl/collectors/distributed/ray.py
+++ b/torchrl/collectors/distributed/ray.py
@@ -277,6 +277,15 @@ class RayCollector(BaseCollector):
             fixed-layout TensorDict payloads, ``transport="distributed"`` is
             the recommended data path (Gloo for CPU tensors and NCCL for CUDA
             tensors). Defaults to ``None``.
+        flatten_data (bool, optional): if ``True``, flatten each remote
+            collector rollout before extending the replay buffer. A rollout
+            with shape ``[N, T]`` is written as ``[N * T]`` transitions to the
+            recommended flat, 1-D replay-buffer layout. Requires
+            ``replay_buffer`` to be set (a ``TypeError`` is raised otherwise).
+            When ``trajs_per_batch`` is set this option is redundant: complete
+            trajectories are already written to the buffer as flat 1-D
+            sequences. Defaults to ``False`` for backward compatibility with
+            multidimensional replay-buffer storage.
         weight_updater (WeightUpdaterBase or constructor, optional): (Deprecated) An instance of :class:`~torchrl.collectors.WeightUpdaterBase`
             or its subclass, responsible for updating the policy weights on remote inference workers managed by Ray.
             If not provided, a :class:`~torchrl.collectors.RayWeightUpdater` will be used by default, leveraging
@@ -376,6 +385,7 @@ def __init__(
         update_after_each_batch: bool = False,
         max_weight_update_interval: int = -1,
         replay_buffer: ReplayBuffer | None = None,
+        flatten_data: bool = False,
         weight_updater: WeightUpdaterBase
         | Callable[[], WeightUpdaterBase]
         | None = None,
@@ -426,6 +436,17 @@ def __init__(
                     "actors. For large fixed-layout TensorDict payloads, consider "
                     "transport='distributed'."
                 )
+        if flatten_data:
+            if replay_buffer is None:
+                raise TypeError(
+                    "flatten_data=True requires a replay buffer to be passed to RayCollector. "
+                    "To flatten yielded batches, reshape them at consumption time with data.reshape(-1)."
+                )
+            if isinstance(collector_kwargs, dict):
+                collector_kwargs.setdefault("flatten_data", True)
+            else:
+                for ck in collector_kwargs:
+                    ck.setdefault("flatten_data", True)
         if trajs_per_batch is not None:
             if isinstance(collector_kwargs, dict):
                 collector_kwargs.setdefault("trajs_per_batch", trajs_per_batch)
@@ -544,6 +565,7 @@ def check_list_length_consistency(*lists):
 
         self.no_cuda_sync = no_cuda_sync
         self.replay_buffer = replay_buffer
+        self.flatten_data = flatten_data
         if not isinstance(policy_factory, Sequence):
             policy_factory = [policy_factory] * len(create_env_fn)
         self.policy_factory = policy_factory
diff --git a/torchrl/trainers/algorithms/configs/collectors.py b/torchrl/trainers/algorithms/configs/collectors.py
index cd38967ef4e..ab4638dfbdc 100644
--- a/torchrl/trainers/algorithms/configs/collectors.py
+++ b/torchrl/trainers/algorithms/configs/collectors.py
@@ -63,6 +63,7 @@ class CollectorConfig(BaseCollectorConfig):
     set_truncated: bool = False
     use_buffers: bool | None = None
     replay_buffer: Any = None
+    flatten_data: bool = False
     extend_buffer: bool = True
     trust_policy: bool | None = None
     compile_policy: Any = None
@@ -127,6 +128,7 @@ class AsyncCollectorConfig(BaseCollectorConfig):
     set_truncated: bool = False
     use_buffers: bool = False
     replay_buffer: ConfigBase | None = None
+    flatten_data: bool = False
     extend_buffer: bool = False
     trust_policy: bool = True
     compile_policy: Any = None
@@ -181,6 +183,7 @@ class MultiSyncCollectorConfig(BaseCollectorConfig):
     set_truncated: bool = False
     use_buffers: bool = False
     replay_buffer: ConfigBase | None = None
+    flatten_data: bool = False
     extend_buffer: bool = False
     trust_policy: bool = True
     compile_policy: Any = None
@@ -244,6 +247,7 @@ class MultiAsyncCollectorConfig(BaseCollectorConfig):
     set_truncated: bool = False
     use_buffers: bool = False
     replay_buffer: ConfigBase | None = None
+    flatten_data: bool = False
     extend_buffer: bool = False
     trust_policy: bool = True
     compile_policy: Any = None