diff --git a/.github/ci-paths.toml b/.github/ci-paths.toml
index f9f2a53bf4..cabd12f66b 100644
--- a/.github/ci-paths.toml
+++ b/.github/ci-paths.toml
@@ -31,10 +31,11 @@ tests = [
 ]
 evaluation = ["evaluation/**"]
 database = [
+  "tests/seekdb_support.py",
   "tests/builtin/persistence/**", "tests/builtin/*native_code*", "tests/builtin/runtime/*recall*",
   "tests/builtin/runtime/test_statistics.py", "tests/builtin/runtime/test_model_usage_recorder.py",
   "tests/builtin/runtime/test_prepared_context.py", "tests/e2e/*atomic*", "tests/e2e/*mysql*",
-  "tests/e2e/*native_code*", "tests/e2e/test_recall_sufficiency_gate.py",
+  "tests/e2e/*native_code*", "tests/e2e/test_seekdb_restart_recovery.py", "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",
   ".github/workflows/e2e-harness.yml", ".github/actions/setup-upgrade-oceanbase/**",
@@ -114,7 +115,13 @@ release_schema = [
   "tests/release_upgrades/test_full_server.py", ".github/workflows/capture-release-schema.yml",
   ".github/actions/setup-upgrade-oceanbase/**",
 ]
-seekdb_lock = ["scripts/probe_seekdb_migration_lock.py", ".github/workflows/seekdb-lock-probe.yml", ".github/workflows/seekdb-lock-observation.yml"]
+seekdb_lock = [
+  "scripts/probe_seekdb_migration_lock.py", ".github/workflows/seekdb-lock-probe.yml",
+  ".github/workflows/seekdb-lock-observation.yml", "tests/seekdb_support.py",
+  "tests/test_seekdb_migration_probe.py", "tests/e2e/test_seekdb_restart_recovery.py",
+  "tests/builtin/persistence/test_mysql_migration_runner.py",
+  "src/powercontext/builtin/persistence/**", "pyproject.toml", "uv.lock",
+]
 skill_guidance = [
   "evaluation/skills/skill-up/**",
   "integrations/claude-code/plugins/powercontext/skills/powercontext-project-context/**",
diff --git a/.github/workflows/master.yml b/.github/workflows/master.yml
index 95679a1bbe..6acc851d96 100644
--- a/.github/workflows/master.yml
+++ b/.github/workflows/master.yml
@@ -477,6 +477,7 @@ jobs:
       fail-fast: false
       matrix:
         python-version: ["3.11", "3.14"]
+        migration-shard: [0, 1]
     steps:
       - name: Check out
         uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
@@ -490,9 +491,11 @@ jobs:
           python-version: ${{ matrix.python-version }}
 
       - name: Test code indexing with embedded seekdb
+        if: matrix.migration-shard == 0
         run: make code-seekdb-test
 
       - name: Test database bootstrap with embedded seekdb without skips
+        if: matrix.migration-shard == 0
         # Two 360s native-process deadlines still leave room for the remaining
         # bootstrap cases and diagnostic upload before the step deadline.
         timeout-minutes: 20
@@ -518,6 +521,7 @@ jobs:
           PYCODE
 
       - name: Test batched archive validation with embedded seekdb
+        if: matrix.migration-shard == 0
         timeout-minutes: 5
         run: |
           uv run --locked --extra seekdb python -c "import pylibseekdb"
@@ -526,10 +530,11 @@ jobs:
 
       - name: Test unified migrations with embedded seekdb
         if: ${{ !cancelled() && steps.seekdb_setup.outcome == 'success' }}
-        # Full-Server cases run in separate processes and exceed fifteen minutes
-        # in aggregate; each runner keeps its 180-second deadline below.
+        # Last-client shutdown adds native startup work. Partition runner nodes;
+        # every node keeps its 180-second deadline.
         timeout-minutes: 25
         env:
+          SEEKDB_MIGRATION_SHARD: ${{ matrix.migration-shard }}
           POWERCONTEXT_TEST_MIGRATION_SEEKDB: "1"
           POWERCONTEXT_TEST_MIGRATION_SQL_LOG: "1"
           PYTHONUNBUFFERED: "1"
@@ -549,16 +554,18 @@ jobs:
           }
           trap collect_diagnostics EXIT
           df -h . /tmp
-          uv run --locked --extra seekdb pytest -vv \
-            -o faulthandler_timeout=120 --durations=20 \
-            tests/builtin/persistence/test_database_migration_backends.py \
-            tests/builtin/persistence/test_database_backup_providers.py \
-            tests/builtin/persistence/test_database_migration_connections.py \
-            tests/e2e/test_topic_memory_recovery_migration.py \
-            tests/builtin/persistence/test_mysql_migration_checks.py \
-            tests/builtin/persistence/test_mysql_migration_schema.py -k seekdb \
-            2>&1 | tee seekdb-migrations.log
-          uv run --locked --extra seekdb pytest --collect-only -q --color=no \
+          if [ "$SEEKDB_MIGRATION_SHARD" -eq 0 ]; then
+            uv run --locked --extra seekdb --extra code pytest -vv \
+              -o faulthandler_timeout=120 --durations=20 \
+              tests/builtin/persistence/test_database_migration_backends.py \
+              tests/builtin/persistence/test_database_backup_providers.py \
+              tests/builtin/persistence/test_database_migration_connections.py \
+              tests/e2e/test_topic_memory_recovery_migration.py \
+              tests/builtin/persistence/test_mysql_migration_checks.py \
+              tests/builtin/persistence/test_mysql_migration_schema.py -k seekdb \
+              2>&1 | tee seekdb-migrations.log
+          fi
+          uv run --locked --extra seekdb --extra code pytest --collect-only -q --color=no \
             tests/builtin/persistence/test_mysql_migration_runner.py -k seekdb \
             2>&1 | tee seekdb-runner-collection.log | tee -a seekdb-migrations.log
           mapfile -t seekdb_runner_nodes < <(
@@ -569,25 +576,31 @@ jobs:
             echo "No seekdb migration runner tests were collected" >&2
             exit 1
           fi
-          for seekdb_runner_node in "${seekdb_runner_nodes[@]}"; do
-            uv run --locked --extra seekdb python .github/scripts/run_seekdb_migration_test.py \
+          for index in "${!seekdb_runner_nodes[@]}"; do
+            if [ "$((index % 2))" -ne "$SEEKDB_MIGRATION_SHARD" ]; then
+              continue
+            fi
+            seekdb_runner_node="${seekdb_runner_nodes[$index]}"
+            uv run --locked --extra seekdb --extra code python .github/scripts/run_seekdb_migration_test.py \
               "$seekdb_runner_node" \
               2>&1 | tee -a seekdb-migrations.log
           done
           # These are required independent runs, not retries: any failure stops
           # the job. Exercise the lifecycle race on both supported matrix ends.
-          for repetition in 2 3; do
-            echo "Baseline adoption stability run $repetition/3" | tee -a seekdb-migrations.log
-            uv run --locked --extra seekdb python .github/scripts/run_seekdb_migration_test.py \
-              'tests/builtin/persistence/test_mysql_migration_runner.py::test_mysql_baseline_adoption_recovers_an_empty_committed_version_table[seekdb]' \
-              2>&1 | tee -a seekdb-migrations.log
-          done
+          if [ "$SEEKDB_MIGRATION_SHARD" -eq 0 ]; then
+            for repetition in 2 3; do
+              echo "Baseline adoption stability run $repetition/3" | tee -a seekdb-migrations.log
+              uv run --locked --extra seekdb --extra code python .github/scripts/run_seekdb_migration_test.py \
+                'tests/builtin/persistence/test_mysql_migration_runner.py::test_mysql_baseline_adoption_recovers_an_empty_committed_version_table[seekdb]' \
+                2>&1 | tee -a seekdb-migrations.log
+            done
+          fi
 
       - name: Upload seekdb bootstrap diagnostics
         if: always()
         uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7
         with:
-          name: seekdb-bootstrap-${{ matrix.python-version }}
+          name: seekdb-bootstrap-${{ matrix.python-version }}-${{ matrix.migration-shard }}
           path: .powercontext/database-bootstrap-ci/
           if-no-files-found: ignore
 
@@ -595,7 +608,7 @@ jobs:
         if: always()
         uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7
         with:
-          name: seekdb-migrations-${{ matrix.python-version }}
+          name: seekdb-migrations-${{ matrix.python-version }}-${{ matrix.migration-shard }}
           path: |
             seekdb-migrations.log
             seekdb-native.log
diff --git a/.github/workflows/seekdb-lock-probe.yml b/.github/workflows/seekdb-lock-probe.yml
index 7d5a4ebc70..a961dc86a4 100644
--- a/.github/workflows/seekdb-lock-probe.yml
+++ b/.github/workflows/seekdb-lock-probe.yml
@@ -28,22 +28,41 @@ jobs:
     needs: changes
     if: ${{ !cancelled() && ((inputs.suites == '' && needs.changes.result != 'success') || fromJSON(inputs.suites || needs.changes.outputs.suites || '{}').seekdb_lock) }}
     runs-on: ubuntu-latest
-    timeout-minutes: 20
+    timeout-minutes: 40
+    strategy:
+      fail-fast: false
+      matrix:
+        python: ['3.11', '3.14']
+        scenario: [migration, restart-recovery]
     steps:
       - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
         with:
           persist-credentials: false
       - uses: ./.github/actions/setup-python-env
         with:
-          python-version: '3.11'
+          python-version: ${{ matrix.python }}
       - name: Reproduce migration lifecycle with complete native evidence
-        run: uv run --locked --extra seekdb python scripts/probe_seekdb_migration_lock.py --attempts 12
+        env:
+          SEEKDB_PROBE_SCENARIO: ${{ matrix.scenario }}
+        run: |
+          restart_args=()
+          if [ "$SEEKDB_PROBE_SCENARIO" = restart-recovery ]; then
+            restart_args+=(--restart)
+          fi
+          uv run --locked --extra seekdb python scripts/probe_seekdb_migration_lock.py \
+            --scenario "$SEEKDB_PROBE_SCENARIO" "${restart_args[@]}" --attempts 12 --timeout 180
       - name: Upload native logs, lifecycle and live transaction snapshots
         if: always()
         uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7
         with:
-          name: seekdb-lock-probe-${{ github.sha }}
-          path: .powercontext/seekdb-lock-probe
+          name: seekdb-lock-probe-${{ matrix.python }}-${{ matrix.scenario }}-${{ github.sha }}
+          path: |
+            .powercontext/seekdb-lock-probe/environment.json
+            .powercontext/seekdb-lock-probe/attempt-*/*.json
+            .powercontext/seekdb-lock-probe/attempt-*/*.jsonl
+            .powercontext/seekdb-lock-probe/attempt-*/*.xml
+            .powercontext/seekdb-lock-probe/attempt-*/*.log
+            .powercontext/seekdb-lock-probe/attempt-*/*.gz
           include-hidden-files: true
           if-no-files-found: error
           retention-days: 7
diff --git a/docs/en/docs/operate/database-migrations.md b/docs/en/docs/operate/database-migrations.md
index 95e76d08de..2ad44358c9 100644
--- a/docs/en/docs/operate/database-migrations.md
+++ b/docs/en/docs/operate/database-migrations.md
@@ -435,6 +435,53 @@ Use the latest plan's `plan_id` and `coordination.pending_run_id`. Reconfirm the
 the original recovery point with a backup of partial state. Normal upgrades still use one confirmation. Recovery requires
 the additional explicit DBA declaration; PC never invents it when showing the plan.
 
+## Native SeekDB restart and recovery acceptance
+
+The embedded engine must use `pylibseekdb>=1.4.0.post2,<2`. The native transaction recovery fix is supplied by
+that dependency. The migration checks retain `SHOW TRIGGERS` and release the last client between operations.
+Run the following from the repository root, separately under Python 3.11 and Python 3.14:
+
+```bash
+uv sync --locked --extra seekdb
+uv run --locked --extra seekdb python scripts/probe_seekdb_migration_lock.py \
+  --scenario migration --attempts 12 --timeout 180 --output .powercontext/seekdb-migration
+uv run --locked --extra seekdb python scripts/probe_seekdb_migration_lock.py \
+  --scenario restart-recovery --restart --attempts 12 --timeout 180 --output .powercontext/seekdb-restart
+```
+
+Use a new output directory for each run. The migration scenario selects
+`test_mysql_committed_revision_recovers_before_evidence_advance[seekdb]` with its natural client lifecycle. Releasing a
+client does not imply that the native process has already exited; a rapid reopen may reuse it. The restart scenario
+checks metadata inspection, commit, confirmed last-client shutdown, baseline DDL and preserved content and tags after
+reopening. Its test waits for kernel process-exit notification even when invoked without the probe. Each attempt must
+report exactly one passing test, with no skip, error or timeout. `--restart` explicitly waits up to 10 seconds for native
+exit on each close, then requires distinct process IDs and an exit record before each subsequent open. A timeout fails
+the attempt; it does not retry. Evidence includes the Python and binding versions, native binary SHA-256, instance
+lifetimes, SQL trace, compressed native logs and live transaction snapshots for slow attempts. The separate extended
+observation workflow is diagnostic; its longer deadline does not replace the ordinary acceptance deadline.
+
+Both explicit restart gates run on Linux and observe kernel process-exit notifications before each reopen;
+neither uses a fixed startup delay. The tests prefer Python's `os.pidfd_open`, then the optional libc wrapper.
+When both are absent, they use the Linux syscall on the locked wheels' 64-bit x86_64 and aarch64 architectures.
+Kernel errors and exit timeouts still fail acceptance. For recovery acceptance, first reproduce the stall with an isolated
+`1.4.0.post1` environment and `--scenario restart-recovery --keep-failed-storage`.
+Stop its engine before freezing the failed `tmp/pc-*/seekdb`
+directory. Keep the original immutable. This scenario creates synthetic data in the `powercontext` database, including
+the `preserved` content and `Keep` tag expected by the fixture test. A healthy database or a fixture with different
+schema or sentinels cannot establish recovery of this defect. Compare the old and fixed engines on separate copies of
+the same frozen fault fixture:
+
+```bash
+uv run --locked --extra seekdb python scripts/probe_seekdb_migration_lock.py \
+  --scenario fixture-recovery --fixture /absolute/path/to/frozen/seekdb \
+  --restart --attempts 1 --timeout 75 --output .powercontext/seekdb-fixture
+```
+
+The fixture gate opens only a disposable copy, creates and commits the baseline revision, verifies the original data,
+reopens to verify it and commit a new write, then reopens once more to read that write. Preserve the fixture and the
+old-engine failure evidence locally; CI uploads diagnostic files, not physical databases. Do not infer upgrade support
+from these gates: the full installed-release upgrade matrix remains required.
+
 ## Backup choices
 
 | Policy | Behavior | Options |
diff --git a/docs/zh/docs/operate/database-migrations.md b/docs/zh/docs/operate/database-migrations.md
index bf6b9ead5f..0f43258f79 100644
--- a/docs/zh/docs/operate/database-migrations.md
+++ b/docs/zh/docs/operate/database-migrations.md
@@ -337,6 +337,47 @@ powercontext server db-migrate verify --env-file oceanbase.env \
 
 使用最新计划中的 `plan_id` 和 `coordination.pending_run_id`。重新声明既有手动备份，不用部分状态的新备份替换原恢复点。正常升级仍只确认一次；中断恢复另需用户显式作出 DBA 声明，PC 展示计划时不会自动替用户确认。
 
+## 原生 SeekDB 重启与恢复验收
+
+嵌入式引擎必须使用 `pylibseekdb>=1.4.0.post2,<2`，事务恢复修复由该原生依赖提供。迁移检查保留
+`SHOW TRIGGERS`，并在操作之间释放最后一个客户端。分别在 Python 3.11 和 Python 3.14 环境中，从仓库根目录执行：
+
+```bash
+uv sync --locked --extra seekdb
+uv run --locked --extra seekdb python scripts/probe_seekdb_migration_lock.py \
+  --scenario migration --attempts 12 --timeout 180 --output .powercontext/seekdb-migration
+uv run --locked --extra seekdb python scripts/probe_seekdb_migration_lock.py \
+  --scenario restart-recovery --restart --attempts 12 --timeout 180 --output .powercontext/seekdb-restart
+```
+
+每次运行使用新的输出目录。migration 场景选择原 issue 的
+`test_mysql_committed_revision_recovers_before_evidence_advance[seekdb]` 节点，保留自然客户端生命周期。释放客户端不代表
+原生进程已经退出，紧接着打开可能复用该进程。restart-recovery 场景检查元数据查询、提交、确认最后一个客户端关闭后
+进程退出、基线 DDL，以及重开后内容和标签的保留；即使不通过探针调用，该测试也会等待内核进程退出通知。
+每次必须恰好一个测试通过，跳过、报错和超时均不算通过。`--restart` 在每次关闭时最多等待 10 秒确认原生进程退出，
+再要求各代 PID 不同，且下一次打开之前存在退出记录。等待超时直接失败，不自动重试。
+证据包括 Python 和绑定版本、原生二进制 SHA-256、实例生命周期、
+SQL 日志、压缩后的原生日志，以及慢执行时的事务快照。独立的延长观察工作流仅用于诊断，不能用它的较长时限替代正常验收时限。
+
+两个显式重开验收均在 Linux 上执行，每次重开前通过内核退出事件确认上一代进程已结束，不使用固定启动等待。
+测试优先使用 Python 的 `os.pidfd_open`，其次使用可选的 libc 封装；两者均缺失时，仅在锁定 wheels 对应的
+64 位 x86_64 和 aarch64 架构上调用 Linux syscall。内核报错或退出等待超时仍会导致验收失败。
+故障库验收应先在隔离的 `1.4.0.post1` 环境中，
+使用 `--scenario restart-recovery --keep-failed-storage` 复现阻塞。
+停止引擎后，冻结失败输出中的 `tmp/pc-*/seekdb` 目录，并保持原件不变。此场景在 `powercontext` 数据库中创建合成数据，
+包括故障库测试要求的 `preserved` 内容和 `Keep` 标签。健康库、结构或哨兵数据不同的库，不能用于证明该缺陷已恢复。
+使用同一冻结故障库的独立副本，对比旧引擎与修复后的引擎：
+
+```bash
+uv run --locked --extra seekdb python scripts/probe_seekdb_migration_lock.py \
+  --scenario fixture-recovery --fixture /absolute/path/to/frozen/seekdb \
+  --restart --attempts 1 --timeout 75 --output .powercontext/seekdb-fixture
+```
+
+故障库门禁只打开临时副本：创建并提交基线版本表、验证原始数据；重开后再次验证并提交新写入；第三次打开时读回该写入。
+故障库原件与旧版失败证据保存在本地；CI 只上传诊断文件，不上传物理数据库。这些门禁不代表完整升级支持，
+发布仍须通过安装产物的全部升级矩阵。
+
 ## 备份选择
 
 | 策略 | 行为 | 参数 |
diff --git a/scripts/probe_seekdb_migration_lock.py b/scripts/probe_seekdb_migration_lock.py
index 9839491bce..31057ac59e 100644
--- a/scripts/probe_seekdb_migration_lock.py
+++ b/scripts/probe_seekdb_migration_lock.py
@@ -18,14 +18,18 @@
 
 import argparse
 import gzip
+import hashlib
 import json
 import os
 import platform
 import shutil
+import signal
 import subprocess
 import sys
 import time
+from contextlib import suppress
 from datetime import UTC, datetime
+from importlib import import_module
 from importlib.metadata import version
 from pathlib import Path
 from types import ModuleType
@@ -36,46 +40,70 @@
 
 NODE = (
     "tests/builtin/persistence/test_mysql_migration_runner.py::"
-    "test_mysql_baseline_adoption_recovers_an_empty_committed_version_table[seekdb]"
+    "test_mysql_committed_revision_recovers_before_evidence_advance[seekdb]"
 )
 
+SCENARIOS = {
+    "migration": NODE,
+    "restart-recovery": (
+        "tests/e2e/test_seekdb_restart_recovery.py::test_metadata_inspection_survives_last_client_restarts"
+    ),
+    "fixture-recovery": (
+        "tests/e2e/test_seekdb_restart_recovery.py::test_saved_fault_recovers_and_remains_usable_after_reopen"
+    ),
+}
+
 
 def record(directory: Path, event: str, **values: object) -> None:
     with (directory / "lifecycle.jsonl").open("a") as output:
-        output.write(json.dumps({"time": datetime.now(UTC).isoformat(), "event": event, **values}) + "\n")
+        output.write(
+            json.dumps({"time": datetime.now(UTC).isoformat(), "event": event, **values}, ensure_ascii=False) + "\n"
+        )
 
 
 def pytest_sessionstart(session: pytest.Session) -> None:
     """Optional pytest plugin: observe only this subprocess's native instance lifetime."""
     from powercontext.builtin.persistence.seekdb import profile
+    from tests.seekdb_support import RestartingInstance
 
     directory = Path(os.environ["POWERCONTEXT_SEEKDB_PROBE"])
+    restart = os.environ.get("POWERCONTEXT_SEEKDB_REQUIRE_RESTART") == "1"
     original = profile._open_instance
 
     class ObservedInstance:
-        def __init__(self, instance: Any, path: Path) -> None:
+        def __init__(self, instance: Any, path: Path, native_pid: int) -> None:
             self.instance = instance
             self.path = path
+            self.native_pid = native_pid
 
         def connection_options(self) -> Any:
             return self.instance.connection_options()
 
         def close(self) -> None:
-            record(directory, "closing", path=str(self.path))
+            record(directory, "closing", path=str(self.path), native_pid=self.native_pid)
             try:
                 self.instance.close()
+                if restart:
+                    record(directory, "native_exited", path=str(self.path), native_pid=self.native_pid)
             finally:
-                record(directory, "close_returned", path=str(self.path))
+                record(directory, "close_returned", path=str(self.path), native_pid=self.native_pid)
 
     async def observed_open(module: ModuleType, path: Path) -> profile._SeekDBInstance:
         record(directory, "opening", path=str(path))
         instance = await original(module, path)
+        try:
+            native_pid = int((path / "run/seekdb.pid").read_text().strip())
+        except (OSError, ValueError):
+            instance.close()
+            raise
+        if restart:
+            instance = RestartingInstance(instance, native_pid)
         options = dict(instance.connection_options())
         # Embedded scratch instance only. Never capture credentials or external connection settings.
         socket = options.get("unix_socket")
         (directory / "socket.json").write_text(json.dumps({"unix_socket": socket}))
-        record(directory, "opened", path=str(path))
-        return ObservedInstance(instance, path)
+        record(directory, "opened", path=str(path), native_pid=native_pid)
+        return ObservedInstance(instance, path, native_pid)
 
     patch = pytest.MonkeyPatch()
     patch.setattr(profile, "_open_instance", observed_open)
@@ -171,7 +199,50 @@ def successful_report(report: Path, code: int) -> bool:
     )
 
 
-def run_attempt(directory: Path, timeout: int) -> bool:
+def stop_process(process: subprocess.Popen[bytes]) -> None:
+    """Bound shutdown of a probe child, including an unresponsive native client."""
+    if process.poll() is None:
+        with suppress(ProcessLookupError):
+            os.killpg(process.pid, signal.SIGTERM)
+    try:
+        process.wait(timeout=10)
+    except subprocess.TimeoutExpired:
+        with suppress(ProcessLookupError):
+            os.killpg(process.pid, signal.SIGKILL)
+        process.wait()
+
+
+def restart_evidence(directory: Path) -> dict[str, Any]:
+    """Require confirmed exit before each subsequent native generation opens."""
+    lifecycle = directory / "lifecycle.jsonl"
+    events = [json.loads(line) for line in lifecycle.read_text().splitlines()] if lifecycle.exists() else []
+    processes: dict[str, list[int]] = {}
+    running: dict[str, int] = {}
+    valid = True
+    for event in events:
+        if event["event"] == "opened":
+            path, pid = event["path"], event["native_pid"]
+            pids = processes.setdefault(path, [])
+            valid = valid and path not in running and pid > 0 and pid not in pids
+            pids.append(pid)
+            running[path] = pid
+        elif event["event"] == "native_exited":
+            path, pid = event["path"], event["native_pid"]
+            valid = valid and running.get(path) == pid
+            running.pop(path, None)
+    passed = valid and not running and any(len(pids) > 1 for pids in processes.values())
+    return {"passed": passed, "processes": processes}
+
+
+def run_attempt(
+    directory: Path,
+    timeout: int,
+    *,
+    node: str = NODE,
+    restart: bool = False,
+    fixture: Path | None = None,
+    keep_failed_storage: bool = False,
+) -> bool:
     directory.mkdir(parents=True)
     temporary = directory / "tmp"
     temporary.mkdir()
@@ -179,8 +250,12 @@ def run_attempt(directory: Path, timeout: int) -> bool:
         os.environ,
         TMPDIR=str(temporary),
         POWERCONTEXT_SEEKDB_PROBE=str(directory),
+        POWERCONTEXT_SEEKDB_REQUIRE_RESTART=str(int(restart)),
         POWERCONTEXT_TEST_MIGRATION_SEEKDB="1",
         POWERCONTEXT_TEST_MIGRATION_SQL_LOG="1",
+        POWERCONTEXT_TEST_SEEKDB_RESTART_RECOVERY="1",
+        POWERCONTEXT_SEEKDB_RECOVERY_FIXTURE=str(fixture or ""),
+        POWERCONTEXT_SEEKDB_KEEP_FAILED_STORAGE=str(int(keep_failed_storage)),
     )
     report = directory / "junit.xml"
     command = [
@@ -194,9 +269,9 @@ def run_attempt(directory: Path, timeout: int) -> bool:
         "-vv",
         "--log-cli-level=INFO",
         "-o",
-        "faulthandler_timeout=120",
+        f"faulthandler_timeout={min(120, max(1, timeout // 2))}",
         f"--junitxml={report}",
-        NODE,
+        node,
     ]
     logs = NativeLogs(temporary, directory)
     snapshots: list[tuple[subprocess.Popen[bytes], Any, float]] = []
@@ -204,16 +279,24 @@ def run_attempt(directory: Path, timeout: int) -> bool:
     next_snapshot = 45.0
     timed_out = False
     process = None
+    code = -1
+    passed = False
+    restarts = None
     try:
         with (directory / "pytest.log").open("w") as output:
-            process = subprocess.Popen(command, env=environment, stdout=output, stderr=subprocess.STDOUT)  # noqa: S603
+            process = subprocess.Popen(  # noqa: S603
+                command, env=environment, stdout=output, stderr=subprocess.STDOUT, start_new_session=True
+            )
             while process.poll() is None:
                 logs.poll()
                 elapsed = time.monotonic() - started
                 if elapsed >= next_snapshot:
                     log = (directory / f"snapshot-{int(next_snapshot)}.jsonl").open("w")
                     probe = subprocess.Popen(  # noqa: S603
-                        [sys.executable, __file__, "--snapshot", str(directory)], stdout=log, stderr=subprocess.STDOUT
+                        [sys.executable, __file__, "--snapshot", str(directory)],
+                        stdout=log,
+                        stderr=subprocess.STDOUT,
+                        start_new_session=True,
                     )
                     snapshots.append((probe, log, time.monotonic()))
                     next_snapshot += 45
@@ -222,60 +305,126 @@ def run_attempt(directory: Path, timeout: int) -> bool:
                         probe.kill()
                 if elapsed >= timeout:
                     timed_out = True
-                    process.terminate()
-                    try:
-                        process.wait(timeout=10)
-                    except subprocess.TimeoutExpired:
-                        process.kill()
+                    stop_process(process)
                     break
                 time.sleep(0.1)
             code = process.wait()
+        passed = not timed_out and successful_report(report, code)
+        if restart and passed:
+            restarts = restart_evidence(directory)
+            passed = restarts["passed"]
     finally:
-        if process is not None and process.poll() is None:
-            process.kill()
-            process.wait()
+        if process is not None:
+            stop_process(process)
         for probe, output, _ in snapshots:
-            if probe.poll() is None:
-                probe.kill()
-            probe.wait()
+            stop_process(probe)
             output.close()
         logs.close()
         # Only remove this attempt's disposable engine files, after preserving native logs.
-        shutil.rmtree(temporary)
-    passed = not timed_out and successful_report(report, code)
-    result = {"passed": passed, "exit_code": code, "timed_out": timed_out, "seconds": time.monotonic() - started}
+        if not keep_failed_storage or passed:
+            shutil.rmtree(temporary)
+    result = {
+        "passed": passed,
+        "exit_code": code,
+        "timed_out": timed_out,
+        "seconds": time.monotonic() - started,
+        "restarts": restarts,
+    }
     (directory / "result.json").write_text(json.dumps(result, indent=2) + "\n")
     print(directory.name, json.dumps(result), flush=True)
     return passed
 
 
-def main() -> int:
+def positive_integer(value: str) -> int:
+    parsed = int(value)
+    if parsed < 1:
+        message = "must be positive"
+        raise argparse.ArgumentTypeError(message)
+    return parsed
+
+
+def parse_arguments() -> argparse.Namespace:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--snapshot", type=Path)
     parser.add_argument("--output", type=Path, default=Path(".powercontext/seekdb-lock-probe"))
-    parser.add_argument("--attempts", type=int, default=12)
-    parser.add_argument("--timeout", type=int, default=180)
+    parser.add_argument("--attempts", type=positive_integer, default=12)
+    parser.add_argument("--node", help="Exact pytest node, overriding the selected scenario.")
+    parser.add_argument("--scenario", choices=SCENARIOS, default="migration")
+    parser.add_argument(
+        "--fixture",
+        type=Path,
+        help="Frozen synthetic baseline with the recovery test's sentinels; copied before opening.",
+    )
+    parser.add_argument(
+        "--keep-failed-storage", action="store_true", help="Retain failed scratch storage for recovery acceptance."
+    )
+    parser.add_argument(
+        "--restart", action="store_true", help="Wait for Linux native process exit before each reopen and verify it."
+    )
+    parser.add_argument("--timeout", type=positive_integer, default=180)
     arguments = parser.parse_args()
+    if arguments.snapshot:
+        return arguments
+    if arguments.scenario == "fixture-recovery" and arguments.fixture is None:
+        parser.error("--scenario fixture-recovery requires --fixture")
+    if arguments.fixture is not None:
+        arguments.fixture = arguments.fixture.resolve()
+        if not (arguments.fixture / "etc/seekdb.data_version.bin").is_file():
+            parser.error("--fixture must be a recognized native seekdb directory")
+    node = arguments.node or SCENARIOS[arguments.scenario]
+    directory = arguments.output.resolve()
+    if arguments.fixture is not None and directory.is_relative_to(arguments.fixture):
+        parser.error("--output must be outside the immutable --fixture directory")
+    if directory.exists():
+        parser.error("--output must be a new directory to preserve previous evidence")
+    arguments.output = directory
+    arguments.node = node
+    return arguments
+
+
+def main() -> int:
+    arguments = parse_arguments()
     if arguments.snapshot:
         snapshot(arguments.snapshot)
         return 0
-    directory = arguments.output.resolve()
-    directory.mkdir(parents=True, exist_ok=True)
+    directory = arguments.output
+    node = arguments.node
+    directory.mkdir(parents=True)
+    binding = import_module("pylibseekdb")
+    binding_file = binding.__file__
+    if binding_file is None:
+        message = "The SeekDB binding has no module file; cannot locate its native engine"
+        raise RuntimeError(message)
+    native_engine = Path(binding_file).with_name("seekdb")
+    with native_engine.open("rb") as engine_file:
+        native_digest = hashlib.file_digest(engine_file, "sha256").hexdigest()
     (directory / "environment.json").write_text(
         json.dumps(
             {
                 "platform": platform.platform(),
                 "python": sys.version,
                 "pylibseekdb": version("pylibseekdb"),
-                "node": NODE,
+                "seekdb_engine": str(native_engine),
+                "seekdb_engine_sha256": native_digest,
+                "node": node,
+                "scenario": arguments.scenario,
+                "restart": arguments.restart,
             },
             indent=2,
+            ensure_ascii=False,
         )
         + "\n"
     )
     for attempt in range(1, arguments.attempts + 1):
         attempt_directory = directory / f"attempt-{attempt:02d}"
-        if not run_attempt(attempt_directory, arguments.timeout):
+        if not run_attempt(
+            attempt_directory,
+            arguments.timeout,
+            node=node,
+            restart=arguments.restart,
+            fixture=arguments.fixture,
+            keep_failed_storage=arguments.keep_failed_storage,
+        ):
             return 1
         # Keep a complete successful baseline and every failure. Other successful
         # repeats retain their Python trace, lifecycle and test result only.
diff --git a/tests/builtin/persistence/test_mysql_migration_checks.py b/tests/builtin/persistence/test_mysql_migration_checks.py
index c0e4a7d6ce..cb50afac8e 100644
--- a/tests/builtin/persistence/test_mysql_migration_checks.py
+++ b/tests/builtin/persistence/test_mysql_migration_checks.py
@@ -30,7 +30,6 @@
 from powercontext.builtin.persistence.migrations.mysql import MySQLMigrationRunner
 from powercontext.builtin.persistence.migrations.mysql_schema import MySQLSchema, _expression_sql, mysql_inventory
 from powercontext.builtin.persistence.seekdb import SeekDBConfig
-from tests.builtin.persistence.test_mysql_migration_runner import _seekdb_scenario_lifetime
 from tests.builtin.persistence.test_mysql_migration_schema import _inventory, _ReflectedDatabase
 
 _FIXTURE = Path(__file__).parents[2] / "fixtures/database_migrations/schemas/seekdb_v1_3_0_checks.json"
@@ -179,9 +178,7 @@ def reflect(connection, _identity, guard):
     [("a % 2 = 0", "INTEGER", 2, 3), ("a = '100%s%%'", "VARCHAR(32)", "100%s%%", "different")],
     ids=["remainder", "percent_literal"],
 )
-def test_real_seekdb_percent_checks_validate_data_and_plan(
-    short_tmp_path, monkeypatch, expression, column_type, valid, invalid
-):
+def test_real_seekdb_percent_checks_validate_data_and_plan(short_tmp_path, expression, column_type, valid, invalid):
     pytest.importorskip("pylibseekdb")
     # Extend an isolated bundle, preserving the shipped layouts and their eligibility.
     directory = short_tmp_path / "bundle"
@@ -222,23 +219,22 @@ def invalidate(connection, _identity, guard):
         with pytest.raises(MigrationError, match=r"data_integrity_error.*CHECK"):
             schema.validate_data(connection, "p0001")
 
-    with _seekdb_scenario_lifetime(monkeypatch):
-        fresh = MySQLMigrationRunner(
-            MaintenanceConnections(SeekDBConfig(path=short_tmp_path / "fresh_seekdb")),
-            bundle,
-            evidence_directory=short_tmp_path / "fresh_evidence",
-        )
-        fresh_plan = fresh.plan(backup_policy="skip")
-        fresh.apply(
-            plan_id=fresh_plan.plan_id,
-            accepted=True,
-            maintenance_confirmed=True,
-            backup_policy="skip",
-            accept_no_backup=True,
-        )
-        assert fresh.verify().state == "ready"
-        connections.run(prepare, writable=True)
-        assert runner.plan().state == "migration_required"
-        connections.run(invalidate, writable=True)
-        with pytest.raises(MigrationError, match=r"data_integrity_error.*CHECK"):
-            runner.plan()
+    fresh = MySQLMigrationRunner(
+        MaintenanceConnections(SeekDBConfig(path=short_tmp_path / "fresh_seekdb")),
+        bundle,
+        evidence_directory=short_tmp_path / "fresh_evidence",
+    )
+    fresh_plan = fresh.plan(backup_policy="skip")
+    fresh.apply(
+        plan_id=fresh_plan.plan_id,
+        accepted=True,
+        maintenance_confirmed=True,
+        backup_policy="skip",
+        accept_no_backup=True,
+    )
+    assert fresh.verify().state == "ready"
+    connections.run(prepare, writable=True)
+    assert runner.plan().state == "migration_required"
+    connections.run(invalidate, writable=True)
+    with pytest.raises(MigrationError, match=r"data_integrity_error.*CHECK"):
+        runner.plan()
diff --git a/tests/builtin/persistence/test_mysql_migration_runner.py b/tests/builtin/persistence/test_mysql_migration_runner.py
index f50ab3d23b..90fe367760 100644
--- a/tests/builtin/persistence/test_mysql_migration_runner.py
+++ b/tests/builtin/persistence/test_mysql_migration_runner.py
@@ -27,6 +27,7 @@
 from contextlib import contextmanager
 from dataclasses import dataclass
 from datetime import UTC, datetime
+from importlib import import_module
 from pathlib import Path
 from typing import Literal, cast
 from unittest.mock import Mock
@@ -126,44 +127,19 @@ async def manage(*, create: bool) -> None:
                 pytest.fail("OceanBase owned test database cleanup failed", pytrace=False)
 
 
-@contextmanager
-def _seekdb_scenario_lifetime(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
-    binding = pytest.importorskip("pylibseekdb")
-    open_instance = binding.aopen
-    instances = {}
-
-    async def open_with_scenario_lifetime(path: str):
-        # Keep a real native lease across this scenario's independent
-        # maintenance connections. Rapid last-client exits in seekdb 1.4
-        # can strand a background DDL lock on restart. Engine restart and
-        # cross-process ownership have separate connection acceptance tests.
-        # Acquire lazily so inspecting a missing target still creates nothing.
-        if path not in instances:
-            instances[path] = await open_instance(path)
-        return await open_instance(path)
-
-    monkeypatch.setattr(binding, "aopen", open_with_scenario_lifetime)
-    try:
-        yield
-    finally:
-        for instance in instances.values():
-            instance.close()
-
-
 @pytest.fixture(params=("seekdb", "oceanbase"), ids=("seekdb", "oceanbase"))
 def backend(
     request: pytest.FixtureRequest,
     short_tmp_path: Path,
     caplog: pytest.LogCaptureFixture,
-    monkeypatch: pytest.MonkeyPatch,
 ) -> Iterator[_BackendTarget]:
     if os.environ.get("POWERCONTEXT_TEST_MIGRATION_SQL_LOG") == "1":
         caplog.set_level(logging.INFO, logger="sqlalchemy.engine")
     if request.param == "seekdb":
         if os.environ.get("POWERCONTEXT_TEST_MIGRATION_SEEKDB") != "1":
             pytest.skip("real seekdb probe not enabled")
-        with _seekdb_scenario_lifetime(monkeypatch):
-            yield _BackendTarget(SeekDBConfig(path=short_tmp_path / "seekdb"), short_tmp_path / "evidence")
+        import_module("pylibseekdb")
+        yield _BackendTarget(SeekDBConfig(path=short_tmp_path / "seekdb"), short_tmp_path / "evidence")
         return
     value = os.environ.get("POWERCONTEXT_TEST_MIGRATION_OCEANBASE_URL")
     if not value:
@@ -1218,9 +1194,24 @@ def interrupt(journal, evidence) -> None:
     recovery = replacement.plan(backup_policy="manual")
     assert recovery.source_revision == "p0002"
     replacement = _acknowledge_known_completed_interruption(backend, recovery)
-    assert _apply(replacement, policy="manual").state == "ready"
+    upgrade = replacement._upgrade
+    recovered_p0003 = False
+
+    def observe_recovery(connection, config, source, target, guard) -> None:
+        nonlocal recovered_p0003
+        upgrade(connection, config, source, target, guard)
+        if target == "p0003":
+            assert connection.exec_driver_sql("SELECT version_num FROM pc_schema_revision").scalar_one() == "p0003"
+            recovered_p0003 = True
+
+    with monkeypatch.context() as patch:
+        patch.setattr(replacement, "_upgrade", observe_recovery)
+        assert _apply(replacement, policy="manual").state == "ready"
+    assert recovered_p0003, "Recovery did not reach p0003"
     assert replacement.verify().revision == "p0004"
-    assert _read_state(backend)["content"] == [(b'{"content":"preserved"}',)]
+    state = _read_state(backend)
+    assert state["content"] == [(b'{"content":"preserved"}',)]
+    assert state["tags"] == [("Keep",)]
 
 
 def test_mysql_completed_head_with_an_interrupted_run_requires_exact_acknowledgement(backend: _BackendTarget) -> None:
diff --git a/tests/e2e/test_seekdb_restart_recovery.py b/tests/e2e/test_seekdb_restart_recovery.py
new file mode 100644
index 0000000000..7a3baa3038
--- /dev/null
+++ b/tests/e2e/test_seekdb_restart_recovery.py
@@ -0,0 +1,192 @@
+# 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.
+
+"""Native restart/recovery release gates; no additional instance lease is held."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import shutil
+import tempfile
+from collections.abc import Callable, Iterator
+from contextlib import contextmanager
+from importlib import import_module
+from pathlib import Path
+
+import pytest
+from sqlalchemy import Connection, text
+
+from powercontext.builtin.persistence.migrations.connections import BackendIdentity, MaintenanceConnections
+from powercontext.builtin.persistence.migrations.deployment import production_bundle
+from powercontext.builtin.persistence.seekdb import SeekDBConfig, profile
+from tests.seekdb_support import RestartingInstance
+
+pytestmark = pytest.mark.skipif(
+    os.environ.get("POWERCONTEXT_TEST_SEEKDB_RESTART_RECOVERY") != "1",
+    reason="native seekdb restart/recovery release gate not enabled",
+)
+
+
+@pytest.fixture(autouse=True)
+def native_restarts(monkeypatch: pytest.MonkeyPatch) -> None:
+    original = profile._open_instance
+
+    async def open_generation(binding, path: Path):
+        instance = await original(binding, path)
+        try:
+            pid = int((path / "run/seekdb.pid").read_text().strip())
+        except (OSError, ValueError):
+            instance.close()
+            raise
+        return RestartingInstance(instance, pid)
+
+    monkeypatch.setattr(profile, "_open_instance", open_generation)
+
+
+@pytest.fixture
+def recovery_path() -> Iterator[Path]:
+    path = Path(tempfile.mkdtemp(prefix="pc-"))
+    try:
+        yield path
+    finally:
+        # The probe owns the parent directory and removes successful attempts.
+        # Keep failures even when the client timeout lets pytest exit normally.
+        if os.environ.get("POWERCONTEXT_SEEKDB_KEEP_FAILED_STORAGE") != "1":
+            shutil.rmtree(path)
+
+
+def test_metadata_inspection_survives_last_client_restarts(recovery_path: Path) -> None:
+    import_module("pylibseekdb")
+    connections = MaintenanceConnections(SeekDBConfig(path=recovery_path / "seekdb"))
+    bundle = production_bundle()
+    schema = json.loads((bundle.snapshot_directory / "schemas/pre_dream.json").read_text())
+
+    def seed(connection: Connection | None, _identity: BackendIdentity, _guard: Callable[[], None]) -> None:
+        assert connection is not None
+        for statement in schema["mysql"]:
+            connection.exec_driver_sql(statement)
+        connection.execute(
+            text(
+                "INSERT INTO pc_artifacts (scope_id,family,artifact_id,revision,content) "
+                "VALUES ('s','memory','m',1,:content)"
+            ),
+            {"content": b'{"content":"preserved"}'},
+        )
+        connection.exec_driver_sql(
+            "INSERT INTO pc_artifact_heads (scope_id,family,artifact_id,revision) VALUES ('s','memory','m',1)"
+        )
+        connection.exec_driver_sql(
+            "INSERT INTO pc_artifact_tags "
+            "(scope_id,family,artifact_id,target_type,target_id,tag_key_hash,tag_key,tag,assigned_at) "
+            "VALUES ('s','memory','m','artifact','m',UNHEX(REPEAT('00',32)),'keep','Keep','2026-09-28 12:00:00')"
+        )
+        connection.commit()
+
+    def inspect(connection: Connection | None, _identity: BackendIdentity, _guard: Callable[[], None]) -> None:
+        assert connection is not None
+        assert connection.exec_driver_sql("SHOW TRIGGERS").all() == []
+        connection.commit()
+
+    def adopt(connection: Connection | None, _identity: BackendIdentity, _guard: Callable[[], None]) -> None:
+        assert connection is not None
+        inspect(connection, _identity, _guard)
+        connection.exec_driver_sql("START TRANSACTION")
+        connection.exec_driver_sql(
+            "CREATE TABLE pc_schema_revision (version_num VARCHAR(32) NOT NULL, "
+            "CONSTRAINT pc_schema_revision_pkc PRIMARY KEY (version_num))"
+        )
+        connection.exec_driver_sql("INSERT INTO pc_schema_revision VALUES ('p0001')")
+        connection.commit()
+
+    def verify(connection: Connection | None, _identity: BackendIdentity, _guard: Callable[[], None]) -> None:
+        assert connection is not None
+        assert connection.exec_driver_sql("SELECT version_num FROM pc_schema_revision").all() == [("p0001",)]
+        assert connection.exec_driver_sql("SELECT content FROM pc_artifacts").all() == [(b'{"content":"preserved"}',)]
+        assert connection.exec_driver_sql("SELECT tag FROM pc_artifact_tags").all() == [("Keep",)]
+
+    connections.run(seed, writable=True)
+    connections.run(inspect)
+    connections.run(inspect)
+    connections.run(adopt, writable=True)
+    connections.run(verify)
+
+
+@contextmanager
+def _native_generation(binding, path: Path):
+    instance = asyncio.run(profile._open_instance(binding, path))
+    try:
+        yield instance
+    finally:
+        instance.close()
+
+
+def test_saved_fault_recovers_and_remains_usable_after_reopen(recovery_path: Path) -> None:
+    binding = import_module("pylibseekdb")
+    pymysql = import_module("pymysql")
+    configured = os.environ.get("POWERCONTEXT_SEEKDB_RECOVERY_FIXTURE")
+    if not configured:
+        pytest.fail("enabled recovery acceptance requires an immutable native fault fixture directory", pytrace=False)
+    source = Path(configured).resolve()
+    assert (source / "etc/seekdb.data_version.bin").is_file(), "unrecognized native fault fixture"
+    assert (source / "store/sstable/meta.db").is_file(), "unrecognized native fault fixture"
+    path = recovery_path / "seekdb"
+    # Open only the disposable clone, including for startup/recovery writes.
+    shutil.copytree(
+        source,
+        path,
+        ignore=lambda directory, _names: ["run"] if Path(directory) == source else [],
+    )
+
+    def verify(connection) -> None:
+        with connection.cursor() as cursor:
+            cursor.execute("SELECT version_num FROM pc_schema_revision")
+            assert cursor.fetchall() == (("p0001",),)
+            cursor.execute("SELECT content FROM pc_artifacts")
+            assert cursor.fetchall() == ((b'{"content":"preserved"}',),)
+            cursor.execute("SELECT tag FROM pc_artifact_tags")
+            assert cursor.fetchall() == (("Keep",),)
+
+    with (
+        _native_generation(binding, path) as instance,
+        pymysql.connect(**instance.connection_options(), database="powercontext", read_timeout=15) as connection,
+    ):
+        with connection.cursor() as cursor:
+            cursor.execute(
+                "CREATE TABLE pc_schema_revision (version_num VARCHAR(32) NOT NULL, "
+                "CONSTRAINT pc_schema_revision_pkc PRIMARY KEY (version_num))"
+            )
+            cursor.execute("INSERT INTO pc_schema_revision VALUES ('p0001')")
+        connection.commit()
+        verify(connection)
+
+    with (
+        _native_generation(binding, path) as instance,
+        pymysql.connect(**instance.connection_options(), database="powercontext", read_timeout=15) as connection,
+    ):
+        verify(connection)
+        with connection.cursor() as cursor:
+            cursor.execute("CREATE TABLE pc_recovery_write_check (id INT PRIMARY KEY)")
+            cursor.execute("INSERT INTO pc_recovery_write_check VALUES (1)")
+        connection.commit()
+
+    with (
+        _native_generation(binding, path) as instance,
+        pymysql.connect(**instance.connection_options(), database="powercontext", read_timeout=15) as connection,
+    ):
+        verify(connection)
+        with connection.cursor() as cursor:
+            cursor.execute("SELECT id FROM pc_recovery_write_check")
+            assert cursor.fetchall() == ((1,),)
diff --git a/tests/e2e/test_topic_memory_recovery_migration.py b/tests/e2e/test_topic_memory_recovery_migration.py
index c0d84cc18b..d463043029 100644
--- a/tests/e2e/test_topic_memory_recovery_migration.py
+++ b/tests/e2e/test_topic_memory_recovery_migration.py
@@ -22,7 +22,7 @@
 import os
 import shutil
 import sqlite3
-from contextlib import closing, nullcontext
+from contextlib import closing
 
 import pytest
 from sqlalchemy import Engine, event
@@ -41,7 +41,6 @@
 from powercontext.builtin.runtime.topic_memory_recovery import TopicMemoryRecoveryService
 from powercontext.cli.app import create_cli
 from powercontext.server.cli import app
-from tests.builtin.persistence.test_mysql_migration_runner import _seekdb_scenario_lifetime
 from tests.builtin.runtime.test_topic_memory_processing import _repositories
 from tests.builtin.runtime.test_topic_memory_scope import BINDING, _capture
 from tests.builtin.runtime.test_topic_memory_terminal_frontier import _budget, _seed
@@ -587,7 +586,7 @@ async def reopen():
 @pytest.mark.parametrize("backend", ["seekdb", "oceanbase"])
 @pytest.mark.parametrize("interrupted", [False, True])
 def test_real_mysql_recovery_upgrade_preserves_receipt_and_resumes_registered_ddl_prefix(
-    backend, interrupted, short_tmp_path, tmp_path, monkeypatch
+    backend, interrupted, short_tmp_path, tmp_path
 ):
     async def scenario():
         async with _isolated_database(backend, short_tmp_path) as (config, open_profile):
@@ -661,8 +660,4 @@ def stop_during_ddl(_connection, _cursor, statement, _parameters, _context, _exe
                         )
                     ).one() == (3, 9, 1000)
 
-    # Exercise committed DDL recovery across independent connections without
-    # triggering seekdb 1.4's rapid last-client-close runtime lock stall.
-    # Native restart/exclusion acceptance remains in the connection tests.
-    with _seekdb_scenario_lifetime(monkeypatch) if backend == "seekdb" else nullcontext():
-        asyncio.run(scenario())
+    asyncio.run(scenario())
diff --git a/tests/seekdb_support.py b/tests/seekdb_support.py
new file mode 100644
index 0000000000..0b95340443
--- /dev/null
+++ b/tests/seekdb_support.py
@@ -0,0 +1,92 @@
+# 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.
+
+"""Linux process-exit ownership for native SeekDB acceptance tests."""
+
+from __future__ import annotations
+
+import ctypes
+import os
+import platform
+import selectors
+import sys
+from typing import Any
+
+
+def _open_pidfd(pid: int) -> int:
+    """Open a Linux pidfd without requiring a recent host libc or Python build."""
+    if pid <= 0:
+        message = "native process ID must be positive"
+        raise ValueError(message)
+    if sys.platform != "linux":
+        message = "native restart acceptance requires Linux pidfd support"
+        raise NotImplementedError(message)
+    python_open = getattr(os, "pidfd_open", None)
+    if python_open is not None:
+        return python_open(pid, 0)
+
+    libc = ctypes.CDLL(None, use_errno=True)
+    libc_open = getattr(libc, "pidfd_open", None)
+    if libc_open is not None:
+        libc_open.argtypes = (ctypes.c_int, ctypes.c_uint)
+        libc_open.restype = ctypes.c_int
+        descriptor = libc_open(pid, 0)
+    else:
+        # Linux UAPI: arch/x86/entry/syscalls/syscall_64.tbl and
+        # include/uapi/asm-generic/unistd.h (used by aarch64).
+        numbers = {"x86_64": 434, "aarch64": 434}
+        number = numbers.get(platform.machine())
+        if number is None or ctypes.sizeof(ctypes.c_void_p) != 8:
+            message = "pidfd syscall fallback supports only the locked 64-bit x86_64 and aarch64 Linux wheels"
+            raise NotImplementedError(message)
+        syscall = libc.syscall
+        syscall.argtypes = (ctypes.c_long, ctypes.c_int, ctypes.c_uint)
+        syscall.restype = ctypes.c_long
+        descriptor = syscall(number, pid, 0)
+    if descriptor < 0:
+        error = ctypes.get_errno()
+        raise OSError(error, os.strerror(error))
+    return descriptor
+
+
+class RestartingInstance:
+    """Release a client and require its native process to exit before returning."""
+
+    def __init__(self, instance: Any, native_pid: int, *, timeout: float = 10) -> None:
+        self.instance = instance
+        self.timeout = timeout
+        self.descriptor: int | None = None
+        try:
+            self.descriptor = _open_pidfd(native_pid)
+        except BaseException:
+            instance.close()
+            raise
+
+    def connection_options(self) -> Any:
+        return self.instance.connection_options()
+
+    def close(self) -> None:
+        descriptor = self.descriptor
+        if descriptor is None:
+            return
+        self.descriptor = None
+        try:
+            self.instance.close()
+            with selectors.DefaultSelector() as selector:
+                selector.register(descriptor, selectors.EVENT_READ)
+                if not selector.select(self.timeout):
+                    message = "native process did not exit after last-client close"
+                    raise TimeoutError(message)
+        finally:
+            os.close(descriptor)
diff --git a/tests/test_ci_changes.py b/tests/test_ci_changes.py
index 9695998396..d616600936 100644
--- a/tests/test_ci_changes.py
+++ b/tests/test_ci_changes.py
@@ -89,6 +89,13 @@
         (["tests/test_ci_installation_roles.py"], {"tests", "packaging"}),
         (["scripts/capture_release_schema.py"], {"upgrade", "release_schema"}),
         (["scripts/probe_seekdb_migration_lock.py"], {"seekdb_lock"}),
+        (["tests/seekdb_support.py"], {"tests", "database", "seekdb_lock"}),
+        (["tests/test_seekdb_migration_probe.py"], {"tests", "database", "upgrade", "seekdb_lock"}),
+        (["tests/e2e/test_seekdb_restart_recovery.py"], {"tests", "database", "seekdb_lock"}),
+        (
+            ["tests/builtin/persistence/test_mysql_migration_runner.py"],
+            {"tests", "database", "upgrade", "candidate_replay", "seekdb_lock"},
+        ),
         (["tests/test_database_migration_cli.py"], {"tests", "database", "upgrade"}),
         ([".github/workflows/skill-guidance.yml"], {"skill_guidance"}),
         ([".licenserc.yaml"], set()),
@@ -123,7 +130,15 @@ def test_shared_or_unclassified_changes_run_every_suite(path: str) -> None:
     assert all(select_suites(["docs/en/docs/quickstart.md", path], RULES).values())
 
 
-@pytest.mark.parametrize("path", ["src/powercontext/server/authz/owner_recovery.py", "uv.lock", "pyproject.toml"])
+@pytest.mark.parametrize(
+    "path",
+    [
+        "src/powercontext/server/authz/owner_recovery.py",
+        "src/powercontext/builtin/persistence/mysql.py",
+        "uv.lock",
+        "pyproject.toml",
+    ],
+)
 def test_shared_runtime_inputs_keep_backend_upgrade_and_consumer_coverage(path: str) -> None:
     selected = select_suites([path], RULES)
     assert {
@@ -140,7 +155,7 @@ def test_shared_runtime_inputs_keep_backend_upgrade_and_consumer_coverage(path:
         "managed_resources",
     } <= {suite for suite, enabled in selected.items() if enabled}
     assert not selected["release_schema"]
-    assert not selected["seekdb_lock"]
+    assert selected["seekdb_lock"] is (path != "src/powercontext/server/authz/owner_recovery.py")
 
 
 def git(repository: Path, *arguments: str) -> str:
diff --git a/tests/test_ops.py b/tests/test_ops.py
index d5c6666d2e..60dcbe19cb 100644
--- a/tests/test_ops.py
+++ b/tests/test_ops.py
@@ -507,6 +507,7 @@ def emit(message):
     # owns the worker whose exit releases both locks.
     bootstrap = """import runpy, site, sys
 packages, repository, script, *arguments = sys.argv[1:]
+sys.path.insert(0, packages)
 site.addsitedir(packages)
 sys.path.insert(0, repository)
 sys.argv = [script, *arguments]
diff --git a/tests/test_seekdb_migration_probe.py b/tests/test_seekdb_migration_probe.py
new file mode 100644
index 0000000000..25a75ca24e
--- /dev/null
+++ b/tests/test_seekdb_migration_probe.py
@@ -0,0 +1,185 @@
+# 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.
+
+"""A skipped test or retained engine must not certify native recovery."""
+
+import ctypes
+import errno
+import json
+import os
+import subprocess
+import sys
+from collections.abc import Iterator
+from pathlib import Path
+from types import ModuleType, SimpleNamespace
+
+import pytest
+
+from scripts import probe_seekdb_migration_lock as probe
+from scripts.probe_seekdb_migration_lock import restart_evidence, successful_report
+from tests import seekdb_support
+from tests.seekdb_support import RestartingInstance
+
+
+@pytest.mark.parametrize("outcome", ["", "<skipped/>", "<failure/>", "<error/>"])
+def test_native_acceptance_requires_one_passing_case(tmp_path: Path, outcome: str) -> None:
+    report = tmp_path / "junit.xml"
+    report.write_text(f"<testsuite><testcase>{outcome}</testcase></testsuite>")
+    assert successful_report(report, 0) is (outcome == "")
+    assert not successful_report(report, 1)
+    report.write_text("<testsuite/>")
+    assert not successful_report(report, 0)
+
+
+@pytest.mark.parametrize("pids, accepted", [([], False), ([101], False), ([101, 101], False), ([101, 102], True)])
+def test_restart_acceptance_rejects_an_engine_kept_alive(tmp_path: Path, pids: list[int], accepted: bool) -> None:
+    events = [
+        {"event": event, "path": "/synthetic/seekdb", "native_pid": pid}
+        for pid in pids
+        for event in ("opened", "native_exited")
+    ]
+    (tmp_path / "lifecycle.jsonl").write_text("\n".join(json.dumps(event) for event in events))
+    assert restart_evidence(tmp_path)["passed"] is accepted
+
+
+@pytest.mark.parametrize("exit_before_reopen", [False, True])
+def test_different_pids_cannot_certify_restart_without_confirmed_exit(tmp_path: Path, exit_before_reopen: bool) -> None:
+    events = [{"event": "opened", "path": "/synthetic/seekdb", "native_pid": 101}]
+    exited = {"event": "native_exited", "path": "/synthetic/seekdb", "native_pid": 101}
+    if exit_before_reopen:
+        events.append(exited)
+    events.append({"event": "opened", "path": "/synthetic/seekdb", "native_pid": 102})
+    if not exit_before_reopen:
+        events.append(exited)
+        events.append({"event": "native_exited", "path": "/synthetic/seekdb", "native_pid": 102})
+    (tmp_path / "lifecycle.jsonl").write_text("\n".join(json.dumps(event) for event in events))
+    assert not restart_evidence(tmp_path)["passed"]
+
+
+def test_probe_rejects_a_binding_without_a_module_file(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    binding = ModuleType("pylibseekdb")
+    binding.__file__ = None
+    monkeypatch.setattr(probe, "import_module", lambda _name: binding)
+    monkeypatch.setattr(sys, "argv", ["probe", "--output", str(tmp_path / "probe")])
+    with pytest.raises(RuntimeError, match="no module file"):
+        probe.main()
+
+
+@pytest.fixture
+def owned_process() -> Iterator[subprocess.Popen[bytes]]:
+    with subprocess.Popen(
+        [sys.executable, "-c", "import sys; sys.stdin.buffer.read()"], stdin=subprocess.PIPE
+    ) as process:
+        try:
+            yield process
+        finally:
+            if process.poll() is None:
+                process.kill()
+            process.wait(timeout=10)
+
+
+@pytest.fixture(params=("default", "python", "libc", "syscall"))
+def pidfd_backend(request: pytest.FixtureRequest, monkeypatch: pytest.MonkeyPatch) -> None:
+    if request.param == "default":
+        return
+    if request.param == "python":
+        if not hasattr(os, "pidfd_open"):
+            pytest.skip("this Python build does not expose os.pidfd_open")
+        # A working Python API must not need any libc wrapper symbol.
+        monkeypatch.setattr(ctypes, "CDLL", lambda *_args, **_kwargs: SimpleNamespace())
+        return
+    libc = ctypes.CDLL(None, use_errno=True)
+    monkeypatch.delattr(os, "pidfd_open", raising=False)
+    if request.param == "libc":
+        if not hasattr(libc, "pidfd_open"):
+            pytest.skip("this libc does not expose a pidfd_open wrapper")
+    else:
+        # Keep the real syscall entry point but hide the optional named wrapper.
+        monkeypatch.setattr(ctypes, "CDLL", lambda *_args, **_kwargs: SimpleNamespace(syscall=libc.syscall))
+
+
+@pytest.mark.skipif(sys.platform != "linux", reason="native restart acceptance uses Linux pidfd")
+def test_last_client_close_waits_for_process_exit(owned_process: subprocess.Popen[bytes], pidfd_backend: None) -> None:
+    assert owned_process.stdin is not None
+    instance = RestartingInstance(SimpleNamespace(close=owned_process.stdin.close), owned_process.pid)
+    instance.close()
+    assert owned_process.poll() == 0
+    instance.close()
+
+
+@pytest.mark.skipif(sys.platform != "linux", reason="native restart acceptance uses Linux pidfd")
+def test_last_client_close_rejects_a_process_that_stays_alive(
+    owned_process: subprocess.Popen[bytes], pidfd_backend: None
+) -> None:
+    instance = RestartingInstance(SimpleNamespace(close=lambda: None), owned_process.pid, timeout=0)
+    with pytest.raises(TimeoutError, match="did not exit"):
+        instance.close()
+    assert owned_process.poll() is None
+
+
+@pytest.mark.skipif(sys.platform != "linux", reason="native restart acceptance uses Linux pidfd")
+@pytest.mark.parametrize("error", [errno.ENOSYS, errno.EPERM])
+def test_pidfd_errors_preserve_the_failure_and_close_the_client(
+    owned_process: subprocess.Popen[bytes], monkeypatch: pytest.MonkeyPatch, error: int
+) -> None:
+    def unavailable(_pid: int, _flags: int) -> int:
+        raise OSError(error, os.strerror(error))
+
+    monkeypatch.setattr(os, "pidfd_open", unavailable, raising=False)
+    # Falling back after a real kernel error would hide the reason for failure.
+    monkeypatch.setattr(ctypes, "CDLL", lambda *_args, **_kwargs: SimpleNamespace())
+    assert owned_process.stdin is not None
+    with pytest.raises(OSError) as failed:
+        RestartingInstance(SimpleNamespace(close=owned_process.stdin.close), owned_process.pid)
+    assert failed.value.errno == error
+    assert owned_process.wait(timeout=10) == 0
+
+
+@pytest.mark.skipif(sys.platform != "linux", reason="native restart acceptance uses Linux pidfd")
+def test_pidfd_syscall_rejects_unknown_architectures(
+    owned_process: subprocess.Popen[bytes], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.delattr(os, "pidfd_open", raising=False)
+    monkeypatch.setattr(ctypes, "CDLL", lambda *_args, **_kwargs: SimpleNamespace())
+    monkeypatch.setattr(seekdb_support.platform, "machine", lambda: "unknown")
+    assert owned_process.stdin is not None
+    with pytest.raises(NotImplementedError, match="locked 64-bit"):
+        RestartingInstance(SimpleNamespace(close=owned_process.stdin.close), owned_process.pid)
+    assert owned_process.wait(timeout=10) == 0
+
+
+@pytest.mark.skipif(sys.platform != "linux", reason="native restart acceptance uses Linux pidfd")
+def test_last_client_close_supports_pidfds_above_select_limit(
+    owned_process: subprocess.Popen[bytes], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    import fcntl
+    import resource
+
+    soft, _hard = resource.getrlimit(resource.RLIMIT_NOFILE)
+    if soft != resource.RLIM_INFINITY and soft <= 1025:
+        pytest.skip("the process descriptor limit cannot exercise fd >= 1024")
+    original = seekdb_support._open_pidfd
+
+    def high_descriptor(pid: int) -> int:
+        descriptor = original(pid)
+        try:
+            return fcntl.fcntl(descriptor, fcntl.F_DUPFD_CLOEXEC, 1024)
+        finally:
+            os.close(descriptor)
+
+    monkeypatch.setattr(seekdb_support, "_open_pidfd", high_descriptor)
+    assert owned_process.stdin is not None
+    instance = RestartingInstance(SimpleNamespace(close=owned_process.stdin.close), owned_process.pid)
+    instance.close()
+    assert owned_process.poll() == 0
diff --git a/tox.ini b/tox.ini
index 3cab314540..bccaa0da49 100644
--- a/tox.ini
+++ b/tox.ini
@@ -12,7 +12,10 @@ python =
 [testenv]
 passenv = PYTHON_VERSION
 allowlist_externals = uv
+setenv =
+    UV_PROJECT_ENVIRONMENT={envdir}
+    COVERAGE_FILE={envtmpdir}/.coverage
 commands =
-    uv sync --python {envpython}
-    uv run python -m pytest --doctest-modules tests --cov --cov-config=pyproject.toml --cov-report=xml
-    ty check
+    uv sync --locked --python {envpython}
+    uv run --no-sync python -m pytest --doctest-modules tests --cov=powercontext --cov-config=pyproject.toml --cov-report=xml:{envtmpdir}/coverage.xml {posargs}
+    uv run --no-sync ty check