Fix non concatenation history observation handling case

source/isaaclab/isaaclab/managers/observation_manager.py

@@ -520,18 +520,20 @@ def combine_group_dims(
 
         For flattened terms (tuples of length 1), it simply sums the single value.
 
+        If concatenation is disabled for the group, returns the original list of dimensions.
+
         Raises a RuntimeError if mixed formats are encountered or if dimensions are incompatible.
         """
+        # If concatenation is disabled, return the original dimensions
+        if not self._group_obs_concatenate[group_name]:
+            return group_term_dims
+
         # Separate flattened and unflattened dimensions.
         flattened = [dims for dims in group_term_dims if len(dims) == 1]
         unflattened = [dims for dims in group_term_dims if len(dims) > 1]
 
         if flattened and unflattened:
-            if self._group_obs_concatenate[group_name]:
-                # Observation shapes should not be mixed if concatenating - raise error.
-                raise RuntimeError(f"In group '{group_name}', mixed dimension formats encountered: {group_term_dims}")
-            else:
-                return group_term_dims
+            raise RuntimeError(f"In group '{group_name}', mixed dimension formats encountered: {group_term_dims}")
 
         if unflattened:
             # Unflattened: dims = (H, d1, d2, ..., d_n)

source/isaaclab/test/managers/test_observation_manager.py

@@ -744,6 +744,86 @@ class PolicyCfg(ObservationGroupCfg):
         with self.assertRaises(ValueError):
             self.obs_man = ObservationManager(cfg, self.env)
 
+    def test_compute_with_non_concatenated_history(self):
+        """Test the observation computation with history buffers when concatenation is disabled."""
+        HISTORY_LENGTH = 5
+
+        @configclass
+        class MyObservationManagerCfg:
+            """Test config class for observation manager."""
+
+            @configclass
+            class PolicyCfg(ObservationGroupCfg):
+                """Test config class for policy observation group with non-concatenated terms."""
+
+                concatenate_terms = False
+
+                term_1 = ObservationTermCfg(
+                    func=grilled_chicken_freerange, history_length=HISTORY_LENGTH, flatten_history_dim=False
+                )
+                term_2 = ObservationTermCfg(
+                    func=lin_vel_w_data, history_length=HISTORY_LENGTH, flatten_history_dim=False
+                )
+
+            policy: ObservationGroupCfg = PolicyCfg()
+
+        # create observation manager
+        cfg = MyObservationManagerCfg()
+        self.obs_man = ObservationManager(cfg, self.env)
+
+        # compute observation using manager
+        observations = self.obs_man.compute()
+
+        # obtain the group observations
+        obs_policy: dict[str, torch.Tensor] = observations["policy"]
+
+        # check the observation shapes
+        self.assertEqual((self.env.num_envs, HISTORY_LENGTH, 4), obs_policy["term_1"].shape)
+        self.assertEqual((self.env.num_envs, HISTORY_LENGTH, 3), obs_policy["term_2"].shape)
+
+        # check initial values
+        expected_term_1_data = torch.arange(4, device=self.env.device).reshape(1, 1, 4).expand(self.env.num_envs, 1, 4)
+        expected_term_1_data = expected_term_1_data.expand(-1, HISTORY_LENGTH, -1)
+        self.assertTrue(torch.equal(expected_term_1_data, obs_policy["term_1"]))
+
+        expected_term_2_data = lin_vel_w_data(self.env).reshape(self.env.num_envs, 1, -1).expand(-1, HISTORY_LENGTH, -1)
+        self.assertTrue(torch.equal(expected_term_2_data, obs_policy["term_2"]))
+
+        # test that history updates correctly for term_1 (should increment)
+        for _ in range(HISTORY_LENGTH - 1):
+            observations = self.obs_man.compute()
+            obs_policy = observations["policy"]
+
+        # The increments that the freerange obs manager would add
+        increments = torch.arange(HISTORY_LENGTH, device=self.env.device).reshape(1, -1, 1)
+        expected_term_1_data = expected_term_1_data + increments
+        self.assertTrue(torch.equal(expected_term_1_data, obs_policy["term_1"]))
+
+        # term_2 should remain constant
+        self.assertTrue(torch.equal(expected_term_2_data, obs_policy["term_2"]))
+
+        # test reset
+        self.obs_man.reset()
+        observations = self.obs_man.compute()
+        obs_policy = observations["policy"]
+
+        # After reset, should be back to initial values
+        expected_term_1_data = torch.arange(4, device=self.env.device).reshape(1, 1, 4).expand(self.env.num_envs, 1, 4)
+        expected_term_1_data = expected_term_1_data.expand(-1, HISTORY_LENGTH, -1)
+        self.assertTrue(torch.equal(expected_term_1_data, obs_policy["term_1"]))
+        self.assertTrue(torch.equal(expected_term_2_data, obs_policy["term_2"]))
+
+        # test reset of specific env ids
+        reset_env_ids = [2, 4, 16]
+        self.obs_man.reset(reset_env_ids)
+        observations = self.obs_man.compute()
+        obs_policy = observations["policy"]
+
+        # Only reset envs should be back to initial values
+        for env_id in reset_env_ids:
+            self.assertTrue(torch.equal(expected_term_1_data[env_id], obs_policy["term_1"][env_id]))
+            self.assertTrue(torch.equal(expected_term_2_data[env_id], obs_policy["term_2"][env_id]))
+
 
 if __name__ == "__main__":
     run_tests()