diff --git a/.github/ci-paths.toml b/.github/ci-paths.toml
index 1127e2d91e..ec0691af3d 100644
--- a/.github/ci-paths.toml
+++ b/.github/ci-paths.toml
@@ -20,6 +20,7 @@ tests = [
   "tests/**", "integrations/**", "examples/**", "scripts/generate_js_operations.py",
   "scripts/sync_dsh_plugin_tree.py", "scripts/generate_integration_manifest_docs.py",
   "scripts/run_atomic_memory_clean_migration.py", "scripts/scan_e2e_evidence.sh",
+  "scripts/check_atomic_memory_migration_shards.py",
   "scripts/ci_role_smoke.py",
   # These evaluator suites still live under the root tests/ directory.
   "evaluation/memory/locomo/**", "evaluation/memory/locomo_plus/**", "evaluation/memory/write_gate/**",
@@ -32,6 +33,7 @@ database = [
   "tests/e2e/*native_code*", "tests/e2e/test_recall_sufficiency_gate.py",
   "tests/e2e/test_context_text_assembly.py", "tests/e2e/test_scope_search_migration.py",
   "scripts/run_atomic_memory_clean_migration.py",
+  "scripts/check_atomic_memory_migration_shards.py",
 ]
 dify = ["integrations/dify/**", "tests/*dify*"]
 hermes = ["integrations/hermes/**", "tests/*hermes*"]
diff --git a/.github/workflows/e2e-harness.yml b/.github/workflows/e2e-harness.yml
index 686b8da638..fa86390ae9 100644
--- a/.github/workflows/e2e-harness.yml
+++ b/.github/workflows/e2e-harness.yml
@@ -297,20 +297,24 @@ jobs:
         if: always()
         run: docker rm --force recall-effort-oceanbase || true
 
-  atomic-memory-clean-oceanbase:
-    name: Atomic Memory clean migration (OceanBase)
+  atomic-memory-clean-oceanbase-shards:
+    name: Atomic Memory clean migration (OceanBase shard ${{ matrix.shard }}/3)
     needs: changes
     if: >-
       ${{ !cancelled() && (github.event_name != 'workflow_dispatch' || inputs.suite != 'recall-effort') &&
           (needs.changes.result != 'success' || fromJSON(needs.changes.outputs.suites || '{}').database) }}
     runs-on: ubuntu-latest
     timeout-minutes: 145
+    strategy:
+      fail-fast: false
+      matrix:
+        shard: [0, 1, 2]
     env:
       OCEANBASE_IMAGE: ghcr.io/oceanbase/oceanbase-ce:4.3.5.6-106000012026040916
       POWERCONTEXT_TEST_MIGRATION_OCEANBASE_URL: >-
         mysql+aoceanbase://root%40test:powercontext-e2e@127.0.0.1:2881/powercontext_test?charset=utf8mb4
       PYTHONUNBUFFERED: "1"
-      MIGRATION_REPORTS: ${{ github.workspace }}/.powercontext/atomic-memory-clean-oceanbase
+      MIGRATION_REPORTS: ${{ github.workspace }}/.powercontext/atomic-memory-clean-oceanbase-shard-${{ matrix.shard }}
     steps:
       - name: Check out
         uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
@@ -381,28 +385,28 @@ jobs:
         timeout-minutes: 10
         run: |
           uv run --locked python scripts/run_atomic_memory_clean_migration.py \
-            oceanbase --output "$MIGRATION_REPORTS/scope-search" --tests \
+            oceanbase --shard "${{ matrix.shard }}/3" --output "$MIGRATION_REPORTS/scope-search" --tests \
             tests/e2e/test_scope_search_migration.py
 
       - name: Test background listener lifecycle without skips
         timeout-minutes: 5
         run: |
           uv run --locked --extra server python scripts/run_atomic_memory_clean_migration.py \
-            oceanbase --output "$MIGRATION_REPORTS/background-metrics" --tests \
+            oceanbase --shard "${{ matrix.shard }}/3" --output "$MIGRATION_REPORTS/background-metrics" --tests \
             tests/e2e/test_background_metrics.py
 
       - name: Run clean migration evidence, grants and vectors without skips
         timeout-minutes: 60
         run: |
           uv run --locked python scripts/run_atomic_memory_clean_migration.py \
-            oceanbase --output "$MIGRATION_REPORTS/clean-migration"
+            oceanbase --shard "${{ matrix.shard }}/3" --output "$MIGRATION_REPORTS/clean-migration"
 
       - name: Test concurrent managed startup without skips
         # Collection and three isolated cases retain their 180s process deadlines.
         timeout-minutes: 15
         run: |
           uv run --locked --extra server python scripts/run_atomic_memory_clean_migration.py \
-            oceanbase --output "$MIGRATION_REPORTS/startup-snapshot" --tests \
+            oceanbase --shard "${{ matrix.shard }}/3" --output "$MIGRATION_REPORTS/startup-snapshot" --tests \
             tests/e2e/test_managed_startup_read_consistency.py
           test "$(wc -l < "$MIGRATION_REPORTS/startup-snapshot/nodes.txt")" -eq 3
 
@@ -411,7 +415,7 @@ jobs:
         timeout-minutes: 30
         run: |
           uv run --locked python scripts/run_atomic_memory_clean_migration.py \
-            oceanbase --output "$MIGRATION_REPORTS/long-query" --tests \
+            oceanbase --shard "${{ matrix.shard }}/3" --output "$MIGRATION_REPORTS/long-query" --tests \
             tests/e2e/test_atomic_artifact_search_http.py \
             tests/e2e/test_atomic_memory_long_fts_query.py
 
@@ -443,7 +447,7 @@ jobs:
         if: always()
         uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7
         with:
-          name: atomic-memory-clean-oceanbase-${{ github.sha }}
+          name: atomic-memory-clean-oceanbase-${{ github.sha }}-shard-${{ matrix.shard }}
           path: ${{ env.MIGRATION_REPORTS }}/
           include-hidden-files: true
           if-no-files-found: error
@@ -453,6 +457,46 @@ jobs:
         if: always()
         run: docker rm --force atomic-memory-clean-oceanbase || true
 
+  atomic-memory-clean-oceanbase:
+    name: Atomic Memory clean migration (OceanBase)
+    needs: [changes, atomic-memory-clean-oceanbase-shards]
+    if: >-
+      ${{ !cancelled() && (github.event_name != 'workflow_dispatch' || inputs.suite != 'recall-effort') &&
+          (needs.changes.result != 'success' || fromJSON(needs.changes.outputs.suites || '{}').database) }}
+    runs-on: ubuntu-latest
+    timeout-minutes: 5
+    steps:
+      - name: Require every shard to succeed
+        env:
+          SHARD_RESULT: ${{ needs.atomic-memory-clean-oceanbase-shards.result }}
+        run: |
+          if [ "$SHARD_RESULT" != success ]; then
+            echo "OceanBase migration shards did not all succeed: $SHARD_RESULT" >&2
+            exit 1
+          fi
+
+      - name: Check out
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
+        with:
+          persist-credentials: false
+
+      - name: Set up Python
+        uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7
+        with:
+          python-version: "3.12"
+
+      - name: Download shard reports into separate directories
+        uses: actions/download-artifact@9000827ccba6bdab643e8b6fd33ac0654aef8333 # v8
+        with:
+          pattern: atomic-memory-clean-oceanbase-${{ github.sha }}-shard-*
+          path: ${{ runner.temp }}/atomic-memory-clean-oceanbase-shards
+          merge-multiple: false
+
+      - name: Verify complete execution coverage
+        run: >-
+          python scripts/check_atomic_memory_migration_shards.py
+          --reports "${{ runner.temp }}/atomic-memory-clean-oceanbase-shards" --shards 3
+
   acceptance:
     needs: validate
     runs-on: ubuntu-latest
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 78d6e4ac85..905013deae 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -115,6 +115,43 @@ git push origin name-of-your-bugfix-or-feature
 
 # Pull Request Guidelines
 
+## OceanBase migration acceptance
+
+The E2E harness runs Scope search migrations, background listener lifecycle, clean
+Atomic Memory migrations, managed startup consistency, and long-query acceptance
+on three isolated OceanBase runners. Every group collects and validates its full
+test collection before selecting zero-based shard `INDEX/COUNT`. Cases within each shard still run
+serially in separate pytest processes; skips and failures fail the shard.
+
+With the migration OceanBase URL configured as in `.github/workflows/e2e-harness.yml`,
+reproduce the clean migration part of one shard with:
+
+```bash
+uv run --locked python scripts/run_atomic_memory_clean_migration.py \
+  oceanbase --shard 0/3 --output .powercontext/migration-shard-0/clean-migration
+```
+
+Omit `--shard` to run the complete collection. Use `--tests` to select the other
+groups' files from the workflow, or `--collect-only` to inspect the full collection
+without running cases. Every invocation writes `summary.json`; only successful
+execution, including a legitimately empty shard, marks the report complete.
+
+The aggregate check retains the name `Atomic Memory clean migration (OceanBase)`.
+It requires all three runners to succeed and verifies that their reports cover
+every collected node exactly once in each of the five groups. To verify downloaded
+artifacts locally, keep each shard artifact in a separate subdirectory and run:
+
+```bash
+uv run --locked python scripts/check_atomic_memory_migration_shards.py \
+  --reports .powercontext/downloaded-migration-shards --shards 3
+```
+
+Collection-only or incomplete reports cannot satisfy the aggregate check. Include
+links and measured before/after job and workflow times in migration CI PRs, and
+report runner queue time separately from execution time.
+
+## Submission checklist
+
 Before you submit a pull request, check that it meets these guidelines:
 
 1. Add behavior tests for new externally observable behavior. Add a regression test when a defect is likely to
diff --git a/docs/en/development/observability.md b/docs/en/development/observability.md
index e3ddf7e0f6..4d7cd31c7f 100644
--- a/docs/en/development/observability.md
+++ b/docs/en/development/observability.md
@@ -133,10 +133,16 @@ These checks exercise real spawn, OTLP/HTTP, a local model fixture, Pydantic AI
 surface. They prove local reception, not storage/query acceptance by an external backend.
 
 The background CLI listener lifecycle requires a disposable OceanBase instance and
-`POWERCONTEXT_TEST_MIGRATION_OCEANBASE_URL`. The `E2E harness` workflow's `atomic-memory-clean-oceanbase` job runs
-`tests/e2e/test_background_metrics.py` through the clean migration acceptance runner, which rejects skipped tests.
-The test initializes its fresh database before starting the real background CLI, then checks listener startup,
-authorized metrics, absence of business routes and clean SIGTERM shutdown. The listener startup budget excludes
-first-run schema and index creation. CI retains `background-metrics/background.log` with the migration reports;
-startup failures include the child return code and a bounded log tail. Backend acceptance requires this step to pass.
-Without the backend setting, default test runs skip that lifecycle test.
+`POWERCONTEXT_TEST_MIGRATION_OCEANBASE_URL`. The `E2E harness` workflow's
+`atomic-memory-clean-oceanbase-shards` matrix runs
+`tests/e2e/test_background_metrics.py` through the clean migration acceptance runner
+on three runners, and rejects skipped tests. The legacy-named
+`atomic-memory-clean-oceanbase` job only downloads the three shard artifacts and
+checks their complete, exactly-once coverage. The test initializes its fresh database
+before starting the real background CLI, then checks listener startup, authorized
+metrics, absence of business routes and clean SIGTERM shutdown. The listener startup
+budget excludes first-run schema and index creation. CI retains
+`background-metrics/background.log` in the shard artifact; this group has one node,
+so the log is in shard `0`. Startup failures include the child return code and a
+bounded log tail. Backend acceptance requires this step to pass. Without the backend
+setting, default test runs skip that lifecycle test.
diff --git a/docs/zh/development/observability.md b/docs/zh/development/observability.md
index 27c0d00142..c7373f0a13 100644
--- a/docs/zh/development/observability.md
+++ b/docs/zh/development/observability.md
@@ -135,9 +135,13 @@ uv run --locked --extra tracing-otlp pytest tests/e2e/test_worker_otlp.py \
 这些证据证明本地 reception，不能代替外部 backend 的存储与查询验收。
 
 Background CLI listener 生命周期测试需要可丢弃的 OceanBase 实例及
-`POWERCONTEXT_TEST_MIGRATION_OCEANBASE_URL`。`E2E harness` workflow 的 `atomic-memory-clean-oceanbase` job
-通过 clean migration acceptance runner 执行 `tests/e2e/test_background_metrics.py`，不允许测试 skip。
-该测试先初始化新建数据库，再启动真实 background CLI，检查 listener 启动、metrics 授权、business routes
-不可用，以及 SIGTERM 后正常退出。Listener 启动预算不包含首次建表与索引创建。
-CI 在 migration reports 中保留 `background-metrics/background.log`；启动失败时输出 child return code
-和有界 log tail。此步骤通过后，才能认定 backend 验收通过。未设置 backend 变量时，默认测试会跳过此用例。
+`POWERCONTEXT_TEST_MIGRATION_OCEANBASE_URL`。`E2E harness` workflow 的
+`atomic-memory-clean-oceanbase-shards` 矩阵在三个 runner 上通过 clean migration
+acceptance runner 执行 `tests/e2e/test_background_metrics.py`，不允许测试 skip。
+沿用旧名称的 `atomic-memory-clean-oceanbase` job 只负责下载三个分片的 artifact，
+并校验完整且恰好一次的覆盖。该测试先初始化新建数据库，再启动真实 background CLI，
+检查 listener 启动、metrics 授权、business routes 不可用，以及 SIGTERM 后正常退出。
+Listener 启动预算不包含首次建表与索引创建。CI 在分片 artifact 的
+`background-metrics/background.log` 中保留日志；该组只有一个 node，因此日志位于
+shard `0`。启动失败时输出 child return code 和有界 log tail。此步骤通过后，才能认定
+backend 验收通过。未设置 backend 变量时，默认测试会跳过此用例。
diff --git a/scripts/check_atomic_memory_migration_shards.py b/scripts/check_atomic_memory_migration_shards.py
new file mode 100644
index 0000000000..4cc4db1f65
--- /dev/null
+++ b/scripts/check_atomic_memory_migration_shards.py
@@ -0,0 +1,107 @@
+# Copyright (c) 2026 OceanBase.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Verify complete, exactly-once OceanBase acceptance across downloaded shard artifacts."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from collections import Counter
+from pathlib import Path
+
+GROUPS = ("scope-search", "background-metrics", "clean-migration", "startup-snapshot", "long-query")
+
+
+def verify_group(paths: list[Path], shards: int) -> int:
+    """Require matching collections and successful, disjoint execution of every node."""
+    collection: list[str] | None = None
+    passed: list[str] = []
+    indices: set[int] = set()
+    for path in paths:
+        report = json.loads(path.read_text(encoding="utf-8"))
+        if not isinstance(report, dict):
+            message = f"{path}: expected a summary object"
+            raise TypeError(message)
+        index = report.get("shard_index")
+        if (
+            type(index) is not int
+            or not 0 <= index < shards
+            or index in indices
+            or type(report.get("shard_count")) is not int
+            or report["shard_count"] != shards
+            or report.get("backend") != "oceanbase"
+            or report.get("complete") is not True
+        ):
+            message = f"{path}: incomplete summary or invalid shard identity"
+            raise ValueError(message)
+        nodes = report.get("collected_nodes")
+        actual = report.get("passed_nodes")
+        if (
+            not isinstance(nodes, list)
+            or not nodes
+            or not all(isinstance(node, str) and node for node in nodes)
+            or len(set(nodes)) != len(nodes)
+            or not isinstance(actual, list)
+            or not all(isinstance(node, str) for node in actual)
+        ):
+            message = f"{path}: invalid collection or passed nodes"
+            raise ValueError(message)
+        if collection is None:
+            collection = nodes
+        if nodes != collection:
+            message = f"{path}: full collections differ between shards"
+            raise ValueError(message)
+        if actual != nodes[index::shards]:
+            message = f"{path}: passed nodes differ from the assigned shard"
+            raise ValueError(message)
+        indices.add(index)
+        passed.extend(actual)
+    if indices != set(range(shards)) or Counter(passed) != Counter(collection or []):
+        message = "shard coverage must contain every collected node exactly once"
+        raise ValueError(message)
+    return len(passed)
+
+
+def verify_reports(directory: Path, shards: int) -> None:
+    """Verify every acceptance group in exactly one artifact per expected shard."""
+    artifacts = sorted(path for path in directory.iterdir() if path.is_dir())
+    if len(artifacts) != shards:
+        message = f"expected {shards} shard artifacts, found {len(artifacts)}"
+        raise ValueError(message)
+    for group in GROUPS:
+        count = verify_group([artifact / group / "summary.json" for artifact in artifacts], shards)
+        print(f"{group}: all {count} OceanBase cases passed exactly once without skips.", flush=True)
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--reports", type=Path, required=True, help="Directory containing one artifact directory per shard"
+    )
+    parser.add_argument("--shards", type=int, required=True)
+    args = parser.parse_args()
+    if args.shards < 1:
+        parser.error("--shards must be positive")
+    try:
+        verify_reports(args.reports, args.shards)
+    except (OSError, TypeError, ValueError) as error:
+        print(f"Shard verification failed: {error}", file=sys.stderr)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/scripts/run_atomic_memory_clean_migration.py b/scripts/run_atomic_memory_clean_migration.py
index d293f393bf..539ac8292a 100644
--- a/scripts/run_atomic_memory_clean_migration.py
+++ b/scripts/run_atomic_memory_clean_migration.py
@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import argparse
+import json
 import subprocess
 import sys
 from pathlib import Path
