diff --git a/.github/workflows/e2e-harness.yml b/.github/workflows/e2e-harness.yml
index 6583227073..c85d618ff3 100644
--- a/.github/workflows/e2e-harness.yml
+++ b/.github/workflows/e2e-harness.yml
@@ -343,7 +343,8 @@ jobs:
       ${{ !cancelled() && (github.event_name != 'workflow_dispatch' || inputs.suite != 'recall-effort') &&
           ((inputs.suites == '' && needs.changes.result != 'success') || fromJSON(inputs.suites || needs.changes.outputs.suites || '{}').database) }}
     runs-on: ubuntu-latest
-    timeout-minutes: 145
+    # Test steps total 155m; reserve 25m for setup, diagnostics and artifact upload.
+    timeout-minutes: 180
     strategy:
       max-parallel: 2
       fail-fast: false
@@ -429,33 +430,35 @@ jobs:
             tests/e2e/test_scope_search_migration.py
 
       - name: Test background listener lifecycle without skips
-        timeout-minutes: 5
+        # One full Server case (300s), collection (180s), and termination grace.
+        timeout-minutes: 10
         run: |
           uv run --locked --extra server python scripts/run_atomic_memory_clean_migration.py \
-            oceanbase --shard "${{ matrix.shard }}/3" --output "$MIGRATION_REPORTS/background-metrics" --tests \
+            oceanbase --test-timeout 300 --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
+        # Up to 16 cases per shard at 300s each, plus collection and termination grace.
+        timeout-minutes: 90
         run: |
           uv run --locked python scripts/run_atomic_memory_clean_migration.py \
-            oceanbase --shard "${{ matrix.shard }}/3" --output "$MIGRATION_REPORTS/clean-migration"
+            oceanbase --test-timeout 300 --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.
+        # Each shard runs one full Server case (300s), plus collection (180s).
         timeout-minutes: 15
         run: |
           uv run --locked --extra server python scripts/run_atomic_memory_clean_migration.py \
-            oceanbase --shard "${{ matrix.shard }}/3" --output "$MIGRATION_REPORTS/startup-snapshot" --tests \
+            oceanbase --test-timeout 300 --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
 
       - name: Test long-query retrieval and admission without skips
-        # Eight isolated cases include full Server setup and retain 180s deadlines.
+        # Up to three full Server cases per shard at 300s each, plus collection (180s).
         timeout-minutes: 30
         run: |
           uv run --locked python scripts/run_atomic_memory_clean_migration.py \
-            oceanbase --shard "${{ matrix.shard }}/3" --output "$MIGRATION_REPORTS/long-query" --tests \
+            oceanbase --test-timeout 300 --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
 
diff --git a/scripts/run_atomic_memory_clean_migration.py b/scripts/run_atomic_memory_clean_migration.py
index 539ac8292a..6c16aac14d 100644
--- a/scripts/run_atomic_memory_clean_migration.py
+++ b/scripts/run_atomic_memory_clean_migration.py
@@ -42,13 +42,13 @@ def parse_shard(value: str) -> tuple[int, int]:
     raise argparse.ArgumentTypeError(message)
 
 
-def run_pytest(arguments: list[str], log: Path) -> int:
+def run_pytest(arguments: list[str], log: Path, *, timeout_seconds: int = 180) -> int:
     # GNU timeout also terminates native child processes; never retry a failed node.
     command = [
         "timeout",
         "--signal=TERM",
         "--kill-after=10s",
-        "180s",
+        f"{timeout_seconds}s",
         sys.executable,
         "-m",
         "pytest",
@@ -59,7 +59,7 @@ def run_pytest(arguments: list[str], log: Path) -> int:
         return subprocess.run(command, stdout=output, stderr=subprocess.STDOUT, check=False).returncode  # noqa: S603
 
 
-def run_node(node: str, directory: Path) -> bool:
+def run_node(node: str, directory: Path, *, timeout_seconds: int) -> bool:
     directory.mkdir()
     (directory / "node.txt").write_text(node + "\n")
     report = directory / "junit.xml"
@@ -81,6 +81,7 @@ def run_node(node: str, directory: Path) -> bool:
             node,
         ],
         log,
