diff --git a/.github/workflows/test_python.yml b/.github/workflows/test_python.yml
index c735cf1135..8b73635fcf 100644
--- a/.github/workflows/test_python.yml
+++ b/.github/workflows/test_python.yml
@@ -56,11 +56,20 @@ jobs:
           # the key must never match, even when restarting workflows, as that
           # will cause durations to get out of sync between groups, the
           # combined durations will be loaded if available
-          key: test2-durations-split-${{ github.run_id }}-${{ github.run_number}}-${{ matrix.python }}-${{ matrix.group }}
+          key: test4-durations-split-${{ github.run_id }}-${{ github.run_number }}-${{ github.run_attempt }}-${{ matrix.python }}-${{ matrix.group }}
           restore-keys: |
-            test2-durations-combined-${{ matrix.python }}-${{ github.sha }}
-            test2-durations-combined-${{ matrix.python }}
-      - run: pytest --cov=deepmd source/tests --ignore=source/tests/tf2 --splits 12 --group ${{ matrix.group }} --store-durations --clean-durations --durations-path=.test_durations --splitting-algorithm least_duration
+            test4-durations-combined-${{ matrix.python }}-${{ github.sha }}
+            test4-durations-combined-${{ matrix.python }}
+      - name: Prepare independent duration inputs
+        run: |
+          for output in main tf2 jax2tf consistent_tf2; do
+            if [ -f .test_durations ]; then
+              cp .test_durations ".test_durations_${output}"
+            fi
+          done
+      # Keep fixture-sharing classes/modules together, then balance those
+      # indivisible units by their recorded durations.
+      - run: python -m pytest -p source.tests.ci_split --cov=deepmd source/tests --ignore=source/tests/tf2 --ci-splits 12 --ci-group ${{ matrix.group }} --store-durations --clean-durations --durations-path=.test_durations_main
         env:
           NUM_WORKERS: 0
           DP_CI_IMPORT_PADDLE_BEFORE_TF: 1
@@ -68,29 +77,38 @@ jobs:
       - name: Test TF2 eager mode
         run: |
           run_pytest_allow_no_tests() {
+            local durations_path=$1
+            shift
             set +e
-            pytest "$@"
+            pytest "$@" \
+              --store-durations \
+              --clean-durations \
+              --durations-path="$durations_path"
             local status=$?
             set -e
             if [ "$status" -eq 5 ]; then
               # pytest-split may leave an individual shard with no selected
               # tests after path/-k filtering. Other shards still cover the
               # selected tests, so do not fail the whole matrix for exit 5.
+              printf '{}\n' > "$durations_path"
               return 0
             fi
             return "$status"
           }
 
-          run_pytest_allow_no_tests --cov=deepmd --cov-append \
+          run_pytest_allow_no_tests .test_durations_tf2 \
+            --cov=deepmd --cov-append \
             source/tests/tf2 \
             --splits 12 \
             --group ${{ matrix.group }}
-          run_pytest_allow_no_tests --cov=deepmd --cov-append \
+          run_pytest_allow_no_tests .test_durations_jax2tf \
+            --cov=deepmd --cov-append \
             source/tests/consistent/io/test_io.py \
             source/jax2tf_tests \
             --splits 12 \
             --group ${{ matrix.group }}
-          run_pytest_allow_no_tests --cov=deepmd --cov-append \
+          run_pytest_allow_no_tests .test_durations_consistent_tf2 \
+            --cov=deepmd --cov-append \
             source/tests/consistent \
             -k tf2 \
             --splits 12 \
@@ -100,7 +118,8 @@ jobs:
           DP_TEST_TF2_ONLY: 1
           DP_DTYPE_PROMOTION_STRICT: 1
           DP_CI_IMPORT_PADDLE_BEFORE_TF: 1
-      - run: mv .test_durations .test_durations_${{ matrix.group }}
+      - name: Combine shard durations
+        run: jq -s 'reduce .[] as $durations ({}; . * $durations)' .test_durations_main .test_durations_tf2 .test_durations_jax2tf .test_durations_consistent_tf2 > .test_durations_${{ matrix.group }}
       - name: Upload partial durations
         uses: actions/upload-artifact@v7
         with:
@@ -132,15 +151,15 @@ jobs:
           # key won't match during the first run for the given commit, but
           # restore-key will if there's a previous stored durations file,
           # so cache will both be loaded and stored
