From d6ad9d280ff0341e1f8324863744866ba8c50a4b Mon Sep 17 00:00:00 2001
From: cxymds <cxymds@gmail.com>
Date: Wed, 11 Nov 2026 11:51:22 +0800
Subject: [PATCH 1/2] =?UTF-8?q?fix(upgrade):=20=E4=BF=9D=E7=95=99=E7=A7=8D?=
 =?UTF-8?q?=E5=AD=90=E5=A4=B1=E8=B4=A5=E8=AF=81=E6=8D=AE=E5=B9=B6=E8=A1=A5?=
 =?UTF-8?q?=E9=BD=90=E5=8D=87=E7=BA=A7=E7=9F=A9=E9=98=B5?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .github/workflows/rustfs-upgrade-test.yml  |  15 +-
 .github/workflows/upgrade-script-tests.yml |  46 +++
 docs/upgrade-test-evidence.md              |  34 ++
 rustfs-upgrade-test.sh                     | 317 +++++++++++++--
 scripts/functional_case_report.py          |  55 ++-
 scripts/issue_manager.py                   |   6 +-
 scripts/upgrade_evidence.py                | 185 +++++++++
 test_rustfs_upgrade.py                     | 426 +++++++++++++++++++++
 8 files changed, 1023 insertions(+), 61 deletions(-)
 create mode 100644 .github/workflows/upgrade-script-tests.yml
 create mode 100644 docs/upgrade-test-evidence.md
 create mode 100644 scripts/upgrade_evidence.py
 create mode 100644 test_rustfs_upgrade.py

diff --git a/.github/workflows/rustfs-upgrade-test.yml b/.github/workflows/rustfs-upgrade-test.yml
index e4916e76..0ce1b07e 100644
--- a/.github/workflows/rustfs-upgrade-test.yml
+++ b/.github/workflows/rustfs-upgrade-test.yml
@@ -174,7 +174,7 @@ jobs:
           TO_VERSION='${{ inputs.to_version }}'
           TOPOLOGY='${{ inputs.topology }}'
           BACKENDS='${{ inputs.backends }}'
-          ARGS=(-y --log-file "${LOG_FILE}")
+          ARGS=(-y --log-file "${LOG_FILE}" --artifacts-dir "${FUNCTIONAL_ARTIFACTS_DIR}/upgrade")
           if [ "${TOPOLOGY}" = "all" ] || [ -z "${TOPOLOGY}" ] || [ "${TOPOLOGY}" = "null" ]; then
             ARGS+=(--all-topologies)
           else
@@ -246,8 +246,10 @@ jobs:
           fi
           CASE_TABLE="${FUNCTIONAL_ARTIFACTS_DIR}/cases.md"
           MATRIX_TABLE="${FUNCTIONAL_ARTIFACTS_DIR}/matrix.md"
+          CASE_LOG="${FUNCTIONAL_ARTIFACTS_DIR}/upgrade/events.log"
+          [ -f "${CASE_LOG}" ] || CASE_LOG="${LOG_FILE}"
           CASE_RESULT=success
-          python3 scripts/functional_case_report.py "${LOG_FILE}" "${CASE_TABLE}" "${MATRIX_TABLE}" || CASE_RESULT=failure
+          python3 scripts/functional_case_report.py "${CASE_LOG}" "${CASE_TABLE}" "${MATRIX_TABLE}" || CASE_RESULT=failure
           RESULT=failure
           if [ '${{ steps.test.outcome }}' = 'success' ] && [ "${CASE_RESULT}" = 'success' ]; then
             RESULT=success
@@ -255,10 +257,12 @@ jobs:
 
           # Report harness completion here; the final verdict checks every
           # case result after diagnostics and cleanup have been retained.
-          CASES_TOTAL="$(grep -cE '^\| [A-Z][A-Z0-9]*-[0-9]+ .*\| (PASS|FAIL|UNSUPPORTED|RUNNING) \|' "${CASE_TABLE}" 2>/dev/null || true)"
+          CASES_TOTAL="$(grep -cE '^\| [A-Z][A-Z0-9]*-[0-9]+ .*\| (PASS|FAIL|UNSUPPORTED|RUNNING|NOT_RUN) \|' "${CASE_TABLE}" 2>/dev/null || true)"
           CASES_FAIL="$(grep -cE '^\| [A-Z][A-Z0-9]*-[0-9]+ .*\| FAIL \|' "${CASE_TABLE}" 2>/dev/null || true)"
+          CASES_PASS="$(grep -cE '^\| [A-Z][A-Z0-9]*-[0-9]+ .*\| PASS \|' "${CASE_TABLE}" 2>/dev/null || true)"
+          CASES_NOT_RUN="$(grep -cE '^\| [A-Z][A-Z0-9]*-[0-9]+ .*\| (NOT_RUN|RUNNING) \|' "${CASE_TABLE}" 2>/dev/null || true)"
           CASES_TOTAL=$(( ${CASES_TOTAL:-0} + 0 )); CASES_FAIL=$(( ${CASES_FAIL:-0} + 0 ))
-          CASES_PASS=$(( CASES_TOTAL - CASES_FAIL ))
+          CASES_PASS=$(( ${CASES_PASS:-0} + 0 )); CASES_NOT_RUN=$(( ${CASES_NOT_RUN:-0} + 0 ))
           HARNESS_OK=0
           if [ '${{ steps.test.outcome }}' = 'success' ]; then
             HARNESS_OK=1
@@ -285,7 +289,7 @@ jobs:
               cat "${CASE_TABLE}"
               echo ""
             fi
-            echo "- Product result: ${CASES_PASS} passed, ${CASES_FAIL} failed (failing cases are tracked in rustfs/backlog)"
+            echo "- 用例结果：${CASES_PASS} 通过，${CASES_FAIL} 断言失败，${CASES_NOT_RUN} 未完成；准备、升级及恢复状态见矩阵。"
             echo ""
             if [ -s "${CASE_TABLE}" ]; then
               echo "## Log tail"
@@ -401,6 +405,7 @@ jobs:
             ${{ env.FUNCTIONAL_ARTIFACTS_DIR }}/suite.log
             ${{ env.FUNCTIONAL_ARTIFACTS_DIR }}/cases.md
             ${{ env.FUNCTIONAL_ARTIFACTS_DIR }}/matrix.md
+            ${{ env.FUNCTIONAL_ARTIFACTS_DIR }}/upgrade/
           if-no-files-found: error
           retention-days: 3
 
diff --git a/.github/workflows/upgrade-script-tests.yml b/.github/workflows/upgrade-script-tests.yml
new file mode 100644
index 00000000..e7346809
--- /dev/null
+++ b/.github/workflows/upgrade-script-tests.yml
@@ -0,0 +1,46 @@
+name: Upgrade Script Tests
+
+on:
+  pull_request:
+    paths:
+      - rustfs-upgrade-test.sh
+      - scripts/upgrade_evidence.py
+      - scripts/functional_case_report.py
+      - scripts/issue_manager.py
+      - scripts/lib/**
+      - test_rustfs_upgrade.py
+      - .github/workflows/rustfs-upgrade-test.yml
+      - .github/workflows/upgrade-script-tests.yml
+  push:
+    branches: [main]
+    paths:
+      - rustfs-upgrade-test.sh
+      - scripts/upgrade_evidence.py
+      - scripts/functional_case_report.py
+      - scripts/issue_manager.py
+      - scripts/lib/**
+      - test_rustfs_upgrade.py
+      - .github/workflows/rustfs-upgrade-test.yml
+      - .github/workflows/upgrade-script-tests.yml
+
+permissions:
+  contents: read
+
+jobs:
+  upgrade-script-tests:
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
+        with:
+          persist-credentials: false
+      - name: Install protocol fixture client
+        run: |
+          python3 -m venv "${RUNNER_TEMP}/upgrade-tests"
+          "${RUNNER_TEMP}/upgrade-tests/bin/pip" install awscli==1.45.19
+          echo "${RUNNER_TEMP}/upgrade-tests/bin" >> "${GITHUB_PATH}"
+          command -v jq
+      - name: Validate upgrade suite and evidence handling
+        run: |
+          bash -n rustfs-upgrade-test.sh
+          python3 -B -m unittest -v test_rustfs_upgrade.py test_sensitive_diagnostics.py test_issue_manager.py
diff --git a/docs/upgrade-test-evidence.md b/docs/upgrade-test-evidence.md
new file mode 100644
index 00000000..5cbb92a3
--- /dev/null
+++ b/docs/upgrade-test-evidence.md
@@ -0,0 +1,34 @@
+# 升级套件的种子与结果证据
+
+`rustfs-upgrade-test.sh` 在每个拓扑和 KMS 后端组合上依次执行旧版准备、原地升级、18 条升级后断言、诊断归档和环境清理。默认六组，共 108 次用例执行。
+
+100MiB 种子固定写入两个对象，避免按时间生成大量对象后再裁剪。1KiB 和 1MiB 种子保留 Warp 生成方式，分别保留五个对象。升级前读回种子，记录完整键清单、字节数和 SHA-256；升级后逐对象比较清单和内容，并保留 HEAD/GET ETag 检查。
+
+裁剪请求每批最多 1000 个键。AWS CLI 退出成功仍需检查响应的 `Errors`，发现逐对象错误立即停止准备，不自动重试或把错误当作成功。每批保存请求键、CLI 退出码、可解析的 HTTP 状态、请求 ID 和安全响应。顶层 S3 错误仅保留 Code、Message 和 RequestId；签名请求、原始 CLI debug、凭据和 golden.tsv 留在私有临时目录，退出时删除。
+
+通过 `--artifacts-dir <新目录>` 指定证据目录，或使用 `RUSTFS_UPGRADE_ARTIFACTS_DIR`。目录必须尚不存在，避免复用旧结果。未指定时默认使用 `<log-file>.artifacts`，无日志路径时使用当前目录中的 UTC 时间路径。
+
+证据目录包含：
+
+- `events.log`：同步写入的逐项用例和阶段结果，不依赖 tee 完成时序。
+- `cases.md`、`matrix.md`、`results.json`：所有选定组合的逐项结果和结构化汇总。
+- `<拓扑>-<后端>/`：种子清单、删除响应、逐项诊断、Warp 输出及节点文件日志和 journal。节点文件日志最多保留每文件末尾 4MiB、每节点合计 16MiB，截断会标明。
+
+报告状态：
+
+| 状态 | 含义 |
+| --- | --- |
+| PASS | 升级后断言通过，诊断归档成功，环境已清理或显式保留 |
+| PREPARE_FAILED | 旧版部署、健康检查或种子准备失败，尚未进入升级 |
+| UPGRADE_FAILED | 原地升级或升级后健康检查失败 |
+| VERIFY_FAILED | 已进入新版验证，至少一个断言失败 |
+| INTERRUPTED | 收到 INT/TERM 中断 |
+| DIAGNOSTICS_FAILED | 诊断证据未完整归档 |
+| CLEANUP_FAILED | 环境清理未完成 |
+| NOT_RUN | 该组或用例未执行完成，不能作为通过证据 |
+
+准备失败和中断都会补齐未完成的用例，先归档再清理当前组，并将后续组标记为未执行。普通断言失败在清理成功后继续其他组合；诊断或清理失败则停止后续组。主失败原因与清理/诊断结果分别保留。`--keep-env` 显式保留环境，仍归档证据并标记清理为 `RETAINED`。
+
+例如三组 local 验证通过后，vault-kv2 第一组种子准备失败，报告应为 54 PASS、54 NOT_RUN，矩阵包含 `PREPARE_FAILED`。这证明本轮升级矩阵未完成，不能据此判定新版 Vault 兼容性失败，也不能自动关闭历史产品 Issue。
+
+GitHub Actions 上传上述安全证据目录。独立的 `Upgrade Script Tests` 使用本地协议 fixture 和模拟环境边界验证脚本，包含真实 AWS CLI 的 HTTP 200 逐对象错误、顶层 503 和批量删除分片。这些检查不连接 VM，不代表产品实机复测。
diff --git a/rustfs-upgrade-test.sh b/rustfs-upgrade-test.sh
index 2b18d1c7..f66e0caa 100755
--- a/rustfs-upgrade-test.sh
+++ b/rustfs-upgrade-test.sh
@@ -46,7 +46,21 @@ SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
 JQ_LIB="${SCRIPT_DIR}/scripts/lib"
 
 TEST_TMP="$(mktemp -d "${TMPDIR:-/tmp}/rustfs-upgrade.XXXXXX")"
-trap 'rm -rf "${TEST_TMP}"' EXIT
+trap 'upgrade_exit "$?"' EXIT
+trap 'exit 130' INT
+trap 'exit 143' TERM
+ARTIFACTS_DIR="${RUSTFS_UPGRADE_ARTIFACTS_DIR:-}"
+UPGRADE_PYTHON="${RUSTFS_UPGRADE_PYTHON:-python3}"
+GROUP_ACTIVE=0
+ARTIFACTS_READY=0
+GROUP_INDEX=0
+UPGRADE_GROUPS=()
+CURRENT_DONE=" "
+CURRENT_STARTED=" "
+CURRENT_PHASE=PREPARE
+CURRENT_RESULT=""
+DIAGNOSTICS_DONE=0
+DELETE_BATCH=0
 
 # ==================== Configuration ====================
 NODES=(vm000 vm001 vm002 vm003)
@@ -461,6 +475,9 @@ vault_stop() {
   [ "${DRY_RUN}" -eq 1 ] && { log "DRY-RUN: remove vault container on ${HEAL_HOST}"; return 0; }
   log "stopping and removing vault container ${VAULT_CONTAINER} on ${HEAL_HOST}"
   heal_run docker rm -f "${VAULT_CONTAINER}" >/dev/null 2>&1 || true
+  local remaining
+  remaining="$(heal_run docker ps -aq --filter "name=^/${VAULT_CONTAINER}$")" || return 1
+  [ -z "${remaining}" ] || { warn "vault container removal failed"; return 1; }
   # rm through heal_run: vault creates core/sys/logical with restrictive
   # modes inside the data dir, so the runner user cannot remove them
   # directly (rm exits 1 under set -e and kills the suite mid-loop).
@@ -627,9 +644,128 @@ upgrade_in_place() {
 }
 
 env_teardown() {
-  local topo="$1"
+  local topo="$1" fail=0
   log "environment teardown: ${topo} (kms=${KMS_BACKEND})"
-  for node in "${NODES[@]}"; do cleanup_node "${node}"; done
+  for node in "${NODES[@]}"; do cleanup_node "${node}" || fail=1; done
+  return "${fail}"
+}
+
+# Evidence is written synchronously; report generation never races tee.
+evidence_line() {
+  printf '%s\n' "$*"
+  if [ "${ARTIFACTS_READY}" -eq 1 ]; then printf '%s\n' "$*" >> "${ARTIFACTS_DIR}/events.log"; fi
+}
+
+collect_upgrade_diagnostics() {
+  [ "${DIAGNOSTICS_DONE}" -eq 0 ] || return 0
+  DIAGNOSTICS_DONE=1
+  [ "${DRY_RUN}" -eq 0 ] || return 0
+  local node fail=0 nodes=()
+  local warp_log
+  for warp_log in "${TEST_TMP}"/warp-put-*.log; do
+    [ -f "${warp_log}" ] || continue
+    sanitize_upgrade_log "${warp_log}" "${GROUP_DIR}/$(basename "${warp_log}")" || fail=1
+  done
+  read -r -a nodes <<<"$(topology_nodes "${CURRENT_TOPO}")"
+  for node in "${nodes[@]}"; do
+    # Only inspect the current suite's log directory, before its teardown.
+    if ! timeout 35 ssh "${SSH_OPTS[@]}" "${SSH_USER}@${node}" 'bash -s' >"${TEST_TMP}/diagnostic.log" 2>&1 <<'UPG_DIAGNOSTICS'
+set -euo pipefail
+SUDO=""; [ "$(id -u)" -eq 0 ] || SUDO="sudo -n"
+${SUDO} python3 - <<'UPG_PY'
+import json, pathlib, subprocess
+budget = 16 * 1024 * 1024
+logs = []
+for path in sorted(pathlib.Path('/var/log/rustfs').rglob('*')):
+    if not path.is_file() or path.is_symlink():
+        continue
+    size = path.stat().st_size
+    take = min(size, 4 * 1024 * 1024, budget)
+    with path.open('rb') as stream:
+        stream.seek(max(0, size - take))
+        content = stream.read(take).decode('utf-8', errors='replace')
+    logs.append({'path': str(path), 'truncated': take < size, 'content': content})
+    budget -= take
+    if budget <= 0:
+        break
+result = subprocess.run(['journalctl', '-u', 'rustfs', '-n', '1500', '--no-pager'],
+                        capture_output=True, text=True, timeout=15)
+print(json.dumps({'file_logs': logs, 'journal_exit': result.returncode,
+                  'journal': result.stdout, 'journal_error': result.stderr}, indent=2))
+raise SystemExit(0 if result.returncode == 0 else 1)
+UPG_PY
+UPG_DIAGNOSTICS
+    then fail=1; fi
+    sanitize_upgrade_log "${TEST_TMP}/diagnostic.log" "${GROUP_DIR}/diagnostics-${node}.log" || fail=1
+  done
+  return "${fail}"
+}
+
+sanitize_upgrade_log() {
+  UPGRADE_REDACT_VAULT_TOKEN="${VAULT_TOKEN}" UPGRADE_REDACT_MASTER_KEY="${KMS_LOCAL_MASTER_KEY}" \
+  UPGRADE_REDACT_ACCESS_KEY="${ACCESS_KEY}" UPGRADE_REDACT_SECRET_KEY="${SECRET_KEY}" \
+  UPGRADE_REDACT_BACKUP_KEK="${KMS_BACKUP_KEK}" \
+    "${UPGRADE_PYTHON}" "${SCRIPT_DIR}/scripts/upgrade_evidence.py" sanitize \
+    --input "$1" --output "$2" --golden "${GOLDEN_FILE}"
+}
+
+finish_upgrade_group() {
+  local fail=0 cleanup=RETAINED diagnostics=PASS
+  collect_upgrade_diagnostics || { fail=1; diagnostics=FAIL; }
+  if [ "${KEEP_ENV}" -eq 0 ]; then
+    cleanup=PASS
+    env_teardown "${CURRENT_TOPO}" || { fail=1; cleanup=FAIL; }
+    if [ "${KMS_BACKEND}" = vault-kv2 ]; then vault_stop || { fail=1; cleanup=FAIL; }; fi
+  fi
+  if [ "${CURRENT_RESULT}" = PASS ]; then
+    if [ "${cleanup}" = FAIL ]; then CURRENT_RESULT=CLEANUP_FAILED
+    elif [ "${diagnostics}" = FAIL ]; then CURRENT_RESULT=DIAGNOSTICS_FAILED; fi
+  fi
+  evidence_line "[UPG-CLEANUP] ${CURRENT_TOPO} ${KMS_BACKEND} ${cleanup}"
+  evidence_line "[UPG-DIAGNOSTICS] ${CURRENT_TOPO} ${KMS_BACKEND} ${diagnostics}"
+  evidence_line "[UPG-STATE] ${CURRENT_TOPO} ${KMS_BACKEND} ${OLD_VERSION_ACTUAL:-unknown} ${NEW_VERSION_ACTUAL:-unknown} ${CURRENT_RESULT}"
+  GROUP_ACTIVE=0
+  return "${fail}"
+}
+
+emit_unrun_cases() {
+  local i
+  for i in "${!UPGRADE_IDS[@]}"; do
+    if [[ "${CURRENT_DONE}" != *" ${UPGRADE_IDS[i]} "* ]]; then
+      if [[ "${CURRENT_STARTED}" != *" ${UPGRADE_IDS[i]} "* ]]; then
+        evidence_line "--- ${UPGRADE_IDS[i]} ${UPGRADE_NAMES[i]} ---"
+      fi
+      evidence_line "[NOT_RUN] ${UPGRADE_IDS[i]} ${UPGRADE_NAMES[i]}"
+    fi
+  done
+}
+
+upgrade_exit() {
+  local rc="$1" index group topo backend
+  trap - EXIT INT TERM
+  set +e
+  if [ "${ARTIFACTS_READY}" -eq 1 ]; then
+    if [ "${GROUP_ACTIVE}" -eq 1 ]; then
+      if [ "${rc}" -eq 130 ] || [ "${rc}" -eq 143 ]; then CURRENT_RESULT=INTERRUPTED; fi
+      [ -n "${CURRENT_RESULT}" ] || CURRENT_RESULT="${CURRENT_PHASE}_FAILED"
+      emit_unrun_cases
+      finish_upgrade_group || rc=1
+      GROUP_INDEX=$((GROUP_INDEX+1))
+    fi
+    for ((index=GROUP_INDEX; index<${#UPGRADE_GROUPS[@]}; index++)); do
+      group="${UPGRADE_GROUPS[index]}"; topo="${group%/*}"; backend="${group#*/}"
+      evidence_line "[INFO] == topology: ${topo}  kms-backend: ${backend} =="
+      CURRENT_DONE=" "
+      CURRENT_STARTED=" "
+      emit_unrun_cases
+      evidence_line "[UPG-STATE] ${topo} ${backend} unknown unknown NOT_RUN"
+    done
+    "${UPGRADE_PYTHON}" "${SCRIPT_DIR}/scripts/functional_case_report.py" \
+      "${ARTIFACTS_DIR}/events.log" "${ARTIFACTS_DIR}/cases.md" "${ARTIFACTS_DIR}/matrix.md" \
+      --json "${ARTIFACTS_DIR}/results.json" || rc=1
+  fi
+  rm -rf "${TEST_TMP}"
+  exit "${rc}"
 }
 
 # ==================== Admin API (SigV4, optional JSON body) ====================
