diff --git a/docs/sphinx/source/en/2-user_guide/4-tasks/2-motion_tracking.md b/docs/sphinx/source/en/2-user_guide/4-tasks/2-motion_tracking.md
index 7b402c533..5bc28206c 100644
--- a/docs/sphinx/source/en/2-user_guide/4-tasks/2-motion_tracking.md
+++ b/docs/sphinx/source/en/2-user_guide/4-tasks/2-motion_tracking.md
@@ -65,8 +65,9 @@ actor keeps the command and anchor-orientation terms at one step while the
`base_ang_vel`, `joint_pos`, `joint_vel`, and `actions` terms declare
`history_length: 5`. `ObservationManager` owns and flattens those per-term
histories; the actor uses the configured encoder-biased joint-position term while
-the critic keeps the clean term. Per-term oldest-first ordering is guarded by
-`tests/scripts/test_obs_alignment_g1_wbt.py`; the hardware-side contract is
+the critic keeps the clean term. Per-term oldest-first ordering is guaranteed by
+the `ObservationManager` per-term history buffers
+(`tests/managers/test_observation_buffers_noise.py`); the hardware-side contract is
documented in the sim-to-real deployment guide. When a Motrix sim2sim replay needs
a checkpoint from another log root, pass the absolute path through `uv run eval`:
diff --git a/docs/sphinx/source/en/3-deployment/1-sim_to_real/2-g1_whole_body.md b/docs/sphinx/source/en/3-deployment/1-sim_to_real/2-g1_whole_body.md
index 60d087cee..fe3f0bf9f 100644
--- a/docs/sphinx/source/en/3-deployment/1-sim_to_real/2-g1_whole_body.md
+++ b/docs/sphinx/source/en/3-deployment/1-sim_to_real/2-g1_whole_body.md
@@ -97,7 +97,8 @@ within the term, and terms are concatenated in declaration order:
The motion command contributes the reference joint position and velocity
(`29 + 29`) ahead of the observation terms. Per-term oldest-first ordering is
-guarded by `tests/scripts/test_obs_alignment_g1_wbt.py`; mirror that ordering
+guaranteed by the `ObservationManager` per-term history buffers
+(`tests/managers/test_observation_buffers_noise.py`); mirror that ordering
on hardware or the policy reads a permuted vector.
## 3. Actuator interface
diff --git a/docs/sphinx/source/en/3-deployment/1-sim_to_real/8-latency_budget.md b/docs/sphinx/source/en/3-deployment/1-sim_to_real/8-latency_budget.md
index 6cecd56ce..8a373e25c 100644
--- a/docs/sphinx/source/en/3-deployment/1-sim_to_real/8-latency_budget.md
+++ b/docs/sphinx/source/en/3-deployment/1-sim_to_real/8-latency_budget.md
@@ -10,7 +10,7 @@ budgets as robot-specific measurements, not UniLab defaults.
| --- | --- | --- |
| One-step action delay | Manager action term `simulate_action_latency` declarations in task owners | Executes the previous action instead of the current action. |
| G1 WBT observation history | Per-term `history_length` in `src/unilab/conf/sac/task/g1_wbt_obs/mujoco.yaml` | Per-term history for `base_ang_vel`, `joint_pos`, `joint_vel`, and `actions`. |
-| Obs history ordering guard | `tests/scripts/test_obs_alignment_g1_wbt.py` | Asserts per-term oldest-first flatten for the G1 WBT actor obs. |
+| Obs history ordering | `ObservationManager` per-term history buffers (`tests/managers/test_observation_buffers_noise.py`) | Per-term oldest-first flatten for the G1 WBT actor obs. |
## Action Latency
diff --git a/docs/sphinx/source/zh_CN/2-user_guide/4-tasks/2-motion_tracking.md b/docs/sphinx/source/zh_CN/2-user_guide/4-tasks/2-motion_tracking.md
index c8b279f49..c4acb0154 100644
--- a/docs/sphinx/source/zh_CN/2-user_guide/4-tasks/2-motion_tracking.md
+++ b/docs/sphinx/source/zh_CN/2-user_guide/4-tasks/2-motion_tracking.md
@@ -61,7 +61,8 @@ uv run train --algo sac --task g1_wbt_obs --sim mujoco training.use_amp=true
orientation term 保持单步，`base_ang_vel`、`joint_pos`、`joint_vel` 和 `actions` term
分别声明 `history_length: 5`。这些逐项历史由 `ObservationManager` 维护并展开；actor
使用配置中的 encoder-biased joint-position term，critic 则保留 clean term。逐项最旧
-优先顺序由 `tests/scripts/test_obs_alignment_g1_wbt.py` 守护；硬件侧契约见仿真到真机
+优先顺序由 `ObservationManager` 的逐项历史缓冲实现保证
+（`tests/managers/test_observation_buffers_noise.py`）；硬件侧契约见仿真到真机
部署指南。当 Motrix sim2sim 回放需要引用其他日志根目录下的 checkpoint 时，用
`uv run eval` 透传绝对路径：
diff --git a/docs/sphinx/source/zh_CN/3-deployment/1-sim_to_real/2-g1_whole_body.md b/docs/sphinx/source/zh_CN/3-deployment/1-sim_to_real/2-g1_whole_body.md
index 52567a1b3..d1ced945e 100644
--- a/docs/sphinx/source/zh_CN/3-deployment/1-sim_to_real/2-g1_whole_body.md
+++ b/docs/sphinx/source/zh_CN/3-deployment/1-sim_to_real/2-g1_whole_body.md
@@ -91,7 +91,8 @@ Actor 观测宽度是 `env.observations.actor.terms` 下各项 `dim * history_le
```
motion command 在观测项之前贡献参考关节位置与速度（`29 + 29`）。逐项的最旧优先
-顺序由 `tests/scripts/test_obs_alignment_g1_wbt.py` 守护；硬件侧必须镜像该顺序，
+顺序由 `ObservationManager` 的逐项历史缓冲实现保证
+（`tests/managers/test_observation_buffers_noise.py`）；硬件侧必须镜像该顺序，
否则策略读到的是被置换过的向量。
## 3. 执行器接口
diff --git a/docs/sphinx/source/zh_CN/3-deployment/1-sim_to_real/8-latency_budget.md b/docs/sphinx/source/zh_CN/3-deployment/1-sim_to_real/8-latency_budget.md
index b2acebb6c..33d8814d9 100644
--- a/docs/sphinx/source/zh_CN/3-deployment/1-sim_to_real/8-latency_budget.md
+++ b/docs/sphinx/source/zh_CN/3-deployment/1-sim_to_real/8-latency_budget.md
@@ -9,7 +9,7 @@
| --- | --- | --- |
| 单步动作延迟 | task owner 中 Manager action term 的 `simulate_action_latency` 声明 | 执行上一步动作而非当前动作。 |
| G1 WBT 观测历史 | `src/unilab/conf/sac/task/g1_wbt_obs/mujoco.yaml` 中逐 term 的 `history_length` | 为 `base_ang_vel`、`joint_pos`、`joint_vel` 与 `actions` 提供逐项历史。 |
-| 观测历史顺序守护 | `tests/scripts/test_obs_alignment_g1_wbt.py` | 断言 G1 WBT actor 观测按逐项最旧优先展平。 |
+| 观测历史顺序 | `ObservationManager` 逐项历史缓冲（`tests/managers/test_observation_buffers_noise.py`） | G1 WBT actor 观测按逐项最旧优先展平。 |
## 动作延迟
diff --git a/src/unilab/tasks/compatibility.py b/src/unilab/tasks/compatibility.py
deleted file mode 100644
index a33106b20..000000000
--- a/src/unilab/tasks/compatibility.py
+++ /dev/null
@@ -1,132 +0,0 @@
-"""Internal, cold-path compatibility seam for legacy task factories.
-
-This module is intentionally task-owned and is not part of the base registry
-contract. It only admits legacy factories that already use UniLab's
-``EnvCfg -> NpEnv`` lifecycle; it does not provide a fallback runtime.
-"""
-
-from __future__ import annotations
-
-from dataclasses import dataclass
-from enum import Enum
-from typing import Generic, Protocol, TypeVar
-
-from unilab.base.base import ABEnv, EnvCfg
-from unilab.base.np_env import NpEnv
-
-
-class CompatibilityStatus(str, Enum):
- """Documented outcome for one legacy compatibility boundary."""
-
- ADAPTED = "Adapted"
- UNSUPPORTED = "Unsupported"
-
-
-@dataclass(frozen=True)
-class LegacyTaskCompatibility:
- """Immutable compatibility evidence attached to one task-family seam."""
-
- task_family: str
- status: CompatibilityStatus
- reason: str
-
- def __post_init__(self) -> None:
- if not self.task_family.strip():
- raise ValueError("legacy compatibility task_family must be non-empty")
- if not self.reason.strip():
- raise ValueError("legacy compatibility reason must be non-empty")
-
-
-TCfg_contra = TypeVar("TCfg_contra", bound=EnvCfg, contravariant=True)
-
-
-class LegacyEnvFactory(Protocol[TCfg_contra]):
- """Existing registry-shaped legacy factory admitted by this seam."""
-
- def __call__(
- self,
- cfg: TCfg_contra,
- *,
- num_envs: int = 1,
- backend_type: str = "mujoco",
- ) -> ABEnv: ...
-
-
-@dataclass(frozen=True)
-class LegacyFactoryAdapter(Generic[TCfg_contra]):
- """Validate one legacy factory at the existing env construction boundary."""
-
- factory: LegacyEnvFactory[TCfg_contra]
- compatibility: LegacyTaskCompatibility
-
- def __post_init__(self) -> None:
- if self.compatibility.status is not CompatibilityStatus.ADAPTED:
- raise ValueError("LegacyFactoryAdapter compatibility status must be Adapted")
-
- def __call__(
- self,
- cfg: TCfg_contra,
- *,
- num_envs: int = 1,
- backend_type: str = "mujoco",
- ) -> NpEnv:
- family = self.compatibility.task_family
- if not isinstance(cfg, EnvCfg):
- raise TypeError(
- f"Legacy task family '{family}' expected EnvCfg, received {type(cfg).__name__}"
- )
-
- env = self.factory(cfg, num_envs=num_envs, backend_type=backend_type)
- if not isinstance(env, ABEnv):
- raise TypeError(
- f"Legacy task family '{family}' factory returned {type(env).__name__}, "
- "expected ABEnv"
- )
- if not isinstance(env, NpEnv):
- raise TypeError(
- f"Legacy task family '{family}' compatibility is Unsupported: "
- f"{type(env).__name__} does not use the NpEnv lifecycle"
- )
- return env
-
-
-def adapt_legacy_factory(
- factory: LegacyEnvFactory[TCfg_contra],
- *,
- task_family: str,
- reason: str,
-) -> LegacyFactoryAdapter[TCfg_contra]:
- """Mark and wrap an existing ``EnvCfg -> NpEnv`` task factory.
-
- The wrapper runs only while the registry constructs an environment. It
- forwards the registry's fixed arguments exactly once and rejects any
- other config or runtime shape instead of probing or falling back.
- """
-
- if not callable(factory):
- raise TypeError(f"legacy task family '{task_family}' factory must be callable")
- compatibility = LegacyTaskCompatibility(
- task_family=task_family,
- status=CompatibilityStatus.ADAPTED,
- reason=reason,
- )
- return LegacyFactoryAdapter(factory=factory, compatibility=compatibility)
-
-
-def unsupported_legacy_task(*, task_family: str, reason: str) -> LegacyTaskCompatibility:
- """Record an explicit unsupported surface without creating a factory."""
-
- return LegacyTaskCompatibility(
- task_family=task_family,
- status=CompatibilityStatus.UNSUPPORTED,
- reason=reason,
- )
-
-
-__all__ = [
- "CompatibilityStatus",
- "LegacyFactoryAdapter",
- "LegacyTaskCompatibility",
- "adapt_legacy_factory",
- "unsupported_legacy_task",
-]
diff --git a/src/unilab/tasks/migration_matrix.py b/src/unilab/tasks/migration_matrix.py
index 366868f73..0e190ad3c 100644
--- a/src/unilab/tasks/migration_matrix.py
+++ b/src/unilab/tasks/migration_matrix.py
@@ -11,7 +11,7 @@
from typing import Literal
MigrationStatus = Literal["Compatible", "Adapted"]
-MigrationTarget = Literal["complete", "mba", "compatibility"]
+MigrationTarget = Literal["complete", "mba"]
@dataclass(frozen=True)
diff --git a/tests/algos/test_appo_runner.py b/tests/algos/test_appo_runner.py
index 0ab1ad143..496979fd1 100644
--- a/tests/algos/test_appo_runner.py
+++ b/tests/algos/test_appo_runner.py
@@ -26,23 +26,6 @@
from unilab.structured_configs import APPOConfig
-@pytest.mark.slow
-def test_appo_runner_init_no_crash(mock_env_name):
- cfg = APPOConfig().to_dict()
- cfg["num_envs"] = 4
- cfg["steps_per_env"] = 4
-
- runner = APPORunner(
- env_name=mock_env_name,
- env_factory=registry_env_factory(mock_env_name, "mujoco"),
- env_cfg_overrides={},
- rl_cfg=cfg,
- num_envs=4,
- steps_per_env=4,
- )
- runner.close()
-
-
@pytest.mark.slow
@pytest.mark.parametrize("env_name", ["Go2JoystickFlat"])
def test_appo_runner_learn_two_iterations(env_name):
diff --git a/tests/algos/test_offpolicy_double_buffer_runner.py b/tests/algos/test_offpolicy_double_buffer_runner.py
index 733c36ee7..a1dd5706c 100644
--- a/tests/algos/test_offpolicy_double_buffer_runner.py
+++ b/tests/algos/test_offpolicy_double_buffer_runner.py
@@ -11,7 +11,6 @@
import pytest
from hydra import compose, initialize_config_dir
from hydra.core.global_hydra import GlobalHydra
-from hydra.errors import ConfigCompositionException
from uni_rl.ipc.dp_launcher import UNILAB_DP_LOG_DIR, UNILAB_DP_RANK, UNILAB_DP_WORLD_SIZE
_ROOT = Path(__file__).parent.parent.parent
@@ -83,35 +82,6 @@ def test_offpolicy_config_has_one_replay_path():
cfg = _offpolicy_cfg()
assert cfg.training.replay_prefetch_mode == "one_tick"
assert cfg.training.env_steps_per_sync == 1
- assert "env_steps_per_sync" not in cfg.algo
- assert "inference_owner" not in cfg.training
- assert "collector_infer_device" not in cfg.training
- assert "no_sync_collection" not in cfg.training
- assert "replay_pipeline" not in cfg.training
- assert "verbose_metrics" not in cfg.training
- assert "replay_pack_layout" not in cfg.training
- assert "replay_pack_executor" not in cfg.training
- assert "replay_h2d_submitter" not in cfg.training
-
-
-@pytest.mark.parametrize(
- "override",
- [
- "training.replay_pipeline=cpu_pinned_double_buffer",
- "training.verbose_metrics=true",
- "training.num_gpus=2",
- "training.multi_gpu_sync_mode=sync_sgd",
- "training.multi_gpu_sync_interval=2",
- "training.device=cuda",
- "training.inference_owner=collector",
- "training.collector_infer_device=cpu",
- "training.no_sync_collection=true",
- "algo.env_steps_per_sync=2",
- ],
-)
-def test_removed_offpolicy_options_fail_hydra_compose(override: str):
- with pytest.raises(ConfigCompositionException, match="Could not override"):
- _offpolicy_cfg([override])
@pytest.mark.parametrize("mode", ["invalid_mode", "same_tick"])
@@ -160,11 +130,6 @@ def test_sac_dispatch_constructs_unique_runner(monkeypatch: pytest.MonkeyPatch):
assert runner.kwargs["algo_type"] == "sac"
assert runner.kwargs["device"] == "cuda:0"
assert runner.kwargs["replay_prefetch_mode"] == "one_tick"
- assert "inference_owner" not in runner.kwargs
- assert "collector_infer_device" not in runner.kwargs
- assert "sync_collection" not in runner.kwargs
- assert "replay_pipeline" not in runner.kwargs
- assert "verbose_metrics" not in runner.kwargs
assert runner.kwargs["learner"].kwargs == {
"device": "cuda:0",
"obs_dim": 4,
@@ -252,10 +217,6 @@ def test_td3_dispatch_constructs_unique_runner(monkeypatch: pytest.MonkeyPatch):
assert isinstance(runner, _FakeRunner)
assert runner.kwargs["algo_type"] == "td3"
assert runner.kwargs["device"] == "cuda:0"
- assert "inference_owner" not in runner.kwargs
- assert "collector_infer_device" not in runner.kwargs
- assert "sync_collection" not in runner.kwargs
- assert "replay_pipeline" not in runner.kwargs
assert runner.kwargs["replay_prefetch_mode"] == "one_tick"
nan_guard_cfg = runner.kwargs["nan_guard_cfg"]
assert nan_guard_cfg.enabled is True
@@ -326,10 +287,6 @@ def test_flashsac_dispatch_constructs_unique_runner(monkeypatch: pytest.MonkeyPa
assert runner.kwargs["algo_type"] == "flashsac"
assert runner.kwargs["device"] == "cuda:0"
assert runner.kwargs["replay_prefetch_mode"] == "one_tick"
- assert "inference_owner" not in runner.kwargs
- assert "collector_infer_device" not in runner.kwargs
- assert "sync_collection" not in runner.kwargs
- assert "replay_pipeline" not in runner.kwargs
def test_flashsac_n_step_is_rejected():
diff --git a/tests/algos/test_offpolicy_dp_sync.py b/tests/algos/test_offpolicy_dp_sync.py
index 3dd427dac..b4ca8ef37 100644
--- a/tests/algos/test_offpolicy_dp_sync.py
+++ b/tests/algos/test_offpolicy_dp_sync.py
@@ -153,15 +153,6 @@ def test_learner_without_initial_sync_tensors_fails_with_type_error():
runner._dp_init_broadcast()
-def test_close_closes_dp_sync_idempotently():
- dp_sync = _FakeDpSync()
- runner = _runner_with(_SyncLearner(), dp_sync)
- # Avoid the full AsyncRunner.close(); only the dp_sync branch is under test.
- runner.dp_sync.close()
- runner.dp_sync.close()
- assert [name for name, _ in dp_sync.calls] == ["close", "close"]
-
-
def test_close_restores_terminal_and_ipc_before_destroying_process_group(monkeypatch):
from uni_rl.offpolicy.runner import OffPolicyRunner
@@ -499,11 +490,6 @@ def __init__(self, *args, **kwargs):
return runner.kwargs
-def test_offpolicy_config_has_no_periodic_parameter_sync_interval():
- cfg = _offpolicy_cfg()
- assert "dp_sync_interval" not in cfg.training
-
-
def test_build_runner_single_rank_keeps_dp_sync_none(monkeypatch: pytest.MonkeyPatch):
monkeypatch.delenv(UNILAB_DP_RANK, raising=False)
kwargs = _build_sac_runner_with_dp_fakes(monkeypatch, [])
diff --git a/tests/algos/test_rsl_rl_runner.py b/tests/algos/test_rsl_rl_runner.py
index a86020b35..95549844d 100644
--- a/tests/algos/test_rsl_rl_runner.py
+++ b/tests/algos/test_rsl_rl_runner.py
@@ -21,90 +21,16 @@
)
rsl_rl = pytest.importorskip("rsl_rl")
-import numpy as np
-import torch
-from tensordict import TensorDict
-from uni_rl.algos.rsl_rl import normalize_ppo_train_cfg
+from uni_rl.algos.rsl_rl import RslRlVecEnvWrapper, normalize_ppo_train_cfg
from unilab.base import registry
from unilab.base.config_adapter import BackendAdapter
from unilab.base.registry import ensure_registries
from unilab.structured_configs import PPOConfig
-from unilab.utils.tensor import to_torch
ensure_registries()
-# ---------------------------------------------------------------------------
-# Minimal wrapper (same as src/unilab/scripts/train_rsl_rl.py)
-# ---------------------------------------------------------------------------
-
-
-class _RslRlVecEnvWrapper:
- """Lightweight RSL-RL wrapper for testing."""
-
- def __init__(self, env, device="cpu"):
- self.env = env
- self.cfg = env.cfg
- self.device = device
- self.num_envs = env.num_envs
- self.observation_space = env.observation_space
- self.action_space = env.action_space
- self.num_obs = int(env.obs_groups_spec["obs"])
- self.num_privileged_obs = int(env.obs_groups_spec.get("critic", self.num_obs))
- self.num_actions = env.action_space.shape[0]
-
- self.episode_returns = torch.zeros(self.num_envs, device=device)
- self.episode_lengths = torch.zeros(self.num_envs, device=device)
- self.episode_length_buf = self.episode_lengths
- self.max_episode_length = np.ceil(env.cfg.max_episode_seconds / env.cfg.ctrl_dt)
- self.reset()
-
- def _obs_to_tensordict(self, obs: dict[str, np.ndarray]) -> TensorDict:
- actor = to_torch(obs["obs"], self.device)
- td = {"actor": actor, "policy": actor}
- if "critic" in obs:
- td["critic"] = to_torch(obs["critic"], self.device)
- return TensorDict(td, batch_size=self.num_envs, device=self.device)
-
- def step(self, actions):
- actions_np = (
- actions.detach().cpu().numpy() if isinstance(actions, torch.Tensor) else actions
- )
- state = self.env.step(actions_np)
- rewards = to_torch(state.reward, self.device)
- dones = to_torch(state.terminated | state.truncated, self.device).bool()
- self.episode_returns += rewards
- self.episode_lengths += 1
- infos = {}
- done_idx = torch.nonzero(dones).flatten()
- if len(done_idx) > 0:
- infos["time_outs"] = to_torch(state.truncated, self.device).bool()
- self.episode_returns[done_idx] = 0
- self.episode_lengths[done_idx] = 0
- if "log" in state.info:
- infos["log"] = state.info["log"]
- return self._obs_to_tensordict(state.obs), rewards, dones, infos
-
- def reset(self):
- if self.env.state is None:
- self.env.init_state()
- env_indices = np.arange(self.num_envs, dtype=np.int32)
- obs_out, _ = self.env.reset(env_indices)
- self.episode_returns[:] = 0
- self.episode_lengths[:] = 0
- return self._obs_to_tensordict(obs_out), {}
-
- def get_observations(self):
- assert self.env.state is not None
- return self._obs_to_tensordict(self.env.state.obs)
-
- def get_privileged_observations(self):
- assert self.env.state is not None
- obs = self.env.state.obs
- return to_torch(obs.get("critic", obs["obs"]), self.device)
-
-
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
@@ -147,7 +73,7 @@ def test_rsl_rl_ppo_one_iteration(
sim_backend="mujoco",
env_cfg_override=env_cfg_override,
)
- wrapped = _RslRlVecEnvWrapper(env, device="cpu")
+ wrapped = RslRlVecEnvWrapper(env, device="cpu")
cfg = PPOConfig()
train_cfg = cfg.to_dict()
diff --git a/tests/base/backend/test_drake_batch_pool.py b/tests/base/backend/test_drake_batch_pool.py
index 1b2766ec4..59f07bc90 100644
--- a/tests/base/backend/test_drake_batch_pool.py
+++ b/tests/base/backend/test_drake_batch_pool.py
@@ -109,43 +109,6 @@ def make_go1_pool(nbatch, nthread):
"""
-def test_batch_import_diagnostic_is_preserved() -> None:
- output = _run_clean_python(
- """
- import json
-
- try:
- from drake_uni.batch_env import batch_available, batch_import_error
- except ImportError as exc:
- captured_error = exc
-
- def batch_available():
- return False
-
- def batch_import_error():
- return captured_error
-
- error = batch_import_error()
- summary = {
- "available": bool(batch_available()),
- "error_type": None if error is None else type(error).__name__,
- "missing_module": getattr(error, "name", None),
- }
- print(json.dumps(summary, sort_keys=True))
- """
- )
- summary = json.loads(output.strip().splitlines()[-1])
- if summary["available"]:
- assert summary["error_type"] is None
- assert summary["missing_module"] is None
- else:
- assert summary["error_type"] == "ModuleNotFoundError"
- assert summary["missing_module"] in {
- "drake_uni",
- "drake_uni.compiled._drake_env_pool",
- }
-
-
def test_batch_backend_mode_rejects_existing_pydrake_module() -> None:
output = _run_clean_python(
"""
@@ -284,41 +247,6 @@ def test_drake_uni_runtime_import_is_lazy_and_pydrake_free() -> None:
}
-def test_unilab_drake_public_surface_excludes_batch_backend_symbol() -> None:
- output = _run_clean_python(
- """
- import json
-
- import unilab.base.backend_factory as backend_root
- import unisim.backend.drake as drake_pkg
- from unisim.backend.drake import backend as backend_module
-
- try:
- from unisim.backend.drake.backend import DrakeUniBatchBackend # noqa: F401
- except ImportError:
- direct_import = "failed"
- else:
- direct_import = "succeeded"
-
- summary = {
- "direct_import": direct_import,
- "root_has_batch": hasattr(backend_root, "DrakeUniBatchBackend"),
- "subpackage_has_batch": hasattr(drake_pkg, "DrakeUniBatchBackend"),
- "module_has_batch": hasattr(backend_module, "DrakeUniBatchBackend"),
- "module_all_has_batch": "DrakeUniBatchBackend" in backend_module.__all__,
- }
- print(json.dumps(summary, sort_keys=True))
- """
- )
- assert json.loads(output.strip().splitlines()[-1]) == {
- "direct_import": "failed",
- "module_all_has_batch": False,
- "module_has_batch": False,
- "root_has_batch": False,
- "subpackage_has_batch": False,
- }
-
-
@pytest.mark.skipif(
not _batch_extension_built(),
reason="optional Drake batch extension has not been built",
@@ -336,7 +264,6 @@ def test_drake_batch_pool_go1_smoke_shapes_and_time() -> None:
output = pool.step(state, 2, control, None, True)
sensor_data = output["sensor_data"]
summary = {
- "forward_removed": not hasattr(pool, "forward"),
"state_only_has_sensor_data": "sensor_data" in state_only,
"state_shape": list(output["state"].shape),
"sensor_shape": list(sensor_data.shape),
@@ -355,7 +282,6 @@ def test_drake_batch_pool_go1_smoke_shapes_and_time() -> None:
summary = json.loads(output.strip().splitlines()[-1])
assert summary.pop("num_filtered_geometries") > 0
assert summary == {
- "forward_removed": True,
"has_sensor_data": True,
"nthread": 1,
"sensor_finite": True,
diff --git a/tests/base/test_body_state_copy.py b/tests/base/test_body_state_copy.py
deleted file mode 100644
index 28a06e290..000000000
--- a/tests/base/test_body_state_copy.py
+++ /dev/null
@@ -1,48 +0,0 @@
-"""Focused parity tests for the shared body-state copy kernel."""
-
-from __future__ import annotations
-
-import numpy as np
-from unisim.backend.body_state import copy_selected_body_state
-
-
-def _outputs(num_envs: int, num_selected: int) -> tuple[np.ndarray, ...]:
- shape = (num_envs, num_selected, 3)
- return (
- np.empty(shape, dtype=np.float32),
- np.empty((num_envs, num_selected, 4), dtype=np.float32),
- np.empty(shape, dtype=np.float32),
- np.empty(shape, dtype=np.float32),
- )
-
-
-def test_shared_body_state_copy_kernel_preserves_selection_and_outputs() -> None:
- rng = np.random.default_rng(7)
- num_envs, num_bodies = 257, 9
- pos = rng.standard_normal((num_envs, num_bodies, 3), dtype=np.float32)
- quat = rng.standard_normal((num_envs, num_bodies, 4), dtype=np.float32)
- lin_vel = rng.standard_normal((num_envs, num_bodies, 3), dtype=np.float32)
- ang_vel = rng.standard_normal((num_envs, num_bodies, 3), dtype=np.float32)
- selected = np.asarray([7, 1, 5], dtype=np.intp)
- out_pos, out_quat, out_lin_vel, out_ang_vel = _outputs(num_envs, len(selected))
-
- copy_selected_body_state(
- pos,
- quat,
- lin_vel,
- ang_vel,
- selected,
- out_pos,
- out_quat,
- out_lin_vel,
- out_ang_vel,
- )
-
- np.testing.assert_array_equal(out_pos, pos[:, selected])
- np.testing.assert_array_equal(out_quat, quat[:, selected])
- np.testing.assert_array_equal(out_lin_vel, lin_vel[:, selected])
- np.testing.assert_array_equal(out_ang_vel, ang_vel[:, selected])
- # The shared helper intentionally stays NumPy-only so the core package has
- # no mandatory Numba dependency. Backend-specific compiled kernels belong
- # to their optional adapter extras.
- assert not hasattr(copy_selected_body_state, "targetoptions")
diff --git a/tests/base/test_dr_legacy_removed.py b/tests/base/test_dr_legacy_removed.py
deleted file mode 100644
index dbc990d02..000000000
--- a/tests/base/test_dr_legacy_removed.py
+++ /dev/null
@@ -1,36 +0,0 @@
-from __future__ import annotations
-
-import importlib
-import subprocess
-import sys
-from pathlib import Path
-
-import pytest
-
-
-def test_legacy_dr_protocol_is_removed() -> None:
- import unilab.base.np_env as np_env
-
- assert not hasattr(np_env.NpEnv, "_init_domain_randomization")
- with pytest.raises(ModuleNotFoundError, match="No module named 'unilab.dr'"):
- importlib.import_module("unilab.dr")
-
- source = Path(np_env.__file__).read_text(encoding="utf-8")
- assert "_dr_manager" not in source
- assert "unilab.dr" not in source
-
-
-def test_fresh_import_graph_does_not_load_legacy_dr_manager() -> None:
- code = "\n".join(
- [
- "import sys",
- "import unilab",
- "assert 'unilab.dr' not in sys.modules",
- "assert 'unilab.dr.manager' not in sys.modules",
- "assert 'unilab.dr.provider' not in sys.modules",
- ]
- )
- result = subprocess.run(
- [sys.executable, "-c", code], capture_output=True, text=True, check=False
- )
- assert result.returncode == 0, result.stderr
diff --git a/tests/base/test_mjwarp_backend.py b/tests/base/test_mjwarp_backend.py
index e31378445..0aa94cb21 100644
--- a/tests/base/test_mjwarp_backend.py
+++ b/tests/base/test_mjwarp_backend.py
@@ -359,9 +359,6 @@ def test_set_state_returns_schema_conformant_timing() -> None:
assert timing["set_state_host_cache_refresh_ms"] > 0.0
assert timing["set_state_mask_ms"] == 0.0
assert timing["set_state_pool_reset_ms"] == 0.0
- # Legacy collapsed keys are gone.
- assert "set_state_reset_ms" not in timing
- assert "set_state_cache_refresh_ms" not in timing
empty = backend.set_state(
np.asarray([], dtype=np.int32),
diff --git a/tests/base/test_mjwarp_cuda_graph.py b/tests/base/test_mjwarp_cuda_graph.py
deleted file mode 100644
index ad5ed0e1d..000000000
--- a/tests/base/test_mjwarp_cuda_graph.py
+++ /dev/null
@@ -1,243 +0,0 @@
-"""Cold-path CUDA graph eligibility, capture, replay, and fallback tests."""
-
-from __future__ import annotations
-
-from typing import Any
-
-import pytest
-from unisim.backend.mjwarp.backend import (
- MjwarpBackend,
- _cuda_graph_eligibility,
- _reset_scratch_capacity_for_batch,
-)
-
-
-class _FakeDevice:
- def __init__(self, *, is_cuda: bool = True) -> None:
- self.is_cuda = is_cuda
-
-
-class _FakeContext:
- def __enter__(self) -> "_FakeContext":
- return self
-
- def __exit__(self, *_args: Any) -> None:
- return None
-
-
-class _FakeCapture(_FakeContext):
- def __init__(self, graph: str, *, failure: Exception | None = None) -> None:
- self.graph = graph
- self._failure = failure
-
- def __enter__(self) -> "_FakeCapture":
- if self._failure is not None:
- raise self._failure
- return self
-
-
-class _FakeWarp:
- def __init__(
- self,
- *,
- driver: tuple[int, int] | None = (13, 0),
- mempool: bool = True,
- fail_capture_index: int | None = None,
- ) -> None:
- self.driver = driver
- self.mempool = mempool
- self.fail_capture_index = fail_capture_index
- self.capture_count = 0
- self.launches: list[str] = []
-
- def get_cuda_driver_version(self) -> tuple[int, int] | None:
- return self.driver
-
- def is_mempool_enabled(self, _device: Any) -> bool:
- return self.mempool
-
- def ScopedDevice(self, _device: Any) -> _FakeContext: # noqa: N802
- return _FakeContext()
-
- def ScopedCapture(self) -> _FakeCapture: # noqa: N802
- capture_index = self.capture_count
- self.capture_count += 1
- failure = (
- RuntimeError("synthetic capture failure")
- if capture_index == self.fail_capture_index
- else None
- )
- return _FakeCapture(f"graph-{capture_index}", failure=failure)
-
- def capture_launch(self, graph: str) -> None:
- self.launches.append(graph)
-
-
-class _FakeMujocoWarp:
- def __init__(self) -> None:
- self.step_calls = 0
- self.forward_calls = 0
- self.reset_calls = 0
-
- def step(self, _model: Any, _data: Any) -> None:
- self.step_calls += 1
-
- def forward(self, _model: Any, _data: Any) -> None:
- self.forward_calls += 1
-
- def reset_data(self, _model: Any, _data: Any, *, reset: Any) -> None:
- del reset
- self.reset_calls += 1
-
-
-def _backend(warp: _FakeWarp, mujoco_warp: _FakeMujocoWarp) -> MjwarpBackend:
- backend = MjwarpBackend.__new__(MjwarpBackend)
- backend._warp = warp
- backend._mujoco_warp = mujoco_warp
- backend._device_model = object()
- backend._device_data = object()
- backend._reset_mask_device = object()
- backend._reset_scratch_capacity = 0
- backend._reset_scratch_data = None
- return backend
-
-
-@pytest.mark.parametrize(
- ("device", "driver", "mempool", "expected_reason"),
- [
- (_FakeDevice(is_cuda=False), (13, 0), True, "not CUDA"),
- (_FakeDevice(), None, True, "unavailable"),
- (_FakeDevice(), (12, 3), True, "older than 12.4"),
- (_FakeDevice(), (12, 4), False, "mempool is disabled"),
- ],
-)
-def test_cuda_graph_eligibility_fails_closed(
- device: _FakeDevice,
- driver: tuple[int, int] | None,
- mempool: bool,
- expected_reason: str,
-) -> None:
- eligible, reason = _cuda_graph_eligibility(
- _FakeWarp(driver=driver, mempool=mempool),
- device,
- )
-
- assert eligible is False
- assert reason is not None and expected_reason in reason
-
-
-def test_cuda_graph_capture_replays_fixed_address_operations() -> None:
- warp = _FakeWarp(driver=(12, 4), mempool=True)
- mujoco_warp = _FakeMujocoWarp()
- backend = _backend(warp, mujoco_warp)
-
- backend._initialize_cuda_graphs(_FakeDevice())
-
- assert backend._cuda_graph_enabled is True
- assert backend._cuda_graph_disable_reason is None
- assert (backend._step_graph, backend._forward_graph, backend._reset_graph) == (
- "graph-0",
- "graph-1",
- "graph-2",
- )
- capture_calls = (
- mujoco_warp.step_calls,
- mujoco_warp.forward_calls,
- mujoco_warp.reset_calls,
- )
-
- backend._execute_device_steps(3)
- backend._execute_device_reset()
- backend._execute_device_forward()
-
- assert warp.launches == ["graph-0", "graph-0", "graph-0", "graph-2", "graph-1"]
- assert (
- mujoco_warp.step_calls,
- mujoco_warp.forward_calls,
- mujoco_warp.reset_calls,
- ) == capture_calls
-
-
-def test_cuda_graph_capture_includes_materialized_reset_scratch() -> None:
- warp = _FakeWarp(driver=(12, 4), mempool=True)
- mujoco_warp = _FakeMujocoWarp()
- backend = _backend(warp, mujoco_warp)
- backend._reset_scratch_capacity = 4
- backend._reset_scratch_data = object()
- backend._reset_scratch_mask_device = object()
-
- backend._initialize_cuda_graphs(_FakeDevice())
-
- assert backend._cuda_graph_enabled is True
- assert backend._reset_scratch_reset_graph == "graph-3"
- assert backend._reset_scratch_forward_graph == "graph-4"
- assert mujoco_warp.reset_calls == 2
- assert mujoco_warp.forward_calls == 2
- assert backend._can_use_reset_scratch(4) is True
- assert backend._can_use_reset_scratch(5) is False
-
-
-@pytest.mark.parametrize(
- ("num_envs", "expected_capacity"),
- [
- (512, 0),
- (1024, 128),
- (2048, 128),
- (4096, 256),
- (8192, 512),
- (16384, 512),
- ],
-)
-def test_reset_scratch_capacity_scales_with_batch_and_stays_bounded(
- num_envs: int, expected_capacity: int
-) -> None:
- assert _reset_scratch_capacity_for_batch(num_envs) == expected_capacity
-
-
-def test_ineligible_cuda_graph_warns_and_uses_eager_operations() -> None:
- warp = _FakeWarp(driver=(12, 3), mempool=True)
- mujoco_warp = _FakeMujocoWarp()
- backend = _backend(warp, mujoco_warp)
-
- with pytest.warns(RuntimeWarning, match="older than 12.4"):
- backend._initialize_cuda_graphs(_FakeDevice())
-
- backend._execute_device_steps(2)
- backend._execute_device_reset()
- backend._execute_device_forward()
-
- assert backend._cuda_graph_enabled is False
- assert warp.capture_count == 0
- assert warp.launches == []
- assert mujoco_warp.step_calls == 2
- assert mujoco_warp.reset_calls == 1
- assert mujoco_warp.forward_calls == 1
-
-
-def test_cuda_graph_capture_failure_atomically_falls_back_to_eager() -> None:
- warp = _FakeWarp(fail_capture_index=1)
- mujoco_warp = _FakeMujocoWarp()
- backend = _backend(warp, mujoco_warp)
-
- with pytest.warns(RuntimeWarning, match="synthetic capture failure"):
- backend._initialize_cuda_graphs(_FakeDevice())
-
- assert backend._cuda_graph_enabled is False
- assert backend._step_graph is None
- assert backend._forward_graph is None
- assert backend._reset_graph is None
- assert "synthetic capture failure" in backend._cuda_graph_disable_reason
- captured_calls = (
- mujoco_warp.step_calls,
- mujoco_warp.forward_calls,
- mujoco_warp.reset_calls,
- )
-
- backend._execute_device_steps(2)
- backend._execute_device_reset()
- backend._execute_device_forward()
-
- assert warp.launches == []
- assert mujoco_warp.step_calls == captured_calls[0] + 2
- assert mujoco_warp.forward_calls == captured_calls[1] + 1
- assert mujoco_warp.reset_calls == captured_calls[2] + 1
diff --git a/tests/base/test_mjwarp_host_cache.py b/tests/base/test_mjwarp_host_cache.py
deleted file mode 100644
index 9a8557188..000000000
--- a/tests/base/test_mjwarp_host_cache.py
+++ /dev/null
@@ -1,180 +0,0 @@
-"""Owner-level tests for MJWarp's explicit pinned host-cache barrier."""
-
-from __future__ import annotations
-
-from types import SimpleNamespace
-from typing import Any
-
-import numpy as np
-from unisim.backend.mjwarp.backend import MjwarpBackend
-
-
-class _FakeStorage:
- def __init__(self, array: np.ndarray) -> None:
- self.array = array
-
- def numpy(self) -> np.ndarray:
- return self.array
-
-
-class _FakeWarp:
- def __init__(self) -> None:
- self.empty_calls: list[dict[str, Any]] = []
- self.events: list[tuple[Any, ...]] = []
-
- def empty(self, shape: tuple[int, ...], **kwargs: Any) -> _FakeStorage:
- self.empty_calls.append({"shape": shape, **kwargs})
- return _FakeStorage(np.empty(shape, dtype=np.float32))
-
- def copy(self, destination: Any, source: Any) -> None:
- self.events.append(("copy", destination, source))
-
-
-def _backend(warp: _FakeWarp) -> MjwarpBackend:
- backend = object.__new__(MjwarpBackend)
- backend._warp = warp
- return backend
-
-
-def test_allocate_host_cache_uses_pinned_cpu_storage() -> None:
- warp = _FakeWarp()
- backend = _backend(warp)
- device_array = SimpleNamespace(shape=(4, 7), dtype="float32")
-
- storage, cache = backend._allocate_pinned_host_cache(device_array)
-
- assert warp.empty_calls == [
- {"shape": (4, 7), "dtype": "float32", "device": "cpu", "pinned": True}
- ]
- assert cache is storage.array
-
-
-def test_refresh_host_cache_batches_all_downloads_before_sync() -> None:
- warp = _FakeWarp()
- backend = _backend(warp)
- backend._device_data = SimpleNamespace(
- qpos="device-qpos", qvel="device-qvel", sensordata="device-sensor"
- )
- backend._qpos_cache_storage = "host-qpos"
- backend._qvel_cache_storage = "host-qvel"
- backend._sensor_cache_storage = "host-sensor"
- backend._synchronize = lambda: warp.events.append(("sync",)) # type: ignore[method-assign]
-
- backend._refresh_host_cache()
-
- assert warp.events == [
- ("copy", "host-qpos", "device-qpos"),
- ("copy", "host-qvel", "device-qvel"),
- ("copy", "host-sensor", "device-sensor"),
- ("sync",),
- ]
-
-
-def test_refresh_reset_scratch_cache_only_scatters_selected_rows() -> None:
- warp = _FakeWarp()
- backend = _backend(warp)
- scratch_values = np.arange(12, dtype=np.float32).reshape(4, 3)
- scratch_storage = _FakeStorage(np.empty_like(scratch_values))
- backend._reset_scratch_data = SimpleNamespace(sensordata=scratch_values)
- backend._reset_scratch_sensor_storage = scratch_storage
- backend._reset_scratch_sensor_cache = scratch_storage.array
- backend._sensor_cache = np.full((6, 3), -1.0, dtype=np.float32)
- backend._download = lambda source, destination: np.copyto( # type: ignore[method-assign]
- destination.array, source
- )
- backend._synchronize = lambda: warp.events.append(("sync",)) # type: ignore[method-assign]
-
- backend._refresh_reset_scratch_cache(np.asarray([4, 1], dtype=np.int32))
-
- np.testing.assert_array_equal(backend._sensor_cache[4], scratch_values[0])
- np.testing.assert_array_equal(backend._sensor_cache[1], scratch_values[1])
- np.testing.assert_array_equal(
- backend._sensor_cache[[0, 2, 3, 5]],
- np.full((4, 3), -1.0, dtype=np.float32),
- )
- assert warp.events == [("sync",)]
-
-
-def test_host_reset_routes_small_row_sets_through_scratch() -> None:
- warp = _FakeWarp()
- backend = _backend(warp)
- backend._reset_mask_host = np.zeros(8, dtype=np.bool_)
- backend._reset_mask_device = "main-mask"
- backend._device_data = SimpleNamespace(qpos="main-qpos", qvel="main-qvel")
- backend._time_cache = np.ones(8, dtype=np.float32)
- events: list[tuple[Any, ...]] = []
- backend._can_use_reset_scratch = lambda _count: True # type: ignore[method-assign]
- backend._upload = lambda target, source: events.append( # type: ignore[method-assign]
- ("upload", target, np.asarray(source).copy())
- )
- backend._execute_device_reset = lambda: events.append( # type: ignore[method-assign]
- ("main-reset",)
- )
- backend._execute_reset_scratch_forward = ( # type: ignore[method-assign]
- lambda qpos, qvel: events.append(("scratch-forward", qpos.copy(), qvel.copy()))
- )
- backend._execute_device_forward = lambda: events.append( # type: ignore[method-assign]
- ("main-forward",)
- )
- backend._synchronize = lambda: events.append(("sync",)) # type: ignore[method-assign]
- backend._refresh_reset_scratch_cache = ( # type: ignore[method-assign]
- lambda rows: events.append(("scratch-refresh", rows.copy()))
- )
- backend._refresh_host_cache = lambda: events.append( # type: ignore[method-assign]
- ("main-refresh",)
- )
- rows = np.asarray([6, 2], dtype=np.int32)
- full_qpos = np.zeros((8, 3), dtype=np.float32)
- full_qvel = np.zeros((8, 2), dtype=np.float32)
- reset_qpos = np.ones((2, 3), dtype=np.float32)
- reset_qvel = np.ones((2, 2), dtype=np.float32)
-
- backend._execute_host_reset(rows, full_qpos, full_qvel, reset_qpos, reset_qvel)
-
- assert np.flatnonzero(backend._reset_mask_host).tolist() == [2, 6]
- assert backend._time_cache[rows].tolist() == [0.0, 0.0]
- assert [event[0] for event in events] == [
- "upload",
- "main-reset",
- "upload",
- "upload",
- "scratch-forward",
- "sync",
- "scratch-refresh",
- ]
- assert not any(event[0] in {"main-forward", "main-refresh"} for event in events)
-
-
-def test_row_body_getters_gather_selected_rows_without_full_batch_reads() -> None:
- """MJWarp's partial-reset getters must not materialize the full env batch."""
- backend = object.__new__(MjwarpBackend)
- backend._body_id_to_tracked_idx = np.asarray([2, 0, 1], dtype=np.intp)
- backend._tracked_pos_w_all = np.arange(4 * 3 * 3, dtype=np.float32).reshape(4, 3, 3)
- backend._tracked_quat_w_all = np.arange(4 * 3 * 4, dtype=np.float32).reshape(4, 3, 4)
- backend._tracked_linvel_w_all = np.arange(4 * 3 * 3, dtype=np.float32).reshape(4, 3, 3) + 1000
- backend._tracked_angvel_w_all = np.arange(4 * 3 * 3, dtype=np.float32).reshape(4, 3, 3) + 2000
-
- def fail_full_getter(*_args: Any, **_kwargs: Any) -> Any:
- raise AssertionError("row getter must not call a full-batch getter")
-
- backend.get_body_pos_w = fail_full_getter # type: ignore[method-assign]
- backend.get_body_quat_w = fail_full_getter # type: ignore[method-assign]
- backend.get_body_lin_vel_w = fail_full_getter # type: ignore[method-assign]
- backend.get_body_ang_vel_w = fail_full_getter # type: ignore[method-assign]
-
- rows = np.asarray([3, 1, 3], dtype=np.int32)
- body_ids = np.asarray([1, 2], dtype=np.int32)
- mapped = np.asarray([0, 1], dtype=np.intp)
- expected_index = (rows[:, None], mapped)
-
- pose_pos, pose_quat = backend.get_body_pose_w_rows(rows, body_ids)
- np.testing.assert_array_equal(pose_pos, backend._tracked_pos_w_all[expected_index])
- np.testing.assert_array_equal(pose_quat, backend._tracked_quat_w_all[expected_index])
- np.testing.assert_array_equal(
- backend.get_body_lin_vel_w_rows(rows, body_ids),
- backend._tracked_linvel_w_all[expected_index],
- )
- np.testing.assert_array_equal(
- backend.get_body_ang_vel_w_rows(rows, body_ids),
- backend._tracked_angvel_w_all[expected_index],
- )
diff --git a/tests/base/test_np_env.py b/tests/base/test_np_env.py
index 505f070e2..f6c6937a7 100644
--- a/tests/base/test_np_env.py
+++ b/tests/base/test_np_env.py
@@ -229,28 +229,6 @@ def test_obs_is_dict(self):
assert "obs" in state.obs
assert "critic" in state.obs
- def test_terminated_and_truncated_are_separate_signals(self):
- state = NpEnvState(
- obs={"a": np.zeros((3, 1))},
- reward=np.zeros(3),
- terminated=np.array([True, False, False]),
- truncated=np.array([False, False, True]),
- info={},
- )
- np.testing.assert_array_equal(state.terminated, [True, False, False])
- np.testing.assert_array_equal(state.truncated, [False, False, True])
- np.testing.assert_array_equal(state.terminated | state.truncated, [True, False, True])
-
- def test_terminated_and_truncated_can_both_be_true(self):
- state = NpEnvState(
- obs={"a": np.zeros((1, 1))},
- reward=np.zeros(1),
- terminated=np.array([True]),
- truncated=np.array([True]),
- info={},
- )
- assert (state.terminated | state.truncated)[0] is np.True_
-
def test_replace_preserves_type(self):
obs = {"obs": np.zeros((2, 3))}
state = NpEnvState(
diff --git a/tests/config/test_config_system.py b/tests/config/test_config_system.py
index 4d7a47cbf..5be759f93 100644
--- a/tests/config/test_config_system.py
+++ b/tests/config/test_config_system.py
@@ -132,14 +132,11 @@ def test_algo_config_composes(algo_dir: str, config_name: str):
assert cfg.training.sim_backend == "mujoco"
-def test_task_files_keep_full_identity_without_hidden_backend_marker():
+def test_backend_task_files_keep_full_identity():
for path in sorted(CONF_DIR.glob("*/task/**/*.yaml")):
cfg = OmegaConf.load(path)
cfg_dict_raw = OmegaConf.to_container(cfg, resolve=True) or {}
assert isinstance(cfg_dict_raw, dict)
- assert "_selected_sim_backend" not in cfg_dict_raw, (
- f"task has hidden backend marker: {path}"
- )
if path.stem not in _BACKENDS:
continue
training_raw = cfg_dict_raw.get("training", {})
@@ -167,15 +164,6 @@ def test_supported_task_composes(
_assert_reward_populated(cfg, task_file)
-def test_offpolicy_g1_walk_flat_motrix_sac_preserves_backend_overrides():
- cfg = _compose("sac", overrides=["task=g1_walk_flat/motrix"])
-
- assert cfg.algo.num_envs == 2048
- assert cfg.algo.max_iterations == 5000
- assert cfg.reward.tracking_lin_vel.weight == pytest.approx(2.2)
- assert cfg.env.events.pd_gains is None
-
-
def test_offpolicy_g1_walk_flat_mujoco_td3_uses_td3_task_owner():
cfg = _compose("td3", overrides=["task=g1_walk_flat/mujoco"])
diff --git a/tests/config/test_locomotion_params.py b/tests/config/test_locomotion_params.py
index 7a0bc5e94..f3ed8a3c9 100644
--- a/tests/config/test_locomotion_params.py
+++ b/tests/config/test_locomotion_params.py
@@ -174,18 +174,6 @@ def test_offpolicy_td3_defaults():
assert cfg.algo.algo_params.log_std_min == pytest.approx(-1.6)
-def test_offpolicy_td3_g1_task_overrides():
- from hydra import compose, initialize_config_dir
- from hydra.core.global_hydra import GlobalHydra
-
- GlobalHydra.instance().clear()
- with initialize_config_dir(config_dir=str(CONF_DIR / "td3"), version_base="1.3"):
- cfg = compose("config", overrides=["task=g1_walk_flat/mujoco"])
- assert cfg.training.task_name == "G1WalkFlat"
- assert cfg.algo.max_iterations == 100000
- assert cfg.env.actions.joint_pos.scale == pytest.approx(1.0)
-
-
def test_offpolicy_flashsac_g1_task_overrides():
from hydra import compose, initialize_config_dir
from hydra.core.global_hydra import GlobalHydra
@@ -294,7 +282,6 @@ def test_appo_g1_task_overrides():
assert cfg.algo.max_iterations == 500
assert cfg.algo.save_interval == 100
assert cfg.training.task_name == "G1WalkFlat"
- assert "obs_profile" not in cfg.env
assert "curriculum" not in cfg.env
@@ -315,22 +302,6 @@ def test_ppo_go1_max_iterations():
assert cfg.algo.algorithm.enable_compile is False
-def test_ppo_compile_overrides():
- from hydra import compose, initialize_config_dir
- from hydra.core.global_hydra import GlobalHydra
-
- GlobalHydra.instance().clear()
- with initialize_config_dir(config_dir=str(CONF_DIR / "ppo"), version_base="1.3"):
- cfg = compose(
- "config",
- overrides=[
- "task=go1_joystick_flat/mujoco",
- "algo.algorithm.enable_compile=false",
- ],
- )
- assert cfg.algo.algorithm.enable_compile is False
-
-
def test_ppo_g1_num_envs():
from hydra import compose, initialize_config_dir
from hydra.core.global_hydra import GlobalHydra
@@ -341,7 +312,6 @@ def test_ppo_g1_num_envs():
assert cfg.algo.num_envs == 2048
assert cfg.algo.max_iterations == 2200
assert cfg.training.task_name == "G1WalkFlat"
- assert "obs_profile" not in cfg.env
assert "curriculum" not in cfg.env
diff --git a/tests/conftest.py b/tests/conftest.py
index c242ad33f..20e4073fe 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -25,8 +25,6 @@
import shutil
import pytest
-import torch
-from uni_rl.ipc.rollout_ring_buffer import RolloutRingBuffer
# ---------------------------------------------------------------------------
# Dummy flat env — no MuJoCo required
@@ -40,9 +38,6 @@
DUMMY_ENV_NAME as _DUMMY_ENV_NAME,
)
-_DUMMY_OBS_DIM = 8
-_DUMMY_ACT_DIM = 3
-
# Make the dummy env discoverable inside spawn collector subprocesses.
_existing = os.environ.get("UNILAB_EXTRA_REGISTRY_PACKAGES", "")
_pkgs = [p.strip() for p in _existing.split(",") if p.strip()]
@@ -91,36 +86,6 @@ def _isolate_training_logs_for_tests(tmp_path_factory: pytest.TempPathFactory):
shutil.rmtree(log_root, ignore_errors=True)
-@pytest.fixture
-def mp_ctx():
- return torch.multiprocessing.get_context("spawn")
-
-
-@pytest.fixture
-def tiny_storage():
- storage = RolloutRingBuffer(
- num_envs=4,
- num_steps=10,
- obs_dim=_DUMMY_OBS_DIM,
- action_dim=_DUMMY_ACT_DIM,
- num_slots=2,
- create=True,
- )
- yield storage
- storage.cleanup()
-
-
-@pytest.fixture
-def tiny_weight_shapes():
- """Small MLP param shapes dict — linear(8,16) + bias, linear(16,3) + bias."""
- return {
- "layer1.weight": torch.Size([16, 8]),
- "layer1.bias": torch.Size([16]),
- "layer2.weight": torch.Size([3, 16]),
- "layer2.bias": torch.Size([3]),
- }
-
-
@pytest.fixture
def mock_env_name() -> str:
return _DUMMY_ENV_NAME
diff --git a/tests/envs/locomotion/a2/test_a2_joystick_contract.py b/tests/envs/locomotion/a2/test_a2_joystick_contract.py
index dfcc3664c..77e4d940a 100644
--- a/tests/envs/locomotion/a2/test_a2_joystick_contract.py
+++ b/tests/envs/locomotion/a2/test_a2_joystick_contract.py
@@ -2,7 +2,6 @@
from __future__ import annotations
-import importlib
from collections.abc import Mapping, Sequence
from dataclasses import fields, is_dataclass
from pathlib import Path
@@ -269,23 +268,12 @@ def test_a2_owner_declares_all_randomization_as_manager_events() -> None:
assert push.is_global_time is True
-def test_a2_registry_has_no_legacy_config_or_runtime_fallback() -> None:
+def test_a2_registry_is_manager_only() -> None:
registry.ensure_registries()
- module = importlib.import_module("unilab.tasks.locomotion.a2.joystick")
- assert not hasattr(module, "A2JoystickCfg")
- assert not hasattr(module, "A2JoystickFlatEnv")
- assert not hasattr(module, "A2JoystickDomainRandomizationProvider")
assert registry.list_registered_envs()["A2JoystickFlat"] == {
"config_factory": "ManagerBasedRlEnvCfg",
"available_backends": ["mujoco"],
}
- for legacy_override in (
- {"reward_config": {}},
- {"domain_rand": {"randomize_kp": True}},
- {"control_config": {"action_scale": 0.4}},
- ):
- with pytest.raises(ValueError, match="has no attribute"):
- apply_cfg_overrides(ManagerBasedRlEnvCfg(), legacy_override)
def test_a2_registry_executes_real_manager_runtime() -> None:
diff --git a/tests/envs/locomotion/g1/test_g1_owner_contract.py b/tests/envs/locomotion/g1/test_g1_owner_contract.py
index eb0e1568b..86c3bb6f3 100644
--- a/tests/envs/locomotion/g1/test_g1_owner_contract.py
+++ b/tests/envs/locomotion/g1/test_g1_owner_contract.py
@@ -591,10 +591,22 @@ def test_g1_owner_materializes_complete_plain_manager_cfg(
assert hydra_cfg.training.play_render_mode == "auto"
if backend == "isaacgym":
assert env_cfg.isaacgym_device_id == 0
+ # The subprocess backend consumes the self-contained MJCF scene
+ # directly; scene fragments and generated terrain stay unset.
+ assert env_cfg.scene.fragment_files == []
+ assert env_cfg.scene.terrain is None
+ # Effort-mode dofs carry no PD gains, so the owner disables kp/kd
+ # randomization like the mjwarp/motrix owners.
+ assert env_cfg.events["pd_gains"] is None
# Native rendering (viewer + camera-sensor record) is supported;
# playback stays on the base config's auto mode.
assert hydra_cfg.training.play_render_mode == "auto"
if backend == "genesis":
+ assert env_cfg.genesis_device_id == 0
+ # The in-process backend consumes the self-contained MJCF scene
+ # directly; scene fragments and generated terrain stay unset.
+ assert env_cfg.scene.fragment_files == []
+ assert env_cfg.scene.terrain is None
# Re-declares the MJCF