-          key: test2-durations-combined-${{ matrix.python }}-${{ github.sha }}
-          restore-keys: test2-durations-combined-${{ matrix.python }}
+          key: test4-durations-combined-${{ matrix.python }}-${{ github.sha }}
+          restore-keys: test4-durations-combined-${{ matrix.python }}
       - name: Download artifacts
         uses: actions/download-artifact@v8
         with:
           pattern: split-${{ matrix.python }}-*
           merge-multiple: true
       - name: Combine test durations
-        run: jq -s add .test_durations_* > .test_durations
+        run: jq -s 'reduce .[] as $durations ({}; . * $durations)' .test_durations_* > .test_durations
   pass:
     name: Pass testing Python
     needs: [testpython, update_durations]
diff --git a/source/tests/ci_split.py b/source/tests/ci_split.py
new file mode 100644
index 0000000000..dce082f2ce
--- /dev/null
+++ b/source/tests/ci_split.py
@@ -0,0 +1,293 @@
+# SPDX-License-Identifier: LGPL-3.0-or-later
+"""Keep fixture-sharing test units together while balancing CI shards.
+
+``pytest-split`` balances individual test items well, but assigning individual
+methods independently repeats class- and module-scoped setup in multiple CI
+processes.  This plugin first groups items into fixture-sharing units, then
+uses longest-processing-time (LPT) bin packing to balance those units by their
+recorded durations.
+
+Classes are kept together by default.  Module-level tests from the same file
+form one unit.  Tests that need to share a module-scoped fixture across class
+boundaries can use ``@pytest.mark.ci_split_group("name")`` to opt into the same
+explicit unit.
+"""
+
+from __future__ import (
+    annotations,
+)
+
+import heapq
+import json
+import math
+from collections import (
+    OrderedDict,
+)
+from dataclasses import (
+    dataclass,
+    field,
+)
+from pathlib import (
+    Path,
+)
+from typing import (
+    TYPE_CHECKING,
+    Protocol,
+)
+
+import pytest
+
+if TYPE_CHECKING:
+    from _pytest import (
+        nodes,
+    )
+    from _pytest.config import (
+        Config,
+    )
+    from _pytest.config.argparsing import (
+        Parser,
+    )
+
+
+class _SplitItem(Protocol):
+    """The subset of a pytest item used by the grouping algorithm."""
+
+    nodeid: str
+    cls: type[object] | None
+
+    def get_closest_marker(self, name: str) -> pytest.Mark | None:
+        """Return the closest marker with the requested name."""
+        raise NotImplementedError
+
+
+@dataclass
+class _TestUnit:
+    """A set of test items that must execute in the same CI shard."""
+
+    key: str
+    items: list[_SplitItem] = field(default_factory=list)
+    estimated_duration: float = 0.0
+
+
+@dataclass
+class _TestGroup:
+    """One CI shard selected by the grouped splitting algorithm."""
+
+    items: list[_SplitItem]
+    estimated_duration: float
+    unit_count: int
+
+
+def _explicit_group_name(item: _SplitItem) -> str | None:
+    """Return and validate an explicitly requested cross-class group name."""
+    marker = item.get_closest_marker("ci_split_group")
+    if marker is None:
+        return None
+    if len(marker.args) != 1 or marker.kwargs or not isinstance(marker.args[0], str):
+        raise pytest.UsageError(
+            "ci_split_group requires exactly one non-empty string argument"
+        )
+    name = marker.args[0].strip()
+    if not name:
+        raise pytest.UsageError(
+            "ci_split_group requires exactly one non-empty string argument"
+        )
+    return name
+
+
+def _base_unit_key(item: _SplitItem) -> str:
+    """Return the class/module unit that must never be split."""
+    # nodeid contains a repository-relative path, unlike item.path which may
+    # contain runner-specific checkout prefixes.
+    path = item.nodeid.split("::", maxsplit=1)[0]
+    if item.cls is not None:
+        return f"class:{path}::{item.cls.__qualname__}"
+    return f"module:{path}"
+
+
+def _relevant_durations(
+    items: list[_SplitItem], durations: dict[str, float]
+) -> dict[str, float]:
+    """Discard stale cache entries before deriving the unknown-test fallback."""
+    nodeids = {item.nodeid for item in items}
+    return {
+        nodeid: float(duration)
+        for nodeid, duration in durations.items()
+        if nodeid in nodeids and duration >= 0 and math.isfinite(duration)
+    }
+
+
+def _build_units(
+    items: list[_SplitItem],
+    durations: dict[str, float],
+) -> list[_TestUnit]:
+    """Build complete class/module units, then merge explicit unit groups."""
+    relevant = _relevant_durations(items, durations)
+    fallback_duration = sum(relevant.values()) / len(relevant) if relevant else 1.0
+    base_units: OrderedDict[str, _TestUnit] = OrderedDict()
+    for item in items:
+        key = _base_unit_key(item)
+        unit = base_units.setdefault(key, _TestUnit(key=key))
+        unit.items.append(item)
+
+    for unit in base_units.values():
+        known_durations = [
+            relevant[item.nodeid] for item in unit.items if item.nodeid in relevant
+        ]
+        if known_durations:
+            missing_count = len(unit.items) - len(known_durations)
+            unit.estimated_duration = sum(known_durations) + (
+                missing_count * fallback_duration
+            )
+        else:
+            unit.estimated_duration = len(unit.items) * fallback_duration
+
+    merged_units: OrderedDict[str, _TestUnit] = OrderedDict()
+    for unit in base_units.values():
+        explicit_names = [
+            name
+            for item in unit.items
+            if (name := _explicit_group_name(item)) is not None
+        ]
+        if explicit_names and len(explicit_names) != len(unit.items):
+            raise pytest.UsageError(
+                "ci_split_group must mark every test in its class/module unit"
+            )
+        if len(set(explicit_names)) > 1:
+            raise pytest.UsageError(
+                "a class/module unit cannot use multiple ci_split_group names"
+            )
+        key = f"explicit:{explicit_names[0]}" if explicit_names else unit.key
+        merged = merged_units.setdefault(key, _TestUnit(key=key))
+        merged.items.extend(unit.items)
+        merged.estimated_duration += unit.estimated_duration
+    return list(merged_units.values())
+
+
+def _split_items(
+    items: list[_SplitItem],
+    durations: dict[str, float],
+    splits: int,
+) -> list[_TestGroup]:
+    """Assign indivisible units using deterministic LPT bin packing."""
+    units = _build_units(items, durations)
+    item_index = {id(item): index for index, item in enumerate(items)}
+
+    # The group index breaks equal-duration ties deterministically.  Sorting
+    # units by key gives all CI runners the same assignment even when several
+    # units have identical estimates, including a completely empty cache.
+    heap: list[tuple[float, int, list[_TestUnit]]] = [
+        (0.0, group_index, []) for group_index in range(splits)
+    ]
+    heapq.heapify(heap)
+    for unit in sorted(units, key=lambda value: (-value.estimated_duration, value.key)):
+        duration, group_index, assigned = heapq.heappop(heap)
+        assigned.append(unit)
+        heapq.heappush(
+            heap,
+            (duration + unit.estimated_duration, group_index, assigned),
+        )
+
+    groups: list[_TestGroup | None] = [None] * splits
+    for duration, group_index, assigned in heap:
+        selected = [item for unit in assigned for item in unit.items]
+        selected.sort(key=lambda item: item_index[id(item)])
+        groups[group_index] = _TestGroup(
+            items=selected,
+            estimated_duration=duration,
+            unit_count=len(assigned),
+        )
+    return [group for group in groups if group is not None]
+
+
+def _load_durations(path: Path) -> dict[str, float]:
+    """Load pytest-split's duration cache, including its legacy list format."""
+    try:
+        data = json.loads(path.read_text(encoding="utf-8"))
+    except FileNotFoundError:
+        return {}
+    except (OSError, json.JSONDecodeError) as exc:
+        raise pytest.UsageError(f"cannot read duration cache {path}: {exc}") from exc
+
+    if isinstance(data, list):
+        try:
+            data = dict(data)
+        except (TypeError, ValueError) as exc:
+            raise pytest.UsageError(
+                f"duration cache {path} contains an invalid legacy list"
+            ) from exc
+    if not isinstance(data, dict):
+        raise pytest.UsageError(f"duration cache {path} must contain a JSON object")
+    try:
+        return {str(nodeid): float(duration) for nodeid, duration in data.items()}
+    except (TypeError, ValueError) as exc:
+        raise pytest.UsageError(
+            f"duration cache {path} contains a non-numeric duration"
+        ) from exc
+
+
+def pytest_addoption(parser: Parser) -> None:
+    """Declare CI-specific split options without activating pytest-split."""
+    group = parser.getgroup("DeePMD grouped CI splitting")
+    group.addoption(
+        "--ci-splits",
+        type=int,
+        help="Number of grouped CI test shards.",
+    )
+    group.addoption(
+        "--ci-group",
+        type=int,
+        help="One-based grouped CI shard to execute.",
+    )
+
+
+@pytest.hookimpl(tryfirst=True)
+def pytest_cmdline_main(config: Config) -> int | None:
+    """Validate grouped split options before pytest starts collection."""
+    splits = config.getoption("ci_splits")
+    group = config.getoption("ci_group")
+    if splits is None and group is None:
+        return None
+    if splits is None or group is None:
+        raise pytest.UsageError("--ci-splits and --ci-group must be used together")
+    if splits < 1:
+        raise pytest.UsageError("--ci-splits must be at least 1")
+    if group < 1 or group > splits:
+        raise pytest.UsageError("--ci-group must be between 1 and --ci-splits")
+    return None
+
+
+def pytest_configure(config: Config) -> None:
+    """Register the marker used to join classes sharing expensive fixtures."""
+    config.addinivalue_line(
+        "markers",
+        "ci_split_group(name): keep marked tests in one duration-balanced CI shard",
+    )
+
+
+@pytest.hookimpl(trylast=True)
+def pytest_collection_modifyitems(config: Config, items: list[nodes.Item]) -> None:
+    """Select one grouped shard after the complete test suite is collected."""
+    splits = config.getoption("ci_splits")
+    group_index = config.getoption("ci_group")
+    if splits is None or group_index is None:
+        return
+
+    durations_path = Path(config.getoption("durations_path"))
+    durations = _load_durations(durations_path)
+    groups = _split_items(items, durations, splits)
+    selected = groups[group_index - 1]
+    selected_ids = {id(item) for item in selected.items}
+    deselected = [item for item in items if id(item) not in selected_ids]
+    items[:] = selected.items
+    config.hook.pytest_deselected(items=deselected)
+
+    reporter = config.pluginmanager.get_plugin("terminalreporter")
+    if reporter is not None:
+        estimate_source = "runtime cache" if durations else "test-count fallback"
+        reporter.write_line(
+            "[deepmd-ci-split] "
+            f"group {group_index}/{splits}: {len(selected.items)} tests in "
+            f"{selected.unit_count} units, estimated "
+            f"{selected.estimated_duration:.2f}s ({estimate_source})"
+        )
diff --git a/source/tests/test_ci_split.py b/source/tests/test_ci_split.py
new file mode 100644
index 0000000000..ef97ce127a
--- /dev/null
+++ b/source/tests/test_ci_split.py
@@ -0,0 +1,173 @@
+# SPDX-License-Identifier: LGPL-3.0-or-later
+"""Unit tests for fixture-aware CI test splitting."""
+
+from dataclasses import (
+    dataclass,
+)
+from pathlib import (
+    Path,
+)
+
+import pytest
+
+from .ci_split import (
+    _load_durations,
+    _split_items,
+    _TestGroup,
+)
+
+
+class FirstClass:
+    pass
+
+
+class SecondClass:
+    pass
+
+
+@dataclass(frozen=True)
+class FakeMarker:
+    args: tuple[object, ...]
+    kwargs: dict[str, object]
+
+
+@dataclass
+class FakeItem:
+    nodeid: str
+    cls: type[object] | None = None
+    explicit_group: str | None = None
+
+    def get_closest_marker(self, name: str) -> FakeMarker | None:
+        if name == "ci_split_group" and self.explicit_group is not None:
+            return FakeMarker((self.explicit_group,), {})
+        return None
+
+
+def _item(
+    nodeid: str,
+    *,
+    cls: type[object] | None = None,
+    explicit_group: str | None = None,
+) -> FakeItem:
+    return FakeItem(
+        nodeid=nodeid,
+        cls=cls,
+        explicit_group=explicit_group,
+    )
+
+
+def _owner(groups: list[_TestGroup], item: FakeItem) -> int:
+    return next(index for index, group in enumerate(groups) if item in group.items)
+
+
+class TestUnitConstruction:
+    def test_class_and_module_items_are_indivisible(self) -> None:
+        class_items = [
+            _item(f"tests/test_a.py::FirstClass::test_{index}", cls=FirstClass)
+            for index in range(3)
+        ]
+        module_items = [
+            _item(f"tests/test_a.py::test_module_{index}") for index in range(2)
+        ]
+        groups = _split_items(class_items + module_items, {}, splits=2)
+
+        assert len({_owner(groups, item) for item in class_items}) == 1
+        assert len({_owner(groups, item) for item in module_items}) == 1
+
+    def test_explicit_group_joins_classes_across_modules(self) -> None:
+        first_class = [
+            _item(
+                f"tests/test_a.py::FirstClass::test_{index}",
+                cls=FirstClass,
+                explicit_group="shared-model",
+            )
+            for index in range(2)
+        ]
+        second_class = [
+            _item(
+                f"tests/test_b.py::SecondClass::test_{index}",
+                cls=SecondClass,
+                explicit_group="shared-model",
+            )
+            for index in range(2)
+        ]
+        groups = _split_items(first_class + second_class, {}, splits=2)
+
+        assert len({_owner(groups, item) for item in first_class + second_class}) == 1
+
+    def test_partial_method_marker_is_rejected(self) -> None:
+        items = [
+            _item(
+                "tests/test_a.py::FirstClass::test_marked",
+                cls=FirstClass,
+                explicit_group="shared-model",
+            ),
+            _item("tests/test_a.py::FirstClass::test_unmarked", cls=FirstClass),
+        ]
+
+        with pytest.raises(pytest.UsageError, match="must mark every test"):
+            _split_items(items, {}, splits=2)
+
+    def test_empty_explicit_group_is_rejected(self) -> None:
+        item = _item(
+            "tests/test_a.py::FirstClass::test_one",
+            cls=FirstClass,
+            explicit_group="   ",
+        )
+
+        with pytest.raises(pytest.UsageError, match="non-empty string"):
+            _split_items([item], {}, splits=1)
+
+
+class TestUnitBalancing:
+    def test_lpt_balances_recorded_unit_durations(self) -> None:
+        items = [_item(f"tests/test_{index}.py::test_value") for index in range(4)]
+        durations = dict(
+            zip(
+                (item.nodeid for item in items),
+                [9.0, 8.0, 7.0, 6.0],
+                strict=True,
+            )
+        )
+        groups = _split_items(items, durations, splits=2)
+
+        assert [group.estimated_duration for group in groups] == [15.0, 15.0]
+
+    def test_empty_cache_balances_by_number_of_tests(self) -> None:
+        items = [
+            *[
+                _item(f"tests/test_a.py::FirstClass::test_{index}", cls=FirstClass)
+                for index in range(5)
+            ],
+            *[
+                _item(f"tests/test_b.py::SecondClass::test_{index}", cls=SecondClass)
+                for index in range(4)
+            ],
+            *[_item(f"tests/test_c.py::test_{index}") for index in range(3)],
+            *[_item(f"tests/test_d.py::test_{index}") for index in range(2)],
+        ]
+        groups = _split_items(items, {}, splits=2)
+
+        assert [len(group.items) for group in groups] == [7, 7]
+
+    def test_assignment_is_deterministic_and_preserves_collection_order(self) -> None:
+        items = [_item(f"tests/test_{index}.py::test_value") for index in range(8)]
+
+        first = _split_items(items, {}, splits=3)
+        second = _split_items(items, {}, splits=3)
+
+        assert [[item.nodeid for item in group.items] for group in first] == [
+            [item.nodeid for item in group.items] for group in second
+        ]
+        for group in first:
+            indices = [items.index(item) for item in group.items]
+            assert indices == sorted(indices)
+
+    def test_malformed_legacy_duration_list_is_usage_error(
+        self, tmp_path: Path
+    ) -> None:
+        path = tmp_path / "durations.json"
+        path.write_text('[["valid", 1.0], ["broken"]]', encoding="utf-8")
+
+        with pytest.raises(pytest.UsageError, match="invalid legacy list"):
+            _load_durations(path)