@@ -834,12 +970,25 @@ seed_warp_bucket() {
   # that exact count as golden state; verification lists and compares counts.
   local bucket="$1" size="$2" duration="$3" keep="$4"
   new_bucket "${bucket}" || die "seed: ${bucket}"
-  log "seeding ${bucket} via warp (obj.size=${size}, duration=${duration}, keep=${keep})"
+  log "seeding ${bucket} (obj.size=${size}, keep=${keep})"
   if [ "${DRY_RUN}" -eq 1 ]; then
     log "DRY-RUN: warp put skipped"
     golden_set "warp:${bucket}" "0"
     return 0
   fi
+  # A time-driven 100MiB run filled nearly 18GiB just to retain 200MiB.
+  # Seed the two large objects directly; keep Warp-generated small objects.
+  if [ "${size}" = 100MiB ]; then
+    local i
+    make_test_file "${TEST_TMP}/large-seed.bin" 104857600
+    for ((i=1; i<=keep; i++)); do
+      aws_run s3api put-object --bucket "${bucket}" --key "seed-${i}" --body "${TEST_TMP}/large-seed.bin"
+      assert_aws_ok || die "seed: ${bucket}/seed-${i} put failed"
+    done
+    golden_set "warp:${bucket}" "${keep}"
+    log "seeded ${bucket}: ${keep} bounded objects of ${size}"
+    return 0
+  fi
   warp put --host "${ENDPOINT#http://}" \
     --bucket "${bucket}" \
     --access-key "${ACCESS_KEY}" \
@@ -860,13 +1009,30 @@ seed_warp_bucket() {
   if [ "${written}" -gt "${keep}" ]; then
     # Trim to <keep> objects: delete the extras in batches of 1000 (the S3
     # delete-objects limit). jq builds one compact JSON payload per batch.
+    # Check payload generation explicitly; process substitution hides jq failures.
+    printf '%s' "${keys_json}" \
+      | jq -c --argjson k "${keep}" -L "${JQ_LIB}" \
+        'include "jq_nwise"; .[0:(length - $k)] | nwise(1000)[] | {Objects: [.[] | {Key: .}], Quiet: true}' \
+      >"${TEST_TMP}/delete-payloads.jsonl" || die "seed: ${bucket}: delete payload generation failed"
     while IFS= read -r payload; do
       [ -n "${payload}" ] || continue
-      aws_run s3api delete-objects --bucket "${bucket}" --delete "${payload}"
-      assert_aws_ok || die "seed: trim ${bucket} (delete batch failed)"
-    done < <(printf '%s' "${keys_json}" \
-             | jq -c --argjson k "${keep}" -L "${JQ_LIB}" \
-               'include "jq_nwise"; .[0:(length - $k)] | nwise(1000)[] | {Objects: [.[] | {Key: .}], Quiet: true}')
+      DELETE_BATCH=$((DELETE_BATCH+1))
+      printf '%s' "${payload}" >"${TEST_TMP}/delete-request.json"
+      local delete_rc=0
+      AWS_MAX_ATTEMPTS=1 AWS_RETRY_MODE=standard aws --debug --endpoint-url "${ENDPOINT}" --region "${REGION}" \
+        --cli-connect-timeout 5 --cli-read-timeout 60 s3api delete-objects --bucket "${bucket}" \
+        --delete "file://${TEST_TMP}/delete-request.json" --output json \
+        >"${TEST_TMP}/delete-response.json" 2>"${TEST_TMP}/delete-debug.log" || delete_rc=$?
+      UPGRADE_REDACT_ACCESS_KEY="${ACCESS_KEY}" UPGRADE_REDACT_SECRET_KEY="${SECRET_KEY}" \
+      UPGRADE_REDACT_VAULT_TOKEN="${VAULT_TOKEN}" UPGRADE_REDACT_MASTER_KEY="${KMS_LOCAL_MASTER_KEY}" \
+      UPGRADE_REDACT_BACKUP_KEK="${KMS_BACKUP_KEK}" \
+      "${UPGRADE_PYTHON}" "${SCRIPT_DIR}/scripts/upgrade_evidence.py" validate-delete \
+        --request "${TEST_TMP}/delete-request.json" --response "${TEST_TMP}/delete-response.json" \
+        --debug "${TEST_TMP}/delete-debug.log" --exit-code "${delete_rc}" \
+        --output "${GROUP_DIR}/delete-${bucket}-${DELETE_BATCH}.json" \
+        || die "seed: trim ${bucket} (delete batch failed)"
+      rm -f "${TEST_TMP}/delete-debug.log"
+    done < "${TEST_TMP}/delete-payloads.jsonl"
     aws_run s3api list-objects-v2 --bucket "${bucket}" --query 'length(Contents)'
     assert_aws_ok || die "seed: post-trim list ${bucket}"
     remaining="$(printf '%s' "${AWS_OUT}" | tr -d '[:space:]')"
@@ -880,6 +1046,23 @@ seed_warp_bucket() {
   fi
 }
 
+capture_seed_manifest() {
+  [ "${DRY_RUN}" -eq 0 ] || return 0
+  local bucket="$1" encoded key
+  aws_run s3api list-objects-v2 --bucket "${bucket}" --query 'Contents[].Key' --output json
+  assert_aws_ok || die "seed: manifest list ${bucket}"
+  printf '%s' "${AWS_OUT}" >"${TEST_TMP}/seed-keys.json"
+  jq -r '.[] | @base64' "${TEST_TMP}/seed-keys.json" >"${TEST_TMP}/seed-keys.txt" || die "seed: invalid key list"
+  while IFS= read -r encoded; do
+    key="$(printf '%s' "${encoded}" | base64 -d)"
+    aws_run s3api get-object --bucket "${bucket}" --key "${key}" "${TEST_TMP}/seed-download.bin"
+    assert_aws_ok || die "seed: manifest read ${bucket}"
+    "${UPGRADE_PYTHON}" "${SCRIPT_DIR}/scripts/upgrade_evidence.py" seed-record \
+      --manifest "${GROUP_DIR}/seed-${bucket}.json" --key "${key}" --body "${TEST_TMP}/seed-download.bin" \
+      || die "seed: manifest ${bucket}"
+  done < "${TEST_TMP}/seed-keys.txt"
+}
+
 # ==================== Golden seed (on OLD version) ====================
 
 seed_golden() {
@@ -970,6 +1153,7 @@ seed_golden() {
   seed_warp_bucket "upg-warp-1kib"  1KiB   2s 5
   seed_warp_bucket "upg-warp-1mib"  1MiB   2s 5
   seed_warp_bucket "upg-warp-100mib" 100MiB 30s 2
+  for bucket in upg-warp-1kib upg-warp-1mib upg-warp-100mib; do capture_seed_manifest "${bucket}"; done
 
   # UPG-105 SSE-S3.
   make_test_file "${TEST_TMP}/sse3.bin" 65536
@@ -1037,19 +1221,25 @@ seed_golden() {
 # ==================== Verification cases (on NEW version) ====================
 
 run_test() {
-  local id="$1" name="$2" fn="$3" out
+  local id="$1" name="$2" fn="$3" case_failed=0
   if [ "${DRY_RUN}" -eq 1 ]; then
     printf '  [DRY-RUN] %s %s\n' "${id}" "${name}"
     return 0
   fi
-  printf '\n--- %s %s ---\n' "${id}" "${name}"
+  CURRENT_STARTED="${CURRENT_STARTED}${id} "
+  evidence_line "--- ${id} ${name} ---"
   if "${fn}" > "${TEST_TMP}/${id}.out" 2>&1; then
-    PASS=$((PASS+1)); printf '\033[1;32m[PASS]\033[0m %s %s\n' "${id}" "${name}"
+    PASS=$((PASS+1)); evidence_line "[PASS] ${id} ${name}"
   else
+    case_failed=1
     FAIL=$((FAIL+1)); FAILED_TESTS+=("${id} ${name}")
-    printf '\033[1;31m[FAIL]\033[0m %s %s\n' "${id}" "${name}"
-    cp "${TEST_TMP}/${id}.out" "${TEST_TMP}/${id}.out.detail" 2>/dev/null || true
-    sed 's/^/      | /' "${TEST_TMP}/${id}.out" | tail -50
+    evidence_line "[FAIL] ${id} ${name}"
+  fi
+  CURRENT_DONE="${CURRENT_DONE}${id} "
+  sanitize_upgrade_log "${TEST_TMP}/${id}.out" "${GROUP_DIR}/${id}.log" \
+    || { CURRENT_RESULT=DIAGNOSTICS_FAILED; die "case diagnostics failed: ${id}"; }
+  if [ "${case_failed}" -eq 1 ]; then
+    sed 's/^/      | /' "${GROUP_DIR}/${id}.log" | tail -50
   fi
 }
 
@@ -1216,12 +1406,16 @@ verify_warp_bucket() {
   # Sample-read every object: full-body GET + ETag comparison between HEAD
   # and GET. Keys come from one JSON listing (jq emits one per line, with a
   # trailing newline), so the iteration cannot drop the last key.
-  local keys_json key etag_head etag_get i=0
+  local keys_json encoded i=0
   aws_run s3api list-objects-v2 --bucket "${bucket}" --query 'Contents[].Key' --output json
   assert_aws_ok || { echo "${bucket}: key listing failed after upgrade"; return 1; }
   keys_json="${AWS_OUT}"
-  while IFS= read -r key; do
-    [ -n "${key}" ] || continue
+  printf '%s' "${keys_json}" >"${TEST_TMP}/verify-keys.json"
+  "${UPGRADE_PYTHON}" "${SCRIPT_DIR}/scripts/upgrade_evidence.py" check-seed \
+    --manifest "${GROUP_DIR}/seed-${bucket}.json" --keys-file "${TEST_TMP}/verify-keys.json" || return 1
+  jq -r '.[] | @base64' "${TEST_TMP}/verify-keys.json" >"${TEST_TMP}/verify-keys.txt" || return 1
+  while IFS= read -r encoded; do
+    key="$(printf '%s' "${encoded}" | base64 -d)" || return 1
     aws_run s3api head-object --bucket "${bucket}" --key "${key}"
     assert_aws_ok || { echo "${bucket}/${key}: HEAD failed after upgrade"; return 1; }
     etag_head="$(printf '%s' "${AWS_OUT}" | jq -r '.ETag')"
@@ -1230,8 +1424,10 @@ verify_warp_bucket() {
     etag_get="$(printf '%s' "${AWS_OUT}" | jq -r '.ETag')"
     [ "${etag_head}" = "${etag_get}" ] \
       || { echo "${bucket}/${key}: ETag mismatch HEAD=${etag_head} GET=${etag_get}"; return 1; }
+    "${UPGRADE_PYTHON}" "${SCRIPT_DIR}/scripts/upgrade_evidence.py" check-body \
+      --manifest "${GROUP_DIR}/seed-${bucket}.json" --key "${key}" --body "${TEST_TMP}/warp-dl.bin" || return 1
     i=$((i+1))
-  done < <(printf '%s' "${keys_json}" | jq -r '.[]')
+  done < "${TEST_TMP}/verify-keys.txt"
   log "${bucket}: ${i} objects readable after upgrade (count=${listed})"
 }
 
@@ -1244,6 +1440,15 @@ verify_warp_100mib() { verify_warp_bucket "upg-warp-100mib"; }
 
 PASS=0; FAIL=0
 FAILED_TESTS=()
+UPGRADE_IDS=(UPG-101 UPG-102 UPG-103 UPG-104 UPG-105 UPG-106 UPG-107 UPG-201 UPG-202 UPG-203 UPG-301 UPG-302 UPG-303 UPG-401 UPG-402 UPG-108 UPG-109 UPG-110)
+UPGRADE_NAMES=("plain object preserved" "versioned object preserved" "multipart object preserved" "large object preserved"
+  "SSE-S3 object preserved" "SSE-KMS object + key preserved" "encryption-context object" "IAM user + policy survive"
+  "service account survives" "STS credential behavior" "node config persisted" "KMS config + keys persisted"
+  "bucket default encryption" "post-upgrade S3 CRUD smoke" "post-upgrade SSE-KMS write"
+  "warp 1KiB objects readable" "warp 1MiB objects readable" "bounded 100MiB objects readable")
+UPGRADE_FUNCTIONS=(verify_plain verify_versioned verify_multipart verify_large verify_sse_s3 verify_sse_kms verify_context
+  verify_identity_user verify_identity_sa verify_identity_sts verify_config_persisted verify_kms_config
+  verify_bucket_default_encryption verify_smoke_crud verify_smoke_kms_write verify_warp_1kib verify_warp_1mib verify_warp_100mib)
 
 emit_topo_result() {
   # Machine-readable line for the workflow report generator:
@@ -1265,24 +1470,8 @@ assert_rc_ok() {
 }
 
 run_verify_cases() {
-  run_test UPG-101 "plain object preserved"          verify_plain
-  run_test UPG-102 "versioned object preserved"      verify_versioned
-  run_test UPG-103 "multipart object preserved"      verify_multipart
-  run_test UPG-104 "large object preserved"          verify_large
-  run_test UPG-105 "SSE-S3 object preserved"         verify_sse_s3
-  run_test UPG-106 "SSE-KMS object + key preserved"  verify_sse_kms
-  run_test UPG-107 "encryption-context object"       verify_context
-  run_test UPG-201 "IAM user + policy survive"       verify_identity_user
-  run_test UPG-202 "service account survives"        verify_identity_sa
-  run_test UPG-203 "STS credential behavior"         verify_identity_sts
-  run_test UPG-301 "node config persisted"           verify_config_persisted
-  run_test UPG-302 "KMS config + keys persisted"     verify_kms_config
-  run_test UPG-303 "bucket default encryption"       verify_bucket_default_encryption
-  run_test UPG-401 "post-upgrade S3 CRUD smoke"      verify_smoke_crud
-  run_test UPG-402 "post-upgrade SSE-KMS write"      verify_smoke_kms_write
-  run_test UPG-108 "warp 1KiB objects readable"      verify_warp_1kib
-  run_test UPG-109 "warp 1MiB objects readable"      verify_warp_1mib
-  run_test UPG-110 "warp 100MiB objects readable"    verify_warp_100mib
+  local i
+  for i in "${!UPGRADE_IDS[@]}"; do run_test "${UPGRADE_IDS[i]}" "${UPGRADE_NAMES[i]}" "${UPGRADE_FUNCTIONS[i]}"; done
 }
 
 results_summary() {
@@ -1333,6 +1522,7 @@ Usage:
   --keep-env                Do not tear down the environment after the last run
   --ssh-user <user>         SSH user for the test nodes (default: azureuser)
   --log-file <path>         Tee output to a log file
+  --artifacts-dir <path>    New directory for seed/delete, phase and cleanup evidence
   --dry-run                 Preview commands without executing them
   -y, --yes                 Skip confirmation prompts
   -h, --help                Show this help
@@ -1357,6 +1547,7 @@ main() {
       --keep-env) KEEP_ENV=1 ;;
       --ssh-user) SSH_USER="$1"; shift ;;
       --log-file) LOG_FILE="$1"; shift ;;
+      --artifacts-dir) ARTIFACTS_DIR="$1"; shift ;;
       --dry-run) DRY_RUN=1 ;;
       -y|--yes) ASSUME_YES=1 ;;
       -h|--help) usage; exit 0 ;;
@@ -1390,32 +1581,74 @@ main() {
   need_cmd jq "JSON processor"
   need_cmd curl "HTTP client"
   need_cmd openssl "OpenSSL (SigV4)"
+  need_cmd "${UPGRADE_PYTHON}" "Python for upgrade evidence"
+  need_cmd timeout "bounded diagnostics"
   log "nodes: ${NODES[*]}  from: $(resolve_from_url)  to: $(resolve_to_url)"
   log "backends: ${BACKENDS[*]}  topologies: ${SELECTED_TOPOLOGIES[*]}"
 
   confirm "Run upgrade compatibility suite (${SELECTED_TOPOLOGIES[*]}, ${BACKENDS[*]})?"
 
+  local group_topo group_backend
+  for group_backend in "${BACKENDS[@]}"; do
+    case "${group_backend}" in local|vault-kv2) ;; *) die "unknown KMS backend: ${group_backend}" ;; esac
+    for group_topo in "${SELECTED_TOPOLOGIES[@]}"; do
+      case "${group_topo}" in single-single|single-multi|multi-multi) ;; *) die "unknown topology: ${group_topo}" ;; esac
+      if [[ " ${UPGRADE_GROUPS[*]} " = *" ${group_topo}/${group_backend} "* ]]; then
+        die "duplicate upgrade group: ${group_topo}/${group_backend}"
+      fi
+      UPGRADE_GROUPS+=("${group_topo}/${group_backend}")
+    done
+  done
+  if [ "${DRY_RUN}" -eq 1 ]; then
+    local preview_group
+    for preview_group in "${UPGRADE_GROUPS[@]}"; do
+      log "DRY-RUN: ${preview_group}: prepare old -> seed -> upgrade -> verify -> diagnostics -> cleanup"
+      run_verify_cases
+    done
+    log "DRY-RUN: no environment or S3 commands executed"
+    return 0
+  fi
+  if [ "${DRY_RUN}" -eq 0 ]; then
+    ARTIFACTS_DIR="${ARTIFACTS_DIR:-${LOG_FILE:-${PWD}/rustfs-upgrade-$(date -u +%Y%m%d-%H%M%S)}.artifacts}"
+    umask 077
+    mkdir -p "$(dirname "${ARTIFACTS_DIR}")"
+    mkdir "${ARTIFACTS_DIR}" || die "artifacts directory must be new: ${ARTIFACTS_DIR}"
+    : >"${ARTIFACTS_DIR}/events.log"
+    ARTIFACTS_READY=1
+  fi
+
   local overall_fail=0 topo backend
   for backend in "${BACKENDS[@]}"; do
     KMS_BACKEND="${backend}"
     for topo in "${SELECTED_TOPOLOGIES[@]}"; do