@@ -29,6 +30,18 @@
 )
 
 
+def parse_shard(value: str) -> tuple[int, int]:
+    """Parse a zero-based INDEX/COUNT shard before collection or execution."""
+    try:
+        index, count = map(int, value.split("/"))
+        if count > 0 and 0 <= index < count:
+            return index, count
+    except ValueError:
+        pass
+    message = "shard must be INDEX/COUNT with COUNT > 0 and 0 <= INDEX < COUNT"
+    raise argparse.ArgumentTypeError(message)
+
+
 def run_pytest(arguments: list[str], log: Path) -> int:
     # GNU timeout also terminates native child processes; never retry a failed node.
     command = [
@@ -92,10 +105,25 @@ def main() -> int:
     parser.add_argument("backend", choices=("seekdb", "oceanbase"))
     parser.add_argument("--output", type=Path, required=True)
     parser.add_argument("--collect-only", action="store_true")
+    parser.add_argument(
+        "--shard", type=parse_shard, default=(0, 1), help="Run zero-based shard INDEX/COUNT (default: 0/1)"
+    )
     parser.add_argument("--tests", nargs="+", default=SUITES, help="Pytest files to run instead of clean migration")
     args = parser.parse_args()
     output = args.output.resolve()
     output.mkdir(parents=True, exist_ok=False)
+    shard_index, shard_count = args.shard
+    passed_nodes: list[str] = []
+    summary = {
+        "backend": args.backend,
+        "shard_index": shard_index,
+        "shard_count": shard_count,
+        "collected_nodes": [],
+        "passed_nodes": passed_nodes,
+        "complete": False,
+    }
+    summary_path = output / "summary.json"
+    summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
     collection = output / "collection.log"
     if run_pytest(["--collect-only", "-q", *args.tests, "-k", args.backend], collection):
         print(collection.read_text(errors="replace"), flush=True)
@@ -103,6 +131,8 @@ def main() -> int:
     nodes = [
         line for line in collection.read_text().splitlines() if line.startswith(tuple(f"{s}::" for s in args.tests))
     ]
+    summary["collected_nodes"] = nodes
+    summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
     if (
         not nodes
         or len(set(nodes)) != len(nodes)
@@ -114,12 +144,21 @@ def main() -> int:
     print(f"Collected {len(nodes)} {args.backend} cases.", flush=True)
     if args.collect_only:
         return 0
+    selected = nodes[shard_index::shard_count]
+    print(f"Shard {shard_index}/{shard_count}: running {len(selected)} cases.", flush=True)
     # SeekDB must exit between cases. OceanBase cases share one tenant, so keep
     # their schema creation and cleanup separate from other test cases.
-    for index, node in enumerate(nodes, 1):
-        if not run_node(node, output / f"{index:03d}"):
-            return 1
-    print(f"All {len(nodes)} {args.backend} cases passed without skips.", flush=True)
+    try:
+        for index, node in enumerate(selected, 1):
+            if not run_node(node, output / f"{index:03d}"):
+                return 1
+            passed_nodes.append(node)
+        summary["complete"] = True
+    finally:
+        # Only record passes after run_node has checked the exit code and JUnit.
+        # A killed process leaves the initial, incomplete report behind.
+        summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    print(f"All {len(selected)} {args.backend} cases passed without skips.", flush=True)
     return 0
 
 
diff --git a/tests/test_atomic_memory_migration_shards.py b/tests/test_atomic_memory_migration_shards.py
new file mode 100644
index 0000000000..633eb1a6bd
--- /dev/null
+++ b/tests/test_atomic_memory_migration_shards.py
@@ -0,0 +1,197 @@
+# Copyright (c) 2026 OceanBase.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Exercise the migration runner through real pytest collection and execution."""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import subprocess
+import sys
+from collections import Counter
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).parents[1]
+RUNNER = ROOT / "scripts/run_atomic_memory_clean_migration.py"
+CHECKER = ROOT / "scripts/check_atomic_memory_migration_shards.py"
+pytestmark = pytest.mark.skipif(sys.platform == "win32" or not shutil.which("timeout"), reason="Linux CI runner")
+
+
+def write_suite(directory: Path, count: int, *, backend: str = "oceanbase", outcome: str = "pass") -> None:
+    (directory / "test_cases.py").write_text(
+        f"""import pathlib
+import pytest
+
+@pytest.mark.parametrize('case', range({count}))
+def test_{backend}_case(case, request):
+    with pathlib.Path('executed.txt').open('a') as output:
+        output.write(request.node.nodeid + '\\n')
+    if case == 1:
+        if {outcome!r} == 'skip':
+            pytest.skip('synthetic skip')
+        if {outcome!r} == 'fail':
+            pytest.fail('synthetic failure')
+        if {outcome!r} == 'interrupt':
+            raise KeyboardInterrupt
+""",
+        encoding="utf-8",
+    )
+
+
+def run_runner(
+    directory: Path,
+    output: Path,
+    *arguments: str,
+    backend: str = "oceanbase",
+    tests: tuple[str, ...] = ("test_cases.py",),
+) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        [sys.executable, str(RUNNER), backend, "--output", str(output), *arguments, "--tests", *tests],
+        cwd=directory,
+        env={**os.environ, "PYTEST_ADDOPTS": "", "PYTEST_DISABLE_PLUGIN_AUTOLOAD": "1"},
+        capture_output=True,
+        text=True,
+        check=False,
+        timeout=60,
+    )
+
+
+def test_three_shards_execute_every_collected_node_exactly_once(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # The temporary suites do not load the repository's CI-only pytest options.
+    monkeypatch.setenv("PYTEST_ADDOPTS", "--require-dsh-config-runtime")
+    write_suite(tmp_path, 5)
+    reports = []
+    for index in range(3):
+        output = tmp_path / f"shard-{index}"
+        result = run_runner(tmp_path, output, "--shard", f"{index}/3")
+        assert result.returncode == 0, result.stdout + result.stderr
+        reports.append(json.loads((output / "summary.json").read_text()))
+
+    nodes = reports[0]["collected_nodes"]
+    assert len(nodes) == 5
+    for index, report in enumerate(reports):
+        assert report["collected_nodes"] == nodes
+        assert report["passed_nodes"] == nodes[index::3]
+        assert report["complete"] is True
+    actual = (tmp_path / "executed.txt").read_text().splitlines()
+    assert Counter(actual) == Counter(nodes)
+    assert Counter(node for report in reports for node in report["passed_nodes"]) == Counter(actual)
+
+    # Verify actual runner reports through the aggregate CLI's artifact layout.
+    for index, report in enumerate(reports):
+        artifact = tmp_path / "artifacts" / f"shard-{index}"
+        for group in ("scope-search", "background-metrics", "clean-migration", "startup-snapshot", "long-query"):
+            destination = artifact / group
+            destination.mkdir(parents=True)
+            (destination / "summary.json").write_text(json.dumps(report), encoding="utf-8")
+    result = subprocess.run(
+        [sys.executable, str(CHECKER), "--reports", str(tmp_path / "artifacts"), "--shards", "3"],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert result.returncode == 0, result.stderr
+
+
+def test_valid_collection_can_produce_an_empty_shard(tmp_path: Path) -> None:
+    write_suite(tmp_path, 1)
+    output = tmp_path / "empty-shard"
+    result = run_runner(tmp_path, output, "--shard", "2/3")
+
+    assert result.returncode == 0, result.stdout + result.stderr
+    report = json.loads((output / "summary.json").read_text())
+    assert len(report["collected_nodes"]) == 1
+    assert report["passed_nodes"] == []
+    assert report["complete"] is True
+    assert not (tmp_path / "executed.txt").exists()
+
+
+@pytest.mark.parametrize("backend", ["seekdb", "oceanbase"])
+def test_default_runs_the_full_collection(tmp_path: Path, backend: str) -> None:
+    write_suite(tmp_path, 2, backend=backend)
+    output = tmp_path / "unsharded"
+    result = run_runner(tmp_path, output, backend=backend)
+
+    assert result.returncode == 0, result.stdout + result.stderr
+    report = json.loads((output / "summary.json").read_text())
+    assert report["backend"] == backend
+    assert (report["shard_index"], report["shard_count"]) == (0, 1)
+    assert len(report["passed_nodes"]) == 2
+    assert report["passed_nodes"] == report["collected_nodes"]
+    assert report["complete"] is True
+
+
+@pytest.mark.parametrize("shard", ["-1/3", "3/3", "0/0", "0/-1", "x/3", "0/3/1", "1"])
+def test_invalid_shard_is_rejected_before_collection(tmp_path: Path, shard: str) -> None:
+    output = tmp_path / "invalid"
+    result = run_runner(tmp_path, output, f"--shard={shard}")
+
+    assert result.returncode == 2
+    assert "INDEX/COUNT" in result.stderr
+    assert not output.exists()
+
+
+@pytest.mark.parametrize("invalid_collection", ["duplicate", "missing-suite", "empty"])
+def test_full_collection_must_be_valid_even_for_an_empty_shard(tmp_path: Path, invalid_collection: str) -> None:
+    write_suite(tmp_path, 1)
+    tests = ("test_cases.py",)
+    if invalid_collection == "duplicate":
+        (tmp_path / "conftest.py").write_text(
+            "def pytest_collection_modifyitems(items): items.append(items[0])\n", encoding="utf-8"
+        )
+    elif invalid_collection == "missing-suite":
+        (tmp_path / "test_other.py").write_text("def test_seekdb_case(): pass\n", encoding="utf-8")
+        tests = ("test_cases.py", "test_other.py")
+    else:
+        write_suite(tmp_path, 1, backend="seekdb")
+    output = tmp_path / "invalid-collection"
+    result = run_runner(tmp_path, output, "--shard", "2/3", tests=tests)
+
+    assert result.returncode == 1
+    report = json.loads((output / "summary.json").read_text())
+    assert report["complete"] is False
+    assert report["passed_nodes"] == []
+    assert not (tmp_path / "executed.txt").exists()
+
+
+def test_collect_only_does_not_claim_execution(tmp_path: Path) -> None:
+    write_suite(tmp_path, 2)
+    output = tmp_path / "collected"
+    result = run_runner(tmp_path, output, "--shard", "0/3", "--collect-only")
+
+    assert result.returncode == 0, result.stdout + result.stderr
+    report = json.loads((output / "summary.json").read_text())
+    assert len(report["collected_nodes"]) == 2
+    assert report["passed_nodes"] == []
+    assert report["complete"] is False
+    assert not (tmp_path / "executed.txt").exists()
+
+
+@pytest.mark.parametrize("outcome", ["fail", "skip", "interrupt"])
+def test_unsuccessful_case_is_not_counted_as_passed_and_is_not_retried(tmp_path: Path, outcome: str) -> None:
+    write_suite(tmp_path, 3, outcome=outcome)
+    output = tmp_path / "failed"
+    result = run_runner(tmp_path, output)
+
+    assert result.returncode == 1
+    report = json.loads((output / "summary.json").read_text())
+    assert report["passed_nodes"] == report["collected_nodes"][:1]
+    assert report["complete"] is False
+    assert (tmp_path / "executed.txt").read_text().splitlines() == report["collected_nodes"][:2]
diff --git a/tests/test_check_atomic_memory_migration_shards.py b/tests/test_check_atomic_memory_migration_shards.py
new file mode 100644
index 0000000000..f8eb4b2ee1
--- /dev/null
+++ b/tests/test_check_atomic_memory_migration_shards.py
@@ -0,0 +1,169 @@
+# Copyright (c) 2026 OceanBase.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+import json
+import shutil
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+CHECKER = Path(__file__).parents[1] / "scripts/check_atomic_memory_migration_shards.py"
+
+
+@pytest.fixture
+def reports(tmp_path: Path) -> Path:
+    root = tmp_path / "reports"
+    for index in range(3):
+        for group in ("scope-search", "background-metrics", "clean-migration", "startup-snapshot", "long-query"):
+            count = {"background-metrics": 1, "startup-snapshot": 3}.get(group, 5)
+            nodes = [f"test_{group}.py::test_oceanbase_case[{case}]" for case in range(count)]
+            path = root / f"shard-{index}" / group / "summary.json"
+            path.parent.mkdir(parents=True)
+            path.write_text(
+                json.dumps({
+                    "backend": "oceanbase",
+                    "shard_index": index,
+                    "shard_count": 3,
+                    "collected_nodes": nodes,
+                    "passed_nodes": nodes[index::3],
+                    "complete": True,
+                }),
+                encoding="utf-8",
+            )
+    return root
+
+
+def run_checker(reports: Path, shards: str = "3") -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        [sys.executable, str(CHECKER), "--reports", str(reports), "--shards", shards],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+
+
+def test_complete_coverage_including_empty_shards_passes(reports: Path) -> None:
+    result = run_checker(reports)
+
+    assert result.returncode == 0, result.stderr
+    assert "background-metrics: all 1 OceanBase cases passed exactly once" in result.stdout
+    assert "clean-migration: all 5 OceanBase cases passed exactly once" in result.stdout
+    assert "startup-snapshot: all 3 OceanBase cases passed exactly once" in result.stdout
+
+
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [
+        ("complete", False),
+        ("shard_count", 4),
+        ("backend", "seekdb"),
+        ("shard_index", False),
+        ("collected_nodes", []),
+        ("collected_nodes", [[]]),
+        ("passed_nodes", []),
+        ("passed_nodes", ["test_other.py::test_oceanbase_case"]),
+        ("passed_nodes", ["test_long-query.py::test_oceanbase_case[0]"]),
+        ("passed_nodes", ["test_long-query.py::test_oceanbase_case[0]"] * 2),
+    ],
+)
+def test_incomplete_or_invalid_execution_fails(reports: Path, field: str, value: object) -> None:
+    path = reports / "shard-0" / "long-query" / "summary.json"
+    report = json.loads(path.read_text())
+    report[field] = value
+    path.write_text(json.dumps(report), encoding="utf-8")
+
+    result = run_checker(reports)
+
+    assert result.returncode == 1
+    assert "Shard verification failed:" in result.stderr
+    assert "Traceback" not in result.stderr
+
+
+@pytest.mark.parametrize(
+    "defect", ["missing-artifact", "extra-artifact", "missing-report", "missing-startup-report", "duplicate-index"]
+)
+def test_missing_or_duplicate_shard_evidence_fails(reports: Path, defect: str) -> None:
+    if defect == "missing-artifact":
+        shutil.rmtree(reports / "shard-2")
+    elif defect == "extra-artifact":
+        shutil.copytree(reports / "shard-2", reports / "shard-3")
+    elif defect == "missing-report":
+        (reports / "shard-2" / "long-query" / "summary.json").unlink()
+    elif defect == "missing-startup-report":
+        (reports / "shard-2" / "startup-snapshot" / "summary.json").unlink()
+    else:
+        shutil.rmtree(reports / "shard-2")
+        shutil.copytree(reports / "shard-0", reports / "shard-2")
+
+    result = run_checker(reports)
+
+    assert result.returncode == 1
+    assert "Shard verification failed:" in result.stderr
+
+
+@pytest.mark.parametrize("content", ["{", "[]"])
+def test_malformed_report_fails_without_a_traceback(reports: Path, content: str) -> None:
+    (reports / "shard-0" / "long-query" / "summary.json").write_text(content, encoding="utf-8")
+
+    result = run_checker(reports)
+
+    assert result.returncode == 1
+    assert "Shard verification failed:" in result.stderr
+    assert "Traceback" not in result.stderr
+
+
+def test_individually_successful_shards_with_different_collections_fail(reports: Path) -> None:
+    path = reports / "shard-1" / "long-query" / "summary.json"
+    report = json.loads(path.read_text())
+    report["collected_nodes"].reverse()
+    report["passed_nodes"] = report["collected_nodes"][1::3]
+    path.write_text(json.dumps(report), encoding="utf-8")
+
+    result = run_checker(reports)
+
+    assert result.returncode == 1
+    assert "full collections differ" in result.stderr
+
+
+def test_swapped_assignments_fail_even_when_the_union_is_complete(reports: Path) -> None:
+    first = reports / "shard-0" / "long-query" / "summary.json"
+    second = reports / "shard-1" / "long-query" / "summary.json"
+    a = json.loads(first.read_text())
+    b = json.loads(second.read_text())
+    a["passed_nodes"], b["passed_nodes"] = b["passed_nodes"], a["passed_nodes"]
+    first.write_text(json.dumps(a), encoding="utf-8")
+    second.write_text(json.dumps(b), encoding="utf-8")
+
+    result = run_checker(reports)
+
+    assert result.returncode == 1
+    assert "passed nodes differ from the assigned shard" in result.stderr
+
+
+def test_missing_download_directory_fails(tmp_path: Path) -> None:
+    result = run_checker(tmp_path / "missing")
+
+    assert result.returncode == 1
+    assert "Shard verification failed:" in result.stderr
+
+
+def test_nonpositive_shard_count_is_rejected(reports: Path) -> None:
+    result = run_checker(reports, "0")
+
+    assert result.returncode == 2
+    assert "--shards must be positive" in result.stderr
diff --git a/tests/test_ci_changes.py b/tests/test_ci_changes.py
index f5b4c9e4fc..90abf00c56 100644
--- a/tests/test_ci_changes.py
+++ b/tests/test_ci_changes.py
@@ -45,6 +45,8 @@
         (["tests/native/test_personal_service_lifecycle.py"], {"tests", "native"}),
         (["e2e/bub/compose.yaml"], {"harness"}),
         (["scripts/scan_e2e_evidence.sh"], {"tests", "harness"}),
+        (["scripts/run_atomic_memory_clean_migration.py"], {"tests", "database"}),
+        (["scripts/check_atomic_memory_migration_shards.py"], {"tests", "database"}),
         (["scripts/sync_dsh_plugin_tree.py"], {"tests", "dsh"}),
         (["integrations/bub/src/powercontext_bub/__init__.py"], {"tests", "harness"}),
         (