Fix Multi-Agent Episode concatenation for sequential environments

[pseudo-rnd-thoughts]

Description

For #59508 and #59581 that using TicTacToe would cause the following error

ray::DQN.train() (pid=88183, ip=127.0.0.1, actor_id=2b775f13e808cc4aaaa23bde01000000, repr=DQN(env=<class 'ray.rllib.examples.envs.classes.multi_agent.tic_tac_toe.TicTacToe'>; env-runners=0; learners=0; multi-agent=True))
  File "ray/python/ray/tune/trainable/trainable.py", line 331, in train
    raise skipped from exception_cause(skipped)
  File "ray/python/ray/tune/trainable/trainable.py", line 328, in train
    result = self.step()
  File "ray/python/ray/rllib/algorithms/algorithm.py", line 1242, in step
    train_results, train_iter_ctx = self._run_one_training_iteration()
  File "ray/python/ray/rllib/algorithms/algorithm.py", line 3666, in _run_one_training_iteration
    training_step_return_value = self.training_step()
  File "ray/python/ray/rllib/algorithms/dqn/dqn.py", line 646, in training_step
    return self._training_step_new_api_stack()
  File "ray/python/ray/rllib/algorithms/dqn/dqn.py", line 668, in _training_step_new_api_stack
    self.local_replay_buffer.add(episodes)
  File "ray/python/ray/rllib/utils/replay_buffers/prioritized_episode_buffer.py", line 314, in add
    existing_eps.concat_episode(eps)
  File "ray/python/ray/rllib/env/multi_agent_episode.py", line 862, in concat_episode
    sa_episode.concat_episode(other.agent_episodes[agent_id])
  File "ray/python/ray/rllib/env/single_agent_episode.py", line 618, in concat_episode
    assert self.t == other.t_started
AssertionError

In the multi-agent-episode concat_episode, we check if any agent hasn't received their next observation from an observation, action. This results in a hanging action where in one episode is the observation, action then in the next is the resulting observation, reward, etc. This code check if this has happened then added an extra step at the beginning to include this hanging data.

However, in testing, the multi-agent episode cut method already implements this (if using slice this will cause a hidden bug) meaning that an extra unnecessary step's data is being added resulting in the episode beginnings not lining up.
Therefore, this PR removes this code and replaces with a simple check to assume that the hanging action is equivalent to the initial action in the next episode.

For testing, I found that the concat_episode test was using slice which doesn't account for hanging data while cut which is used in the env-runner does. I modified the test to be more functional based where I created a custom environment that has agents taking actions at different frequencies then returning as an observation the agent's timestep. This allows us to test through concatenating all episodes of the same ID and checking that the observations increase 0, 1, 2, ... and ensures that no data goes missing for users.

[gemini-code-assist]

Code Review

This pull request addresses an AssertionError during multi-agent episode concatenation by removing an unnecessary add_env_step call. The new approach of verifying the consistency of the hanging action with the next episode's initial action is sound.

I've identified a couple of issues:

1. A debug print statement has been left in the code.
2. The action comparison logic is not robust for numpy arrays or nested structures, which could lead to runtime errors.

I've provided suggestions to fix these. Also, as noted in the PR description, adding tests to cover this fix would be crucial to prevent future regressions.

[simonsays1980]

Thanks for fixing this @pseudo-rnd-thoughts ! Small things: could you raise an issue for the slice method to be fixed in the future? And it needs a small change in the docstring - did not came from your change though

[simonsays1980]

This is something we need to fix in the near future. Could you raise another issue on Ray OSS please?

[pseudo-rnd-thoughts]

I'm not sure if this is a bug or an inherent limitation of the slice method

[simonsays1980]

LGTM. Thanks @pseudo-rnd-thoughts !