-      log "== topology: ${topo}  kms-backend: ${KMS_BACKEND} =="
+      CURRENT_TOPO="${topo}"; CURRENT_PHASE=PREPARE; CURRENT_RESULT=""
+      CURRENT_DONE=" "; CURRENT_STARTED=" "; DIAGNOSTICS_DONE=0
+      OLD_VERSION_ACTUAL=unknown; NEW_VERSION_ACTUAL=unknown
+      rm -f "${TEST_TMP}"/warp-put-*.log
+      GROUP_DIR="${ARTIFACTS_DIR}/${topo}-${backend}"
+      if [ "${ARTIFACTS_READY}" -eq 1 ]; then mkdir "${GROUP_DIR}"; GROUP_ACTIVE=1; fi
+      evidence_line "[INFO] == topology: ${topo}  kms-backend: ${KMS_BACKEND} =="
       if [ "${KMS_BACKEND}" = "vault-kv2" ]; then
         vault_stop
         vault_start
       fi
       env_setup_old "${topo}"
       seed_golden
+      CURRENT_PHASE=UPGRADE
       upgrade_in_place "${topo}"
+      CURRENT_PHASE=VERIFY
       PASS=0; FAIL=0; FAILED_TESTS=()
       run_verify_cases
       results_summary
       record_topology_result "${topo}/${KMS_BACKEND}"
       emit_topo_result "${topo}" "${KMS_BACKEND}"
-      if [ "${KEEP_ENV}" -eq 0 ]; then
-        env_teardown "${topo}"
-        if [ "${KMS_BACKEND}" = "vault-kv2" ]; then vault_stop; fi
-      fi
+      CURRENT_RESULT=PASS
+      if [ "${FAIL}" -gt 0 ]; then CURRENT_RESULT=VERIFY_FAILED; fi
+      CURRENT_PHASE=CLEANUP
+      local cleanup_failed=0
+      finish_upgrade_group || cleanup_failed=1
+      GROUP_INDEX=$((GROUP_INDEX+1))
+      [ "${cleanup_failed}" -eq 0 ] || die "upgrade diagnostics or cleanup failed"
       if [ "${FAIL}" -gt 0 ]; then overall_fail=1; fi
     done
   done
diff --git a/scripts/functional_case_report.py b/scripts/functional_case_report.py
index 7036eaad..89987f88 100644
--- a/scripts/functional_case_report.py
+++ b/scripts/functional_case_report.py
@@ -4,19 +4,25 @@
 from __future__ import annotations
 
 import argparse
+import json
 from pathlib import Path
 import re
 
 
-def generate_report(log_file: Path, case_file: Path, matrix_file: Path | None = None) -> bool:
+def generate_report(log_file: Path, case_file: Path, matrix_file: Path | None = None,
+                    json_file: Path | None = None) -> bool:
     ansi = re.compile(r"\x1b\[[0-9;]*m")
     start_re = re.compile(r"^---\s+([A-Z][A-Z0-9]*-[0-9]+)\s+(.+?)\s+---$")
-    done_re = re.compile(r"^\[(PASS|FAIL|UNSUPPORTED)\]\s+([A-Z][A-Z0-9]*-[0-9]+)\b")
+    done_re = re.compile(r"^\[(PASS|FAIL|UNSUPPORTED|NOT_RUN)\]\s+([A-Z][A-Z0-9]*-[0-9]+)\b")
     context_re = re.compile(r"^(?:\[INFO\]\s+)?==\s+((?:topology|suite):.+?)\s+==$")
     topo_re = re.compile(r"^\[UPG-TOPO\]\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+PASS=(\d+)\s+FAIL=(\d+)\s*$")
+    state_re = re.compile(r"^\[UPG-STATE\]\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s*$")
+    detail_re = re.compile(r"^\[UPG-(CLEANUP|DIAGNOSTICS)\]\s+(\S+)\s+(\S+)\s+(\S+)\s*$")
     rows = []
     pending = {}
     topo_rows = []
+    groups = []
+    details = {}
     context = "context not recorded"
     complete = True
     try:
@@ -28,6 +34,14 @@ def generate_report(log_file: Path, case_file: Path, matrix_file: Path | None =
                     pending.clear()
                 elif match := topo_re.match(line):
                     topo_rows.append(match.groups())
+                elif match := detail_re.match(line):
+                    kind, topo, backend, result = match.groups()
+                    details.setdefault((topo, backend), {})[kind.lower()] = result
+                elif match := state_re.match(line):
+                    topo, backend, old_v, new_v, result = match.groups()
+                    groups.append(dict(topology=topo, backend=backend, from_version=old_v,
+                                       to_version=new_v, result=result,
+                                       **details.pop((topo, backend), {})))
                 elif match := start_re.match(line):
                     case_id, name = match.groups()
                     pending[case_id] = len(rows)
@@ -43,7 +57,8 @@ def generate_report(log_file: Path, case_file: Path, matrix_file: Path | None =
     except FileNotFoundError:
         pass
 
-    counts = {status: sum(row[2] == status for row in rows) for status in ("PASS", "FAIL", "UNSUPPORTED", "RUNNING")}
+    counts = {status: sum(row[2] == status for row in rows)
+              for status in ("PASS", "FAIL", "UNSUPPORTED", "RUNNING", "NOT_RUN")}
     with case_file.open("w", encoding="utf-8") as out:
         out.write(f"## Case Summary\n\n- Total: {len(rows)}\n")
         for status, count in counts.items():
@@ -54,17 +69,32 @@ def generate_report(log_file: Path, case_file: Path, matrix_file: Path | None =
         if not rows:
             out.write("\nNo case execution was recorded; the log is missing, empty, or stopped before the cases.\n")
 
-    valid = complete and bool(rows) and not counts["FAIL"] and not counts["RUNNING"]
+    valid = complete and bool(rows) and not any(counts[status] for status in ("FAIL", "RUNNING", "NOT_RUN"))
+    if groups:
+        valid = valid and all(group["result"] == "PASS" and group.get("cleanup") in ("PASS", "RETAINED")
+                             and group.get("diagnostics") == "PASS" for group in groups)
     if matrix_file is not None:
         with matrix_file.open("w", encoding="utf-8") as out:
-            out.write("## Upgrade Matrix\n\n| Topology | KMS Backend | From Version | To Version | Result |\n")
-            out.write("| --- | --- | --- | --- | --- |\n")
-            for topo, backend, old_v, new_v, npass, nfail in topo_rows:
-                result = "PASS" if nfail == "0" else "FAIL"
-                out.write(f"| {topo} | {backend} | {old_v} | {new_v} | {result} (PASS={npass} FAIL={nfail}) |\n")
-            if not topo_rows:
+            out.write("## Upgrade Matrix\n\n| Topology | KMS Backend | From Version | To Version | Result | Cleanup | Diagnostics |\n")
+            out.write("| --- | --- | --- | --- | --- | --- | --- |\n")
+            if groups:
+                for group in groups:
+                    out.write("| " + " | ".join(group.get(field, "NOT_RUN") for field in
+                              ("topology", "backend", "from_version", "to_version", "result", "cleanup", "diagnostics")) + " |\n")
+                out.write("\nPREPARE_FAILED: 旧版环境或种子准备失败；UPGRADE_FAILED: 升级阶段失败；"
+                          "VERIFY_FAILED: 升级后断言失败；NOT_RUN: 未执行；INTERRUPTED: 被中断。"
+                          "DIAGNOSTICS_FAILED / CLEANUP_FAILED: 证据归档或环境恢复失败。\n")
+            else:
+                for topo, backend, old_v, new_v, npass, nfail in topo_rows:
+                    result = "PASS" if nfail == "0" else "FAIL"
+                    out.write(f"| {topo} | {backend} | {old_v} | {new_v} | {result} (PASS={npass} FAIL={nfail}) | unknown | unknown |\n")
+            if not groups and not topo_rows:
                 out.write("| - | - | - | - | NOT RUN (suite failed before upgrade) |\n")
-        valid = valid and bool(topo_rows) and all(row[-1] == "0" for row in topo_rows)
+        valid = valid and (bool(groups) or (bool(topo_rows) and all(row[-1] == "0" for row in topo_rows)))
+    if json_file is not None:
+        json_file.write_text(json.dumps(dict(schema=1, status="PASS" if valid else "FAIL", counts=counts,
+                             cases=[dict(id=cid, name=name, status=status) for cid, name, status in rows],
+                             groups=groups), ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
     return valid
 
 
@@ -73,5 +103,6 @@ def generate_report(log_file: Path, case_file: Path, matrix_file: Path | None =
     parser.add_argument("log_file", type=Path)
     parser.add_argument("case_file", type=Path)
     parser.add_argument("matrix_file", type=Path, nargs="?")
+    parser.add_argument("--json", dest="json_file", type=Path)
     args = parser.parse_args()
-    raise SystemExit(0 if generate_report(args.log_file, args.case_file, args.matrix_file) else 1)
+    raise SystemExit(0 if generate_report(args.log_file, args.case_file, args.matrix_file, args.json_file) else 1)
diff --git a/scripts/issue_manager.py b/scripts/issue_manager.py
index dba0ef4d..77bb9949 100644
--- a/scripts/issue_manager.py
+++ b/scripts/issue_manager.py
@@ -29,9 +29,9 @@
 CASE_ID_RE = re.compile(r"^[A-Z][A-Z0-9]*-[0-9]+$")
 # Case rows in cases.md as produced by scripts/functional_case_report.py:
 # | S3C-101 | CreateBucket (topology:single-single) | PASS |
-TABLE_ROW_RE = re.compile(r"^\|\s*([A-Z][A-Z0-9]*-[0-9]+)\s*\|(.+)\|\s*(PASS|FAIL|UNSUPPORTED|RUNNING|SKIP)\s*\|")
+TABLE_ROW_RE = re.compile(r"^\|\s*([A-Z][A-Z0-9]*-[0-9]+)\s*\|(.+)\|\s*(PASS|FAIL|UNSUPPORTED|RUNNING|NOT_RUN|SKIP)\s*\|")
 # Suite-log verdict lines: [FAIL] STOR-115 InlineEncryption / [PASS] KMS-317 ...
-LOG_VERDICT_RE = re.compile(r"^\[(PASS|FAIL|UNSUPPORTED)\]\s+([A-Z][A-Z0-9]*-[0-9]+)\b")
+LOG_VERDICT_RE = re.compile(r"^\[(PASS|FAIL|UNSUPPORTED|NOT_RUN)\]\s+([A-Z][A-Z0-9]*-[0-9]+)\b")
 # Fault-tolerance suite verdicts: FT-CASE: <case> verdict=pass|known-divergence|UNEXPECTED
 FT_VERDICT_RE = re.compile(r"^FT-CASE:\s+(\S+)\s+verdict=(\S+)")
 # Heal suite step lines carry no case IDs: [HEAL-STEP] 5 start cluster heal PASS
@@ -167,6 +167,8 @@ def classify(outcome: str, cases: dict, log: Path | None) -> tuple[str, str]:
         return "env", "no case results were recorded"
     fail = [c for c, v in cases.items() if "FAIL" in v["statuses"]]
     pas = [c for c, v in cases.items() if "PASS" in v["statuses"] and "FAIL" not in v["statuses"]]
+    if not fail and any("NOT_RUN" in value["statuses"] for value in cases.values()):
+        return "env", "upgrade matrix incomplete; cases were not executed"
     if not pas and len(fail) >= 3:
         return "env", f"wholesale failure (0 passed, {len(fail)} failed)"
     if log and log.exists():
diff --git a/scripts/upgrade_evidence.py b/scripts/upgrade_evidence.py
new file mode 100644
index 00000000..20985d3c
--- /dev/null
+++ b/scripts/upgrade_evidence.py
@@ -0,0 +1,185 @@
+#!/usr/bin/env python3
+"""Retain safe upgrade diagnostics and validate seed/delete evidence."""
+
+from __future__ import annotations
+
+import argparse
+import ast
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import xml.etree.ElementTree as ET
+
+
+def redact(text: str, golden: Path | None = None) -> str:
+    values = [value for key, value in os.environ.items()
+              if re.search(r"SECRET|TOKEN|PASSWORD|MASTER_KEY|ACCESS_KEY|BACKUP_KEK", key) and value]
+    if golden and golden.exists():
+        for line in golden.read_text().splitlines():
+            key, _, value = line.partition("\t")
+            if key.startswith("creds:"):
+                values.extend(value.split("|"))
+    for value in sorted(set(values), key=len, reverse=True):
+        if value:
+            text = text.replace(value, "[REDACTED]")
+    return re.sub(r"(?im)^.*(?:authorization|secret[ _-]?(?:access[ _-]?)?key|"
+                  r"session[ _-]?token|password|client[ _-]?secret|vault[ _-]?token|"
+                  r"master[ _-]?key)\s*[=:].*$", "[REDACTED SENSITIVE LINE]", text)
+
+
+def write_json(path: Path, value) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n")
+
+
+def safe_value(value, golden=None):
+    if isinstance(value, dict):
+        return {redact(key, golden): ("[REDACTED]" if re.search(r"secret|token|password|authorization|master.?key|access.?key", key, re.I)
+                      else safe_value(item, golden)) for key, item in value.items()}
+    if isinstance(value, list):
+        return [safe_value(item, golden) for item in value]
+    return redact(value, golden) if isinstance(value, str) else value
+
+
+def validate_delete(request: Path, response: Path, debug: Path, exit_code: int, output: Path) -> bool:
+    metadata = {"http_status": None, "request_id": None}
+    # Debug contains signed requests. Only retain these two response fields.
+    if debug.exists():
+        with debug.open(errors="replace") as stream:
+            for line in stream:
+                if match := re.search(r'"POST .*? HTTP/1\.1" (\d{3})\b', line):
+                    metadata["http_status"] = int(match[1])
+                if "Response headers:" in line:
+                    try:
+                        headers = ast.literal_eval(line.split("Response headers:", 1)[1].strip())
+                        metadata["request_id"] = next((value for key, value in headers.items()
+                                                       if key.lower() == "x-amz-request-id"), None)
+                        if isinstance(metadata["request_id"], bytes):
+                            metadata["request_id"] = metadata["request_id"].decode(errors="replace")
+                    except (ValueError, SyntaxError, AttributeError):
+                        pass
+    receipt = {"schema": 1, "cli_exit": exit_code, **metadata, "valid": False}
+    if exit_code and debug.exists():
+        # The top-level S3 error is only in debug/stderr when the CLI exits
+        # nonzero. Retain its allowlisted XML fields, never signed requests.
+        lines = debug.read_text(errors="replace").splitlines()
+        for index, line in enumerate(lines):
+            if "Response body:" not in line:
+                continue
+            try:
+                raw_body = line.split("Response body:", 1)[1].strip()
+                if not raw_body and index + 1 < len(lines):
+                    raw_body = lines[index + 1].strip()
+                body = ast.literal_eval(raw_body)
+                root = ET.fromstring(body)
+                if root.tag.rsplit("}", 1)[-1] == "Error":
+                    receipt["service_error"] = {child.tag.rsplit("}", 1)[-1]: child.text
+                                                for child in root if child.tag.rsplit("}", 1)[-1]
+                                                in ("Code", "Message", "RequestId")}
+            except (ValueError, SyntaxError, TypeError, ET.ParseError):
+                pass
+    try:
+        payload = json.loads(request.read_text())
+        raw_response = response.read_text()
+        # AWS CLI emits no stdout for a successful Quiet DeleteResult.
+        result = json.loads(raw_response) if raw_response.strip() else {}
+        receipt.update(request=payload, response=result)
+        keys = [item["Key"] for item in payload["Objects"]]
+        if not keys or len(keys) > 1000 or any(not isinstance(key, str) for key in keys):
+            raise ValueError("invalid delete request")
+        if not isinstance(result, dict) or not isinstance(result.get("Errors", []), list):
+            raise ValueError("invalid delete response")
+        errors = result.get("Errors", [])
+        receipt["error_count"] = len(errors)
+        receipt["valid"] = exit_code == 0 and not errors and metadata["http_status"] in (None, 200)
+        receipt["reason"] = ("CLI failure" if exit_code else "per-object deletion failure" if errors
+                             else "unexpected HTTP status" if not receipt["valid"] else "PASS")
+    except (OSError, ValueError, KeyError, TypeError) as error:
+        receipt["reason"] = str(error)
+    write_json(output, safe_value(receipt))
+    if not receipt["valid"]:
+        print(f"delete validation failed: {redact(receipt['reason'])}; see {output}")
+    return receipt["valid"]
+
+
+def body_identity(body: Path):
+    digest = hashlib.sha256()
+    size = 0
+    with body.open("rb") as stream:
+        while chunk := stream.read(1024 * 1024):
+            size += len(chunk)
+            digest.update(chunk)
+    return {"bytes": size, "sha256": digest.hexdigest()}
+
+
+def seed_record(manifest: Path, key: str, body: Path):
+    rows = json.loads(manifest.read_text()) if manifest.exists() else []
+    if any(row["key"] == key for row in rows):
+        raise ValueError("duplicate seed key")
+    rows.append({"key": key, **body_identity(body)})
+    write_json(manifest, rows)
+
+
+def check_seed(manifest: Path, keys_file: Path):
+    rows, keys = json.loads(manifest.read_text()), json.loads(keys_file.read_text())
+    expected = [row["key"] for row in rows]
+    if not isinstance(keys, list) or len(keys) != len(expected) or set(keys) != set(expected):
+        raise ValueError("seed key inventory changed")
+
+
+def check_body(manifest: Path, key: str, body: Path):
+    rows = json.loads(manifest.read_text())
+    expected = next((row for row in rows if row["key"] == key), None)
+    if expected is None or body_identity(body) != {name: expected[name] for name in ("bytes", "sha256")}:
+        raise ValueError("seed content changed")
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    commands = parser.add_subparsers(dest="command", required=True)
+    delete = commands.add_parser("validate-delete")
+    for name in ("request", "response", "debug", "output"):
+        delete.add_argument("--" + name, type=Path, required=True)
+    delete.add_argument("--exit-code", type=int, required=True)
+    sanitize = commands.add_parser("sanitize")
+    sanitize.add_argument("--input", type=Path, required=True)
+    sanitize.add_argument("--output", type=Path, required=True)
+    sanitize.add_argument("--golden", type=Path)
+    record = commands.add_parser("seed-record")
+    record.add_argument("--manifest", type=Path, required=True)
+    record.add_argument("--key", required=True)
+    record.add_argument("--body", type=Path, required=True)
+    inventory = commands.add_parser("check-seed")
+    inventory.add_argument("--manifest", type=Path, required=True)
+    inventory.add_argument("--keys-file", type=Path, required=True)
+    body = commands.add_parser("check-body")
+    body.add_argument("--manifest", type=Path, required=True)
+    body.add_argument("--key", required=True)
+    body.add_argument("--body", type=Path, required=True)
+    args = parser.parse_args()
+    try:
+        if args.command == "validate-delete":
+            return 0 if validate_delete(args.request, args.response, args.debug, args.exit_code, args.output) else 1
+        if args.command == "sanitize":
+            raw = args.input.read_text(errors="replace")
+            try:
+                raw = json.dumps(safe_value(json.loads(raw), args.golden), ensure_ascii=False, indent=2)
+            except ValueError:
+                raw = redact(raw, args.golden)
+            args.output.write_text(raw)
+        elif args.command == "seed-record":
+            seed_record(args.manifest, args.key, args.body)
+        elif args.command == "check-seed":
+            check_seed(args.manifest, args.keys_file)
+        elif args.command == "check-body":
+            check_body(args.manifest, args.key, args.body)
+    except (OSError, ValueError, KeyError, TypeError) as error:
+        print(f"upgrade evidence failed: {error}")
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/test_rustfs_upgrade.py b/test_rustfs_upgrade.py
new file mode 100644
index 00000000..064223da
--- /dev/null
+++ b/test_rustfs_upgrade.py
@@ -0,0 +1,426 @@
+"""Upgrade harness regression tests; no SSH, Docker or RustFS is contacted."""
+
+import json
+import os
+from pathlib import Path
+import re
+import shlex
+import shutil
+import subprocess
+import sys
+import tempfile
+import threading
+import unittest
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from unittest import mock
+import xml.etree.ElementTree as ET
+from xml.sax.saxutils import escape
+
+sys.path.insert(0, str(Path(__file__).resolve().parent / "scripts"))
+from scripts import functional_case_report as report
+from scripts import functional_chain_evidence as chain
+from scripts import upgrade_evidence as evidence
+from scripts import issue_manager as issues
+
+
+ROOT = Path(__file__).resolve().parent
+SOURCE = (ROOT / "rustfs-upgrade-test.sh").read_text()
+BASH = shutil.which("bash")
+CANARY = "UPGRADE_DIAGNOSTIC_SECRET_CANARY"
+
+
+def function(name):
+    found = re.search(rf"(?ms)^{name}\(\) \{{.*?^\}}", SOURCE)
+    if not found:
+        raise AssertionError(f"real function {name} missing")
+    return found.group(0)
+
+
+class UpgradeTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.directory = Path(self.temp.name)
+
+    def bash(self, body, **env):
+        script = self.directory / "test.sh"
+        script.write_text(body)
+        child_env = {**os.environ, "UPGRADE_TEST_TOKEN": CANARY, **env}
+        for name in ("AWS_PROFILE", "AWS_DEFAULT_PROFILE", "AWS_SESSION_TOKEN", "AWS_SECURITY_TOKEN"):
+            child_env.pop(name, None)
+        return subprocess.run([BASH, str(script)], capture_output=True, text=True, timeout=45, env=child_env)
+
+    def receipt(self, response, code=0, http=200, debug_extra=""):
+        request = self.directory / "request.json"
+        response_file = self.directory / "response.json"
+        debug = self.directory / "debug.log"
+        output = self.directory / "receipt.json"
+        request.write_text(json.dumps({"Objects": [{"Key": "key"}], "Quiet": True}))
+        response_file.write_text(response if isinstance(response, str) else json.dumps(response))
+        debug.write_text(f'"POST /bucket?delete HTTP/1.1" {http} 0\nResponse headers: '
+                         "{'x-amz-request-id': b'fixture-id'}\n" + debug_extra)
+        with mock.patch.dict(os.environ, {"UPGRADE_TEST_TOKEN": CANARY}):
+            valid = evidence.validate_delete(request, response_file, debug, code, output)
+        self.assertNotIn(CANARY, output.read_text())
+        return valid, json.loads(output.read_text())
+
+    def test_delete_receipts_reject_failures_and_preserve_error_identity(self):
+        for response in ("", {}, {"Deleted": [{"Key": "key"}]}):
+            self.assertTrue(self.receipt(response)[0])
+        for response, code, http in (({"Errors": [{"Key": "key", "Code": "ServiceUnavailable", "Message": CANARY}]}, 0, 200),
+                                     ("", 255, 503), ("not json", 0, 200), ([], 0, 200), ({"Errors": {}}, 0, 200), ({}, 0, 503)):
+            with self.subTest(response=response, code=code):
+                valid, receipt = self.receipt(response, code, http)
+                self.assertFalse(valid)
+                self.assertEqual(receipt["http_status"], http)
+                self.assertEqual(receipt["request_id"], "fixture-id")
+        valid, receipt = self.receipt("", 255, 503, "Response body: b'<Error><Code>ServiceUnavailable</Code><Message>" + CANARY + "</Message><RequestId>fixture-id</RequestId></Error>'\n")
+        self.assertFalse(valid)
+        self.assertEqual(receipt["service_error"]["Code"], "ServiceUnavailable")
+
+    def test_manifest_detects_replaced_keys_and_corrupt_contents(self):
+        body, manifest, keys = (self.directory / name for name in ("body", "manifest.json", "keys.json"))
+        body.write_bytes(b"baseline body")
+        evidence.seed_record(manifest, "original", body)
+        keys.write_text('["original"]')
+        evidence.check_seed(manifest, keys)
+        evidence.check_body(manifest, "original", body)
+        keys.write_text('["replacement"]')
+        with self.assertRaisesRegex(ValueError, "inventory"):
+            evidence.check_seed(manifest, keys)
+        body.write_bytes(b"corrupt! body")
+        with self.assertRaisesRegex(ValueError, "content"):
+            evidence.check_body(manifest, "original", body)
+        with self.assertRaisesRegex(ValueError, "duplicate"):
+            evidence.seed_record(manifest, "original", body)
+
+    def test_large_seed_is_bounded_and_does_not_call_warp_or_delete(self):
+        body = "set -Eeuo pipefail\n" + function("seed_warp_bucket") + r'''
+log() { :; }
+die() { echo "$*" >&2; exit 1; }
+new_bucket() { :; }
+warp() { echo 'unexpected warp' >&2; exit 90; }
+make_test_file() { printf '%s\n' "$2" >> "$TEST_RECORD"; printf fixture > "$1"; }
+aws_run() { printf '%s\n' "$*" >> "$TEST_RECORD"; }
+assert_aws_ok() { :; }
+golden_set() { printf '%s %s\n' "$1" "$2" >> "$TEST_RECORD"; }
+DRY_RUN=0
+TEST_TMP="$TEST_DIRECTORY"
+seed_warp_bucket upg-warp-100mib 100MiB 30s 2
+'''
+        record = self.directory / "calls"
+        result = self.bash(body, TEST_RECORD=str(record), TEST_DIRECTORY=str(self.directory))
+        self.assertEqual(result.returncode, 0, result.stderr)
+        calls = record.read_text()
+        self.assertEqual(calls.count("s3api put-object"), 2)
+        self.assertIn("104857600", calls)
+        self.assertIn("warp:upg-warp-100mib 2", calls)
+        self.assertNotIn("delete-objects", calls)
+
+    def lifecycle(self, mode="pass", extra=()):
+        artifacts = self.directory / "artifacts"
+        # Source all real lifecycle/report functions; every external boundary
+        # is stubbed to forbid accidental access to the shared environment.
+        source = SOURCE.removesuffix('main "$@"\n')
+        stubs = r'''
+SCRIPT_DIR="$TEST_ROOT"
+UPGRADE_PYTHON="$TEST_PYTHON"
+ssh() { echo 'unexpected SSH' >&2; exit 90; }
+docker() { echo 'unexpected Docker' >&2; exit 90; }
+rc() { echo 'unexpected rc' >&2; exit 90; }
+aws() { echo 'unexpected aws' >&2; exit 90; }
+warp() { echo 'unexpected warp' >&2; exit 90; }
+need_cmd() { :; }
+vault_stop() { printf 'vault-stop\n' >> "$TEST_TRACE"; }
+vault_start() { :; }
+env_setup_old() { OLD_VERSION_ACTUAL=old; }
+seed_golden() {
+  if [ "$TEST_MODE" = prepare ] && [ "$KMS_BACKEND" = vault-kv2 ]; then die 'fixture seed failure'; fi
+  if [ "$TEST_MODE" = interrupted ]; then kill -TERM "$$"; fi
+}
+upgrade_in_place() {
+  [ "$TEST_MODE" != upgrade ] || die 'fixture upgrade failure'
+  NEW_VERSION_ACTUAL=new
+}
+collect_upgrade_diagnostics() {
+  printf 'diagnostics %s/%s\n' "$CURRENT_TOPO" "$KMS_BACKEND" >> "$TEST_TRACE"
+  printf 'private value %s\n' "$UPGRADE_TEST_TOKEN" > "$TEST_TMP/fixture.log"
+  sanitize_upgrade_log "$TEST_TMP/fixture.log" "$GROUP_DIR/diagnostics.log"
+  [ "$TEST_MODE" != diagnostics ]
+}
+cleanup_node() {
+  printf 'cleanup %s\n' "$1" >> "$TEST_TRACE"
+  [ "$TEST_MODE" != cleanup ]
+}
+fixture_verify() {
+  printf 'case value %s\n' "$UPGRADE_TEST_TOKEN"
+  [ "$TEST_MODE" != verify ] || [ "$CURRENT_CASE" != UPG-101 ]
+}
+run_verify_cases() {
+  local i
+  for i in "${!UPGRADE_IDS[@]}"; do
+    CURRENT_CASE="${UPGRADE_IDS[i]}"
+    run_test "${UPGRADE_IDS[i]}" "${UPGRADE_NAMES[i]}" fixture_verify
+  done
+}
+main --all-topologies -y --artifacts-dir "$TEST_ARTIFACTS" "$@"
+'''
+        script = self.directory / "lifecycle.sh"
+        script.write_text(source + stubs)
+        trace = self.directory / "trace"
+        result = subprocess.run([BASH, str(script), *extra], capture_output=True, text=True, timeout=30,
+                                env={**os.environ, "TEST_ROOT": str(ROOT), "TEST_PYTHON": sys.executable,
+                                     "TEST_MODE": mode, "TEST_ARTIFACTS": str(artifacts), "TEST_TRACE": str(trace),
+                                     "UPGRADE_TEST_TOKEN": CANARY, "TMPDIR": str(self.directory)})
+        self.assertNotIn(CANARY, result.stdout + result.stderr)
+        self.assertFalse(list(self.directory.glob("rustfs-upgrade.*")), "private scratch must be removed")
+        for path in artifacts.rglob("*"):
+            if path.is_file():
+                self.assertNotIn(CANARY, path.read_text())
+        results = json.loads((artifacts / "results.json").read_text()) if (artifacts / "results.json").exists() else None
+        return result, results, trace.read_text() if trace.exists() else ""
+
+    def test_complete_matrix_and_preparation_failure(self):
+        result, data, trace = self.lifecycle("pass")
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(data["status"], "PASS")
+        self.assertEqual(data["counts"]["PASS"], 108)
+        self.assertEqual(len(data["groups"]), 6)
+        self.assertLess(trace.index("diagnostics"), trace.index("cleanup"))
+        shutil.rmtree(self.directory / "artifacts")
+        result, data, trace = self.lifecycle("prepare")
+        self.assertNotEqual(result.returncode, 0)
+        self.assertEqual(data["counts"]["PASS"], 54)
+        self.assertEqual(data["counts"]["NOT_RUN"], 54)
+        self.assertEqual(data["counts"]["FAIL"], 0)
+        self.assertEqual([g["result"] for g in data["groups"]], ["PASS"] * 3 + ["PREPARE_FAILED", "NOT_RUN", "NOT_RUN"])
+        self.assertEqual(len(data["cases"]), 108)
+        parsed = issues.parse_cases(self.directory / "artifacts/cases.md", None)
+        self.assertTrue(all(row["statuses"] == {"PASS", "NOT_RUN"} for row in parsed.values()))
+        with mock.patch.object(issues, "search_open") as search:
+            issues.handle_success(mock.Mock(), parsed)
+            search.assert_not_called()
+        self.assertEqual(issues.classify("failure", parsed, None)[0], "env")
+        with self.assertRaises(ValueError):
+            chain.verify_report("upgrade", self.directory / "artifacts/cases.md")
+
+    def test_upgrade_interruption_and_recovery_failures_never_pass(self):
+        for mode, state in (("upgrade", "UPGRADE_FAILED"), ("interrupted", "INTERRUPTED"),
+                            ("cleanup", "CLEANUP_FAILED"), ("diagnostics", "DIAGNOSTICS_FAILED")):
+            with self.subTest(mode=mode):
+                shutil.rmtree(self.directory / "artifacts", ignore_errors=True)
+                result, data, trace = self.lifecycle(mode)
+                self.assertNotEqual(result.returncode, 0)
+                self.assertEqual(data["groups"][0]["result"], state)
+                self.assertEqual(data["groups"][1]["result"], "NOT_RUN")
+                self.assertEqual(len(data["cases"]), 108)
+                self.assertIn("cleanup vm003", trace, "teardown attempts all nodes even after failure")
+
+    def test_failed_assertions_keep_all_cases_and_continue_after_cleanup(self):
+        result, data, _ = self.lifecycle("verify")
+        self.assertNotEqual(result.returncode, 0)
+        self.assertEqual(data["counts"]["FAIL"], 6)
+        self.assertEqual(data["counts"]["PASS"], 102)
+        self.assertEqual(data["counts"]["NOT_RUN"], 0)
+        self.assertTrue(all(g["result"] == "VERIFY_FAILED" for g in data["groups"]))
+
+    def test_keep_environment_and_existing_artifact_directory(self):
+        result, data, trace = self.lifecycle(extra=("--keep-env",))
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertNotIn("cleanup vm", trace)
+        self.assertTrue(all(g["cleanup"] == "RETAINED" for g in data["groups"]))
+        old_report = (self.directory / "artifacts/results.json").read_text()
+        (self.directory / "trace").unlink()
+        result, _, trace = self.lifecycle()
+        self.assertNotEqual(result.returncode, 0)
+        self.assertEqual(trace, "")
+        self.assertEqual((self.directory / "artifacts/results.json").read_text(), old_report)
+
+    def test_report_legacy_context_and_missing_cleanup_evidence(self):
+        log, cases, matrix = (self.directory / name for name in ("log", "cases.md", "matrix.md"))
+        log.write_text("[INFO] == topology: single kms-backend: local ==\n--- UPG-101 case ---\n[PASS] UPG-101\n[UPG-TOPO] single local old new PASS=1 FAIL=0\n")
+        self.assertTrue(report.generate_report(log, cases, matrix))
+        log.write_text(log.read_text() + "[UPG-STATE] single local old new PASS\n")
+        self.assertFalse(report.generate_report(log, cases, matrix))
+
+    def test_diagnostics_archive_is_sanitized_even_when_ssh_fails(self):
+        golden = self.directory / "golden.tsv"
+        golden.write_text("creds:sa\tsa-access-canary|sa-secret-canary\n")
+        body = "set -Eeuo pipefail\n" + "\n".join(function(name) for name in ("collect_upgrade_diagnostics", "sanitize_upgrade_log")) + r'''
+TEST_TMP="$TEST_DIRECTORY"
+GROUP_DIR="$TEST_DIRECTORY"
+GOLDEN_FILE="$TEST_DIRECTORY/golden.tsv"
+SCRIPT_DIR="$TEST_ROOT"
+UPGRADE_PYTHON="$TEST_PYTHON"
+VAULT_TOKEN="$UPGRADE_TEST_TOKEN"
+KMS_LOCAL_MASTER_KEY="$UPGRADE_TEST_TOKEN"
+KMS_BACKUP_KEK="$UPGRADE_TEST_TOKEN"
+ACCESS_KEY=fixture-access
+SECRET_KEY="$UPGRADE_TEST_TOKEN"
+DIAGNOSTICS_DONE=0
+DRY_RUN=0
+CURRENT_TOPO=multi-multi
+SSH_OPTS=()
+SSH_USER=fixture
+topology_nodes() { echo 'node1 node2'; }
+timeout() { shift; "$@"; }
+ssh() {
+  cat > /dev/null
+  printf '{"journal": "credential %s sa-secret-canary", "file_logs": [{"content": "sa-access-canary"}]}\n' "$UPGRADE_TEST_TOKEN"
+  return 1
+}
+printf 'warp value %s\n' "$UPGRADE_TEST_TOKEN" > "$TEST_TMP/warp-put-test.log"
+collect_upgrade_diagnostics
+'''
+        result = self.bash(body, TEST_DIRECTORY=str(self.directory), TEST_ROOT=str(ROOT), TEST_PYTHON=sys.executable)
+        self.assertEqual(result.returncode, 1)
+        for name in ("diagnostics-node1.log", "diagnostics-node2.log", "warp-put-test.log"):
+            text = (self.directory / name).read_text()
+            for secret in (CANARY, "sa-access-canary", "sa-secret-canary"):
+                self.assertNotIn(secret, text)
+            self.assertIn("REDACTED", text)
+
+    def test_real_seed_verifier_compares_baseline_beyond_etags(self):
+        body = "set -Eeuo pipefail\n" + "\n".join(function(name) for name in ("capture_seed_manifest", "verify_warp_bucket", "assert_aws_ok")) + r'''
+TEST_TMP="$TEST_DIRECTORY"
+GROUP_DIR="$TEST_DIRECTORY"
+SCRIPT_DIR="$TEST_ROOT"
+UPGRADE_PYTHON="$TEST_PYTHON"
+DRY_RUN=0
+log() { :; }
+die() { echo "$*" >&2; exit 1; }
+golden_get() { echo 2; }
+aws_run() {
+  AWS_CODE=0
+  case "$2" in
+    list-objects-v2)
+      if [[ "$*" == *'length(Contents)'* ]]; then AWS_OUT=2
+      elif [ "${VERIFY_MODE:-}" = replace ]; then AWS_OUT='["one", "replacement"]'
+      else AWS_OUT='["one", "two"]'; fi ;;
+    get-object)
+      if [ "${VERIFY_MODE:-}" = corrupt ]; then printf changed > "${@: -1}"
+      else printf baseline > "${@: -1}"; fi
+      AWS_OUT='{"ETag":"same-etag"}' ;;
+    head-object) AWS_OUT='{"ETag":"same-etag"}' ;;
+    *) exit 90 ;;
+  esac
+}
+capture_seed_manifest fixture
+verify_warp_bucket fixture
+VERIFY_MODE=replace
+if verify_warp_bucket fixture; then exit 91; fi
+VERIFY_MODE=corrupt
+if verify_warp_bucket fixture; then exit 92; fi
+'''
+        result = self.bash(body, TEST_DIRECTORY=str(self.directory), TEST_ROOT=str(ROOT), TEST_PYTHON=sys.executable)
+        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+        self.assertIn("inventory changed", result.stdout)
+        self.assertIn("content changed", result.stdout)
+
+    def test_dry_run_skips_every_environment_boundary(self):
+        result, data, trace = self.lifecycle(extra=("--dry-run",))
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertIsNone(data)
+        self.assertEqual(trace, "")
+        self.assertEqual(result.stdout.count("[DRY-RUN] UPG-"), 108)
+        self.assertNotIn("[PASS]", result.stdout)
+
+
+@unittest.skipUnless(shutil.which("aws") and shutil.which("jq"), "real AWS CLI and jq required")
+class DeleteProtocolTests(unittest.TestCase):
+    setUp = UpgradeTests.setUp
+    bash = UpgradeTests.bash
+    # These tests use the real CLI against a loopback S3 XML fixture. Warp
+    # uploads are stubbed; they do not claim product/VM verification.
+    def test_real_cli_quiet_errors_and_batching(self):
+        for mode in ("success", "errors", "partial", "http503", "jqfail"):
+            with self.subTest(mode=mode):
+                directory = self.directory / mode
+                directory.mkdir()
+                original = ['alpha/001', '/leading', 'quote"and\\slash', '中文/key'] + [f"key-{i}" for i in range(2501)]
+                state = {"keys": original.copy(), "batches": []}
+
+                class Handler(BaseHTTPRequestHandler):
+                    def log_message(self, *_):
+                        pass
+
+                    def xml(self, status, value):
+                        body = value.encode()
+                        self.send_response(status)
+                        self.send_header("Content-Length", str(len(body)))
+                        self.send_header("Content-Type", "application/xml")
+                        self.send_header("x-amz-request-id", "fixture-delete-id")
+                        self.end_headers()
+                        self.wfile.write(body)
+
+                    def do_PUT(self):
+                        self.rfile.read(int(self.headers.get("Content-Length", 0)))
+                        self.xml(200, "")
+
+                    def do_GET(self):
+                        contents = "".join(f'<Contents><Key>{escape(key)}</Key><Size>1024</Size><LastModified>2026-01-01T00:00:00Z</LastModified><ETag>"fixture"</ETag></Contents>' for key in state["keys"])
+                        self.xml(200, f'<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Name>upg-fixture</Name><IsTruncated>false</IsTruncated>{contents}</ListBucketResult>')
+
+                    def do_POST(self):
+                        root = ET.fromstring(self.rfile.read(int(self.headers.get("Content-Length", 0))))
+                        keys = [e.text for e in root.iter() if e.tag.rsplit("}", 1)[-1] == "Key"]
+                        state["batches"].append(keys)
+                        if mode == "http503":
+                            self.xml(503, '<Error><Code>ServiceUnavailable</Code><Message>The service is unavailable</Message><RequestId>fixture-delete-id</RequestId></Error>')
+                            return
+                        failed = keys if mode == "errors" else keys[:1] if mode == "partial" else []
+                        state["keys"] = [key for key in state["keys"] if key not in keys or key in failed]
+                        errors = "".join(f'<Error><Key>{escape(key)}</Key><Code>ServiceUnavailable</Code><Message>{CANARY}</Message></Error>' for key in failed)
+                        self.xml(200, f'<DeleteResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">{errors}</DeleteResult>')
+
+                server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+                worker = threading.Thread(target=server.serve_forever, daemon=True)
+                worker.start()
+                variables = dict(TEST_TMP=str(directory), GOLDEN_FILE=str(directory / "golden.tsv"), GROUP_DIR=str(directory),
+                                 SCRIPT_DIR=str(ROOT), UPGRADE_PYTHON=sys.executable, JQ_LIB=str(ROOT / "scripts/lib"),
+                                 ENDPOINT=f"http://127.0.0.1:{server.server_port}", REGION="us-east-1", DRY_RUN="0", DELETE_BATCH="0",
+                                 ACCESS_KEY="fixture-access", SECRET_KEY=CANARY, VAULT_TOKEN=CANARY,
+                                 KMS_LOCAL_MASTER_KEY=CANARY, KMS_BACKUP_KEK=CANARY)
+                body = "set -Eeuo pipefail\n" + "".join(f"{name}={shlex.quote(value)}\n" for name, value in variables.items())
+                body += "\n".join(function(name) for name in ("aws_run", "assert_aws_ok", "new_bucket", "golden_set", "seed_warp_bucket"))
+                body += '\nlog() { :; }\ndie() { echo "$*" >&2; exit 1; }\nwarp() { :; }\n'
+                if mode == "jqfail":
+                    body += 'jq() { if [[ "$*" == *--argjson* ]]; then return 42; else command jq "$@"; fi; }\n'
+                body += 'seed_warp_bucket upg-fixture 1KiB 1s 2\n'
+                env = {"AWS_ACCESS_KEY_ID": "fixture-access", "AWS_SECRET_ACCESS_KEY": CANARY,
+                       "AWS_CONFIG_FILE": "/dev/null", "AWS_SHARED_CREDENTIALS_FILE": "/dev/null",
+                       "AWS_EC2_METADATA_DISABLED": "true", "AWS_PAGER": "", "AWS_MAX_ATTEMPTS": "1",
+                       "NO_PROXY": "*", "no_proxy": "*", "AWS_SESSION_TOKEN": "", "AWS_PROFILE": "", "AWS_DEFAULT_PROFILE": ""}
+                try:
+                    result = self.bash(body, **env)
+                finally:
+                    server.shutdown()
+                    server.server_close()
+                    worker.join()
+                receipts = [json.loads(path.read_text()) for path in sorted(directory.glob("delete-upg-fixture-*.json"))]
+                self.assertNotIn(CANARY, result.stdout + result.stderr + json.dumps(receipts))
+                if mode == "success":
+                    self.assertEqual(result.returncode, 0, result.stderr)
+                    self.assertEqual([len(keys) for keys in state["batches"]], [1000, 1000, 503])
+                    self.assertEqual(state["keys"], original[-2:])
+                    self.assertTrue(all(receipt["valid"] for receipt in receipts))
+                elif mode == "jqfail":
+                    self.assertNotEqual(result.returncode, 0)
+                    self.assertEqual(state["batches"], [])
+                    self.assertIn("payload generation failed", result.stderr)
+                else:
+                    self.assertNotEqual(result.returncode, 0)
+                    self.assertEqual(len(state["batches"]), 1, "stop on first failing batch")
+                    self.assertFalse(receipts[0]["valid"])
+                    self.assertEqual(receipts[0]["request_id"], "fixture-delete-id")
+                    if mode == "http503":
+                        self.assertEqual(receipts[0]["http_status"], 503)
+                        self.assertEqual(receipts[0]["service_error"]["Code"], "ServiceUnavailable")
+                    else:
+                        self.assertEqual(receipts[0]["http_status"], 200)
+                        self.assertEqual(receipts[0]["error_count"], 1000 if mode == "errors" else 1)
+                        self.assertEqual(receipts[0]["response"]["Errors"][0]["Code"], "ServiceUnavailable")
+
+
+if __name__ == "__main__":
+    unittest.main()

From d8341f9c52023bd6327d8f58e6f657d28f8f35f3 Mon Sep 17 00:00:00 2001
From: cxymds <cxymds@gmail.com>
Date: Wed, 11 Nov 2026 14:34:21 +0800
Subject: [PATCH 2/2] =?UTF-8?q?fix(upgrade):=20=E5=A4=84=E7=90=86=E7=9F=A9?=
 =?UTF-8?q?=E9=98=B5=E8=AF=81=E6=8D=AE=E4=B8=8E=E7=8E=AF=E5=A2=83=E9=9A=94?=
 =?UTF-8?q?=E7=A6=BB=E8=AF=84=E5=AE=A1=E6=84=8F=E8=A7=81?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .github/workflows/rustfs-upgrade-test.yml  |  15 +-
 .github/workflows/upgrade-script-tests.yml |   4 +
 docs/upgrade-test-evidence.md              |   7 +
 rustfs-upgrade-test.sh                     |  93 +++++++---
 scripts/functional_case_report.py          |  75 +++++++-
 scripts/issue_manager.py                   |   4 +-
 scripts/upgrade_evidence.py                |  21 ++-
 test_issue_manager.py                      |   8 +-
 test_rustfs_upgrade.py                     | 199 ++++++++++++++++++++-
 9 files changed, 381 insertions(+), 45 deletions(-)

diff --git a/.github/workflows/rustfs-upgrade-test.yml b/.github/workflows/rustfs-upgrade-test.yml
index 0ce1b07e..4dc8f2f6 100644
--- a/.github/workflows/rustfs-upgrade-test.yml
+++ b/.github/workflows/rustfs-upgrade-test.yml
@@ -74,7 +74,7 @@ defaults:
 env:
   RUSTFS_ACCESS_KEY: ${{ secrets.RUSTFS_ACCESS_KEY }}
   RUSTFS_SECRET_KEY: ${{ secrets.RUSTFS_SECRET_KEY }}
-  RUSTFS_NODES: ${{ secrets.RUSTFS_NODES || vars.RUSTFS_NODES }}
+  RUSTFS_NODES: ${{ secrets.RUSTFS_NODES || vars.RUSTFS_NODES || 'vm000 vm001 vm002 vm003' }}
   RUSTFS_SSH_USER: ${{ secrets.RUSTFS_SSH_USER || vars.RUSTFS_SSH_USER }}
   RUSTFS_NIGHTLY_PACKAGE_URL: ${{ vars.RUSTFS_NIGHTLY_PACKAGE_URL || 'https://dl.rustfs.com/artifacts/rustfs/packages/nightly/rustfs-nightly-latest.deb' }}
   PF_TESTING_GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
@@ -133,7 +133,7 @@ jobs:
         if: ${{ inputs.cleanup_before != 'false' || github.event_name != 'workflow_dispatch' }}
         run: |
           set -euo pipefail
-          read -r -a NODES <<< "${RUSTFS_NODES:-vm000 vm001 vm002}"
+          read -r -a NODES <<< "${RUSTFS_NODES:-vm000 vm001 vm002 vm003}"
           SSH_USER="${RUSTFS_SSH_USER:-azureuser}"
           for node in "${NODES[@]}"; do
             ssh -o BatchMode=yes -o ConnectTimeout=10 -o StrictHostKeyChecking=accept-new "${SSH_USER}@${node}" '
@@ -415,7 +415,7 @@ jobs:
         if: ${{ always() && (inputs.cleanup_after != 'false' || github.event_name != 'workflow_dispatch') && steps.environment.outcome == 'success' }}
         run: |
           set -euo pipefail
-          read -r -a NODES <<< "${RUSTFS_NODES:-vm000 vm001 vm002}"
+          read -r -a NODES <<< "${RUSTFS_NODES:-vm000 vm001 vm002 vm003}"
           SSH_USER="${RUSTFS_SSH_USER:-azureuser}"
           for node in "${NODES[@]}"; do
             ssh -o BatchMode=yes -o ConnectTimeout=10 -o StrictHostKeyChecking=accept-new "${SSH_USER}@${node}" '
@@ -436,7 +436,14 @@ jobs:
         env:
           FUNCTIONAL_ENVIRONMENT_RECEIPT: ${{ runner.temp }}/environment-upgrade.json
           CLEANUP_OUTCOME: ${{ steps.environment_cleanup.outcome }}
-        run: python3 scripts/functional_environment.py finish --suite upgrade --cleanup-outcome "$CLEANUP_OUTCOME"
+        run: |
+          # The outer node cleanup cannot prove Vault cleanup or that the
+          # interrupted suite cleaned every resource it touched.
+          if ! python3 scripts/upgrade_evidence.py check-cleanup \
+            --state-file "${FUNCTIONAL_ARTIFACTS_DIR}/upgrade/fleet-cleanup.state"; then
+            CLEANUP_OUTCOME=failure
+          fi
+          python3 scripts/functional_environment.py finish --suite upgrade --cleanup-outcome "$CLEANUP_OUTCOME"
 
       - name: Retain environment ownership and cleanup receipt
         if: ${{ always() }}
diff --git a/.github/workflows/upgrade-script-tests.yml b/.github/workflows/upgrade-script-tests.yml
index e7346809..cbdbae9f 100644
--- a/.github/workflows/upgrade-script-tests.yml
+++ b/.github/workflows/upgrade-script-tests.yml
@@ -9,6 +9,8 @@ on:
       - scripts/issue_manager.py
       - scripts/lib/**
       - test_rustfs_upgrade.py
+      - test_issue_manager.py
+      - scripts/functional_environment.py
       - .github/workflows/rustfs-upgrade-test.yml
       - .github/workflows/upgrade-script-tests.yml
   push:
@@ -20,6 +22,8 @@ on:
       - scripts/issue_manager.py
       - scripts/lib/**
       - test_rustfs_upgrade.py
+      - test_issue_manager.py
+      - scripts/functional_environment.py
       - .github/workflows/rustfs-upgrade-test.yml
       - .github/workflows/upgrade-script-tests.yml
 
diff --git a/docs/upgrade-test-evidence.md b/docs/upgrade-test-evidence.md
index 5cbb92a3..61b7b1ec 100644
--- a/docs/upgrade-test-evidence.md
+++ b/docs/upgrade-test-evidence.md
@@ -11,6 +11,8 @@
 证据目录包含：
 
 - `events.log`：同步写入的逐项用例和阶段结果，不依赖 tee 完成时序。
+- `plan.tsv`：依赖检查前原子保存的选定组合及其全部用例 ID。解析器对照计划检查每组的用例、终态、清理及诊断记录，缺失或重复都会失败；缺失用例补为 NOT_RUN。
+- `fleet-cleanup.state`：触及环境前标记 PENDING，所有目标节点及 Vault 清理验证完成后才标记 PASS。缺失、PENDING、FAIL 或 RETAINED 均不能授权 Actions 释放共享环境。
 - `cases.md`、`matrix.md`、`results.json`：所有选定组合的逐项结果和结构化汇总。
 - `<拓扑>-<后端>/`：种子清单、删除响应、逐项诊断、Warp 输出及节点文件日志和 journal。节点文件日志最多保留每文件末尾 4MiB、每节点合计 16MiB，截断会标明。
 
@@ -23,12 +25,17 @@
 | UPGRADE_FAILED | 原地升级或升级后健康检查失败 |
 | VERIFY_FAILED | 已进入新版验证，至少一个断言失败 |
 | INTERRUPTED | 收到 INT/TERM 中断 |
+| INCOMPLETE | 进程或日志中断，计划中的终态证据缺失 |
 | DIAGNOSTICS_FAILED | 诊断证据未完整归档 |
 | CLEANUP_FAILED | 环境清理未完成 |
 | NOT_RUN | 该组或用例未执行完成，不能作为通过证据 |
 
 准备失败和中断都会补齐未完成的用例，先归档再清理当前组，并将后续组标记为未执行。普通断言失败在清理成功后继续其他组合；诊断或清理失败则停止后续组。主失败原因与清理/诊断结果分别保留。`--keep-env` 显式保留环境，仍归档证据并标记清理为 `RETAINED`。
 
+依赖检查失败会保留整份计划、准备失败记录及所有未执行用例，不触发环境清理。Python 不可用时仍保留 Bash 写入的事件供后续解析。诊断收集中断保留失败状态，退出处理不得把不完整归档改为通过。dry-run 不打开或覆盖指定日志文件。
+
+Actions 的隔离标记、套件目标节点和外层清理使用同一份 `RUSTFS_NODES`，默认包含 vm000–vm003。释放环境同时要求外层清理成功与套件清理证据通过，任一失败均保留隔离所有权，需在共享锁下完成恢复再解除。
+
 例如三组 local 验证通过后，vault-kv2 第一组种子准备失败，报告应为 54 PASS、54 NOT_RUN，矩阵包含 `PREPARE_FAILED`。这证明本轮升级矩阵未完成，不能据此判定新版 Vault 兼容性失败，也不能自动关闭历史产品 Issue。
 
 GitHub Actions 上传上述安全证据目录。独立的 `Upgrade Script Tests` 使用本地协议 fixture 和模拟环境边界验证脚本，包含真实 AWS CLI 的 HTTP 200 逐对象错误、顶层 503 和批量删除分片。这些检查不连接 VM，不代表产品实机复测。
diff --git a/rustfs-upgrade-test.sh b/rustfs-upgrade-test.sh
index f66e0caa..6fccdcfd 100755
--- a/rustfs-upgrade-test.sh
+++ b/rustfs-upgrade-test.sh
@@ -39,7 +39,7 @@
 #
 # Environment:
 #   RUSTFS_ACCESS_KEY / RUSTFS_SECRET_KEY   admin credentials
-#   RUSTFS_NODES / RUSTFS_SSH_USER          shared test VMs (default vm000..002 / azureuser)
+#   RUSTFS_NODES / RUSTFS_SSH_USER          shared test VMs (default vm000..003 / azureuser)
 set -Eeuo pipefail
 
 SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
@@ -60,10 +60,11 @@ CURRENT_STARTED=" "
 CURRENT_PHASE=PREPARE
 CURRENT_RESULT=""
 DIAGNOSTICS_DONE=0
+DIAGNOSTICS_STATUS=NOT_RUN
 DELETE_BATCH=0
 
 # ==================== Configuration ====================
-NODES=(vm000 vm001 vm002 vm003)
+read -r -a NODES <<<"${RUSTFS_NODES:-vm000 vm001 vm002 vm003}"
 SSH_USER="${RUSTFS_SSH_USER:-azureuser}"
 SSH_PORT="${RUSTFS_SSH_PORT:-22}"
 SSH_OPTS=(-o BatchMode=yes -o ConnectTimeout=10 -o StrictHostKeyChecking=accept-new -p "${SSH_PORT}")
@@ -262,6 +263,10 @@ cleanup_node() {
     printf 'if ${SUDO} dpkg -l %s 2>/dev/null | grep -q "^ii"; then ${SUDO} dpkg -P %s; fi\n' "${RUSTFS_PACKAGE_NAME}" "${RUSTFS_PACKAGE_NAME}"
     printf 'for i in 1 2 3 4; do ${SUDO} rm -rf /data/rustfs${i}/mnmd; done\n'
     printf '${SUDO} rm -rf /var/log/rustfs /var/lib/rustfs/kms /var/lib/rustfs/kms-backup\n'
+    printf 'if ${SUDO} systemctl is-active --quiet rustfs; then exit 1; fi\n'
+    printf 'if ${SUDO} dpkg -l rustfs 2>/dev/null | grep -q "^ii"; then exit 1; fi\n'
+    printf 'for i in 1 2 3 4; do [ ! -e /data/rustfs${i}/mnmd ] || exit 1; done\n'
+    printf 'for p in /var/log/rustfs /var/lib/rustfs/kms /var/lib/rustfs/kms-backup; do [ ! -e "$p" ] || exit 1; done\n'
   } | run_remote "${node}"
 }
 
@@ -482,9 +487,10 @@ vault_stop() {
   # modes inside the data dir, so the runner user cannot remove them
   # directly (rm exits 1 under set -e and kills the suite mid-loop).
   if vault_local; then
-    heal_run rm -rf "${VAULT_DATA_DIR}" "${VAULT_CONFIG_DIR}"
+    heal_run rm -rf "${VAULT_DATA_DIR}" "${VAULT_CONFIG_DIR}" || return 1
+    heal_run bash -c '[ ! -e "$1" ] && [ ! -e "$2" ]' -- "${VAULT_DATA_DIR}" "${VAULT_CONFIG_DIR}"
   else
-    vault_remote_bash "rm -rf '${VAULT_DATA_DIR}' '${VAULT_CONFIG_DIR}'"
+    vault_remote_bash "rm -rf '${VAULT_DATA_DIR}' '${VAULT_CONFIG_DIR}' && [ ! -e '${VAULT_DATA_DIR}' ] && [ ! -e '${VAULT_CONFIG_DIR}' ]"
   fi
 }
 
@@ -658,7 +664,12 @@ evidence_line() {
 
 collect_upgrade_diagnostics() {
   [ "${DIAGNOSTICS_DONE}" -eq 0 ] || return 0
-  DIAGNOSTICS_DONE=1
+  if [ "${DIAGNOSTICS_STATUS}" != NOT_RUN ]; then
+    # Re-entry after interruption must preserve an incomplete/failed archive.
+    DIAGNOSTICS_STATUS=FAIL
+    return 1
+  fi
+  DIAGNOSTICS_STATUS=RUNNING
   [ "${DRY_RUN}" -eq 0 ] || return 0
   local node fail=0 nodes=()
   local warp_log
@@ -698,7 +709,15 @@ UPG_DIAGNOSTICS
     then fail=1; fi
     sanitize_upgrade_log "${TEST_TMP}/diagnostic.log" "${GROUP_DIR}/diagnostics-${node}.log" || fail=1
   done
-  return "${fail}"
+  if [ "${fail}" -ne 0 ]; then DIAGNOSTICS_STATUS=FAIL; return 1; fi
+  DIAGNOSTICS_STATUS=PASS
+  DIAGNOSTICS_DONE=1
+  return 0
+}
+
+record_upgrade_cleanup() {
+  printf '%s\n' "$1" >"${ARTIFACTS_DIR}/fleet-cleanup.state.tmp" && \
+    mv "${ARTIFACTS_DIR}/fleet-cleanup.state.tmp" "${ARTIFACTS_DIR}/fleet-cleanup.state"
 }
 
 sanitize_upgrade_log() {
@@ -724,6 +743,7 @@ finish_upgrade_group() {
   evidence_line "[UPG-CLEANUP] ${CURRENT_TOPO} ${KMS_BACKEND} ${cleanup}"
   evidence_line "[UPG-DIAGNOSTICS] ${CURRENT_TOPO} ${KMS_BACKEND} ${diagnostics}"
   evidence_line "[UPG-STATE] ${CURRENT_TOPO} ${KMS_BACKEND} ${OLD_VERSION_ACTUAL:-unknown} ${NEW_VERSION_ACTUAL:-unknown} ${CURRENT_RESULT}"
+  record_upgrade_cleanup "${cleanup}" || fail=1
   GROUP_ACTIVE=0
   return "${fail}"
 }
@@ -758,7 +778,11 @@ upgrade_exit() {
       CURRENT_DONE=" "
       CURRENT_STARTED=" "
       emit_unrun_cases
-      evidence_line "[UPG-STATE] ${topo} ${backend} unknown unknown NOT_RUN"
+      evidence_line "[UPG-CLEANUP] ${topo} ${backend} NOT_TOUCHED"
+      evidence_line "[UPG-DIAGNOSTICS] ${topo} ${backend} NOT_RUN"
+      local unrun_state=NOT_RUN
+      if [ "${index}" -eq 0 ] && [ "${rc}" -ne 0 ]; then unrun_state=PREPARE_FAILED; fi
+      evidence_line "[UPG-STATE] ${topo} ${backend} unknown unknown ${unrun_state}"
     done
     "${UPGRADE_PYTHON}" "${SCRIPT_DIR}/scripts/functional_case_report.py" \
       "${ARTIFACTS_DIR}/events.log" "${ARTIFACTS_DIR}/cases.md" "${ARTIFACTS_DIR}/matrix.md" \
@@ -1554,7 +1578,7 @@ main() {
       *) die "unknown option: ${opt} (see --help)" ;;
     esac
   done
-  if [ -n "${LOG_FILE:-}" ]; then
+  if [ "${DRY_RUN}" -eq 0 ] && [ -n "${LOG_FILE:-}" ]; then
     mkdir -p "$(dirname "${LOG_FILE}")"
     if ! touch "${LOG_FILE}" 2>/dev/null; then
       LOG_FILE="$(mktemp "${TMPDIR:-/tmp}/rustfs-upgrade.XXXXXX.log")"
@@ -1569,23 +1593,6 @@ main() {
   IFS=',' read -r -a BACKENDS <<<"${backends_arg}"
   [ "${#BACKENDS[@]}" -gt 0 ] || die "no backends selected (--backends local,vault-kv2)"
 
-  local _p
-  for _p in /root/.local/bin /home/*/.local/bin /usr/local/bin; do
-    if [ -d "${_p}" ] && [[ ":${PATH}:" != *":${_p}:"* ]]; then PATH="${_p}:${PATH}"; fi
-  done
-  export PATH
-  need_cmd ssh "SSH client"
-  need_cmd rc "RustFS client CLI"
-  need_cmd aws "AWS CLI (s3api channel)"
-  need_cmd warp "warp benchmark tool (mixed-size data seeding)"
-  need_cmd jq "JSON processor"
-  need_cmd curl "HTTP client"
-  need_cmd openssl "OpenSSL (SigV4)"
-  need_cmd "${UPGRADE_PYTHON}" "Python for upgrade evidence"
-  need_cmd timeout "bounded diagnostics"
-  log "nodes: ${NODES[*]}  from: $(resolve_from_url)  to: $(resolve_to_url)"
-  log "backends: ${BACKENDS[*]}  topologies: ${SELECTED_TOPOLOGIES[*]}"
-
   confirm "Run upgrade compatibility suite (${SELECTED_TOPOLOGIES[*]}, ${BACKENDS[*]})?"
 
   local group_topo group_backend
@@ -1614,19 +1621,51 @@ main() {
     mkdir -p "$(dirname "${ARTIFACTS_DIR}")"
     mkdir "${ARTIFACTS_DIR}" || die "artifacts directory must be new: ${ARTIFACTS_DIR}"
     : >"${ARTIFACTS_DIR}/events.log"
+    local planned_group
+    for planned_group in "${UPGRADE_GROUPS[@]}"; do
+      (IFS=,; printf '%s\t%s\t%s\n' "${planned_group%/*}" "${planned_group#*/}" "${UPGRADE_IDS[*]}")
+    done >"${ARTIFACTS_DIR}/plan.tsv.tmp"
+    mv "${ARTIFACTS_DIR}/plan.tsv.tmp" "${ARTIFACTS_DIR}/plan.tsv"
+    record_upgrade_cleanup NOT_TOUCHED
     ARTIFACTS_READY=1
+    evidence_line "[UPG-PLAN-BEGIN] ${#UPGRADE_GROUPS[@]}"
+    while IFS=$'\t' read -r group_topo group_backend planned_group; do
+      evidence_line "[UPG-PLAN] ${group_topo} ${group_backend} ${planned_group}"
+    done <"${ARTIFACTS_DIR}/plan.tsv"
+    evidence_line "[UPG-PLAN-END] ${#UPGRADE_GROUPS[@]}"
   fi
 
+  local _p
+  for _p in /root/.local/bin /home/*/.local/bin /usr/local/bin; do
+    if [ -d "${_p}" ] && [[ ":${PATH}:" != *":${_p}:"* ]]; then PATH="${_p}:${PATH}"; fi
+  done
+  export PATH
+  need_cmd ssh "SSH client"
+  need_cmd rc "RustFS client CLI"
+  need_cmd aws "AWS CLI (s3api channel)"
+  need_cmd warp "warp benchmark tool (mixed-size data seeding)"
+  need_cmd jq "JSON processor"
+  need_cmd curl "HTTP client"
+  need_cmd openssl "OpenSSL (SigV4)"
+  need_cmd "${UPGRADE_PYTHON}" "Python for upgrade evidence"
+  need_cmd timeout "bounded diagnostics"
+  log "nodes: ${NODES[*]}  from: $(resolve_from_url)  to: $(resolve_to_url)"
+  log "backends: ${BACKENDS[*]}  topologies: ${SELECTED_TOPOLOGIES[*]}"
+
   local overall_fail=0 topo backend
   for backend in "${BACKENDS[@]}"; do
     KMS_BACKEND="${backend}"
     for topo in "${SELECTED_TOPOLOGIES[@]}"; do
       CURRENT_TOPO="${topo}"; CURRENT_PHASE=PREPARE; CURRENT_RESULT=""
-      CURRENT_DONE=" "; CURRENT_STARTED=" "; DIAGNOSTICS_DONE=0
+      CURRENT_DONE=" "; CURRENT_STARTED=" "; DIAGNOSTICS_DONE=0; DIAGNOSTICS_STATUS=NOT_RUN
       OLD_VERSION_ACTUAL=unknown; NEW_VERSION_ACTUAL=unknown
       rm -f "${TEST_TMP}"/warp-put-*.log
       GROUP_DIR="${ARTIFACTS_DIR}/${topo}-${backend}"
-      if [ "${ARTIFACTS_READY}" -eq 1 ]; then mkdir "${GROUP_DIR}"; GROUP_ACTIVE=1; fi
+      if [ "${ARTIFACTS_READY}" -eq 1 ]; then
+        mkdir "${GROUP_DIR}"
+        record_upgrade_cleanup PENDING
+        GROUP_ACTIVE=1
+      fi
       evidence_line "[INFO] == topology: ${topo}  kms-backend: ${KMS_BACKEND} =="
       if [ "${KMS_BACKEND}" = "vault-kv2" ]; then
         vault_stop
diff --git a/scripts/functional_case_report.py b/scripts/functional_case_report.py
index 89987f88..bc6e8d8b 100644
--- a/scripts/functional_case_report.py
+++ b/scripts/functional_case_report.py
@@ -10,7 +10,7 @@
 
 
 def generate_report(log_file: Path, case_file: Path, matrix_file: Path | None = None,
-                    json_file: Path | None = None) -> bool:
+                    json_file: Path | None = None, plan_file: Path | None = None) -> bool:
     ansi = re.compile(r"\x1b\[[0-9;]*m")
     start_re = re.compile(r"^---\s+([A-Z][A-Z0-9]*-[0-9]+)\s+(.+?)\s+---$")
     done_re = re.compile(r"^\[(PASS|FAIL|UNSUPPORTED|NOT_RUN)\]\s+([A-Z][A-Z0-9]*-[0-9]+)\b")
@@ -18,11 +18,19 @@ def generate_report(log_file: Path, case_file: Path, matrix_file: Path | None =
     topo_re = re.compile(r"^\[UPG-TOPO\]\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+PASS=(\d+)\s+FAIL=(\d+)\s*$")
     state_re = re.compile(r"^\[UPG-STATE\]\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s*$")
     detail_re = re.compile(r"^\[UPG-(CLEANUP|DIAGNOSTICS)\]\s+(\S+)\s+(\S+)\s+(\S+)\s*$")
+    plan_re = re.compile(r"^\[UPG-PLAN\]\s+(\S+)\s+(\S+)\s+(\S+)\s*$")
+    boundary_re = re.compile(r"^\[UPG-PLAN-(BEGIN|END)\]\s+(\d+)\s*$")
+    group_context_re = re.compile(r"^topology:\s*(\S+)\s+kms-backend:\s*(\S+)\s*$")
     rows = []
+    row_groups = []
     pending = {}
     topo_rows = []
     groups = []
     details = {}
+    detail_counts = {}
+    logged_plan = []
+    boundaries = []
+    current_group = None
     context = "context not recorded"
     complete = True
     try:
@@ -31,12 +39,21 @@ def generate_report(log_file: Path, case_file: Path, matrix_file: Path | None =
                 line = ansi.sub("", raw).strip()
                 if match := context_re.match(line):
                     context = match[1]
+                    group_match = group_context_re.match(context)
+                    current_group = group_match.groups() if group_match else None
                     pending.clear()
+                elif match := plan_re.match(line):
+                    topo, backend, ids = match.groups()
+                    logged_plan.append((topo, backend, ids.split(",")))
+                elif match := boundary_re.match(line):
+                    boundaries.append((match[1], int(match[2])))
                 elif match := topo_re.match(line):
                     topo_rows.append(match.groups())
                 elif match := detail_re.match(line):
                     kind, topo, backend, result = match.groups()
                     details.setdefault((topo, backend), {})[kind.lower()] = result
+                    key = (topo, backend, kind)
+                    detail_counts[key] = detail_counts.get(key, 0) + 1
                 elif match := state_re.match(line):
                     topo, backend, old_v, new_v, result = match.groups()
                     groups.append(dict(topology=topo, backend=backend, from_version=old_v,
@@ -46,17 +63,63 @@ def generate_report(log_file: Path, case_file: Path, matrix_file: Path | None =
                     case_id, name = match.groups()
                     pending[case_id] = len(rows)
                     rows.append([case_id, f"{name} ({context})", "RUNNING"])
+                    row_groups.append(current_group)
                 elif match := done_re.match(line):
                     status, case_id = match.groups()
                     index = pending.pop(case_id, None)
                     if index is None:
                         complete = False
                         rows.append([case_id, f"{case_id} ({context}; start not recorded)", status])
+                        row_groups.append(current_group)
                     else:
                         rows[index][2] = status
     except FileNotFoundError:
         pass
 
+    # Modern upgrade runs persist the selected matrix before preflight. A
+    # partial log must not invent a smaller successful matrix from its prefix.
+    plan = logged_plan
+    modern = bool(groups or logged_plan or boundaries)
+    if plan_file is None:
+        candidate = log_file.parent / "plan.tsv"
+        plan_file = candidate if candidate.exists() else None
+    if plan_file is not None:
+        modern = True
+        try:
+            plan = []
+            for line in plan_file.read_text(encoding="utf-8").splitlines():
+                topo, backend, ids = line.split("\t")
+                plan.append((topo, backend, ids.split(",")))
+        except (OSError, ValueError):
+            plan = []
+            complete = False
+        if logged_plan:
+            complete = complete and logged_plan == plan
+            complete = complete and boundaries == [("BEGIN", len(plan)), ("END", len(plan))]
+    elif modern:
+        complete = complete and boundaries == [("BEGIN", len(plan)), ("END", len(plan))]
+    if modern:
+        identities = [(topo, backend) for topo, backend, _ in plan]
+        complete = complete and bool(plan) and len(identities) == len(set(identities))
+        for topo, backend, expected in plan:
+            identity = (topo, backend)
+            complete = complete and bool(expected) and len(expected) == len(set(expected)) and all(
+                re.fullmatch(r"UPG-[0-9]+", cid) for cid in expected)
+            actual = [row[0] for row, group in zip(rows, row_groups) if group == identity]
+            complete = complete and sorted(actual) == sorted(expected)
+            for cid in expected:
+                if cid not in actual:
+                    rows.append([cid, f"{cid} (topology: {topo} kms-backend: {backend}; execution not recorded)", "NOT_RUN"])
+                    row_groups.append(identity)
+            terminal = [group for group in groups if (group["topology"], group["backend"]) == identity]
+            complete = complete and len(terminal) == 1 and all(
+                detail_counts.get((topo, backend, kind)) == 1 for kind in ("CLEANUP", "DIAGNOSTICS"))
+            if not terminal:
+                groups.append(dict(topology=topo, backend=backend, from_version="unknown", to_version="unknown",
+                                   result="INCOMPLETE", **details.get(identity, {})))
+        complete = complete and all(group in identities for group in row_groups) and all(
+            (group["topology"], group["backend"]) in identities for group in groups)
+
     counts = {status: sum(row[2] == status for row in rows)
               for status in ("PASS", "FAIL", "UNSUPPORTED", "RUNNING", "NOT_RUN")}
     with case_file.open("w", encoding="utf-8") as out:
@@ -73,6 +136,8 @@ def generate_report(log_file: Path, case_file: Path, matrix_file: Path | None =
     if groups:
         valid = valid and all(group["result"] == "PASS" and group.get("cleanup") in ("PASS", "RETAINED")
                              and group.get("diagnostics") == "PASS" for group in groups)
+        # Unsupported/partial upgrade cases do not prove compatibility.
+        valid = valid and all(row[2] == "PASS" for row in rows)
     if matrix_file is not None:
         with matrix_file.open("w", encoding="utf-8") as out:
             out.write("## Upgrade Matrix\n\n| Topology | KMS Backend | From Version | To Version | Result | Cleanup | Diagnostics |\n")
@@ -83,7 +148,7 @@ def generate_report(log_file: Path, case_file: Path, matrix_file: Path | None =
                               ("topology", "backend", "from_version", "to_version", "result", "cleanup", "diagnostics")) + " |\n")
                 out.write("\nPREPARE_FAILED: 旧版环境或种子准备失败；UPGRADE_FAILED: 升级阶段失败；"
                           "VERIFY_FAILED: 升级后断言失败；NOT_RUN: 未执行；INTERRUPTED: 被中断。"
-                          "DIAGNOSTICS_FAILED / CLEANUP_FAILED: 证据归档或环境恢复失败。\n")
+                          "INCOMPLETE: 计划中的终态证据缺失；DIAGNOSTICS_FAILED / CLEANUP_FAILED: 证据归档或环境恢复失败。\n")
             else:
                 for topo, backend, old_v, new_v, npass, nfail in topo_rows:
                     result = "PASS" if nfail == "0" else "FAIL"
@@ -94,7 +159,8 @@ def generate_report(log_file: Path, case_file: Path, matrix_file: Path | None =
     if json_file is not None:
         json_file.write_text(json.dumps(dict(schema=1, status="PASS" if valid else "FAIL", counts=counts,
                              cases=[dict(id=cid, name=name, status=status) for cid, name, status in rows],
-                             groups=groups), ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+                             groups=groups, plan=[dict(topology=topo, backend=backend, cases=ids) for topo, backend, ids in plan]),
+                             ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
     return valid
 
 
@@ -104,5 +170,6 @@ def generate_report(log_file: Path, case_file: Path, matrix_file: Path | None =
     parser.add_argument("case_file", type=Path)
     parser.add_argument("matrix_file", type=Path, nargs="?")
     parser.add_argument("--json", dest="json_file", type=Path)
+    parser.add_argument("--plan", dest="plan_file", type=Path)
     args = parser.parse_args()
-    raise SystemExit(0 if generate_report(args.log_file, args.case_file, args.matrix_file, args.json_file) else 1)
+    raise SystemExit(0 if generate_report(args.log_file, args.case_file, args.matrix_file, args.json_file, args.plan_file) else 1)
diff --git a/scripts/issue_manager.py b/scripts/issue_manager.py
index 77bb9949..e3e0dbdd 100644
--- a/scripts/issue_manager.py
+++ b/scripts/issue_manager.py
@@ -167,8 +167,8 @@ def classify(outcome: str, cases: dict, log: Path | None) -> tuple[str, str]:
         return "env", "no case results were recorded"
     fail = [c for c, v in cases.items() if "FAIL" in v["statuses"]]
     pas = [c for c, v in cases.items() if "PASS" in v["statuses"] and "FAIL" not in v["statuses"]]
-    if not fail and any("NOT_RUN" in value["statuses"] for value in cases.values()):
-        return "env", "upgrade matrix incomplete; cases were not executed"
+    if not fail:
+        return "env", "suite failed without failing case IDs; preparation, diagnostics or cleanup incomplete"
     if not pas and len(fail) >= 3:
         return "env", f"wholesale failure (0 passed, {len(fail)} failed)"
     if log and log.exists():
diff --git a/scripts/upgrade_evidence.py b/scripts/upgrade_evidence.py
index 20985d3c..2d5ce6d7 100644
--- a/scripts/upgrade_evidence.py
+++ b/scripts/upgrade_evidence.py
@@ -12,6 +12,8 @@
 import re
 import xml.etree.ElementTree as ET
 
+SENSITIVE_FIELD = r"(?:authorization|(?:secret|access)[ _-]?(?:access[ _-]?)?key(?:[ _-]?id)?|(?:session[ _-]?|vault[ _-]?|root[ _-]?)?token|password|client[ _-]?secret|master[ _-]?key)"
+
 
 def redact(text: str, golden: Path | None = None) -> str:
     values = [value for key, value in os.environ.items()
@@ -24,9 +26,10 @@ def redact(text: str, golden: Path | None = None) -> str:
     for value in sorted(set(values), key=len, reverse=True):
         if value:
             text = text.replace(value, "[REDACTED]")
-    return re.sub(r"(?im)^.*(?:authorization|secret[ _-]?(?:access[ _-]?)?key|"
-                  r"session[ _-]?token|password|client[ _-]?secret|vault[ _-]?token|"
-                  r"master[ _-]?key)\s*[=:].*$", "[REDACTED SENSITIVE LINE]", text)
+    # Node logs arrive as strings, including JSON embedded in journal lines
+    # and JSON-escaped quotes. Unknown credentials cannot rely on golden/env.
+    return re.sub(r"(?im)^.*\b" + SENSITIVE_FIELD + r"(?:\\?[\"'])?\s*[=:].*$",
+                  "[REDACTED SENSITIVE LINE]", text)
 
 
 def write_json(path: Path, value) -> None:
@@ -136,9 +139,17 @@ def check_body(manifest: Path, key: str, body: Path):
         raise ValueError("seed content changed")
 
 
+def check_cleanup(state_file: Path):
+    # A missing or interrupted receipt must never authorize fleet release.
+    if state_file.read_text().strip() not in ("PASS", "NOT_TOUCHED"):
+        raise ValueError("upgrade cleanup incomplete; fleet remains quarantined")
+
+
 def main():
     parser = argparse.ArgumentParser(description=__doc__)
     commands = parser.add_subparsers(dest="command", required=True)
+    cleanup = commands.add_parser("check-cleanup")
+    cleanup.add_argument("--state-file", type=Path, required=True)
     delete = commands.add_parser("validate-delete")
     for name in ("request", "response", "debug", "output"):
         delete.add_argument("--" + name, type=Path, required=True)
@@ -162,7 +173,9 @@ def main():
     try:
         if args.command == "validate-delete":
             return 0 if validate_delete(args.request, args.response, args.debug, args.exit_code, args.output) else 1
-        if args.command == "sanitize":
+        if args.command == "check-cleanup":
+            check_cleanup(args.state_file)
+        elif args.command == "sanitize":
             raw = args.input.read_text(errors="replace")
             try:
                 raw = json.dumps(safe_value(json.loads(raw), args.golden), ensure_ascii=False, indent=2)
diff --git a/test_issue_manager.py b/test_issue_manager.py
index 2841e523..367fdb16 100644
--- a/test_issue_manager.py
+++ b/test_issue_manager.py
@@ -58,6 +58,12 @@ def test_cancelled_or_skipped_runs_are_noop(self):
         for outcome in ("cancelled", "skipped"):
             self.assertEqual(manager.classify(outcome, {}, None)[0], "noop")
 
+    def test_failed_run_with_only_passing_cases_is_harness_failure(self):
+        for statuses in ({"PASS"}, {"PASS", "NOT_RUN"}, {"RUNNING"}):
+            with self.subTest(statuses=statuses):
+                cases = {"UPG-101": {"statuses": statuses}}
+                self.assertEqual(manager.classify("failure", cases, None)[0], "env")
+
 
 if __name__ == "__main__":
-    unittest.main()
\ No newline at end of file
+    unittest.main()
diff --git a/test_rustfs_upgrade.py b/test_rustfs_upgrade.py
index 064223da..545850ba 100644
--- a/test_rustfs_upgrade.py
+++ b/test_rustfs_upgrade.py
@@ -130,8 +130,17 @@ def lifecycle(self, mode="pass", extra=()):
 rc() { echo 'unexpected rc' >&2; exit 90; }
 aws() { echo 'unexpected aws' >&2; exit 90; }
 warp() { echo 'unexpected warp' >&2; exit 90; }
-need_cmd() { :; }
-vault_stop() { printf 'vault-stop\n' >> "$TEST_TRACE"; }
+need_cmd() {
+  if [ "$TEST_MODE" = preflight ] && [ "$1" = aws ]; then die 'fixture missing AWS CLI'; fi
+  if [ "$TEST_MODE" = preflight-python ] && [ "$1" = "$UPGRADE_PYTHON" ]; then
+    UPGRADE_PYTHON=fixture-nonexistent-python
+    die 'fixture missing Python'
+  fi
+}
+vault_stop() {
+  printf 'vault-stop\n' >> "$TEST_TRACE"
+  if [ "$TEST_MODE" = vault-cleanup ] && [ "$CURRENT_PHASE" = CLEANUP ]; then return 1; fi
+}
 vault_start() { :; }
 env_setup_old() { OLD_VERSION_ACTUAL=old; }
 seed_golden() {
@@ -236,6 +245,94 @@ def test_keep_environment_and_existing_artifact_directory(self):
         self.assertEqual(trace, "")
         self.assertEqual((self.directory / "artifacts/results.json").read_text(), old_report)
 
+    def test_preflight_records_the_plan_without_touching_environment(self):
+        for mode in ("preflight", "preflight-python"):
+            with self.subTest(mode=mode):
+                shutil.rmtree(self.directory / "artifacts", ignore_errors=True)
+                result, data, trace = self.lifecycle(mode)
+                self.assertNotEqual(result.returncode, 0)
+                self.assertEqual(trace, "")
+                artifacts = self.directory / "artifacts"
+                self.assertEqual(len((artifacts / "plan.tsv").read_text().splitlines()), 6)
+                self.assertEqual((artifacts / "fleet-cleanup.state").read_text().strip(), "NOT_TOUCHED")
+                log = (artifacts / "events.log").read_text()
+                self.assertEqual(log.count("[NOT_RUN]"), 108)
+                self.assertIn("PREPARE_FAILED", log)
+                if mode == "preflight-python":
+                    self.assertIsNone(data)
+                    self.assertFalse(report.generate_report(artifacts / "events.log", artifacts / "cases.md",
+                                                           artifacts / "matrix.md", artifacts / "results.json"))
+                    data = json.loads((artifacts / "results.json").read_text())
+                self.assertEqual(data["counts"]["NOT_RUN"], 108)
+
+    def test_cleanup_receipt_blocks_release_for_nodes_and_vault(self):
+        for mode in ("pass", "cleanup", "vault-cleanup"):
+            with self.subTest(mode=mode):
+                shutil.rmtree(self.directory / "artifacts", ignore_errors=True)
+                _, _, _ = self.lifecycle(mode)
+                path = self.directory / "artifacts/fleet-cleanup.state"
+                if mode == "pass":
+                    evidence.check_cleanup(path)
+                else:
+                    self.assertEqual(path.read_text().strip(), "FAIL")
+                    with self.assertRaises(ValueError):
+                        evidence.check_cleanup(path)
+        for state in ("PENDING", "RETAINED", "", "garbage"):
+            path.write_text(state)
+            with self.assertRaises(ValueError):
+                evidence.check_cleanup(path)
+        path.unlink()
+        with self.assertRaises(OSError):
+            evidence.check_cleanup(path)
+
+    def test_workflow_release_gate_leaves_fleet_quarantined(self):
+        workflow = (ROOT / ".github/workflows/rustfs-upgrade-test.yml").read_text()
+        gate = workflow.split("      - name: Release environment only after successful cleanup", 1)[1].split("      - name:", 1)[0]
+        block = gate.split("        run: |\n", 1)[1]
+        command = "\n".join(line[10:] for line in block.splitlines())
+        state = self.directory / "upgrade/fleet-cleanup.state"
+        state.parent.mkdir()
+        # Execute the actual workflow gate through the real transition function
+        # with node operations replaced by an in-memory quarantine fixture.
+        script = self.directory / "python3"
+        script.write_text(f'''#!{sys.executable}
+import json,os,sys
+from pathlib import Path
+sys.path.insert(0, {str(ROOT / 'scripts')!r})
+import functional_environment as fleet
+if sys.argv[1].endswith('functional_environment.py'):
+    calls=[]
+    try:
+        fleet.transition('finish','upgrade',os.environ,sys.argv[-1],execute=lambda *args:calls.append(args))
+    except ValueError:
+        Path(os.environ['TEST_GATE_CALLS']).write_text(json.dumps(calls))
+        raise SystemExit(1)
+    Path(os.environ['TEST_GATE_CALLS']).write_text(json.dumps(calls))
+else:
+    os.execv(sys.executable,[sys.executable,*sys.argv[1:]])
+''')
+        script.chmod(0o755)
+        for clean in ("PASS", "FAIL", "PENDING", None):
+            if clean is None:
+                state.unlink()
+            else:
+                state.write_text(clean)
+            result = self.bash(command, PATH=str(self.directory) + os.pathsep + os.environ["PATH"],
+                               FUNCTIONAL_ARTIFACTS_DIR=str(self.directory), CLEANUP_OUTCOME="success",
+                               GITHUB_RUN_ID="123", GITHUB_RUN_ATTEMPT="1", RUSTFS_NODES="vm000 vm001 vm002 vm003",
+                               FUNCTIONAL_ENVIRONMENT_RECEIPT=str(self.directory / "ownership.json"),
+                               TEST_GATE_CALLS=str(self.directory / "calls.json"))
+            receipt = json.loads((self.directory / "ownership.json").read_text())
+            calls = json.loads((self.directory / "calls.json").read_text())
+            if clean == "PASS":
+                self.assertEqual(result.returncode, 0, result.stderr)
+                self.assertEqual(receipt["state"], "clean")
+                self.assertEqual(receipt["completed_nodes"], ["vm000", "vm001", "vm002", "vm003"])
+            else:
+                self.assertNotEqual(result.returncode, 0)
+                self.assertEqual(receipt["state"], "quarantined")
+                self.assertEqual(calls, [])
+
     def test_report_legacy_context_and_missing_cleanup_evidence(self):
         log, cases, matrix = (self.directory / name for name in ("log", "cases.md", "matrix.md"))
         log.write_text("[INFO] == topology: single kms-backend: local ==\n--- UPG-101 case ---\n[PASS] UPG-101\n[UPG-TOPO] single local old new PASS=1 FAIL=0\n")
@@ -243,6 +340,40 @@ def test_report_legacy_context_and_missing_cleanup_evidence(self):
         log.write_text(log.read_text() + "[UPG-STATE] single local old new PASS\n")
         self.assertFalse(report.generate_report(log, cases, matrix))
 
+    def test_truncated_or_duplicate_matrix_never_reports_success(self):
+        artifacts = self.directory / "artifacts"
+        _, data, _ = self.lifecycle()
+        log = artifacts / "events.log"
+        original = log.read_text()
+        second = original.index("[INFO] == topology: single-multi")
+        last_state = original.rindex("[UPG-STATE]")
+        mutations = {
+            "later prepare": original[:second],
+            "last cleanup": original[:last_state],
+            "all missing states": "\n".join(line for line in original.splitlines() if not line.startswith("[UPG-STATE]")),
+            "missing diagnostic": "\n".join(line for line in original.splitlines() if not line.startswith("[UPG-DIAGNOSTICS]")),
+            "missing cleanup": "\n".join(line for line in original.splitlines() if not line.startswith("[UPG-CLEANUP]")),
+            "duplicate group": original + "[UPG-STATE] single-single local old new PASS\n",
+            "duplicate case": original + "--- UPG-101 duplicate ---\n[PASS] UPG-101\n",
+            "truncated plan": original[:original.index("[UPG-PLAN-END]")],
+        }
+        for label, text in mutations.items():
+            with self.subTest(label=label):
+                log.write_text(text)
+                self.assertFalse(report.generate_report(log, artifacts / "cases.md", artifacts / "matrix.md",
+                                                       artifacts / "results.json"))
+                data = json.loads((artifacts / "results.json").read_text())
+                self.assertEqual(data["status"], "FAIL")
+                self.assertGreaterEqual(len(data["cases"]), 108)
+        log.write_text(original)
+        self.assertTrue(report.generate_report(log, artifacts / "cases.md", artifacts / "matrix.md"))
+        # With no sidecar, only a completely bounded embedded plan is usable.
+        standalone = self.directory / "standalone.log"
+        standalone.write_text(original)
+        self.assertTrue(report.generate_report(standalone, artifacts / "cases.md", artifacts / "matrix.md"))
+        standalone.write_text(original.replace("[UPG-PLAN-END] 6", ""))
+        self.assertFalse(report.generate_report(standalone, artifacts / "cases.md", artifacts / "matrix.md"))
+
     def test_diagnostics_archive_is_sanitized_even_when_ssh_fails(self):
         golden = self.directory / "golden.tsv"
         golden.write_text("creds:sa\tsa-access-canary|sa-secret-canary\n")
@@ -258,6 +389,7 @@ def test_diagnostics_archive_is_sanitized_even_when_ssh_fails(self):
 ACCESS_KEY=fixture-access
 SECRET_KEY="$UPGRADE_TEST_TOKEN"
 DIAGNOSTICS_DONE=0
+DIAGNOSTICS_STATUS=NOT_RUN
 DRY_RUN=0
 CURRENT_TOPO=multi-multi
 SSH_OPTS=()
@@ -280,6 +412,64 @@ def test_diagnostics_archive_is_sanitized_even_when_ssh_fails(self):
                 self.assertNotIn(secret, text)
             self.assertIn("REDACTED", text)
 
+    def test_unknown_quoted_credentials_and_generic_tokens_are_redacted(self):
+        unknown = "UNKNOWN_NEW_CREDENTIAL_CANARY"
+        for key in ("secretKey", "secret_key", "SecretAccessKey", "token", "root_token", "sessionToken",
+                    "vaultToken", "password", "masterKey", "accessKeyId", "Authorization"):
+            for line in (f'journal prefix {{"{key}":"{unknown}"}}',
+                         f'journal prefix {{\\"{key}\\":\\"{unknown}\\"}}',
+                         f'{key}={unknown}', f"'{key}': '{unknown}'"):
+                with self.subTest(key=key, line=line):
+                    self.assertNotIn(unknown, evidence.redact(line))
+        raw = self.directory / "raw.log"
+        safe = self.directory / "safe.log"
+        raw.write_text(json.dumps({"journal": f'prefix {{"token":"{unknown}"}}',
+                                   "file_logs": [{"content": f'{{"secretKey":"{unknown}"}}'}]}))
+        result = subprocess.run([sys.executable, str(ROOT / "scripts/upgrade_evidence.py"), "sanitize",
+                                 "--input", str(raw), "--output", str(safe)], capture_output=True, text=True)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertNotIn(unknown, safe.read_text())
+
+    def test_interrupted_diagnostics_do_not_become_pass_on_exit_reentry(self):
+        source = SOURCE.removesuffix('main "$@"\n')
+        body = source + r'''
+SCRIPT_DIR="$TEST_ROOT"
+UPGRADE_PYTHON="$TEST_PYTHON"
+ARTIFACTS_DIR="$TEST_DIRECTORY/artifacts"
+mkdir "$ARTIFACTS_DIR"
+GROUP_DIR="$ARTIFACTS_DIR/current"
+mkdir "$GROUP_DIR"
+ARTIFACTS_READY=1
+: > "$ARTIFACTS_DIR/events.log"
+UPGRADE_GROUPS=(multi-multi/local)
+GROUP_ACTIVE=1
+CURRENT_TOPO=multi-multi
+KMS_BACKEND=local
+CURRENT_PHASE=CLEANUP
+CURRENT_RESULT=PASS
+DIAGNOSTICS_DONE=0
+DIAGNOSTICS_STATUS=NOT_RUN
+topology_nodes() { echo 'node1 node2'; }
+timeout() { shift; "$@"; }
+ssh() {
+  cat > /dev/null
+  printf 'first-node\n' >> "$TEST_DIRECTORY/trace"
+  kill -TERM "$$"
+}
+env_teardown() { printf 'cleanup\n' >> "$TEST_DIRECTORY/trace"; }
+record_upgrade_cleanup PENDING
+finish_upgrade_group
+'''
+        result = self.bash(body, TEST_DIRECTORY=str(self.directory), TEST_ROOT=str(ROOT), TEST_PYTHON=sys.executable,
+                           TMPDIR=str(self.directory))
+        self.assertNotEqual(result.returncode, 0)
+        log = (self.directory / "artifacts/events.log").read_text()
+        self.assertIn("[UPG-DIAGNOSTICS] multi-multi local FAIL", log)
+        self.assertNotIn("[UPG-DIAGNOSTICS] multi-multi local PASS", log)
+        self.assertIn("INTERRUPTED", log)
+        self.assertEqual((self.directory / "trace").read_text(), "first-node\ncleanup\n")
+        self.assertFalse(list(self.directory.glob("rustfs-upgrade.*")))
+
     def test_real_seed_verifier_compares_baseline_beyond_etags(self):
         body = "set -Eeuo pipefail\n" + "\n".join(function(name) for name in ("capture_seed_manifest", "verify_warp_bucket", "assert_aws_ok")) + r'''
 TEST_TMP="$TEST_DIRECTORY"
@@ -318,12 +508,15 @@ def test_real_seed_verifier_compares_baseline_beyond_etags(self):
         self.assertIn("content changed", result.stdout)
 
     def test_dry_run_skips_every_environment_boundary(self):
-        result, data, trace = self.lifecycle(extra=("--dry-run",))
+        log = self.directory / "prior.log"
+        log.write_text("prior test evidence\n")
+        result, data, trace = self.lifecycle(extra=("--dry-run", "--log-file", str(log)))
         self.assertEqual(result.returncode, 0, result.stderr)
         self.assertIsNone(data)
         self.assertEqual(trace, "")
         self.assertEqual(result.stdout.count("[DRY-RUN] UPG-"), 108)
         self.assertNotIn("[PASS]", result.stdout)
+        self.assertEqual(log.read_text(), "prior test evidence\n")
 
 
 @unittest.skipUnless(shutil.which("aws") and shutil.which("jq"), "real AWS CLI and jq required")