diff --git a/infra/ansible/playbooks/110-backup-runtime-deploy.yml b/infra/ansible/playbooks/110-backup-runtime-deploy.yml
index b7cade204..679f1c3c9 100644
--- a/infra/ansible/playbooks/110-backup-runtime-deploy.yml
+++ b/infra/ansible/playbooks/110-backup-runtime-deploy.yml
@@ -8,64 +8,269 @@
 
   vars:
     backup_runtime_root: /backup/scripts
-    backup_runtime_files:
+    backup_runtime_stage_root: /backup/.agent99-backup-runtime-stage
+    backup_health_exporter_root: /home/wooo/scripts
+    backup_health_exporter_name: backup-health-textfile-exporter.py
+    backup_runtime_stage1_files:
       - common.sh
       - backup-all.sh
+      - backup-host188-products.sh
+      - verify-host188-products-backup.sh
+      - verify-host188-products-archive.py
+    backup_runtime_stage2_files:
       - backup-awoooi.sh
       - backup-awoooi-frequent.sh
       - backup-clawbot.sh
       - backup-sentry.sh
-      - backup-host188-products.sh
+      - check-backup-integrity.sh
+      - sync-offsite-backups.sh
+      - backup-offsite-readiness-gate.sh
+      - verify-offsite-full-sync.sh
+      - enforce-latest-only-retention.sh
+    backup_runtime_files: "{{ backup_runtime_stage1_files + backup_runtime_stage2_files }}"
+    backup_runtime_fault_injection_after_stage1: false
 
   pre_tasks:
-    - name: Verify the fixed host boundary
+    - name: Verify the fixed host and transaction boundary
       ansible.builtin.assert:
         that:
           - inventory_hostname == "host_110"
-          - backup_runtime_files | length == 7
+          - backup_runtime_files | length == 14
           - backup_runtime_files | unique | length == backup_runtime_files | length
+          - backup_runtime_fault_injection_after_stage1 | type_debug == "bool"
         fail_msg: backup_runtime_deploy_boundary_failed
 
   tasks:
-    - name: Ensure the backup runtime directory exists
+    - name: Ensure fixed runtime directories exist
       ansible.builtin.file:
-        path: "{{ backup_runtime_root }}"
+        path: "{{ item.path }}"
         state: directory
-        owner: wooo
-        group: wooo
-        mode: "0755"
+        owner: "{{ item.owner }}"
+        group: "{{ item.group }}"
+        mode: "{{ item.mode }}"
+      loop:
+        - { path: "{{ backup_runtime_root }}", owner: wooo, group: wooo, mode: "0755" }
+        - { path: "{{ backup_health_exporter_root }}", owner: wooo, group: wooo, mode: "0755" }
 
-    - name: Validate and deploy the fixed backup runtime files
+    - name: Reset the transaction staging directory
+      ansible.builtin.file:
+        path: "{{ backup_runtime_stage_root }}"
+        state: absent
+
+    - name: Create the transaction staging directory
+      ansible.builtin.file:
+        path: "{{ backup_runtime_stage_root }}"
+        state: directory
+        owner: root
+        group: root
+        mode: "0700"
+
+    - name: Stage and shell-validate backup runtime files
       ansible.builtin.copy:
         src: "{{ playbook_dir }}/../../../scripts/backup/{{ item }}"
-        dest: "{{ backup_runtime_root }}/{{ item }}"
-        owner: wooo
-        group: wooo
-        mode: "0755"
-        backup: true
+        dest: "{{ backup_runtime_stage_root }}/{{ item }}"
+        owner: root
+        group: root
+        mode: "0700"
         force: true
         validate: "/bin/bash -n %s"
-      loop: "{{ backup_runtime_files }}"
-      register: backup_runtime_copy
+      loop: "{{ backup_runtime_files | reject('match', '.*[.]py$') | list }}"
 
-    - name: Read back deployed backup runtime metadata
+    - name: Stage and Python-validate backup runtime files
+      ansible.builtin.copy:
+        src: "{{ playbook_dir }}/../../../scripts/backup/{{ item }}"
+        dest: "{{ backup_runtime_stage_root }}/{{ item }}"
+        owner: root
+        group: root
+        mode: "0700"
+        force: true
+        validate: "/usr/bin/python3 -m py_compile %s"
+      loop: "{{ backup_runtime_files | select('match', '.*[.]py$') | list }}"
+
+    - name: Stage and Python-validate backup health exporter
+      ansible.builtin.copy:
+        src: "{{ playbook_dir }}/../../../scripts/ops/{{ backup_health_exporter_name }}"
+        dest: "{{ backup_runtime_stage_root }}/{{ backup_health_exporter_name }}"
+        owner: root
+        group: root
+        mode: "0700"
+        force: true
+        validate: "/usr/bin/python3 -m py_compile %s"
+
+    - name: Read back staged backup runtime metadata
       ansible.builtin.stat:
-        path: "{{ backup_runtime_root }}/{{ item }}"
+        path: "{{ backup_runtime_stage_root }}/{{ item }}"
         checksum_algorithm: sha256
       loop: "{{ backup_runtime_files }}"
-      register: backup_runtime_stat
+      register: backup_runtime_stage_stat
       changed_when: false
 
-    - name: Verify deployed backup runtime identity
+    - name: Verify staged backup runtime identity
       ansible.builtin.assert:
         that:
           - item.stat.exists
           - item.stat.isreg
-          - item.stat.pw_name == "wooo"
-          - item.stat.gr_name == "wooo"
-          - item.stat.mode == "0755"
           - item.stat.checksum == (lookup('ansible.builtin.file', playbook_dir + '/../../../scripts/backup/' + item.item, rstrip=false) | hash('sha256'))
-        fail_msg: "backup_runtime_identity_failed:{{ item.item }}"
-      loop: "{{ backup_runtime_stat.results }}"
+        fail_msg: "backup_runtime_stage_identity_failed:{{ item.item }}"
+      loop: "{{ backup_runtime_stage_stat.results }}"
       loop_control:
         label: "{{ item.item }}"