+        timeout_seconds=timeout_seconds,
     )
     passed = False
     try:
@@ -105,11 +106,19 @@ 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(
+        "--test-timeout",
+        type=int,
+        default=180,
+        help="Positive per-test process deadline in seconds, including setup and cleanup (default: 180)",
+    )
     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()
+    if args.test_timeout <= 0:
+        parser.error("--test-timeout must be a positive number of seconds")
     output = args.output.resolve()
     output.mkdir(parents=True, exist_ok=False)
     shard_index, shard_count = args.shard
@@ -150,7 +159,7 @@ def main() -> int:
     # their schema creation and cleanup separate from other test cases.
     try:
         for index, node in enumerate(selected, 1):
-            if not run_node(node, output / f"{index:03d}"):
+            if not run_node(node, output / f"{index:03d}", timeout_seconds=args.test_timeout):
                 return 1
             passed_nodes.append(node)
         summary["complete"] = True
diff --git a/tests/test_atomic_memory_migration_shards.py b/tests/test_atomic_memory_migration_shards.py
index 633eb1a6bd..3f227d6c44 100644
--- a/tests/test_atomic_memory_migration_shards.py
+++ b/tests/test_atomic_memory_migration_shards.py
@@ -124,10 +124,11 @@ def test_valid_collection_can_produce_an_empty_shard(tmp_path: Path) -> None:
 
 
 @pytest.mark.parametrize("backend", ["seekdb", "oceanbase"])
-def test_default_runs_the_full_collection(tmp_path: Path, backend: str) -> None:
+@pytest.mark.parametrize("arguments", [(), ("--test-timeout", "300")])
+def test_default_runs_the_full_collection(tmp_path: Path, backend: str, arguments: tuple[str, ...]) -> None:
     write_suite(tmp_path, 2, backend=backend)
     output = tmp_path / "unsharded"
-    result = run_runner(tmp_path, output, backend=backend)
+    result = run_runner(tmp_path, output, *arguments, backend=backend)
 
     assert result.returncode == 0, result.stdout + result.stderr
     report = json.loads((output / "summary.json").read_text())
@@ -138,6 +139,44 @@ def test_default_runs_the_full_collection(tmp_path: Path, backend: str) -> None:
     assert report["complete"] is True
 
 
+@pytest.mark.parametrize("seconds", ["0", "-1"])
+def test_nonpositive_test_timeout_is_rejected_before_collection(tmp_path: Path, seconds: str) -> None:
+    output = tmp_path / "invalid-timeout"
+    result = run_runner(tmp_path, output, f"--test-timeout={seconds}")
+
+    assert result.returncode == 2
+    assert "positive" in result.stderr
+    assert not output.exists()
+
+
+@pytest.mark.parametrize("seconds,complete", [(5, False), (15, True)])
+def test_test_timeout_includes_cleanup_but_not_collection(tmp_path: Path, seconds: int, complete: bool) -> None:
+    write_suite(tmp_path, 1)
+    (tmp_path / "conftest.py").write_text(
+        "import time\n"
+        "import pytest\n"
+        "def pytest_collection_modifyitems(config, items):\n"
+        "    if config.option.collectonly: time.sleep(6)\n"
+        "@pytest.fixture(autouse=True)\n"
+        "def cleanup():\n"
+        "    yield\n"
+        "    time.sleep(8)\n",
+        encoding="utf-8",
+    )
+    output = tmp_path / "deadline"
+    result = run_runner(tmp_path, output, "--test-timeout", str(seconds))
+
+    assert result.returncode == (0 if complete else 1), result.stdout + result.stderr
+    report = json.loads((output / "summary.json").read_text())
+    assert report["complete"] is complete
+    assert len(report["collected_nodes"]) == 1
+    assert len(report["passed_nodes"]) == int(complete)
+    assert (tmp_path / "executed.txt").read_text().splitlines() == report["collected_nodes"]
+    if not complete:
+        assert "FAIL exit=124" in result.stdout
+        assert "PASSED" in (output / "001" / "pytest.log").read_text()
+
+
 @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"