+
+    - name: Read current backup runtime metadata for rollback
+      ansible.builtin.stat:
+        path: "{{ backup_runtime_root }}/{{ item }}"
+      loop: "{{ backup_runtime_files }}"
+      register: backup_runtime_before_stat
+      changed_when: false
+
+    - name: Capture current backup runtime content for rollback
+      ansible.builtin.slurp:
+        path: "{{ backup_runtime_root }}/{{ item.item }}"
+      loop: "{{ backup_runtime_before_stat.results }}"
+      when: item.stat.exists
+      register: backup_runtime_before_content
+      no_log: true
+
+    - name: Read current backup exporter metadata for rollback
+      ansible.builtin.stat:
+        path: "{{ backup_health_exporter_root }}/{{ backup_health_exporter_name }}"
+      register: backup_exporter_before_stat
+      changed_when: false
+
+    - name: Capture current backup exporter content for rollback
+      ansible.builtin.slurp:
+        path: "{{ backup_health_exporter_root }}/{{ backup_health_exporter_name }}"
+      when: backup_exporter_before_stat.stat.exists
+      register: backup_exporter_before_content
+      no_log: true
+
+    - name: Promote the staged backup runtime transaction
+      block:
+        - name: Promote transaction stage one
+          ansible.builtin.copy:
+            src: "{{ backup_runtime_stage_root }}/{{ item }}"
+            dest: "{{ backup_runtime_root }}/{{ item }}"
+            remote_src: true
+            owner: wooo
+            group: wooo
+            mode: "0755"
+            backup: true
+            force: true
+          loop: "{{ backup_runtime_stage1_files }}"
+
+        - name: Exercise bounded rollback fault injection when explicitly requested
+          ansible.builtin.fail:
+            msg: backup_runtime_fault_injection_after_stage1
+          when: backup_runtime_fault_injection_after_stage1
+
+        - name: Promote transaction stage two
+          ansible.builtin.copy:
+            src: "{{ backup_runtime_stage_root }}/{{ item }}"
+            dest: "{{ backup_runtime_root }}/{{ item }}"
+            remote_src: true
+            owner: wooo
+            group: wooo
+            mode: "0755"
+            backup: true
+            force: true
+          loop: "{{ backup_runtime_stage2_files }}"
+
+        - name: Promote the staged backup health exporter
+          ansible.builtin.copy:
+            src: "{{ backup_runtime_stage_root }}/{{ backup_health_exporter_name }}"
+            dest: "{{ backup_health_exporter_root }}/{{ backup_health_exporter_name }}"
+            remote_src: true
+            owner: wooo
+            group: wooo
+            mode: "0755"
+            backup: true
+            force: true
+
+        - name: Read back promoted backup runtime metadata
+          ansible.builtin.stat:
+            path: "{{ backup_runtime_root }}/{{ item }}"
+            checksum_algorithm: sha256
+          loop: "{{ backup_runtime_files }}"
+          register: backup_runtime_after_stat
+          changed_when: false
+
+        - name: Verify promoted backup runtime identity
+          ansible.builtin.assert:
+            that:
+              - item.stat.exists
+              - item.stat.isreg
+              - item.stat.pw_name == "wooo"
+              - item.stat.gr_name == "wooo"
+              - item.stat.mode == "0755"
+              - item.stat.checksum == (lookup('ansible.builtin.file', playbook_dir + '/../../../scripts/backup/' + item.item, rstrip=false) | hash('sha256'))
+            fail_msg: "backup_runtime_identity_failed:{{ item.item }}"
+          loop: "{{ backup_runtime_after_stat.results }}"
+          loop_control:
+            label: "{{ item.item }}"
+
+        - name: Read back promoted backup exporter metadata
+          ansible.builtin.stat:
+            path: "{{ backup_health_exporter_root }}/{{ backup_health_exporter_name }}"
+            checksum_algorithm: sha256
+          register: backup_exporter_after_stat
+          changed_when: false
+
+        - name: Verify promoted backup exporter identity
+          ansible.builtin.assert:
+            that:
+              - backup_exporter_after_stat.stat.exists
+              - backup_exporter_after_stat.stat.pw_name == "wooo"
+              - backup_exporter_after_stat.stat.gr_name == "wooo"
+              - backup_exporter_after_stat.stat.mode == "0755"
+              - backup_exporter_after_stat.stat.checksum == (lookup('ansible.builtin.file', playbook_dir + '/../../../scripts/ops/' + backup_health_exporter_name, rstrip=false) | hash('sha256'))
+            fail_msg: backup_exporter_identity_failed
+      rescue:
+        - name: Restore runtime files that existed before promotion
+          ansible.builtin.copy:
+            content: "{{ item.content | b64decode }}"
+            dest: "{{ backup_runtime_root }}/{{ item.item.item }}"
+            owner: wooo
+            group: wooo
+            mode: "0755"
+            force: true
+          loop: "{{ backup_runtime_before_content.results | default([]) }}"
+          when: not (item.skipped | default(false))
+          no_log: true
+
+        - name: Remove runtime files that did not exist before promotion
+          ansible.builtin.file:
+            path: "{{ backup_runtime_root }}/{{ item.item }}"
+            state: absent
+          loop: "{{ backup_runtime_before_stat.results }}"
+          when: not item.stat.exists
+
+        - name: Restore the exporter that existed before promotion
+          ansible.builtin.copy:
+            content: "{{ backup_exporter_before_content.content | b64decode }}"
+            dest: "{{ backup_health_exporter_root }}/{{ backup_health_exporter_name }}"
+            owner: wooo
+            group: wooo
+            mode: "0755"
+            force: true
+          when: backup_exporter_before_stat.stat.exists
+          no_log: true
+
+        - name: Remove exporter created by the failed promotion
+          ansible.builtin.file:
+            path: "{{ backup_health_exporter_root }}/{{ backup_health_exporter_name }}"
+            state: absent
+          when: not backup_exporter_before_stat.stat.exists
+
+        - name: Remove failed transaction staging data
+          ansible.builtin.file:
+            path: "{{ backup_runtime_stage_root }}"
+            state: absent
+
+        - name: Fail closed after transactional rollback
+          ansible.builtin.fail:
+            msg: backup_runtime_promotion_failed_and_rolled_back
+
+    - name: Remove successful transaction staging data
+      ansible.builtin.file:
+        path: "{{ backup_runtime_stage_root }}"
+        state: absent
diff --git a/infra/ansible/playbooks/110-devops.yml b/infra/ansible/playbooks/110-devops.yml
index 7c3c6745c..29205ef4f 100644
--- a/infra/ansible/playbooks/110-devops.yml
+++ b/infra/ansible/playbooks/110-devops.yml
@@ -239,6 +239,8 @@
         - backup-ai-artifacts.sh
         - backup-public-routes.sh
         - backup-host188-products.sh
+        - verify-host188-products-backup.sh
+        - verify-host188-products-archive.py
         - clickhouse-native-backup.sh
         - clickhouse-native-restore-drill.sh
         - clickhouse-restore-inventory.py
@@ -250,6 +252,7 @@
         - backup-offsite-readiness-gate.sh
         - offsite-escrow-evidence-report.sh
         - verify-offsite-full-sync.sh
+        - enforce-latest-only-retention.sh
       tags: backup_jobs
 
     - name: "Backup | 建立 SigNoz native backup/restore config 目錄"
diff --git a/scripts/backup/backup-host188-products.sh b/scripts/backup/backup-host188-products.sh
index c639f10f2..4fa1d601e 100644
--- a/scripts/backup/backup-host188-products.sh
+++ b/scripts/backup/backup-host188-products.sh
@@ -9,8 +9,12 @@ source "$(dirname "$0")/common.sh"
 SERVICE="host188-products"
 REMOTE_HOST="ollama@192.168.0.188"
 LOCAL_REPO="${BACKUP_BASE}/host188-products"
-DUMP_DIR="/tmp/host188-products-backup-$$"
+RESTORE_VERIFIER="$(dirname "$0")/verify-host188-products-backup.sh"
+DUMP_DIR="${BACKUP_BASE}/staging/host188-products-backup-$$"
+LOCK_FILE="/tmp/host188-products-backup.lock"
 MIN_POSTGRES_DUMP_BYTES="${MIN_POSTGRES_DUMP_BYTES:-4096}"
+MIN_FREE_BYTES="${HOST188_BACKUP_MIN_FREE_BYTES:-8589934592}"
+SSH_TIMEOUT_SECONDS="${HOST188_BACKUP_SSH_TIMEOUT_SECONDS:-1800}"
 SSH_OPTIONS=(
     -o BatchMode=yes
     -o StrictHostKeyChecking=yes
@@ -35,7 +39,7 @@ on_error() {
 
 require_commands() {
     local command_name
-    for command_name in restic ssh python3; do
+    for command_name in restic ssh python3 flock timeout; do
         if ! command -v "${command_name}" >/dev/null 2>&1; then
             log_error "Required command missing: ${command_name}"
             return 69
@@ -45,6 +49,27 @@ require_commands() {
         log_error "Restic password reference is unavailable"
         return 69
     fi
+    case "${MIN_FREE_BYTES}:${SSH_TIMEOUT_SECONDS}" in
+        *[!0-9:]*|:*|*:) log_error "Host188 backup numeric limits are invalid"; return 64 ;;
+    esac
+    if [ "${MIN_FREE_BYTES}" -le 0 ] || [ "${SSH_TIMEOUT_SECONDS}" -le 0 ]; then
+        log_error "Host188 backup numeric limits must be positive"
+        return 64
+    fi
+}
+
+run_remote() {
+    timeout --signal=TERM "${SSH_TIMEOUT_SECONDS}s" \
+        ssh "${SSH_OPTIONS[@]}" "${REMOTE_HOST}" "$@"
+}
+
+require_disk_headroom() {
+    local available_bytes
+    available_bytes="$(df -Pk "${BACKUP_BASE}" 2>/dev/null | awk 'NR==2 {printf "%.0f", $4 * 1024}')"
+    if [ -z "${available_bytes}" ] || [ "${available_bytes}" -lt "${MIN_FREE_BYTES}" ]; then
+        log_error "Host188 backup staging headroom is insufficient"
+        return 75
+    fi
 }
 
 assert_fixed_container() {
@@ -69,7 +94,7 @@ dump_redis_container() {
         'test "$(docker inspect -f '\''{{.State.Running}}'\'' %q 2>/dev/null)" = true && docker exec %q sh -eu -c %q' \
         "${container}" "${container}" "${inner_command}"
 
-    if ! ssh "${SSH_OPTIONS[@]}" "${REMOTE_HOST}" "${remote_command}" >"${temporary_file}"; then
+    if ! run_remote "${remote_command}" >"${temporary_file}"; then
         log_error "Redis RDB dump failed: ${container}"
         return 1
     fi
@@ -95,7 +120,7 @@ dump_postgres_container() {
         'test "$(docker inspect -f '\''{{.State.Running}}'\'' %q 2>/dev/null)" = true && docker exec %q sh -eu -c %q' \
         "${container}" "${container}" "${dump_inner}"
 
-    if ! ssh "${SSH_OPTIONS[@]}" "${REMOTE_HOST}" "${dump_command}" >"${temporary_file}"; then
+    if ! run_remote "${dump_command}" >"${temporary_file}"; then
         log_error "PostgreSQL dump failed: ${container}"
         return 1
     fi
@@ -108,7 +133,7 @@ dump_postgres_container() {
     fi
 
     printf -v verify_command 'docker exec -i %q pg_restore --list >/dev/null' "${container}"
-    if ! ssh "${SSH_OPTIONS[@]}" "${REMOTE_HOST}" "${verify_command}" <"${temporary_file}"; then
+    if ! run_remote "${verify_command}" <"${temporary_file}"; then
         log_error "PostgreSQL archive verifier failed: ${container}"
         return 1
     fi
@@ -131,7 +156,7 @@ backup_n8n_online() {
         'test "$(docker inspect -f '\''{{.State.Running}}'\'' n8n 2>/dev/null)" = true && docker exec n8n sh -eu -c %q' \
         "${inner_command}"
 
-    if ! ssh "${SSH_OPTIONS[@]}" "${REMOTE_HOST}" "${remote_command}" >"${temporary_file}"; then
+    if ! run_remote "${remote_command}" >"${temporary_file}"; then
         log_error "n8n online backup failed"
         return 1
     fi
@@ -246,6 +271,9 @@ for name in (
             "credentialsDecrypted": False,
             "serviceStopped": False,
             "offsiteWritePerformed": False,
+            "recoveryReadinessClaimed": False,
+            "n8nEncryptionKeyEscrowRequired": True,
+            "n8nEncryptionKeyEscrowVerified": False,
         },
         separators=(",", ":"),
     )
@@ -269,6 +297,12 @@ main() {
     trap on_error ERR
 
     require_commands
+    exec 9>"${LOCK_FILE}"
+    if ! flock -n 9; then
+        log_error "Host188 product backup is already running"
+        return 75
+    fi
+    require_disk_headroom
     install -d -m 700 "${DUMP_DIR}"
     log_info "Host188 product backup started"
 
@@ -291,10 +325,14 @@ main() {
         --password-file "${RESTIC_PASSWORD_FILE}" \
         --tag "service:${SERVICE}" --tag "source:host188" --tag "timestamp:${timestamp}"
     snapshot_id="$(latest_snapshot_id)"
-    verify_backup "${LOCAL_REPO}" "${snapshot_id}"
+    [ -r "${RESTORE_VERIFIER}" ] || {
+        log_error "Host188 snapshot restore verifier missing"
+        return 69
+    }
+    bash "${RESTORE_VERIFIER}" --snapshot "${snapshot_id}" --data-only
 
     duration="$(($(date +%s) - start_time))"
-    log_success "Host188 product backup completed snapshot=${snapshot_id} duration=${duration}s"
+    log_success "Host188 product data snapshot verified snapshot=${snapshot_id} duration=${duration}s; DR escrow remains a separate gate"
     if [ "${BACKUP_NOTIFY_SUCCESS:-0}" = "1" ]; then
         notify_clawbot "success" "${SERVICE}" "Host188 product backup completed and verified" "${duration}"
     fi
diff --git a/scripts/backup/check-backup-integrity.sh b/scripts/backup/check-backup-integrity.sh
index 76176f4cc..ad60fa103 100755
--- a/scripts/backup/check-backup-integrity.sh
+++ b/scripts/backup/check-backup-integrity.sh
@@ -202,6 +202,18 @@ main() {
                 fi
                 continue
             fi
+            if [ "${name}" = "host188-products" ] && [ -r "${BACKUP_BASE}/scripts/verify-host188-products-backup.sh" ]; then
+                log_info "Host188 specialized isolated restore drill: PostgreSQL, Redis, n8n, manifest, and escrow"
+                if bash "${BACKUP_BASE}/scripts/verify-host188-products-backup.sh" >> "${LOG_FILE}" 2>&1; then
+                    log_success "repo ${name} specialized restore drill OK"
+                    echo "repo=${name} status=restore_drill_specialized_ok snapshots=${count} latest=${latest_ts}" >> "${LOG_FILE}"
+                else
+                    log_error "repo ${name} specialized restore drill blocked or failed"
+                    echo "repo=${name} status=restore_drill_specialized_failed snapshots=${count} latest=${latest_ts}" >> "${LOG_FILE}"
+                    failed=$((failed + 1))
+                fi
+                continue
+            fi
             local sample
             local sample_out
             sample=$(MAX_SAMPLE_BYTES="${MAX_SAMPLE_BYTES}" sample_path_for_repo "${repo}")
diff --git a/scripts/backup/common.sh b/scripts/backup/common.sh
index 69c0e4e87..79c56acce 100644
--- a/scripts/backup/common.sh
+++ b/scripts/backup/common.sh
@@ -8,9 +8,9 @@
 # -----------------------------------------------------------------------------
 # 配置區 (待 CEO 提供 B2 帳號後更新)
 # -----------------------------------------------------------------------------
-export BACKUP_BASE="/backup"
-export BACKUP_LOG_DIR="${BACKUP_BASE}/logs"
-export RESTIC_PASSWORD_FILE="${BACKUP_BASE}/scripts/.restic-password"
+export BACKUP_BASE="${BACKUP_BASE:-/backup}"
+export BACKUP_LOG_DIR="${BACKUP_LOG_DIR:-${BACKUP_BASE}/logs}"
+export RESTIC_PASSWORD_FILE="${RESTIC_PASSWORD_FILE:-${BACKUP_BASE}/scripts/.restic-password}"
 
 # Backblaze B2 配置 (待填入)
 export B2_ACCOUNT_ID=""          # 待 CEO 提供
@@ -115,10 +115,35 @@ build_tags() {
 verify_backup() {
     local repo="$1"
     local snapshot_id="$2"
-    
+    local snapshot_rows snapshot_files
+
+    if ! [[ "${snapshot_id}" =~ ^[0-9a-fA-F]{8,64}$ ]]; then
+        log_error "無效 snapshot id"
+        return 64
+    fi
     log_info "驗證備份快照: ${snapshot_id}"
-    restic -r "${repo}" check --read-data-subset=1% 2>&1
-    return $?
+    snapshot_rows="$(restic -r "${repo}" snapshots "${snapshot_id}" --json \
+        --password-file "${RESTIC_PASSWORD_FILE}" 2>/dev/null)" || return 1
+    if ! printf '%s' "${snapshot_rows}" | python3 -c \
+        'import json,sys; rows=json.load(sys.stdin); raise SystemExit(0 if len(rows) == 1 else 1)'; then
+        log_error "指定 snapshot 無法唯一讀回: ${snapshot_id}"
+        return 1
+    fi
+    snapshot_files="$(restic -r "${repo}" ls "${snapshot_id}" --json \
+        --password-file "${RESTIC_PASSWORD_FILE}" 2>/dev/null | \
+        python3 -c 'import json,sys
+count=0
+for line in sys.stdin:
+    try: row=json.loads(line)
+    except json.JSONDecodeError: continue
+    if row.get("type") == "file": count += 1
+print(count)')" || return 1
+    if [ "${snapshot_files}" -le 0 ]; then
+        log_error "指定 snapshot 不含可還原檔案: ${snapshot_id}"
+        return 1
+    fi
+    restic -r "${repo}" check --read-data-subset=1% \
+        --password-file "${RESTIC_PASSWORD_FILE}" 2>&1
 }
 
 # -----------------------------------------------------------------------------
@@ -126,13 +151,31 @@ verify_backup() {
 # -----------------------------------------------------------------------------
 cleanup_old_backups() {
     local repo="$1"
-    
+    local mode="${BACKUP_RETENTION_MODE:-gfs}"
+    local keep_last="${BACKUP_KEEP_LAST:-1}"
+
+    if [ "${mode}" = "latest" ]; then
+        case "${keep_last}" in
+            ''|*[!0-9]*|0) log_error "BACKUP_KEEP_LAST 必須是正整數"; return 64 ;;
+        esac
+        log_info "執行 latest-only 清理策略 keep-last=${keep_last}"
+        restic -r "${repo}" forget \
+            --keep-last "${keep_last}" \
+            --prune \
+            --password-file "${RESTIC_PASSWORD_FILE}" 2>&1
+        return $?
+    fi
+    if [ "${mode}" != "gfs" ]; then
+        log_error "未知 BACKUP_RETENTION_MODE: ${mode}"
+        return 64
+    fi
     log_info "執行 GFS 清理策略"
     restic -r "${repo}" forget \
-        --keep-daily ${KEEP_DAILY} \
-        --keep-weekly ${KEEP_WEEKLY} \
-        --keep-monthly ${KEEP_MONTHLY} \
-        --prune 2>&1
+        --keep-daily "${KEEP_DAILY}" \
+        --keep-weekly "${KEEP_WEEKLY}" \
+        --keep-monthly "${KEEP_MONTHLY}" \
+        --prune \
+        --password-file "${RESTIC_PASSWORD_FILE}" 2>&1
 }
 
 # -----------------------------------------------------------------------------
diff --git a/scripts/backup/enforce-latest-only-retention.sh b/scripts/backup/enforce-latest-only-retention.sh
index 772323ff1..d4200ee9e 100755
--- a/scripts/backup/enforce-latest-only-retention.sh
+++ b/scripts/backup/enforce-latest-only-retention.sh
@@ -13,6 +13,7 @@ source "$(dirname "$0")/common.sh"
 
 EXPECTED_REPOS_DEFAULT="awoooi configs gitea harbor momo langfuse monitoring signoz open-webui clawbot sentry ai-artifacts public-routes host188-products"
 REPOS="${BACKUP_RETENTION_REPOS:-${EXPECTED_REPOS_DEFAULT}}"
+KEEP_LAST="${BACKUP_KEEP_LAST:-1}"
 
 main() {
     local failed=0
@@ -26,7 +27,8 @@ main() {
         fi
 
         log_info "Enforce latest-only retention: ${name}"
-        if ! BACKUP_RETENTION_MODE=latest cleanup_old_backups "${repo}"; then
+        if ! BACKUP_RETENTION_MODE=latest BACKUP_KEEP_LAST="${KEEP_LAST}" \
+            cleanup_old_backups "${repo}"; then
             failed=$((failed + 1))
         fi
     done
diff --git a/scripts/backup/sync-offsite-backups.sh b/scripts/backup/sync-offsite-backups.sh
index 5cf31d5a7..03a582061 100755
--- a/scripts/backup/sync-offsite-backups.sh
+++ b/scripts/backup/sync-offsite-backups.sh
@@ -19,7 +19,7 @@
 
 set -euo pipefail
 
-source "$(dirname "$0")/common.sh"
+source "$(dirname "${BASH_SOURCE[0]}")/common.sh"
 
 SERVICE="offsite-backup"
 MODE="status"
@@ -183,7 +183,7 @@ backup_disk_used_pct() {
 active_backup_processes() {
     ps -eo pid=,args= | awk -v self="$$" '
         $1 == self { next }
-        /\/backup\/scripts\/backup-(all|awoooi|awoooi-frequent|gitea|harbor|momo|langfuse|monitoring|signoz|open-webui|clawbot|sentry|ai-artifacts|public-routes|configs)\.sh/ {
+        /\/backup\/scripts\/backup-(all|awoooi|awoooi-frequent|gitea|harbor|momo|langfuse|monitoring|signoz|open-webui|clawbot|sentry|ai-artifacts|public-routes|configs|host188-products)\.sh/ {
             print
         }
     '
@@ -411,4 +411,6 @@ main() {
     return "${failed}"
 }
 
-main "$@"
+if [[ "${BASH_SOURCE[0]}" == "$0" ]]; then
+    main "$@"
+fi
diff --git a/scripts/backup/tests/test_110_backup_runtime_deploy_playbook.py b/scripts/backup/tests/test_110_backup_runtime_deploy_playbook.py
index ec3f91ed0..7cbfd38c7 100644
--- a/scripts/backup/tests/test_110_backup_runtime_deploy_playbook.py
+++ b/scripts/backup/tests/test_110_backup_runtime_deploy_playbook.py
@@ -16,10 +16,17 @@ EXPECTED_FILES = {
     "backup-clawbot.sh",
     "backup-sentry.sh",
     "backup-host188-products.sh",
+    "verify-host188-products-backup.sh",
+    "verify-host188-products-archive.py",
+    "check-backup-integrity.sh",
+    "sync-offsite-backups.sh",
+    "backup-offsite-readiness-gate.sh",
+    "verify-offsite-full-sync.sh",
+    "enforce-latest-only-retention.sh",
 }
 
 
-def test_backup_runtime_deploy_is_fixed_to_host_110_and_seven_files() -> None:
+def test_backup_runtime_deploy_is_fixed_to_host_110_and_fourteen_files() -> None:
     plays = yaml.safe_load(PLAYBOOK.read_text(encoding="utf-8"))
     assert len(plays) == 1
     play = plays[0]
@@ -30,7 +37,11 @@ def test_backup_runtime_deploy_is_fixed_to_host_110_and_seven_files() -> None:
     assert play["serial"] == 1
     assert play["any_errors_fatal"] is True
     assert play["vars"]["backup_runtime_root"] == "/backup/scripts"
-    assert set(play["vars"]["backup_runtime_files"]) == EXPECTED_FILES
+    stage1 = set(play["vars"]["backup_runtime_stage1_files"])
+    stage2 = set(play["vars"]["backup_runtime_stage2_files"])
+    assert stage1 | stage2 == EXPECTED_FILES
+    assert not stage1 & stage2
+    assert play["vars"]["backup_runtime_fault_injection_after_stage1"] is False
 
 
 def test_backup_runtime_deploy_has_validation_rollback_and_identity_readback() -> None:
@@ -42,6 +53,13 @@ def test_backup_runtime_deploy_has_validation_rollback_and_identity_readback() -
     assert "checksum_algorithm: sha256" in source
     assert "rstrip=false" in source
     assert "backup_runtime_identity_failed" in source
+    assert "backup_runtime_stage_identity_failed" in source
+    assert "backup_runtime_fault_injection_after_stage1" in source
+    assert "backup_runtime_promotion_failed_and_rolled_back" in source
+    assert "rescue:" in source
+    assert "ansible.builtin.slurp:" in source
+    assert "remote_src: true" in source
+    assert "backup-health-textfile-exporter.py" in source
     assert "changed_when: false" in source
 
     for forbidden in (
@@ -55,3 +73,25 @@ def test_backup_runtime_deploy_has_validation_rollback_and_identity_readback() -
         "kubectl",
     ):
         assert forbidden not in source
+
+
+def test_fault_injection_occurs_mid_promotion_and_rescue_restores_prior_state() -> None:
+    play = yaml.safe_load(PLAYBOOK.read_text(encoding="utf-8"))[0]
+    transaction = next(
+        task
+        for task in play["tasks"]
+        if task["name"] == "Promote the staged backup runtime transaction"
+    )
+    block_names = [task["name"] for task in transaction["block"]]
+    rescue_names = [task["name"] for task in transaction["rescue"]]
+
+    assert block_names.index("Promote transaction stage one") < block_names.index(
+        "Exercise bounded rollback fault injection when explicitly requested"
+    )
+    assert block_names.index(
+        "Exercise bounded rollback fault injection when explicitly requested"
+    ) < block_names.index("Promote transaction stage two")
+    assert "Restore runtime files that existed before promotion" in rescue_names
+    assert "Remove runtime files that did not exist before promotion" in rescue_names
+    assert "Restore the exporter that existed before promotion" in rescue_names
+    assert rescue_names[-1] == "Fail closed after transactional rollback"
diff --git a/scripts/backup/tests/test_backup_common_retention_contract.py b/scripts/backup/tests/test_backup_common_retention_contract.py
new file mode 100644
index 000000000..842101091
--- /dev/null
+++ b/scripts/backup/tests/test_backup_common_retention_contract.py
@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+import os
+import subprocess
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[3]
+COMMON = ROOT / "scripts" / "backup" / "common.sh"
+
+
+def test_latest_only_retention_uses_keep_last_one_and_explicit_password(tmp_path: Path) -> None:
+    bin_dir = tmp_path / "bin"
+    bin_dir.mkdir()
+    event_log = tmp_path / "restic.args"
+    restic = bin_dir / "restic"
+    restic.write_text(
+        '#!/bin/sh\nprintf "%s\\n" "$*" >> "$TEST_EVENT_LOG"\n',
+        encoding="utf-8",
+    )
+    restic.chmod(0o755)
+    backup_base = tmp_path / "backup"
+    password_file = backup_base / "scripts" / ".restic-password"
+    password_file.parent.mkdir(parents=True)
+    password_file.write_text("test-only\n", encoding="utf-8")
+
+    env = os.environ.copy()
+    env.update(
+        {
+            "BACKUP_BASE": str(backup_base),
+            "RESTIC_PASSWORD_FILE": str(password_file),
+            "PATH": f"{bin_dir}:{env['PATH']}",
+            "TEST_EVENT_LOG": str(event_log),
+        }
+    )
+    subprocess.run(
+        [
+            "bash",
+            "-c",
+            f'source "{COMMON}"; BACKUP_RETENTION_MODE=latest '
+            'BACKUP_KEEP_LAST=1 cleanup_old_backups "$BACKUP_BASE/demo"',
+        ],
+        env=env,
+        check=True,
+        text=True,
+        capture_output=True,
+    )
+
+    args = event_log.read_text(encoding="utf-8")
+    assert "forget --keep-last 1 --prune --password-file" in args
+
+
+def test_retention_rejects_invalid_keep_last_before_restic(tmp_path: Path) -> None:
+    backup_base = tmp_path / "backup"
+    env = os.environ.copy()
+    env["BACKUP_BASE"] = str(backup_base)
+    result = subprocess.run(
+        [
+            "bash",
+            "-c",
+            f'source "{COMMON}"; BACKUP_RETENTION_MODE=latest '
+            'BACKUP_KEEP_LAST=0 cleanup_old_backups "$BACKUP_BASE/demo"',
+        ],
+        env=env,
+        text=True,
+        capture_output=True,
+    )
+
+    assert result.returncode == 64
+    assert "BACKUP_KEEP_LAST" in result.stdout
+
+
+def test_verify_backup_reads_the_requested_snapshot_before_repository_check(
+    tmp_path: Path,
+) -> None:
+    bin_dir = tmp_path / "bin"
+    bin_dir.mkdir()
+    event_log = tmp_path / "restic.args"
+    restic = bin_dir / "restic"
+    restic.write_text(
+        """#!/bin/sh
+printf '%s\n' "$*" >> "$TEST_EVENT_LOG"
+case " $* " in
+  *" snapshots "*) printf '%s\n' '[{"short_id":"abcdef12"}]' ;;
+  *" ls "*) printf '%s\n' '{"type":"file","path":"/backup/data.dump"}' ;;
+  *" check "*) ;;
+esac
+""",
+        encoding="utf-8",
+    )
+    restic.chmod(0o755)
+    backup_base = tmp_path / "backup"
+    password_file = backup_base / "scripts" / ".restic-password"
+    password_file.parent.mkdir(parents=True)
+    password_file.write_text("test-only\n", encoding="utf-8")
+    env = os.environ.copy()
+    env.update(
+        {
+            "BACKUP_BASE": str(backup_base),
+            "RESTIC_PASSWORD_FILE": str(password_file),
+            "PATH": f"{bin_dir}:{env['PATH']}",
+            "TEST_EVENT_LOG": str(event_log),
+        }
+    )
+
+    subprocess.run(
+        ["bash", "-c", f'source "{COMMON}"; verify_backup /repo abcdef12'],
+        env=env,
+        check=True,
+        text=True,
+        capture_output=True,
+    )
+
+    calls = event_log.read_text(encoding="utf-8")
+    assert "snapshots abcdef12 --json" in calls
+    assert "ls abcdef12 --json" in calls
+    assert "check --read-data-subset=1%" in calls
diff --git a/scripts/backup/tests/test_backup_host188_products_contract.py b/scripts/backup/tests/test_backup_host188_products_contract.py
index 4b4b2278e..0ef49e71d 100644
--- a/scripts/backup/tests/test_backup_host188_products_contract.py
+++ b/scripts/backup/tests/test_backup_host188_products_contract.py
@@ -47,7 +47,8 @@ def test_host188_product_backup_has_fixed_sources_and_online_verifiers() -> None
         "n8n export:workflow --backup",
         "n8n export:credentials --backup",
         'LOCAL_REPO="${BACKUP_BASE}/host188-products"',
-        "verify_backup",
+        "verify-host188-products-backup.sh",
+        '--snapshot "${snapshot_id}" --data-only',
     ):
         assert expected in source
 
@@ -63,6 +64,13 @@ def test_host188_product_backup_preserves_secret_and_runtime_boundaries() -> Non
     assert "REDISCLI_AUTH" in source
     assert 'if [ "${BACKUP_NOTIFY_SUCCESS:-0}" = "1" ]' in source
     assert "handle.read(1024 * 1024)" in source
+    assert '"recoveryReadinessClaimed": False' in source
+    assert '"n8nEncryptionKeyEscrowRequired": True' in source
+    assert 'DUMP_DIR="${BACKUP_BASE}/staging/' in source
+    assert "flock -n 9" in source
+    assert "HOST188_BACKUP_MIN_FREE_BYTES" in source
+    assert "HOST188_BACKUP_SSH_TIMEOUT_SECONDS" in source
+    assert "timeout --signal=TERM" in source
 
     for forbidden in (
         "--decrypted",
@@ -89,7 +97,9 @@ def test_host188_product_backup_is_wired_into_health_integrity_and_offsite() ->
 
     exporter = EXPORTER.read_text(encoding="utf-8")
     assert '"backup-host188-products.sh"' in exporter
-    assert '("host188_products", "/backup/host188-products", 48)' in exporter
+    assert '"verify-host188-products-backup.sh"' in exporter
+    assert '"verify-host188-products-archive.py"' in exporter
+    assert '("host188_products", "/backup/host188-products", 30)' in exporter
 
     agent99 = AGENT99.read_text(encoding="utf-8")
     assert 'name = "host188-products"' in agent99
@@ -99,3 +109,37 @@ def test_host188_product_backup_is_wired_into_health_integrity_and_offsite() ->
     for path in (FULL_PLAYBOOK, BOUNDED_PLAYBOOK):
         assert "backup-host188-products.sh" in path.read_text(encoding="utf-8")
     assert "backup-host188-products.sh" in REBOOT_READINESS.read_text(encoding="utf-8")
+
+    offsite_sync = OFFSITE_SCRIPTS[0].read_text(encoding="utf-8")
+    assert "configs|host188-products" in offsite_sync
+
+
+def test_host188_specialized_restore_drill_is_snapshot_specific_and_fail_closed() -> None:
+    verifier = (
+        ROOT / "scripts" / "backup" / "verify-host188-products-backup.sh"
+    ).read_text(encoding="utf-8")
+
+    for expected in (
+        'verify_backup "${REPO}" "${SNAPSHOT_ID}"',
+        'restic -r "${REPO}" dump "${SNAPSHOT_ID}"',
+        "verify-host188-products-archive.py",
+        "pg_restore --list",
+        "redis-check-rdb",
+        "n8n_encryption_key.last_verified",
+        "n8n_encryption_key_escrow_missing_or_stale",
+        "HOST188_RESTORE_DRILL_DATA_VERIFIED=1",
+        "HOST188_RESTORE_DRILL_RECOVERY_READY=0",
+    ):
+        assert expected in verifier
+    for forbidden in ("psql ", "redis-cli FLUSH", "docker restart", "docker stop"):
+        assert forbidden not in verifier
+
+    archive_validator = (
+        ROOT / "scripts" / "backup" / "verify-host188-products-archive.py"
+    ).read_text(encoding="utf-8")
+    assert "handle.read(1024 * 1024)" in archive_validator
+    assert ".read_bytes()[:5]" not in archive_validator
+
+    integrity = INTEGRITY.read_text(encoding="utf-8")
+    assert 'name}" = "host188-products"' in integrity
+    assert "verify-host188-products-backup.sh" in integrity
diff --git a/scripts/backup/tests/test_sync_offsite_active_backup_guard.py b/scripts/backup/tests/test_sync_offsite_active_backup_guard.py
new file mode 100644
index 000000000..87b2de6a0
--- /dev/null
+++ b/scripts/backup/tests/test_sync_offsite_active_backup_guard.py
@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+import os
+import subprocess
+import time
+from pathlib import Path
+
+import pytest
+
+
+ROOT = Path(__file__).resolve().parents[3]
+SCRIPT = ROOT / "scripts" / "backup" / "sync-offsite-backups.sh"
+
+
+@pytest.mark.parametrize("name", ["backup-host188-products.sh", "backup-all.sh"])
+def test_active_backup_guard_detects_direct_and_aggregate_runs(
+    tmp_path: Path, name: str
+) -> None:
+    process = subprocess.Popen(
+        ["bash", "-c", f"exec -a /backup/scripts/{name} sleep 30"],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    try:
+        time.sleep(0.1)
+        env = os.environ.copy()
+        env["BACKUP_BASE"] = str(tmp_path / "backup")
+        result = subprocess.run(
+            ["bash", "-c", f'source "{SCRIPT}"; active_backup_processes'],
+            env=env,
+            text=True,
+            capture_output=True,
+            check=True,
+        )
+        assert f"/backup/scripts/{name}" in result.stdout
+    finally:
+        process.terminate()
+        process.wait(timeout=5)
diff --git a/scripts/backup/tests/test_verify_host188_products_archive.py b/scripts/backup/tests/test_verify_host188_products_archive.py
new file mode 100644
index 000000000..c5717f199
--- /dev/null
+++ b/scripts/backup/tests/test_verify_host188_products_archive.py
@@ -0,0 +1,115 @@
+from __future__ import annotations
+
+import hashlib
+import importlib.util
+import json
+import sqlite3
+import tarfile
+from pathlib import Path
+from types import ModuleType
+
+import pytest
+
+
+ROOT = Path(__file__).resolve().parents[3]
+SCRIPT = ROOT / "scripts" / "backup" / "verify-host188-products-archive.py"
+
+
+def load_module() -> ModuleType:
+    spec = importlib.util.spec_from_file_location("host188_archive_verifier", SCRIPT)
+    assert spec and spec.loader
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def sha256(path: Path) -> str:
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def build_restore(root: Path) -> None:
+    timestamp = "20260718_120000"
+    vibe = root / f"vibework-postgres-{timestamp}.dump"
+    fifa = root / f"fifa-postgres-{timestamp}.dump"
+    redis = root / f"fifa-redis-{timestamp}.rdb"
+    n8n = root / f"n8n-online-{timestamp}.tar.gz"
+    validation = root / "n8n-validation.json"
+    vibe.write_bytes(b"PGDMP" + b"v" * 32)
+    fifa.write_bytes(b"PGDMP" + b"f" * 32)
+    redis.write_bytes(b"REDIS0011" + b"r" * 32)
+
+    n8n_root = root / "n8n-source"
+    (n8n_root / "workflows").mkdir(parents=True)
+    (n8n_root / "credentials").mkdir()
+    (n8n_root / "workflows" / "one.json").write_text('{"id":"one"}\n')
+    (n8n_root / "credentials" / "one.json").write_text('{"id":"encrypted"}\n')
+    with sqlite3.connect(n8n_root / "database.sqlite") as connection:
+        connection.execute("CREATE TABLE workflow_entity (id TEXT PRIMARY KEY)")
+        connection.execute("INSERT INTO workflow_entity VALUES ('one')")
+    with tarfile.open(n8n, "w:gz") as archive:
+        for path in sorted(n8n_root.rglob("*")):
+            archive.add(path, arcname=path.relative_to(n8n_root))
+    validation.write_text(
+        json.dumps(
+            {
+                "sqliteIntegrity": "ok",
+                "workflowFiles": 1,
+                "credentialFiles": 1,
+                "credentialsDecrypted": False,
+            }
+        )
+    )
+    for path in sorted(n8n_root.rglob("*"), reverse=True):
+        if path.is_file():
+            path.unlink()
+        else:
+            path.rmdir()
+    n8n_root.rmdir()
+
+    assets = []
+    for path in (vibe, fifa, redis, n8n, validation):
+        assets.append(
+            {
+                "name": path.name,
+                "sizeBytes": path.stat().st_size,
+                "sha256": sha256(path),
+            }
+        )
+    (root / "backup-manifest.json").write_text(
+        json.dumps(
+            {
+                "schemaVersion": "host188_product_backup_manifest_v1",
+                "sourceHost": "192.168.0.188",
+                "controllerHost": "192.168.0.110",
+                "assets": assets,
+                "credentialsDecrypted": False,
+                "serviceStopped": False,
+                "offsiteWritePerformed": False,
+                "recoveryReadinessClaimed": False,
+                "n8nEncryptionKeyEscrowRequired": True,
+            }
+        )
+    )
+
+
+def test_validates_all_restored_host188_assets(tmp_path: Path) -> None:
+    module = load_module()
+    build_restore(tmp_path)
+
+    receipt = module.validate_restore(tmp_path)
+
+    assert receipt["dataVerified"] is True
+    assert receipt["assetCount"] == 5
+    assert receipt["postgresArchiveCount"] == 2
+    assert receipt["redisArchiveCount"] == 1
+    assert receipt["n8nSqliteIntegrity"] == "ok"
+    assert receipt["recoveryReadinessClaimed"] is False
+
+
+def test_rejects_manifest_hash_mismatch(tmp_path: Path) -> None:
+    module = load_module()
+    build_restore(tmp_path)
+    next(tmp_path.glob("fifa-redis-*.rdb")).write_bytes(b"REDIS0011tampered")
+
+    with pytest.raises(ValueError, match="asset_(size|hash)_mismatch"):
+        module.validate_restore(tmp_path)
diff --git a/scripts/backup/tests/test_verify_host188_products_backup_shell.py b/scripts/backup/tests/test_verify_host188_products_backup_shell.py
new file mode 100644
index 000000000..201ac0ace
--- /dev/null
+++ b/scripts/backup/tests/test_verify_host188_products_backup_shell.py
@@ -0,0 +1,125 @@
+from __future__ import annotations
+
+import importlib.util
+import os
+import subprocess
+import time
+from pathlib import Path
+from types import ModuleType
+
+
+ROOT = Path(__file__).resolve().parents[3]
+SCRIPT = ROOT / "scripts" / "backup" / "verify-host188-products-backup.sh"
+FIXTURE_MODULE = ROOT / "scripts" / "backup" / "tests" / "test_verify_host188_products_archive.py"
+
+
+def load_fixture_module() -> ModuleType:
+    spec = importlib.util.spec_from_file_location("host188_restore_fixture", FIXTURE_MODULE)
+    assert spec and spec.loader
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def prepare_runtime(tmp_path: Path) -> tuple[dict[str, str], Path]:
+    restored = tmp_path / "restored"
+    restored.mkdir()
+    load_fixture_module().build_restore(restored)
+    bin_dir = tmp_path / "bin"
+    bin_dir.mkdir()
+    restic = bin_dir / "restic"
+    restic.write_text(
+        """#!/usr/bin/env python3
+import json
+import os
+import sys
+from pathlib import Path
+
+args = sys.argv[1:]
+root = Path(os.environ["FAKE_RESTORE_ROOT"])
+prefix = "/tmp/host188-products-backup-test"
+if "snapshots" in args:
+    print(json.dumps([{"id": "abcdef1234567890", "short_id": "abcdef12"}]))
+elif "ls" in args:
+    print(json.dumps({"struct_type": "snapshot", "id": "abcdef1234567890"}))
+    for path in sorted(root.iterdir()):
+        if path.is_file():
+            print(json.dumps({"type": "file", "path": f"{prefix}/{path.name}"}))
+elif "dump" in args:
+    selected = next((root / Path(value).name for value in args if (root / Path(value).name).is_file()), None)
+    if selected is None:
+        raise SystemExit(2)
+    sys.stdout.buffer.write(selected.read_bytes())
+elif "check" in args:
+    pass
+else:
+    raise SystemExit(2)
+""",
+        encoding="utf-8",
+    )
+    restic.chmod(0o755)
+    ssh = bin_dir / "ssh"
+    ssh.write_text("#!/bin/sh\ncat >/dev/null\n", encoding="utf-8")
+    ssh.chmod(0o755)
+    timeout = bin_dir / "timeout"
+    timeout.write_text(
+        '#!/bin/sh\ncase "$1" in --*) shift ;; esac\nshift\nexec "$@"\n',
+        encoding="utf-8",
+    )
+    timeout.chmod(0o755)
+    backup_base = tmp_path / "backup"
+    password = backup_base / "scripts" / ".restic-password"
+    password.parent.mkdir(parents=True)
+    password.write_text("test-only\n", encoding="utf-8")
+    env = os.environ.copy()
+    env.update(
+        {
+            "BACKUP_BASE": str(backup_base),
+            "BACKUP_LOG_DIR": str(backup_base / "logs"),
+            "RESTIC_PASSWORD_FILE": str(password),
+            "FAKE_RESTORE_ROOT": str(restored),
+            "HOST188_RESTORE_MIN_FREE_BYTES": "1",
+            "PATH": f"{bin_dir}:{env['PATH']}",
+        }
+    )
+    return env, backup_base
+
+
+def run_verifier(env: dict[str, str], *args: str) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        ["bash", str(SCRIPT), "--snapshot", "abcdef12", *args],
+        env=env,
+        text=True,
+        capture_output=True,
+    )
+
+
+def test_data_only_snapshot_restore_replay_passes_without_recovery_claim(tmp_path: Path) -> None:
+    env, _ = prepare_runtime(tmp_path)
+
+    result = run_verifier(env, "--data-only")
+
+    assert result.returncode == 0, result.stderr
+    assert "HOST188_RESTORE_DRILL_DATA_VERIFIED=1" in result.stdout
+    assert "HOST188_RESTORE_DRILL_RECOVERY_READY=0" in result.stdout
+    assert "HOST188_RESTORE_DRILL_SCOPE=data_only" in result.stdout
+
+
+def test_full_restore_replay_blocks_until_matching_escrow_marker(tmp_path: Path) -> None:
+    env, backup_base = prepare_runtime(tmp_path)
+
+    blocked = run_verifier(env)
+    assert blocked.returncode == 78
+    assert "n8n_encryption_key_escrow_missing_or_stale" in blocked.stdout
+
+    marker = backup_base / "escrow-evidence" / "n8n_encryption_key.last_verified"
+    marker.parent.mkdir(parents=True)
+    marker.write_text(
+        f"timestamp={int(time.time())}\n"
+        "item=n8n_encryption_key\n"
+        "evidence_id=TEST-NON-SECRET-EVIDENCE\n",
+        encoding="utf-8",
+    )
+    ready = run_verifier(env)
+    assert ready.returncode == 0, ready.stderr
+    assert "HOST188_RESTORE_DRILL_RECOVERY_READY=1" in ready.stdout
diff --git a/scripts/backup/verify-host188-products-archive.py b/scripts/backup/verify-host188-products-archive.py
new file mode 100644
index 000000000..44bdcd799
--- /dev/null
+++ b/scripts/backup/verify-host188-products-archive.py
@@ -0,0 +1,164 @@
+#!/usr/bin/env python3
+"""Validate an isolated Host188 product backup restore without data writes."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import sqlite3
+import tarfile
+import tempfile
+from pathlib import Path, PurePosixPath
+from typing import Any
+
+
+EXPECTED_PREFIXES = {
+    "vibework-postgres-": ".dump",
+    "fifa-postgres-": ".dump",
+    "fifa-redis-": ".rdb",
+    "n8n-online-": ".tar.gz",
+}
+FIXED_ASSETS = {"n8n-validation.json"}
+
+
+def sha256_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def read_magic(path: Path) -> bytes:
+    with path.open("rb") as handle:
+        return handle.read(5)
+
+
+def _single_prefixed_asset(names: set[str], prefix: str, suffix: str) -> str:
+    matches = sorted(name for name in names if name.startswith(prefix) and name.endswith(suffix))
+    if len(matches) != 1:
+        raise ValueError(f"asset_cardinality_invalid:{prefix}")
+    return matches[0]
+
+
+def _extract_n8n_archive(archive_path: Path, target: Path) -> None:
+    with tarfile.open(archive_path, "r:gz") as archive:
+        members = archive.getmembers()
+        for member in members:
+            path = PurePosixPath(member.name)
+            if path.is_absolute() or ".." in path.parts:
+                raise ValueError("n8n_archive_unsafe_path")
+            if not (member.isfile() or member.isdir()):
+                raise ValueError("n8n_archive_unsafe_member")
+        archive.extractall(target, members=members)
+
+
+def validate_restore(root: Path) -> dict[str, Any]:
+    manifest_path = root / "backup-manifest.json"
+    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    if manifest.get("schemaVersion") != "host188_product_backup_manifest_v1":
+        raise ValueError("manifest_schema_invalid")
+    if manifest.get("sourceHost") != "192.168.0.188":
+        raise ValueError("manifest_source_invalid")
+    if manifest.get("controllerHost") != "192.168.0.110":
+        raise ValueError("manifest_controller_invalid")
+    if manifest.get("credentialsDecrypted") is not False:
+        raise ValueError("credential_boundary_invalid")
+    if manifest.get("serviceStopped") is not False:
+        raise ValueError("service_stop_boundary_invalid")
+    if manifest.get("offsiteWritePerformed") is not False:
+        raise ValueError("offsite_boundary_invalid")
+    if manifest.get("recoveryReadinessClaimed") is not False:
+        raise ValueError("recovery_readiness_false_green")
+    if manifest.get("n8nEncryptionKeyEscrowRequired") is not True:
+        raise ValueError("n8n_escrow_requirement_missing")
+
+    assets = manifest.get("assets")
+    if not isinstance(assets, list):
+        raise ValueError("manifest_assets_invalid")
+    by_name: dict[str, dict[str, Any]] = {}
+    for item in assets:
+        if not isinstance(item, dict):
+            raise ValueError("manifest_asset_invalid")
+        name = str(item.get("name") or "")
+        if not name or name in by_name or Path(name).name != name:
+            raise ValueError("manifest_asset_name_invalid")
+        by_name[name] = item
+
+    names = set(by_name)
+    if not FIXED_ASSETS.issubset(names):
+        raise ValueError("fixed_asset_missing")
+    selected = {
+        prefix: _single_prefixed_asset(names, prefix, suffix)
+        for prefix, suffix in EXPECTED_PREFIXES.items()
+    }
+    expected_names = set(selected.values()) | FIXED_ASSETS
+    if names != expected_names:
+        raise ValueError("unexpected_manifest_asset")
+
+    for name, item in by_name.items():
+        path = root / name
+        if not path.is_file():
+            raise ValueError(f"asset_missing:{name}")
+        if path.stat().st_size != int(item.get("sizeBytes") or -1):
+            raise ValueError(f"asset_size_mismatch:{name}")
+        if sha256_file(path) != str(item.get("sha256") or ""):
+            raise ValueError(f"asset_hash_mismatch:{name}")
+
+    for prefix in ("vibework-postgres-", "fifa-postgres-"):
+        if read_magic(root / selected[prefix]) != b"PGDMP":
+            raise ValueError(f"postgres_magic_invalid:{prefix}")
+    if read_magic(root / selected["fifa-redis-"]) != b"REDIS":
+        raise ValueError("redis_magic_invalid")
+
+    validation = json.loads((root / "n8n-validation.json").read_text(encoding="utf-8"))
+    if validation.get("sqliteIntegrity") != "ok":
+        raise ValueError("n8n_prior_validation_invalid")
+    if validation.get("credentialsDecrypted") is not False:
+        raise ValueError("n8n_prior_credential_boundary_invalid")
+
+    with tempfile.TemporaryDirectory(
+        prefix="host188-n8n-restore-", dir=root.parent
+    ) as temp_dir:
+        extracted = Path(temp_dir)
+        _extract_n8n_archive(root / selected["n8n-online-"], extracted)
+        database = extracted / "database.sqlite"
+        with sqlite3.connect(f"file:{database}?mode=ro", uri=True) as connection:
+            integrity = connection.execute("PRAGMA integrity_check").fetchone()[0]
+        if integrity != "ok":
+            raise ValueError("n8n_sqlite_integrity_invalid")
+        counts: dict[str, int] = {}
+        for category in ("workflows", "credentials"):
+            directory = extracted / category
+            rows = sorted(directory.rglob("*.json")) if directory.is_dir() else []
+            for path in rows:
+                json.loads(path.read_text(encoding="utf-8"))
+            counts[category] = len(rows)
+        if counts["workflows"] != int(validation.get("workflowFiles") or 0):
+            raise ValueError("n8n_workflow_count_mismatch")
+        if counts["credentials"] != int(validation.get("credentialFiles") or 0):
+            raise ValueError("n8n_credential_count_mismatch")
+
+    return {
+        "schemaVersion": "host188_product_restore_validation_v1",
+        "dataVerified": True,
+        "assetCount": len(by_name),
+        "postgresArchiveCount": 2,
+        "redisArchiveCount": 1,
+        "n8nSqliteIntegrity": "ok",
+        "credentialsDecrypted": False,
+        "recoveryReadinessClaimed": False,
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--root", type=Path, required=True)
+    args = parser.parse_args()
+    print(json.dumps(validate_restore(args.root), separators=(",", ":")))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/scripts/backup/verify-host188-products-backup.sh b/scripts/backup/verify-host188-products-backup.sh
new file mode 100644
index 000000000..ff3377c9c
--- /dev/null
+++ b/scripts/backup/verify-host188-products-backup.sh
@@ -0,0 +1,201 @@
+#!/usr/bin/env bash
+# Snapshot-specific isolated restore drill for Host188 product data.
+
+set -Eeuo pipefail
+umask 077
+
+source "$(dirname "$0")/common.sh"
+
+SERVICE="host188-products-restore-drill"
+REPO="${BACKUP_BASE}/host188-products"
+REMOTE_HOST="ollama@192.168.0.188"
+VALIDATOR="$(dirname "$0")/verify-host188-products-archive.py"
+ESCROW_MARKER="${N8N_ESCROW_EVIDENCE_FILE:-${BACKUP_BASE}/escrow-evidence/n8n_encryption_key.last_verified}"
+ESCROW_MAX_AGE_SECONDS="${N8N_ESCROW_MAX_AGE_SECONDS:-2678400}"
+SSH_TIMEOUT_SECONDS="${HOST188_RESTORE_SSH_TIMEOUT_SECONDS:-1800}"
+MIN_FREE_BYTES="${HOST188_RESTORE_MIN_FREE_BYTES:-8589934592}"
+SNAPSHOT_ID="latest"
+DATA_ONLY=0
+RESTORE_DIR="${BACKUP_BASE}/restore-drill/host188-products-$$"
+SSH_OPTIONS=(
+    -o BatchMode=yes
+    -o StrictHostKeyChecking=yes
+    -o ConnectTimeout=8
+    -o ServerAliveInterval=10
+    -o ServerAliveCountMax=2
+)
+
+cleanup() {
+    rm -rf "${RESTORE_DIR}"
+}
+
+usage() {
+    echo "Usage: verify-host188-products-backup.sh [--snapshot <id|latest>] [--data-only]"
+}
+
+while [ "$#" -gt 0 ]; do
+    case "$1" in
+        --snapshot) SNAPSHOT_ID="${2:-}"; shift 2 ;;
+        --data-only) DATA_ONLY=1; shift ;;
+        -h|--help) usage; exit 0 ;;
+        *) usage >&2; exit 2 ;;
+    esac
+done
+
+latest_snapshot_id() {
+    restic -r "${REPO}" snapshots --latest 1 --json \
+        --password-file "${RESTIC_PASSWORD_FILE}" | \
+        python3 -c 'import json,sys; rows=json.load(sys.stdin); print(rows[-1]["short_id"] if rows else "")'
+}
+
+snapshot_paths() {
+    restic -r "${REPO}" ls "${SNAPSHOT_ID}" --json \
+        --password-file "${RESTIC_PASSWORD_FILE}" | python3 -c '
+import json
+import re
+import sys
+
+patterns = (
+    re.compile(r"vibework-postgres-[0-9]{8}_[0-9]{6}[.]dump$"),
+    re.compile(r"fifa-postgres-[0-9]{8}_[0-9]{6}[.]dump$"),
+    re.compile(r"fifa-redis-[0-9]{8}_[0-9]{6}[.]rdb$"),
+    re.compile(r"n8n-online-[0-9]{8}_[0-9]{6}[.]tar[.]gz$"),
+    re.compile(r"n8n-validation[.]json$"),
+    re.compile(r"backup-manifest[.]json$"),
+)
+found = {pattern.pattern: [] for pattern in patterns}
+for line in sys.stdin:
+    try:
+        row = json.loads(line)
+    except json.JSONDecodeError:
+        continue
+    if row.get("type") != "file":
+        continue
+    path = str(row.get("path") or "")
+    for pattern in patterns:
+        if pattern.search(path):
+            found[pattern.pattern].append(path)
+for pattern in patterns:
+    rows = found[pattern.pattern]
+    if len(rows) != 1:
+        raise SystemExit("snapshot_asset_cardinality_invalid")
+    print(rows[0])
+'
+}
+
+verify_postgres_archive() {
+    local container="$1"
+    local archive="$2"
+    local command
+    printf -v command 'docker exec -i %q pg_restore --list >/dev/null' "${container}"
+    timeout --signal=TERM "${SSH_TIMEOUT_SECONDS}s" \
+        ssh "${SSH_OPTIONS[@]}" "${REMOTE_HOST}" "${command}" <"${archive}"
+}
+
+verify_redis_archive() {
+    local archive="$1"
+    local inner command
+    inner='set -eu; umask 077; tmp=/tmp/agent99-fifa-redis-restore-drill.rdb; rm -f "$tmp"; trap '\''rm -f "$tmp"'\'' EXIT HUP INT TERM; cat >"$tmp"; redis-check-rdb "$tmp" >/dev/null'
+    printf -v command \
+        'test "$(docker inspect -f '\''{{.State.Running}}'\'' current-fifa2026-redis-1 2>/dev/null)" = true && docker exec -i current-fifa2026-redis-1 sh -eu -c %q' \
+        "${inner}"
+    timeout --signal=TERM "${SSH_TIMEOUT_SECONDS}s" \
+        ssh "${SSH_OPTIONS[@]}" "${REMOTE_HOST}" "${command}" <"${archive}"
+}
+
+escrow_marker_fresh() {
+    python3 - "${ESCROW_MARKER}" "${ESCROW_MAX_AGE_SECONDS}" <<'PY'
+import sys
+import time
+from pathlib import Path
+
+path = Path(sys.argv[1])
+max_age = int(sys.argv[2])
+if not path.is_file():
+    raise SystemExit(1)
+values = {}
+for line in path.read_text(encoding="utf-8").splitlines():
+    key, separator, value = line.partition("=")
+    if separator:
+        values[key] = value
+timestamp = int(values.get("timestamp") or 0)
+valid = (
+    values.get("item") == "n8n_encryption_key"
+    and bool(values.get("evidence_id"))
+    and 0 <= int(time.time()) - timestamp <= max_age
+)
+raise SystemExit(0 if valid else 1)
+PY
+}
+
+main() {
+    local path name available_bytes
+    local paths=()
+    trap cleanup EXIT HUP INT TERM
+    install -d -m 700 "${RESTORE_DIR}"
+    for command_name in restic ssh python3 timeout; do
+        command -v "${command_name}" >/dev/null 2>&1 || {
+            log_error "Host188 restore command missing: ${command_name}"
+            return 69
+        }
+    done
+    case "${SSH_TIMEOUT_SECONDS}:${ESCROW_MAX_AGE_SECONDS}:${MIN_FREE_BYTES}" in
+        *[!0-9:]*|:*|*:) log_error "Host188 restore numeric limits are invalid"; return 64 ;;
+    esac
+    if [ "${SSH_TIMEOUT_SECONDS}" -le 0 ] || [ "${ESCROW_MAX_AGE_SECONDS}" -le 0 ] || [ "${MIN_FREE_BYTES}" -le 0 ]; then
+        log_error "Host188 restore numeric limits must be positive"
+        return 64
+    fi
+    available_bytes="$(df -Pk "${BACKUP_BASE}" 2>/dev/null | awk 'NR==2 {printf "%.0f", $4 * 1024}')"
+    if [ -z "${available_bytes}" ] || [ "${available_bytes}" -lt "${MIN_FREE_BYTES}" ]; then
+        log_error "Host188 isolated restore headroom is insufficient"
+        return 75
+    fi
+    [ -r "${VALIDATOR}" ] || { log_error "Host188 restore validator missing"; return 69; }
+    [ -r "${RESTIC_PASSWORD_FILE}" ] || { log_error "Restic password reference unavailable"; return 69; }
+
+    if [ "${SNAPSHOT_ID}" = "latest" ]; then
+        SNAPSHOT_ID="$(latest_snapshot_id)"
+    fi
+    [[ "${SNAPSHOT_ID}" =~ ^[0-9a-fA-F]{8,64}$ ]] || {
+        log_error "Host188 snapshot id invalid"
+        return 64
+    }
+    verify_backup "${REPO}" "${SNAPSHOT_ID}"
+    while IFS= read -r path; do
+        paths+=("${path}")
+    done < <(snapshot_paths)
+    [ "${#paths[@]}" -eq 6 ] || { log_error "Host188 snapshot asset list incomplete"; return 1; }
+    for path in "${paths[@]}"; do
+        name="${path##*/}"
+        restic -r "${REPO}" dump "${SNAPSHOT_ID}" "${path}" \
+            --password-file "${RESTIC_PASSWORD_FILE}" >"${RESTORE_DIR}/${name}"
+    done
+
+    python3 "${VALIDATOR}" --root "${RESTORE_DIR}" >"${RESTORE_DIR}/validation-receipt.json"
+    verify_postgres_archive \
+        "vibework-production-postgres-1" \
+        "$(find "${RESTORE_DIR}" -maxdepth 1 -name 'vibework-postgres-*.dump' -print -quit)"
+    verify_postgres_archive \
+        "current-fifa2026-postgres-1" \
+        "$(find "${RESTORE_DIR}" -maxdepth 1 -name 'fifa-postgres-*.dump' -print -quit)"
+    verify_redis_archive \
+        "$(find "${RESTORE_DIR}" -maxdepth 1 -name 'fifa-redis-*.rdb' -print -quit)"
+
+    if [ "${DATA_ONLY}" = "1" ]; then
+        echo "HOST188_RESTORE_DRILL_DATA_VERIFIED=1"
+        echo "HOST188_RESTORE_DRILL_RECOVERY_READY=0"
+        echo "HOST188_RESTORE_DRILL_SCOPE=data_only"
+        return 0
+    fi
+    if ! escrow_marker_fresh; then
+        echo "HOST188_RESTORE_DRILL_DATA_VERIFIED=1"
+        echo "HOST188_RESTORE_DRILL_RECOVERY_READY=0"
+        echo "HOST188_RESTORE_DRILL_BLOCKER=n8n_encryption_key_escrow_missing_or_stale"
+        return 78
+    fi
+    echo "HOST188_RESTORE_DRILL_DATA_VERIFIED=1"
+    echo "HOST188_RESTORE_DRILL_RECOVERY_READY=1"
+}
+
+main "$@"
diff --git a/scripts/ops/ansible-validate.sh b/scripts/ops/ansible-validate.sh
index 4c03d54be..327b253f2 100755
--- a/scripts/ops/ansible-validate.sh
+++ b/scripts/ops/ansible-validate.sh
@@ -65,6 +65,7 @@ bash -n \
   scripts/backup/backup-ai-artifacts.sh \
   scripts/backup/backup-public-routes.sh \
   scripts/backup/backup-host188-products.sh \
+  scripts/backup/verify-host188-products-backup.sh \
   scripts/backup/configure-offsite-rclone.sh \
   scripts/backup/configure-offsite-b2.sh \
   scripts/backup/sync-offsite-backups.sh \
@@ -86,6 +87,7 @@ python3 -m py_compile \
   scripts/ops/backup-alert-live-visibility-check.py \
   scripts/ops/recovery-scorecard-contract-check.py \
   scripts/backup/clickhouse-restore-inventory.py \
+  scripts/backup/verify-host188-products-archive.py \
   scripts/ops/doc-secrets-sanity-check.py
 echo "Python 語法 OK"
 
@@ -107,6 +109,7 @@ echo "== Ansible syntax-check =="
 for playbook in \
   infra/ansible/playbooks/site.yml \
   infra/ansible/playbooks/110-devops.yml \
+  infra/ansible/playbooks/110-backup-runtime-deploy.yml \
   infra/ansible/playbooks/188-ai-web.yml \
   infra/ansible/playbooks/nginx-sync.yml; do
   ansible-playbook -i infra/ansible/inventory/hosts.yml "$playbook" --syntax-check
diff --git a/scripts/ops/backup-health-textfile-exporter.py b/scripts/ops/backup-health-textfile-exporter.py
index ad2063111..ebc955bc2 100755
--- a/scripts/ops/backup-health-textfile-exporter.py
+++ b/scripts/ops/backup-health-textfile-exporter.py
@@ -1193,6 +1193,9 @@ def _collect_110(host: str) -> list[str]:
         "backup-ai-artifacts.sh",
         "backup-public-routes.sh",
         "backup-host188-products.sh",
+        "verify-host188-products-backup.sh",
+        "verify-host188-products-archive.py",
+        "enforce-latest-only-retention.sh",
         "configure-offsite-rclone.sh",
         "configure-offsite-b2.sh",
         "sync-offsite-backups.sh",
@@ -1231,7 +1234,7 @@ def _collect_110(host: str) -> list[str]:
         ("clawbot", "/backup/clawbot", 48),
         ("ai_artifacts", "/backup/ai-artifacts", 48),
         ("public_routes", "/backup/public-routes", 168),
-        ("host188_products", "/backup/host188-products", 48),
+        ("host188_products", "/backup/host188-products", 30),
     ]:
         timestamp, count = _latest_restic_snapshot(repo)
         age = int(time.time()) - timestamp if timestamp else 0
diff --git a/scripts/reboot-recovery/reboot-recovery-readiness-audit.sh b/scripts/reboot-recovery/reboot-recovery-readiness-audit.sh
index 43752b81b..6162e72a3 100755
--- a/scripts/reboot-recovery/reboot-recovery-readiness-audit.sh
+++ b/scripts/reboot-recovery/reboot-recovery-readiness-audit.sh
@@ -279,6 +279,8 @@ require_file scripts/backup/backup-sentry.sh "Sentry dedicated data backup"
 require_file scripts/backup/backup-ai-artifacts.sh "AI artifacts and Ollama manifest backup"
 require_file scripts/backup/backup-public-routes.sh "Public routes DNS/TLS evidence backup"
 require_file scripts/backup/backup-host188-products.sh "Host188 product data-plane backup"
+require_file scripts/backup/verify-host188-products-backup.sh "Host188 snapshot-specific restore drill"
+require_file scripts/backup/verify-host188-products-archive.py "Host188 isolated restore archive validator"
 require_file scripts/backup/configure-offsite-rclone.sh "Offsite Google Drive/rclone host-local config helper"
 require_pattern "create-root-remote" scripts/backup/configure-offsite-rclone.sh "Offsite Google Drive root-scoped remote helper"
 require_pattern "gdrive_awoooi_restic" docs/runbooks/OFFSITE-BACKUP-ESCROW-RUNBOOK.md "Offsite Google Drive root-scoped remote runbook"