diff --git a/.github/workflows/environment-compatibility-tests.yml b/.github/workflows/environment-compatibility-tests.yml
index eb31d15..cc6e202 100644
--- a/.github/workflows/environment-compatibility-tests.yml
+++ b/.github/workflows/environment-compatibility-tests.yml
@@ -8,6 +8,10 @@ on:
       - rustfs-security-test.sh
       - rustfs-upgrade-test.sh
       - rustfs_performance_test.sh
+      - rustfs_performance_testing.sh
+      - scripts/performance_s3.py
+      - scripts/performance_delete_probe.py
+      - test_performance_s3.py
       - scripts/check_execution_environment.py
       - scripts/check_shell_syntax.py
       - scripts/prepare_table_smoke.py
@@ -27,6 +31,10 @@ on:
       - rustfs-security-test.sh
       - rustfs-upgrade-test.sh
       - rustfs_performance_test.sh
+      - rustfs_performance_testing.sh
+      - scripts/performance_s3.py
+      - scripts/performance_delete_probe.py
+      - test_performance_s3.py
       - scripts/check_execution_environment.py
       - scripts/check_shell_syntax.py
       - scripts/prepare_table_smoke.py
@@ -55,6 +63,8 @@ jobs:
           python3 -B scripts/check_shell_syntax.py --report "${RUNNER_TEMP}/shell-syntax.json"
           test -x rustfs-security-test.sh
           test -x rustfs-upgrade-test.sh
+      - name: Install S3 test client
+        run: python3 -m pip install 'boto3>=1.35,<2'
       - name: Test portable execution helpers
         run: >-
           python3 -B -m unittest -v
@@ -62,6 +72,7 @@ jobs:
           test_check_shell_syntax.py
           test_rustfs_table.py
           test_rustfs_performance_testing.py
+          test_performance_s3.py
           test_rustfs_replication.py
       - name: Test missing HOME
         run: env -u HOME bash test_rustfs_pool_expand.sh
diff --git a/.github/workflows/performance-script-tests.yml b/.github/workflows/performance-script-tests.yml
index 35cf6cf..eee518e 100644
--- a/.github/workflows/performance-script-tests.yml
+++ b/.github/workflows/performance-script-tests.yml
@@ -4,12 +4,20 @@ on:
   pull_request:
     paths:
       - rustfs_performance_testing.sh
+      - rustfs_performance_test.sh
+      - scripts/performance_s3.py
+      - scripts/performance_delete_probe.py
+      - test_performance_s3.py
       - test_rustfs_performance_testing.py
       - .github/workflows/performance-script-tests.yml
   push:
     branches: [main]
     paths:
       - rustfs_performance_testing.sh
+      - rustfs_performance_test.sh
+      - scripts/performance_s3.py
+      - scripts/performance_delete_probe.py
+      - test_performance_s3.py
       - test_rustfs_performance_testing.py
       - .github/workflows/performance-script-tests.yml
 
@@ -25,6 +33,10 @@ jobs:
         with:
           persist-credentials: false
       - name: Validate shell syntax
-        run: bash -n rustfs_performance_testing.sh
+        run: |
+          bash -n rustfs_performance_testing.sh
+          bash -n rustfs_performance_test.sh
+      - name: Install S3 test client
+        run: python3 -m pip install 'boto3>=1.35,<2'
       - name: Test benchmark result contracts without a server
-        run: python3 test_rustfs_performance_testing.py
+        run: python3 -B -m unittest -v test_rustfs_performance_testing.py test_performance_s3.py
diff --git a/.github/workflows/rustfs-performance-test.yml b/.github/workflows/rustfs-performance-test.yml
index 2759386..8d3eec8 100644
--- a/.github/workflows/rustfs-performance-test.yml
+++ b/.github/workflows/rustfs-performance-test.yml
@@ -155,6 +155,8 @@ jobs:
           uname -a
           jq --version
           warp --version || true
+          python3 -c 'import boto3; from botocore.config import Config'
+          command -v timeout
           df -h /data | tail -1
 
       - name: Admit clean shared environment
@@ -330,6 +332,11 @@ jobs:
             ${{ env.FUNCTIONAL_ARTIFACTS_DIR }}/results/get_*.txt
             ${{ env.FUNCTIONAL_ARTIFACTS_DIR }}/results/put_*.txt
             ${{ env.FUNCTIONAL_ARTIFACTS_DIR }}/results/mixed_*.txt
+            ${{ env.FUNCTIONAL_ARTIFACTS_DIR }}/results/*.benchdata*
+            ${{ env.FUNCTIONAL_ARTIFACTS_DIR }}/results/*.prepare.log
+            ${{ env.FUNCTIONAL_ARTIFACTS_DIR }}/results/*.cleanup.log
+            ${{ env.FUNCTIONAL_ARTIFACTS_DIR }}/results/*.s3.json
+            ${{ env.FUNCTIONAL_ARTIFACTS_DIR }}/results/*.s3-events.jsonl
           if-no-files-found: error
 
       - name: Reset test environment (after)
diff --git a/README.md b/README.md
index bae7b17..15e783f 100644
--- a/README.md
+++ b/README.md
@@ -26,6 +26,66 @@ port that does not conflict with the local package server or another service.
 All top-level suite entrypoints are directly executable; `bash <script>` also
 works when a local checkout or transport drops executable modes.
 
+## Performance benchmark lifecycle
+
+`rustfs_performance_testing.sh` needs Warp, GNU `timeout`, Python 3 and boto3
+(`python3 -m pip install 'boto3>=1.35,<2'`). It creates a different bucket for
+every selected round; `WARP_BUCKET` is now a name prefix. Warp runs with
+`--noclear`, and the harness owns preparation and cleanup. Credentials are
+passed through environment variables instead of Warp command-line arguments.
+
+`WARP_ROUND_TIMEOUT` (default 900 seconds) bounds Warp's fixture preparation
+and measured workload. `WARP_CLEANUP_TIMEOUT` (default 180 seconds) reserves a
+separate cleanup budget; `WARP_CLEANUP_ATTEMPTS` (default 3) bounds retries of
+retryable delete failures. The deployment wrapper exposes these as
+`RUSTFS_WARP_CLEANUP_TIMEOUT` and `RUSTFS_WARP_CLEANUP_ATTEMPTS`. Ordinary
+Warp failures and timeouts still run cleanup. An interrupted shell makes a
+best-effort cleanup attempt on exit; SIGKILL cannot run that handler. Keep
+the receipt if recovery is interrupted or incomplete.
+
+Each round retains its original `.txt`, native `.benchdata*`, `.prepare.log`,
+`.cleanup.log`, `.s3.json` receipt and `.s3-events.jsonl`. The events record
+HTTP status, request id, SDK retry count, RustFS error cause when supplied,
+and per-key delete errors. HTTP 200 with `DeleteResult/Errors` is a failure
+even though the HTTP request succeeded. `cleanup_http_503_attempts` counts
+HTTP 503 responses; `cleanup_error_entries` counts affected key failure
+entries across attempts, including keys affected by a failed request. These
+are different quantities. The event's `error_origin` distinguishes server
+per-key responses from request/protocol failures. The helper disables SDK
+retries; Warp's native measured-operation results do not prove that its SDK
+never retried an HTTP attempt.
+
+The round stays FAIL if cleanup retries encountered any error, even when
+cleanup ultimately succeeds with zero residue. Incomplete cleanup stops the
+matrix and marks later rounds NOT_RUN. `summary.tsv` contains only certified
+rounds, while `rounds.tsv` and `summary.md` include every selected round and
+its cleanup verdict. Delete error counters are saved after every attempt,
+before backoff or the next request. The sticky `cleanup_had_errors` field
+also retains deadline, malformed-response and interrupted-cleanup failures;
+successful recovery cannot clear it or certify the original round. Cleanup
+receipts carry `cleanup_in_progress` so recovery detects a process killed
+before its final checkpoint. Older receipts with a historical error code
+remain failed even if their final cleanup status is PASS.
+A new execution refuses to overwrite an existing
+`master.log`. `--parse-only` writes separate `.reparsed.*` reports; old outputs
+without a lifecycle marker remain readable, with unknown cleanup status.
+
+```bash
+python3 -B -m unittest -v test_rustfs_performance_testing.py test_performance_s3.py
+```
+
+These tests validate harness behavior, including real boto3 parsing against
+a local HTTP stub. They do not claim live RustFS coverage.
+
+For a separate bulk-delete investigation, `scripts/performance_delete_probe.py`
+compares batches of 1, 32, 128, 500 and 1000 objects at idle and under 32 GET
+workers. Supply each endpoint with `--endpoint http://node:9000` and a fresh
+`--result-dir`. It uses the same environment credentials as Warp and records
+each one-shot delete verdict before any recovery retry, UTC timestamps,
+per-key failures, request ids, GET results and residue before/after cleanup.
+Cleanup runs in `finally`; an incomplete cleanup aborts the remaining cases.
+The caller must hold the environment's test lock throughout this probe.
+
 ## Release validation migration
 
 `validation-cases.json` snapshots all 158 canonical cases from
diff --git a/rustfs_performance_test.sh b/rustfs_performance_test.sh
index d5c2e46..0e6b5e3 100755
--- a/rustfs_performance_test.sh
+++ b/rustfs_performance_test.sh
@@ -96,6 +96,8 @@ BENCH_SCRIPT="${RUSTFS_BENCH_SCRIPT:-${_BENCH_RESOLVED}}"
 RESULT_DIR="${RUSTFS_RESULT_DIR:-$(pwd)/warp-bench-results-$(date +%Y%m%d-%H%M%S)}"
 WARP_HOST="${RUSTFS_WARP_HOST:-rustfs-node1:9000,rustfs-node2:9000,rustfs-node3:9000,rustfs-node4:9000}"
 WARP_BUCKET="${RUSTFS_WARP_BUCKET:-warp-benchmark-bucket}"
+WARP_CLEANUP_TIMEOUT="${RUSTFS_WARP_CLEANUP_TIMEOUT:-180}"
+WARP_CLEANUP_ATTEMPTS="${RUSTFS_WARP_CLEANUP_ATTEMPTS:-3}"
 WARP_CONCURRENCY="${RUSTFS_WARP_CONCURRENCY:-64}"
 WARP_DURATION="${RUSTFS_WARP_DURATION:-5m}"
 WARP_GET_OBJECTS="${RUSTFS_WARP_GET_OBJECTS:-2500}"
@@ -265,6 +267,15 @@ preflight() {
   need_cmd ssh "openssh client"
   need_cmd curl "http client"
   need_cmd warp "warp benchmark tool (for step 5)"
+  local step
+  for step in "${SELECTED_STEPS[@]}"; do
+    if [[ "${step}" == 5 && "${DRY_RUN}" -eq 0 ]]; then
+      need_cmd python3 "verified benchmark cleanup"
+      need_cmd timeout "bounded benchmark cleanup"
+      python3 -c 'import boto3; from botocore.config import Config' \
+        || die "Python 3 with boto3 is required for verified benchmark cleanup"
+    fi
+  done
   if [ ! -x "${BENCH_SCRIPT}" ]; then
     die "benchmark script not found or not executable: ${BENCH_SCRIPT}"
   fi
@@ -480,6 +491,8 @@ step5_benchmark() {
   WARP_SLEEP_BETWEEN_ROUNDS="${WARP_SLEEP}" \
   WARP_ROUND_TIMEOUT="${WARP_ROUND_TIMEOUT}" \
   WARP_MAX_CONSECUTIVE_FAILURES="${WARP_MAX_CONSECUTIVE_FAILURES}" \
+  WARP_CLEANUP_TIMEOUT="${WARP_CLEANUP_TIMEOUT}" \
+  WARP_CLEANUP_ATTEMPTS="${WARP_CLEANUP_ATTEMPTS}" \
   WARP_METHODS="${WARP_METHODS}" \
   WARP_SIZES="${WARP_SIZES}" \
   WARP_RESULT_DIR="${RESULT_DIR}" \
diff --git a/rustfs_performance_testing.sh b/rustfs_performance_testing.sh
index 65d37eb..1dae7d4 100755
--- a/rustfs_performance_testing.sh
+++ b/rustfs_performance_testing.sh
@@ -11,7 +11,7 @@
 #   5) Result parsing: after each round, parse the warp output into
 #      summary.tsv (machine-readable) and summary.md (Markdown tables)
 #
-# Requirements: warp >= v1.6 (MinIO warp), bash, awk/sed/grep
+# Requirements: warp >= v1.6 (MinIO warp), bash, Python 3 + boto3, awk/sed/grep
 # Note: warp v1.6.1's put does not support --objects; handled automatically
 #       (the flag is only passed for get/mixed)
 #
@@ -21,6 +21,7 @@
 #   WARP_RESULT_DIR
 #   WARP_METHODS WARP_SIZES   # restrict methods/sizes (comma/space separated); empty = full run
 #   WARP_ROUND_TIMEOUT WARP_MAX_CONSECUTIVE_FAILURES
+#   WARP_CLEANUP_TIMEOUT WARP_CLEANUP_ATTEMPTS
 
 set -u -o pipefail
 
@@ -34,7 +35,15 @@ GET_OBJECTS="${WARP_GET_OBJECTS:-2500}"
 SLEEP_BETWEEN_ROUNDS="${WARP_SLEEP_BETWEEN_ROUNDS:-60}"
 ROUND_TIMEOUT="${WARP_ROUND_TIMEOUT:-900}"
 MAX_CONSECUTIVE_FAILURES="${WARP_MAX_CONSECUTIVE_FAILURES:-0}"
+CLEANUP_TIMEOUT="${WARP_CLEANUP_TIMEOUT:-180}"
+CLEANUP_ATTEMPTS="${WARP_CLEANUP_ATTEMPTS:-3}"
 RESULT_DIR="${WARP_RESULT_DIR:-$(pwd)/warp-bench-results-$(date +%Y%m%d-%H%M%S)}"
+SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
+S3_HELPER="${SCRIPT_DIR}/scripts/performance_s3.py"
+S3_ENDPOINT="${HOST%%,*}"
+[[ "${S3_ENDPOINT}" == *://* ]] || S3_ENDPOINT="http://${S3_ENDPOINT}"
+export WARP_ACCESS_KEY="${ACCESS_KEY}" WARP_SECRET_KEY="${SECRET_KEY}"
+umask 077
 
 if [ -n "${WARP_SIZES:-}" ] && [ "${WARP_SIZES}" != "all" ] && [ "${WARP_SIZES}" != "ALL" ]; then
   read -r -a SIZES <<<"${WARP_SIZES//,/ }"
@@ -51,6 +60,7 @@ ROUND=0
 PARSED_ROUNDS=0
 FAILED_ROUNDS=0
 CONSECUTIVE_FAILURES=0
+ABORT_REASON=aborted
 
 # --parse-only <result-dir>: parse an existing result dir (${method}_${size}.txt) without running
 # Only round files with a successful terminal execution status can be certified.
@@ -73,6 +83,24 @@ if [[ "${1:-}" != "--parse-only" ]] && ! command -v warp >/dev/null 2>&1; then
   echo "error: warp command not found; install MinIO warp first" >&2
   exit 1
 fi
+if [[ "${1:-}" != "--parse-only" ]]; then
+  if ! python3 -c 'import boto3; from botocore.config import Config' >/dev/null 2>&1; then
+    echo "error: Python 3 with boto3 is required for verified benchmark cleanup" >&2
+    exit 1
+  fi
+  if ! command -v timeout >/dev/null 2>&1; then
+    echo "error: GNU timeout is required for bounded benchmark and cleanup phases" >&2
+    exit 1
+  fi
+  if [[ ! "${CLEANUP_TIMEOUT}" =~ ^[1-9][0-9]*$ || ! "${CLEANUP_ATTEMPTS}" =~ ^[1-9][0-9]*$ ]]; then
+    echo "error: cleanup timeout and attempts must be positive integers" >&2
+    exit 1
+  fi
+  if [[ -f "${LOG_FILE}" ]]; then
+    echo "error: result directory already contains an execution; use a fresh directory or --parse-only" >&2
+    exit 1
+  fi
+fi
 
 log() {
   echo "$(date -u '+%Y-%m-%dT%H:%M:%SZ') $*" | tee -a "${LOG_FILE}"
@@ -102,8 +130,12 @@ record_round() {
     reason=killed_before_report
     [[ "${reported}" == 0 ]] || reason=killed_after_report
   fi
-  printf '%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n' \
-    "${method}" "${size}" "${verdict}" "${reason}" "${warp_rc}" "${capture_rc}" "${elapsed}" "${reported}" "${errors}" >> "${ROUNDS_TSV}"
+  local lifecycle=$'unknown\tunknown\tunknown\tunknown\tunknown\tunknown\tunknown\tunknown'
+  if [[ -f "${RESULT_DIR}/${method}_${size}.s3.json" ]]; then
+    lifecycle=$(python3 "${S3_HELPER}" fields --state "${RESULT_DIR}/${method}_${size}.s3.json" 2>/dev/null) || lifecycle=$'unknown\tunknown\tFAIL\tunknown\tunknown\tunknown\tunknown\tunknown'
+  fi
+  printf '%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n' \
+    "${method}" "${size}" "${verdict}" "${reason}" "${warp_rc}" "${capture_rc}" "${elapsed}" "${reported}" "${errors}" "${lifecycle}" >> "${ROUNDS_TSV}"
 }
 
 # Extract the raw Average / Reqs / TTFB lines of a section (GET/PUT/Total etc.)
@@ -151,6 +183,19 @@ parse_round() {
     log "PARSE_FAILED method=${method} size=${size} reason=logged_errors file=${file} (metrics retained in raw output)"
     return 1
   fi
+  if grep -q '^RustFS lifecycle schema: 1$' "$file"; then
+    local lifecycle bucket prepare cleanup cleanup_errors remaining http_503 request_failures had_cleanup_errors
+    lifecycle=$(python3 "${S3_HELPER}" fields --state "${RESULT_DIR}/${method}_${size}.s3.json" 2>/dev/null) || lifecycle=""
+    IFS=$'\t' read -r bucket prepare cleanup cleanup_errors remaining http_503 request_failures had_cleanup_errors <<< "${lifecycle}"
+    if [[ "${prepare}" != PASS || "${cleanup}" != PASS || "${remaining}" != 0 ]]; then
+      PARSE_FAILURE_REASON=cleanup_unverified
+      return 1
+    fi
+    if [[ "${cleanup_errors}" != 0 || "${http_503}" != 0 || "${request_failures}" != 0 || "${had_cleanup_errors}" != False ]]; then
+      PARSE_FAILURE_REASON=cleanup_errors
+      return 1
+    fi
+  fi
   local line
   if [[ "${method}" == "mixed" ]]; then
     local total get put
@@ -205,13 +250,15 @@ gen_summary_md() {
     if [[ -n "${OUTPUT_TAG}" ]]; then
       echo "- Reparsed raw outputs: elapsed time is unknown; configuration above is from this parser invocation. See the original rounds.tsv, master.log and summary files for execution evidence."
     fi
-    echo "- Timeout covers preparation, workload and teardown. Report presence only distinguishes timeout before/after a report; it does not prove the internal phase."
+    echo "- Each round owns a fresh bucket. The Warp phase includes its fixture preparation and workload; native benchdata records the measured workload. Setup and verified cleanup have separate logs."
+    echo "- Warp timeout: ${ROUND_TIMEOUT}s; cleanup has a separate ${CLEANUP_TIMEOUT}s recovery budget. Report presence alone does not identify the internal Warp phase."
+    echo "- Any cleanup error keeps the round failed even when retries remove the residue. An incomplete cleanup aborts the remaining matrix. Legacy results without a lifecycle receipt have unknown cleanup status."
     echo ""
     echo "## Every selected round"
     echo ""
-    echo "| Method | Size | Verdict | Reason | Warp exit | Capture exit | Elapsed seconds | Report present | Error lines |"
-    echo "|---|---|---|---|---|---|---|---|---|"
-    awk -F '\t' 'NR > 1 { printf "| %s | %s | %s | %s | %s | %s | %s | %s | %s |\n", $1,$2,$3,$4,$5,$6,$7,$8,$9 }' "${ROUNDS_TSV}"
+    echo "| Method | Size | Verdict | Reason | Warp exit | Capture exit | Elapsed seconds | Report present | Warp error lines | Bucket | Setup | Cleanup | Delete key failure entries | Remaining objects | Cleanup HTTP 503 attempts | Cleanup request failures | Cleanup had errors |"
+    echo "|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|"
+    awk -F '\t' 'NR > 1 { printf "| %s | %s | %s | %s | %s | %s | %s | %s | %s | %s | %s | %s | %s | %s | %s | %s | %s |\n", $1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17 }' "${ROUNDS_TSV}"
     echo ""
   } > "${SUMMARY_MD}"
 
@@ -247,7 +294,7 @@ gen_summary_md() {
 mkdir -p "${RESULT_DIR}" || exit 1
 log "CONFIG host=${HOST} bucket=${BUCKET} concurrency=${CONCURRENCY} duration=${DURATION} get_objects=${GET_OBJECTS} sleep_between_rounds=${SLEEP_BETWEEN_ROUNDS}s" || exit 1
 printf 'method\tsize\tthroughput\tobj_per_s\treq_avg\treq_p50\treq_p90\treq_p99\tttfb_avg\tttfb_p99\tttfb_worst\n' > "${SUMMARY_TSV}" || exit 1
-printf 'method\tsize\tverdict\treason\twarp_exit\tcapture_exit\telapsed_seconds\treport_present\terror_lines\n' > "${ROUNDS_TSV}" || exit 1
+printf 'method\tsize\tverdict\treason\twarp_exit\tcapture_exit\telapsed_seconds\treport_present\terror_lines\tbucket\tprepare\tcleanup\tcleanup_error_entries\tremaining_objects\tcleanup_http_503_attempts\tcleanup_request_failures\tcleanup_had_errors\n' > "${ROUNDS_TSV}" || exit 1
 
 if [[ ${TOTAL_ROUNDS} -eq 0 ]]; then
   log "NO_ROUNDS_SELECTED"
@@ -285,10 +332,32 @@ if [[ "${1:-}" == "--parse-only" ]]; then
   exit $?
 fi
 
+active_state=""
+active_events=""
+run_cleanup() {
+  timeout --signal=TERM --kill-after=5 "${CLEANUP_TIMEOUT}" python3 "${S3_HELPER}" cleanup --endpoint "${S3_ENDPOINT}" \
+    --state "${active_state}" --events "${active_events}" --budget "${CLEANUP_TIMEOUT}" \
+    --attempts "${CLEANUP_ATTEMPTS}" > "${active_state%.s3.json}.cleanup.log" 2>&1
+}
+cleanup_on_exit() {
+  local exit_status=$?
+  trap - EXIT INT TERM
+  if [[ -n "${active_state}" ]]; then
+    log "RECOVERY cleanup_on_exit receipt=${active_state}"
+    run_cleanup || log "RECOVERY_FAILED receipt=${active_state}"
+  fi
+  exit "${exit_status}"
+}
+trap cleanup_on_exit EXIT
+trap 'exit 130' INT
+trap 'exit 143' TERM
+
 for method in "${METHODS[@]}"; do
   for size in "${SIZES[@]}"; do
     ROUND=$((ROUND + 1))
     outfile="${RESULT_DIR}/${method}_${size}.txt"
+    state_file="${RESULT_DIR}/${method}_${size}.s3.json"
+    events_file="${RESULT_DIR}/${method}_${size}.s3-events.jsonl"
 
     log "START round=${ROUND}/${TOTAL_ROUNDS} method=${method} size=${size} concurrency=${CONCURRENCY} duration=${DURATION}"
 
@@ -297,17 +366,30 @@ for method in "${METHODS[@]}"; do
       extra_args=(--objects "${GET_OBJECTS}")
     fi
 
+    start_epoch=$(date +%s)
+    log "PHASE round=${ROUND} stage=setup"
+    active_state="${state_file}"
+    active_events="${events_file}"
+    if ! timeout --signal=TERM --kill-after=5 30 python3 "${S3_HELPER}" prepare --endpoint "${S3_ENDPOINT}" --prefix "${BUCKET}" \
+      --state "${state_file}" --events "${events_file}" --budget 30 > "${RESULT_DIR}/${method}_${size}.prepare.log" 2>&1; then
+      FAILED_ROUNDS=$((FAILED_ROUNDS + 1))
+      record_round "${method}" "${size}" "" FAIL setup_failed 125 0 "$(( $(date +%s) - start_epoch ))"
+      ABORT_REASON=setup_failed
+      log "ABORT round=${ROUND} reason=setup_failed"
+      break 2
+    fi
+    round_bucket=$(python3 "${S3_HELPER}" bucket --state "${state_file}") || exit 1
     warp_args=("${method}" \
       --host "${HOST}" \
-      --access-key "${ACCESS_KEY}" \
-      --secret-key "${SECRET_KEY}" \
-      --bucket "${BUCKET}" \
+      --bucket "${round_bucket}" \
+      --noclear \
+      --benchdata "${RESULT_DIR}/${method}_${size}.benchdata" \
       --concurrent "${CONCURRENCY}" \
       --duration "${DURATION}" \
       --obj.size "${size}" \
       ${extra_args[@]+"${extra_args[@]}"} \
       --no-color)
-    start_epoch=$(date +%s)
+    log "PHASE round=${ROUND} stage=warp bucket=${round_bucket}"
     if [[ "${ROUND_TIMEOUT}" =~ ^[1-9][0-9]*$ ]]; then
       timeout --signal=TERM --kill-after=30 "${ROUND_TIMEOUT}" \
         bash -c 'warp "$@"' bash "${warp_args[@]}" 2>&1 | tee "${outfile}"
@@ -317,12 +399,17 @@ for method in "${METHODS[@]}"; do
     round_status=("${PIPESTATUS[@]}")
     rc=${round_status[0]}
     capture_rc=${round_status[1]}
+    log "PHASE round=${ROUND} stage=cleanup bucket=${round_bucket}"
+    cleanup_rc=0
+    run_cleanup || cleanup_rc=$?
+    active_state=""
     end_epoch=$(date +%s)
-    printf '\nRustFS round status: warp=%s capture=%s\n' "${rc}" "${capture_rc}" >> "${outfile}" || capture_rc=1
+    printf '\nRustFS lifecycle schema: 1\nRustFS round status: warp=%s capture=%s\n' "${rc}" "${capture_rc}" >> "${outfile}" || capture_rc=1
     PARSE_FAILURE_REASON=warp_exit
     [[ ${capture_rc} -eq 0 ]] || PARSE_FAILURE_REASON=output_capture
+    [[ ${cleanup_rc} -eq 0 ]] || PARSE_FAILURE_REASON=cleanup_unverified
 
-    if [[ ${rc} -eq 0 && ${capture_rc} -eq 0 ]] && parse_round "${method}" "${size}" "${outfile}"; then
+    if [[ ${rc} -eq 0 && ${capture_rc} -eq 0 && ${cleanup_rc} -eq 0 ]] && parse_round "${method}" "${size}" "${outfile}"; then
       PARSED_ROUNDS=$((PARSED_ROUNDS + 1))
       CONSECUTIVE_FAILURES=0
       log "END round=${ROUND}/${TOTAL_ROUNDS} method=${method} size=${size} rc=${rc} elapsed=$((end_epoch - start_epoch))s parsed=ok"
@@ -332,6 +419,11 @@ for method in "${METHODS[@]}"; do
       CONSECUTIVE_FAILURES=$((CONSECUTIVE_FAILURES + 1))
       log "END round=${ROUND}/${TOTAL_ROUNDS} method=${method} size=${size} rc=${rc} capture_rc=${capture_rc} elapsed=$((end_epoch - start_epoch))s parsed=failed"
       record_round "${method}" "${size}" "${outfile}" FAIL "${PARSE_FAILURE_REASON}" "${rc}" "${capture_rc}" "$((end_epoch - start_epoch))"
+      if [[ ${cleanup_rc} -ne 0 ]]; then
+        log "ABORT round=${ROUND} reason=cleanup_incomplete receipt=${state_file}"
+        ABORT_REASON=cleanup_incomplete
+        break 2
+      fi
       if [[ "${MAX_CONSECUTIVE_FAILURES}" =~ ^[1-9][0-9]*$ ]] \
         && (( CONSECUTIVE_FAILURES >= MAX_CONSECUTIVE_FAILURES )); then
         log "ABORT consecutive_failures=${CONSECUTIVE_FAILURES} limit=${MAX_CONSECUTIVE_FAILURES}"
@@ -351,7 +443,7 @@ done
 for method in "${METHODS[@]}"; do
   for size in "${SIZES[@]}"; do
     if ! awk -F '\t' -v m="${method}" -v s="${size}" 'NR>1 && $1==m && $2==s {found=1} END {exit !found}' "${ROUNDS_TSV}"; then
-      record_round "${method}" "${size}" "" NOT_RUN aborted unknown unknown 0
+      record_round "${method}" "${size}" "" NOT_RUN "${ABORT_REASON}" unknown unknown 0
     fi
   done
 done
diff --git a/scripts/performance_delete_probe.py b/scripts/performance_delete_probe.py
new file mode 100644
index 0000000..ad47611
--- /dev/null
+++ b/scripts/performance_delete_probe.py
@@ -0,0 +1,181 @@
+#!/usr/bin/env python3
+"""Compare one-shot bulk deletes at idle and under GET load on explicit endpoints.
+
+The caller must hold its environment's exclusive test lock for the entire run.
+Only fresh receipt-owned buckets are mutated; nodes and configuration are untouched.
+"""
+
+from concurrent.futures import ThreadPoolExecutor
+import argparse
+import datetime
+import hashlib
+import json
+import os
+from pathlib import Path
+import signal
+import threading
+import time
+
+try:
+    from .performance_s3 import Session, RequestFailure, cleanup, delete_batch, prepare, remaining_objects, save
+except ImportError:
+    from performance_s3 import Session, RequestFailure, cleanup, delete_batch, prepare, remaining_objects, save
+
+
+def read_load(endpoint, events, bucket, key, expected_hash, stop, ready, deadline):
+    session = Session(endpoint, events, 10)
+    counts = {"endpoint": endpoint, "successes": 0, "failures": 0, "http_503_attempts": 0,
+              "first_success": False, "error_samples": []}
+    try:
+        client = session.client(5)
+        while not stop.is_set() and time.monotonic() < deadline:
+            try:
+                response = client.get_object(Bucket=bucket, Key=key)
+                body = response["Body"]
+                try:
+                    actual = hashlib.sha256(body.read()).hexdigest()
+                finally:
+                    body.close()
+                if actual != expected_hash:
+                    raise RequestFailure("ReadIntegrityMismatch")
+                counts["successes"] += 1
+                counts["first_success"] = True
+                ready.set()
+            except Exception as exc:
+                counts["failures"] += 1
+                response = getattr(exc, "response", {})
+                code = getattr(exc, "code", response.get("Error", {}).get("Code", type(exc).__name__))
+                counts["http_503_attempts"] += int(response.get("ResponseMetadata", {}).get("HTTPStatusCode") == 503)
+                if len(counts["error_samples"]) < 5:
+                    counts["error_samples"].append(code)
+                    session.record_response("probe_get", response, code)
+                stop.wait(0.1)
+    finally:
+        ready.set()
+    return counts
+
+
+def run_case(endpoints, endpoint, count, loaded, result_dir, workers=32, object_size=131072,
+             budget=300, cleanup_budget=180):
+    state = result_dir / "bucket.s3.json"
+    events = result_dir / "s3-events.jsonl"
+    result = {"endpoint": endpoint, "batch_size": count, "load": "get" if loaded else "idle",
+              "read_workers": workers if loaded else 0, "verdict": "FAIL", "cleanup": "NOT_RUN",
+              "started_at": datetime.datetime.now(datetime.timezone.utc).isoformat()}
+    session = Session(endpoint, events, budget)
+    stop = threading.Event()
+    futures, ready, pool = [], [], None
+    try:
+        document = prepare(session, state, "bulk-delete-probe")
+        bucket = document["bucket"]
+        result["bucket"] = bucket
+        payload = b"x" * object_size
+        for index in range(count):
+            session.call("put_object", Bucket=bucket, Key=f"delete/{index:04}", Body=payload)
+        if loaded:
+            session.call("put_object", Bucket=bucket, Key="read-source", Body=payload)
+            pool = ThreadPoolExecutor(max_workers=workers)
+            load_deadline = min(session.deadline, time.monotonic() + 90)
+            for index in range(workers):
+                event = threading.Event()
+                ready.append(event)
+                futures.append(pool.submit(read_load, endpoints[index % len(endpoints)], events, bucket,
+                                           "read-source", hashlib.sha256(payload).hexdigest(), stop, event, load_deadline))
+            warm_deadline = time.monotonic() + 10
+            for event in ready:
+                event.wait(max(0, warm_deadline - time.monotonic()))
+            result["load_started_at"] = datetime.datetime.now(datetime.timezone.utc).isoformat()
+            result["load_ready_workers"] = sum(event.is_set() and not future.done()
+                                                for event, future in zip(ready, futures))
+        keys = [f"delete/{index:04}" for index in range(count)]
+        http_before, failed_before = session.http_503, session.request_failures
+        started = time.monotonic()
+        result["delete_started_at"] = datetime.datetime.now(datetime.timezone.utc).isoformat()
+        # No cleanup retry is allowed to turn this diagnostic delete into a success.
+        pending, errors = delete_batch(session, bucket, keys, attempts=1)
+        result.update(delete_seconds=round(time.monotonic() - started, 3),
+                      delete_finished_at=datetime.datetime.now(datetime.timezone.utc).isoformat(),
+                      delete_error_entries=errors, failed_keys=pending,
+                      delete_http_503_attempts=session.http_503 - http_before,
+                      delete_request_failures=session.request_failures - failed_before)
+        stop.set()
+        reads = [future.result() for future in futures]
+        result["reads"] = reads
+        count_left, sample = remaining_objects(session, bucket)
+        result.update(remaining_objects_before_cleanup=count_left, remaining_sample=sample)
+        expected_residue = int(loaded)
+        if (not pending and errors == 0 and count_left == expected_residue
+                and (not loaded or result["load_ready_workers"] == workers)
+                and all(row["first_success"] and row["failures"] == 0 for row in reads)):
+            result["verdict"] = "PASS"
+    except Exception as exc:
+        result["error_code"] = getattr(exc, "code", type(exc).__name__)
+    finally:
+        stop.set()
+        if pool:
+            pool.shutdown(wait=True)
+        if state.exists():
+            try:
+                document = cleanup(Session(endpoint, events, cleanup_budget), state)
+                result.update(cleanup=document["cleanup"],
+                              remaining_objects_after_cleanup=document.get("remaining_objects"),
+                              cleanup_error_entries=document.get("cleanup_error_entries", 0),
+                              cleanup_http_503_attempts=document.get("cleanup_http_503_attempts", 0),
+                              cleanup_had_errors=document.get("cleanup_had_errors", False))
+                if (document["cleanup"] != "PASS" or document.get("cleanup_error_entries", 0)
+                        or document.get("cleanup_had_errors", False)):
+                    result["verdict"] = "FAIL"
+            except Exception as exc:
+                result.update(verdict="FAIL", cleanup="FAIL",
+                              cleanup_error_code=getattr(exc, "code", type(exc).__name__))
+        result["finished_at"] = datetime.datetime.now(datetime.timezone.utc).isoformat()
+        save(result_dir / "result.json", result)
+    return result
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--endpoint", action="append", required=True)
+    parser.add_argument("--result-dir", type=Path, required=True)
+    parser.add_argument("--batch-sizes", default="1,32,128,500,1000")
+    parser.add_argument("--read-workers", type=int, default=32)
+    parser.add_argument("--object-size", type=int, default=131072)
+    args = parser.parse_args()
+    os.umask(0o077)
+    try:
+        sizes = [int(value) for value in args.batch_sizes.split(",")]
+    except ValueError:
+        parser.error("batch sizes must be integers")
+    if (not sizes or any(not 1 <= size <= 1000 for size in sizes)
+            or len(sizes) != len(set(sizes)) or not 1 <= args.read_workers <= 64
+            or not 1 <= args.object_size <= 4194304
+            or any(not endpoint.startswith(("http://", "https://")) for endpoint in args.endpoint)):
+        parser.error("invalid endpoint, batch sizes, worker count or object size")
+    args.result_dir.mkdir(parents=True, exist_ok=False)
+    def on_term(*_):
+        raise KeyboardInterrupt()
+    signal.signal(signal.SIGTERM, on_term)
+    results, aborted = [], False
+    for index, endpoint in enumerate(args.endpoint):
+        for loaded in (False, True):
+            for count in sizes:
+                directory = args.result_dir / f"endpoint-{index + 1}-{'get' if loaded else 'idle'}-{count}"
+                directory.mkdir()
+                result = run_case(args.endpoint, endpoint, count, loaded, directory,
+                                  args.read_workers, args.object_size)
+                results.append(result)
+                print(json.dumps({key: result[key] for key in ("endpoint", "load", "batch_size", "verdict", "cleanup")}), flush=True)
+                if result["cleanup"] != "PASS":
+                    aborted = True
+                    break
+            if aborted:
+                break
+        if aborted:
+            break
+    save(args.result_dir / "summary.json", {"results": results, "aborted": aborted,
+                                            "expected_cases": len(args.endpoint) * len(sizes) * 2})
+    return int(aborted or any(result["verdict"] != "PASS" for result in results))
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/scripts/performance_s3.py b/scripts/performance_s3.py
new file mode 100644
index 0000000..f063399
--- /dev/null
+++ b/scripts/performance_s3.py
@@ -0,0 +1,307 @@
+#!/usr/bin/env python3
+"""Own one fresh benchmark bucket and verify bounded S3 cleanup."""
+
+from __future__ import annotations
+
+import argparse
+import datetime
+import json
+import math
+import os
+from pathlib import Path
+import re
+import time
+import uuid
+
+
+RETRYABLE = {"ServiceUnavailable", "SlowDown", "InternalError", "RequestTimeout",
+             "ReadTimeoutError", "ConnectTimeoutError", "EndpointConnectionError"}
+
+
+class RequestFailure(Exception):
+    def __init__(self, code):
+        self.code = code
+        super().__init__(code)
+
+
+def save(path, document):
+    temporary = path.with_suffix(path.suffix + ".tmp")
+    temporary.write_text(json.dumps(document, indent=2) + "\n")
+    temporary.replace(path)
+
+
+def new_bucket(prefix):
+    if not re.fullmatch(r"[a-z0-9][a-z0-9.-]{0,61}[a-z0-9]|[a-z0-9]", prefix):
+        raise ValueError("invalid bucket prefix")
+    if any(part in prefix for part in ("..", ".-", "-.")):
+        raise ValueError("invalid bucket prefix")
+    return prefix[:30].rstrip(".-") + "-" + uuid.uuid4().hex
+
+
+class Session:
+    def __init__(self, endpoint, events, budget, factory=None):
+        self.endpoint = endpoint
+        self.events = events
+        self.deadline = time.monotonic() + budget
+        self.factory = factory or self.client
+        self.s3 = None
+        self.http_503 = 0
+        self.request_failures = 0
+
+    def client(self, remaining):
+        import boto3
+        from botocore.config import Config
+        return boto3.client(
+            "s3", endpoint_url=self.endpoint, region_name="us-east-1",
+            aws_access_key_id=os.environ["WARP_ACCESS_KEY"],
+            aws_secret_access_key=os.environ["WARP_SECRET_KEY"],
+            aws_session_token=os.environ.get("WARP_SESSION_TOKEN") or None,
+            config=Config(s3={"addressing_style": "path"}, proxies={},
+                          connect_timeout=min(3, remaining), read_timeout=min(60, remaining),
+                          retries={"total_max_attempts": 1}),
+        )
+
+    def emit(self, **record):
+        record["time"] = datetime.datetime.now(datetime.timezone.utc).isoformat()
+        with self.events.open("a") as stream:
+            stream.write(json.dumps(record) + "\n")
+
+    def call(self, operation, **arguments):
+        remaining = self.deadline - time.monotonic()
+        if remaining <= 0:
+            raise RequestFailure("CleanupDeadlineExceeded")
+        try:
+            # Rebuild near the deadline so a call cannot retain a longer old timeout.
+            if self.s3 is None or remaining < 60:
+                self.s3 = self.factory(max(0.01, remaining))
+            response = getattr(self.s3, operation)(**arguments)
+        except Exception as exc:
+            response = getattr(exc, "response", {})
+            if not isinstance(response, dict):
+                response = {}
+            code = response.get("Error", {}).get("Code", type(exc).__name__)
+            self.request_failures += 1
+            self.record_response(operation, response, code)
+            raise RequestFailure(code) from exc
+        if not isinstance(response, dict):
+            raise RequestFailure("InvalidS3Response")
+        self.record_response(operation, response)
+        return response
+
+    def record_response(self, operation, response, code=None):
+        metadata = response.get("ResponseMetadata", {})
+        if not isinstance(metadata, dict):
+            metadata = {}
+        headers = metadata.get("HTTPHeaders", {})
+        if not isinstance(headers, dict):
+            headers = {}
+        status = metadata.get("HTTPStatusCode")
+        self.http_503 += int(status == 503)
+        self.emit(kind="s3_request", operation=operation, endpoint=self.endpoint,
+                  http_status=status, request_id=metadata.get("RequestId"),
+                  sdk_retries=metadata.get("RetryAttempts"), error_code=code,
+                  error_cause=headers.get("x-rustfs-error-cause"),
+                  retry_after=headers.get("retry-after"))
+
+
+def prepare(session, state, prefix):
+    bucket = new_bucket(prefix)
+    document = {"version": 1, "bucket": bucket, "endpoint": session.endpoint,
+                "created": False, "prepare": "FAIL", "cleanup": "NOT_RUN"}
+    save(state, document)
+    session.call("create_bucket", Bucket=bucket)
+    document.update(created=True, prepare="PASS")
+    save(state, document)
+    return document
+
+
+def page_keys(response):
+    contents = response.get("Contents", [])
+    if not isinstance(contents, list) or any(not isinstance(row, dict) or not isinstance(row.get("Key"), str) for row in contents):
+        raise RequestFailure("InvalidListResponse")
+    keys = [row["Key"] for row in contents]
+    if len(keys) != len(set(keys)) or (response.get("IsTruncated") and not keys):
+        raise RequestFailure("InvalidListResponse")
+    return keys
+
+
+def delete_batch(session, bucket, keys, attempts, on_attempt=None):
+    if (not 1 <= len(keys) <= 1000 or any(not isinstance(key, str) for key in keys)
+            or len(set(keys)) != len(keys) or attempts < 1):
+        raise ValueError("delete batch requires 1..1000 distinct keys and positive attempts")
+    pending = list(keys)
+    error_entries = 0
+    for attempt in range(1, attempts + 1):
+        origin = "per_object_response"
+        try:
+            response = session.call("delete_objects", Bucket=bucket,
+                                    Delete={"Objects": [{"Key": key} for key in pending], "Quiet": False})
+            deleted_rows, error_rows = response.get("Deleted", []), response.get("Errors", [])
+            if (not isinstance(deleted_rows, list) or not isinstance(error_rows, list)
+                    or any(not isinstance(row, dict) or not isinstance(row.get("Key"), str)
+                           for row in deleted_rows + error_rows)
+                    or any(not isinstance(row.get("Code"), str) for row in error_rows)):
+                raise RequestFailure("InvalidDeleteResponse")
+            deleted = [row["Key"] for row in deleted_rows]
+            errors = [{"Key": row["Key"], "Code": row["Code"]} for row in error_rows]
+            answered = deleted + [row["Key"] for row in errors]
+            if set(answered) != set(pending) or len(answered) != len(pending):
+                raise RequestFailure("IncompleteDeleteResponse")
+        except RequestFailure as exc:
+            origin = "request_or_protocol_failure"
+            errors = [{"Key": key, "Code": exc.code} for key in pending]
+        error_entries += len(errors)
+        if on_attempt:
+            # Persist before logging, backoff or the next possibly interrupted RPC.
+            on_attempt(len(errors))
+        session.emit(kind="delete_batch", bucket=bucket, attempt=attempt,
+                     requested=len(pending), error_entries=len(errors), errors=errors,
+                     error_origin=origin)
+        if not errors:
+            return [], error_entries
+        if any(row["Code"] not in RETRYABLE for row in errors):
+            return [row["Key"] for row in errors], error_entries
+        pending = [row["Key"] for row in errors]
+        if attempt < attempts:
+            time.sleep(min(0.25 * attempt, max(0, session.deadline - time.monotonic())))
+    return pending, error_entries
+
+
+def abort_uploads(session, bucket):
+    # An interrupted PUT may leave no listed object but still leave an upload.
+    while True:
+        response = session.call("list_multipart_uploads", Bucket=bucket, MaxUploads=1000)
+        uploads = response.get("Uploads", [])
+        if (not isinstance(uploads, list)
+                or any(not isinstance(row, dict) or not isinstance(row.get("Key"), str)
+                       or not isinstance(row.get("UploadId"), str) for row in uploads)):
+            raise RequestFailure("InvalidUploadListResponse")
+        if not uploads:
+            if response.get("IsTruncated"):
+                raise RequestFailure("InvalidUploadListResponse")
+            return
+        for upload in uploads:
+            session.call("abort_multipart_upload", Bucket=bucket, Key=upload["Key"], UploadId=upload["UploadId"])
+
+
+def remaining_objects(session, bucket):
+    count, sample, token, seen = 0, [], None, set()
+    while True:
+        arguments = {"Bucket": bucket, "MaxKeys": 1000}
+        if token:
+            arguments["ContinuationToken"] = token
+        response = session.call("list_objects_v2", **arguments)
+        keys = page_keys(response)
+        count += len(keys)
+        sample.extend(keys[:max(0, 20 - len(sample))])
+        if not response.get("IsTruncated"):
+            return count, sample
+        token = response.get("NextContinuationToken")
+        if not isinstance(token, str) or not token or token in seen:
+            raise RequestFailure("InvalidListContinuation")
+        seen.add(token)
+
+
+def read_state(state):
+    document = json.loads(state.read_text())
+    if (not isinstance(document, dict) or document.get("version") != 1
+            or not isinstance(document.get("endpoint"), str)
+            or not isinstance(document.get("bucket"), str)
+            or not re.fullmatch(r"[a-z0-9][a-z0-9.-]*-[a-f0-9]{32}", document["bucket"])
+            or len(document["bucket"]) > 63
+            or (document.get("prepare") == "PASS" and document.get("created") is not True)):
+        raise ValueError("invalid benchmark bucket receipt")
+    if any(type(document.get(key, False)) is not bool
+           for key in ("cleanup_had_errors", "cleanup_in_progress")):
+        raise ValueError("invalid cleanup history in benchmark bucket receipt")
+    # Older receipts may carry a historical error code even after successful recovery.
+    document["cleanup_had_errors"] = bool(document.get("cleanup_had_errors")
+                                           or document.get("cleanup_error_code")
+                                           or document.get("cleanup_in_progress")
+                                           or document.get("cleanup") == "FAIL")
+    return document
+
+
+def cleanup(session, state, attempts=3):
+    document = read_state(state)
+    bucket = document["bucket"]
+    if document.get("created") is not True or document.get("endpoint") != session.endpoint:
+        raise ValueError("cleanup requires a receipt for a freshly created benchmark bucket")
+    if document.get("cleanup") == "PASS":
+        return document
+    previous_http_503 = document.get("cleanup_http_503_attempts", 0)
+    previous_failures = document.get("cleanup_request_failures", 0)
+    initial_http_503, initial_failures = session.http_503, session.request_failures
+    document.update(cleanup="FAIL", cleanup_in_progress=True, remaining_objects=None,
+                    cleanup_error_entries=document.get("cleanup_error_entries", 0))
+    save(state, document)
+    started = time.monotonic()
+
+    def checkpoint(errors=0):
+        document["cleanup_error_entries"] += errors
+        document["cleanup_had_errors"] |= bool(errors)
+        document["cleanup_http_503_attempts"] = previous_http_503 + session.http_503 - initial_http_503
+        document["cleanup_request_failures"] = previous_failures + session.request_failures - initial_failures
+        save(state, document)
+
+    try:
+        while True:
+            response = session.call("list_objects_v2", Bucket=bucket, MaxKeys=1000)
+            keys = page_keys(response)
+            if not keys:
+                break
+            pending, _ = delete_batch(session, bucket, keys, attempts, on_attempt=checkpoint)
+            if pending:
+                break
+        count, sample = remaining_objects(session, bucket)
+        document.update(remaining_objects=count, remaining_sample=sample)
+        if count == 0:
+            abort_uploads(session, bucket)
+            session.call("delete_bucket", Bucket=bucket)
+            document["cleanup"] = "PASS"
+    except RequestFailure as exc:
+        document["cleanup_error_code"] = exc.code
+    finally:
+        document["cleanup_had_errors"] |= document["cleanup"] != "PASS"
+        document["cleanup_in_progress"] = False
+        document["cleanup_seconds"] = round(time.monotonic() - started, 3)
+        checkpoint()
+    return document
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("action", choices=("prepare", "cleanup", "bucket", "fields"))
+    parser.add_argument("--state", type=Path, required=True)
+    parser.add_argument("--events", type=Path)
+    parser.add_argument("--endpoint")
+    parser.add_argument("--prefix", default="warp-benchmark-bucket")
+    parser.add_argument("--budget", type=float, default=180)
+    parser.add_argument("--attempts", type=int, default=3)
+    args = parser.parse_args()
+    if args.action in {"bucket", "fields"}:
+        document = read_state(args.state)
+        if args.action == "bucket":
+            print(document["bucket"])
+        else:
+            print("\t".join(str(document.get(key, "unknown")) for key in
+                            ("bucket", "prepare", "cleanup", "cleanup_error_entries",
+                             "remaining_objects", "cleanup_http_503_attempts",
+                             "cleanup_request_failures", "cleanup_had_errors")))
+        return 0
+    if not math.isfinite(args.budget) or args.budget <= 0 or args.attempts < 1 or not args.endpoint or not args.events:
+        parser.error("positive budget/attempts, endpoint and events are required")
+    os.umask(0o077)
+    session = Session(args.endpoint, args.events, args.budget)
+    try:
+        document = (prepare(session, args.state, args.prefix) if args.action == "prepare"
+                    else cleanup(session, args.state, args.attempts))
+    except (RequestFailure, ValueError, OSError) as exc:
+        print(f"S3 {args.action} failed: {getattr(exc, 'code', type(exc).__name__)}", flush=True)
+        return 1
+    print(f"S3 {args.action}: {document[args.action]}; bucket={document['bucket']}", flush=True)
+    return int(document[args.action] != "PASS")
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/test_performance_s3.py b/test_performance_s3.py
new file mode 100644
index 0000000..c9da491
--- /dev/null
+++ b/test_performance_s3.py
@@ -0,0 +1,367 @@
+#!/usr/bin/env python3
+"""Cleanup oracles, including an actual HTTP 200 S3 DeleteResult error body."""
+
+from collections import deque
+import json
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+import unittest
+from unittest.mock import patch
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import xml.etree.ElementTree as ET
+
+from scripts.performance_s3 import (Session, RequestFailure, cleanup, delete_batch,
+                                    new_bucket, prepare, remaining_objects, save)
+
+
+class FakeS3:
+    def __init__(self, keys=()):
+        self.keys = set(keys)
+        self.batches = []
+        self.responses = deque()
+        self.uploads = [{"Key": "unfinished", "UploadId": "upload-1"}]
+        self.bucket_deleted = False
+
+    def create_bucket(self, **kwargs):
+        return {}
+
+    def put_object(self, **kwargs):
+        self.keys.add(kwargs["Key"])
+        return {}
+
+    def list_objects_v2(self, **kwargs):
+        offset = int(kwargs.get("ContinuationToken", "0"))
+        keys = sorted(self.keys)[offset:offset + 1000]
+        more = len(self.keys) > offset + len(keys)
+        return {"Contents": [{"Key": key} for key in keys], "IsTruncated": more,
+                "NextContinuationToken": str(offset + len(keys))}
+
+    def delete_objects(self, **kwargs):
+        keys = [row["Key"] for row in kwargs["Delete"]["Objects"]]
+        self.batches.append(keys)
+        response = self.responses.popleft() if self.responses else {"Deleted": [{"Key": key} for key in keys]}
+        if isinstance(response, Exception):
+            raise response
+        for row in response.get("Deleted", []):
+            if isinstance(row, dict):
+                self.keys.discard(row.get("Key"))
+        return response
+
+    def list_multipart_uploads(self, **kwargs):
+        return {"Uploads": list(self.uploads)}
+
+    def abort_multipart_upload(self, **kwargs):
+        self.uploads.clear()
+        return {}
+
+    def delete_bucket(self, **kwargs):
+        assert not self.keys and not self.uploads
+        self.bucket_deleted = True
+        return {}
+
+
+class CleanupTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        self.state = self.root / "state.json"
+        self.events = self.root / "events.jsonl"
+        self.fake = FakeS3()
+        self.session = Session("http://example.invalid", self.events, 30, lambda _: self.fake)
+        self.document = prepare(self.session, self.state, "test")
+        self.bucket = self.document["bucket"]
+
+    def records(self):
+        return [json.loads(line) for line in self.events.read_text().splitlines()]
+
+    def test_2500_objects_use_three_batches_and_abort_uploads_before_bucket_delete(self):
+        self.fake.keys = {f"key-{i:04}" for i in range(2500)}
+        document = cleanup(self.session, self.state)
+        self.assertEqual([len(batch) for batch in self.fake.batches], [1000, 1000, 500])
+        self.assertEqual(document["cleanup"], "PASS")
+        self.assertEqual(document["remaining_objects"], 0)
+        self.assertEqual(document["cleanup_error_entries"], 0)
+        self.assertTrue(self.fake.bucket_deleted)
+        self.assertEqual(document, json.loads(self.state.read_text()))
+
+    def test_retry_only_failed_keys_and_keep_error_evidence_after_recovery(self):
+        self.fake.keys = {"a", "b"}
+        self.fake.responses.append({"Deleted": [{"Key": "a"}],
+                                    "Errors": [{"Key": "b", "Code": "ServiceUnavailable"}],
+                                    "ResponseMetadata": {"HTTPStatusCode": 200}})
+        with patch("scripts.performance_s3.time.sleep"):
+            document = cleanup(self.session, self.state)
+        self.assertEqual(self.fake.batches, [["a", "b"], ["b"]])
+        self.assertEqual(document["cleanup"], "PASS")
+        self.assertEqual(document["cleanup_error_entries"], 1)
+        self.assertEqual(document["cleanup_http_503_attempts"], 0)
+        self.assertEqual(document["cleanup_request_failures"], 0)
+
+    def test_one_http_503_is_one_request_failure_and_has_1000_affected_keys(self):
+        failure = RuntimeError()
+        failure.response = {"Error": {"Code": "ServiceUnavailable"},
+                            "ResponseMetadata": {"HTTPStatusCode": 503, "RetryAttempts": 0,
+                                                 "RequestId": "test-id",
+                                                 "HTTPHeaders": {"x-rustfs-error-cause": "lock", "retry-after": "1"}}}
+        self.fake.keys = {f"key-{i:04}" for i in range(1000)}
+        self.fake.responses.append(failure)
+        with patch("scripts.performance_s3.time.sleep"):
+            document = cleanup(self.session, self.state)
+        self.assertEqual(document["cleanup_http_503_attempts"], 1)
+        self.assertEqual(document["cleanup_request_failures"], 1)
+        self.assertEqual(document["cleanup_error_entries"], 1000)
+        request = next(row for row in self.records() if row.get("http_status") == 503)
+        self.assertEqual(request["request_id"], "test-id")
+        self.assertEqual(request["error_cause"], "lock")
+        batch = next(row for row in self.records() if row["kind"] == "delete_batch")
+        self.assertEqual(batch["error_origin"], "request_or_protocol_failure")
+
+    def test_non_retryable_error_leaves_exact_residue_and_does_not_delete_bucket(self):
+        self.fake.keys = {"a", "b"}
+        self.fake.responses.append({"Deleted": [{"Key": "a"}], "Errors": [{"Key": "b", "Code": "AccessDenied"}]})
+        document = cleanup(self.session, self.state)
+        self.assertEqual(document["cleanup"], "FAIL")
+        self.assertEqual(document["remaining_objects"], 1)
+        self.assertEqual(document["remaining_sample"], ["b"])
+        self.assertEqual(len(self.fake.batches), 1)
+        self.assertFalse(self.fake.bucket_deleted)
+
+    def test_finite_retries_exhaust_without_clearing_or_hiding_errors(self):
+        self.fake.keys = {"a"}
+        self.fake.responses.extend([{"Errors": [{"Key": "a", "Code": "ServiceUnavailable"}]}] * 3)
+        with patch("scripts.performance_s3.time.sleep"):
+            document = cleanup(self.session, self.state)
+        self.assertEqual(document["cleanup_error_entries"], 3)
+        self.assertEqual(document["remaining_objects"], 1)
+        self.assertEqual(len(self.fake.batches), 3)
+
+    def test_missing_duplicate_foreign_or_malformed_delete_acknowledgments_fail_closed(self):
+        for response in ({}, {"Deleted": [{"Key": "a"}, {"Key": "a"}]},
+                         {"Deleted": [{"Key": "foreign"}]}, {"Errors": [{"Key": "a"}]},
+                         {"Deleted": None}, {"Deleted": [{}]}, {"Errors": "invalid"}):
+            with self.subTest(response=response):
+                self.fake.responses.append(response)
+                pending, errors = delete_batch(self.session, self.bucket, ["a"], 3)
+                self.assertEqual(pending, ["a"])
+                self.assertEqual(errors, 1)
+                self.assertEqual(len(self.fake.responses), 0)
+
+    def test_invalid_receipt_never_calls_delete_or_list(self):
+        for mutation in ({"created": False}, {"bucket": "existing-user-bucket"},
+                         {"endpoint": "http://elsewhere.invalid"}, {"version": 9}):
+            with self.subTest(mutation=mutation):
+                save(self.state, {**self.document, **mutation})
+                with self.assertRaises(ValueError):
+                    cleanup(self.session, self.state)
+                self.assertFalse(self.fake.bucket_deleted)
+                self.assertEqual(self.fake.batches, [])
+
+    def test_deadline_saves_failure_with_unknown_residue(self):
+        self.session.deadline = time.monotonic() - 1
+        document = cleanup(self.session, self.state)
+        self.assertEqual(document["cleanup"], "FAIL")
+        self.assertIsNone(document["remaining_objects"])
+        self.assertEqual(document["cleanup_error_code"], "CleanupDeadlineExceeded")
+
+    def test_resume_preserves_previous_error_counts(self):
+        self.fake.keys = {"a"}
+        self.fake.responses.append({"Errors": [{"Key": "a", "Code": "ServiceUnavailable"}]})
+        first = cleanup(self.session, self.state, attempts=1)
+        second = cleanup(Session(self.session.endpoint, self.events, 30, lambda _: self.fake), self.state)
+        self.assertEqual(first["cleanup"], "FAIL")
+        self.assertEqual(second["cleanup"], "PASS")
+        self.assertEqual(second["cleanup_error_entries"], 1)
+
+    def test_sigkill_during_second_attempt_preserves_first_attempt_errors_on_disk(self):
+        script = '''import os, signal, sys
+from pathlib import Path
+from scripts.performance_s3 import Session, cleanup
+class Client:
+    calls = 0
+    def list_objects_v2(self, **kwargs):
+        return {"Contents": [{"Key": "a"}], "IsTruncated": False}
+    def delete_objects(self, **kwargs):
+        self.calls += 1
+        if self.calls == 2:
+            os.kill(os.getpid(), signal.SIGKILL)
+        return {"Errors": [{"Key": "a", "Code": "ServiceUnavailable"}],
+                "ResponseMetadata": {"HTTPStatusCode": 200}}
+client = Client()
+cleanup(Session("http://example.invalid", Path(sys.argv[2]), 30, lambda _: client), Path(sys.argv[1]))
+'''
+        process = subprocess.run([sys.executable, "-B", "-c", script, str(self.state), str(self.events)],
+                                 cwd=Path(__file__).resolve().parent, capture_output=True, text=True, timeout=10)
+        self.assertEqual(process.returncode, -9, process.stderr)
+        interrupted = json.loads(self.state.read_text())
+        self.assertEqual(interrupted["cleanup_error_entries"], 1)
+        self.assertTrue(interrupted["cleanup_had_errors"])
+        self.assertTrue(interrupted["cleanup_in_progress"])
+        self.fake.keys = {"a"}
+        recovered = cleanup(self.session, self.state)
+        self.assertEqual(recovered["cleanup"], "PASS")
+        self.assertEqual(recovered["cleanup_error_entries"], 1)
+        self.assertTrue(recovered["cleanup_had_errors"])
+        self.assertFalse(recovered["cleanup_in_progress"])
+
+    def test_deadline_and_invalid_list_failures_remain_sticky_after_successful_recovery(self):
+        for failure in ("deadline", "invalid-list"):
+            with self.subTest(failure=failure):
+                save(self.state, self.document)
+                failed_session = Session(self.session.endpoint, self.events, 30, lambda _: FakeS3())
+                if failure == "deadline":
+                    failed_session.deadline = time.monotonic() - 1
+                else:
+                    fake = FakeS3()
+                    fake.list_objects_v2 = lambda **_: {"Contents": [{}]}
+                    failed_session.factory = lambda _: fake
+                failed = cleanup(failed_session, self.state)
+                self.assertTrue(failed["cleanup_had_errors"])
+                recovered = cleanup(self.session, self.state)
+                self.assertEqual(recovered["cleanup"], "PASS")
+                self.assertEqual(recovered["remaining_objects"], 0)
+                self.assertTrue(recovered["cleanup_had_errors"])
+                for counter in ("cleanup_error_entries", "cleanup_request_failures", "cleanup_http_503_attempts"):
+                    self.assertEqual(recovered[counter], 0)
+
+    def test_remaining_count_uses_validated_pagination_and_bounded_sample(self):
+        self.fake.keys = {f"key-{i:04}" for i in range(2500)}
+        count, sample = remaining_objects(self.session, self.bucket)
+        self.assertEqual(count, 2500)
+        self.assertEqual(len(sample), 20)
+        self.fake.list_objects_v2 = lambda **_: {"Contents": [{"Key": "a"}], "IsTruncated": True}
+        with self.assertRaisesRegex(RequestFailure, "InvalidListContinuation"):
+            remaining_objects(self.session, self.bucket)
+
+    def test_malformed_object_or_upload_lists_are_saved_as_failure(self):
+        for operation, response in (("list_objects_v2", {"Contents": [{}]}),
+                                    ("list_objects_v2", {"Contents": [], "IsTruncated": True}),
+                                    ("list_multipart_uploads", {"Uploads": [{}]})):
+            with self.subTest(operation=operation, response=response):
+                fake = FakeS3()
+                setattr(fake, operation, lambda **_: response)
+                save(self.state, self.document)
+                document = cleanup(Session(self.session.endpoint, self.events, 30, lambda _: fake), self.state)
+                self.assertEqual(document["cleanup"], "FAIL")
+                self.assertIn("Invalid", document["cleanup_error_code"])
+
+    def test_invalid_batches_and_prefixes_never_reach_s3(self):
+        for keys in ([], ["a"] * 2, list(map(str, range(1001))), [{}]):
+            with self.assertRaises(ValueError):
+                delete_batch(self.session, self.bucket, keys, 3)
+        for prefix in ("Bad", "-bad", "a..b", "a.-b", "a/b", ""):
+            with self.assertRaises(ValueError):
+                new_bucket(prefix)
+        self.assertLessEqual(len(new_bucket("a" * 63)), 63)
+
+    def test_probe_reports_original_delete_failure_even_when_finally_cleanup_recovers(self):
+        from scripts.performance_delete_probe import run_case
+        self.fake.responses.append({"Errors": [{"Key": "delete/0000", "Code": "ServiceUnavailable"}],
+                                    "ResponseMetadata": {"HTTPStatusCode": 200}})
+        def session_factory(endpoint, events, budget):
+            return Session(endpoint, events, budget, lambda _: self.fake)
+        with patch("scripts.performance_delete_probe.Session", side_effect=session_factory):
+            result = run_case([self.session.endpoint], self.session.endpoint, 1, False, self.root)
+        self.assertEqual(result["verdict"], "FAIL")
+        self.assertEqual(result["delete_error_entries"], 1)
+        self.assertEqual(result["delete_http_503_attempts"], 0)
+        self.assertEqual(result["remaining_objects_before_cleanup"], 1)
+        self.assertEqual(result["cleanup"], "PASS")
+        self.assertEqual(result["remaining_objects_after_cleanup"], 0)
+
+    def test_probe_cleanup_runs_after_setup_put_failure(self):
+        from scripts.performance_delete_probe import run_case
+        self.fake.put_object = lambda **_: (_ for _ in ()).throw(RequestFailure("PutFailed"))
+        with patch("scripts.performance_delete_probe.Session", side_effect=lambda endpoint, events, budget:
+                   Session(endpoint, events, budget, lambda _: self.fake)):
+            result = run_case([self.session.endpoint], self.session.endpoint, 1, False, self.root)
+        self.assertEqual(result["verdict"], "FAIL")
+        self.assertEqual(result["cleanup"], "PASS")
+        self.assertTrue(self.fake.bucket_deleted)
+
+    def test_successful_probe_is_failed_when_cleanup_raises(self):
+        from scripts.performance_delete_probe import run_case
+        with patch("scripts.performance_delete_probe.Session", side_effect=lambda endpoint, events, budget:
+                   Session(endpoint, events, budget, lambda _: self.fake)), \
+             patch("scripts.performance_delete_probe.cleanup", side_effect=OSError("unwritable receipt")):
+            result = run_case([self.session.endpoint], self.session.endpoint, 1, False, self.root)
+        self.assertEqual(result["delete_error_entries"], 0)
+        self.assertEqual(result["remaining_objects_before_cleanup"], 0)
+        self.assertEqual(result["cleanup"], "FAIL")
+        self.assertEqual(result["verdict"], "FAIL")
+        self.assertEqual(json.loads((self.root / "result.json").read_text())["verdict"], "FAIL")
+
+    def test_successful_probe_is_failed_when_cleanup_reports_prior_errors(self):
+        from scripts.performance_delete_probe import run_case
+        with patch("scripts.performance_delete_probe.Session", side_effect=lambda endpoint, events, budget:
+                   Session(endpoint, events, budget, lambda _: self.fake)), \
+             patch("scripts.performance_delete_probe.cleanup", return_value={"cleanup": "PASS", "remaining_objects": 0,
+                                                                           "cleanup_had_errors": True}):
+            result = run_case([self.session.endpoint], self.session.endpoint, 1, False, self.root)
+        self.assertEqual(result["delete_error_entries"], 0)
+        self.assertEqual(result["cleanup"], "PASS")
+        self.assertEqual(result["verdict"], "FAIL")
+
+
+class HttpDeleteTests(unittest.TestCase):
+    def test_boto3_distinguishes_http_200_xml_key_errors_from_http_503(self):
+        try:
+            import boto3  # noqa: F401
+        except ImportError:
+            self.skipTest("boto3 required for HTTP integration test")
+        replies = deque([
+            (200, '<DeleteResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">'
+             '<Deleted><Key>a</Key></Deleted><Error><Key>b</Key><Code>ServiceUnavailable</Code>'
+             '<Message>The service is unavailable. Please retry.</Message></Error></DeleteResult>'),
+            (200, '<DeleteResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">'
+             '<Deleted><Key>b</Key></Deleted></DeleteResult>'),
+            (503, '<Error><Code>ServiceUnavailable</Code><Message>Please retry</Message></Error>'),
+        ])
+        requests = []
+
+        class Handler(BaseHTTPRequestHandler):
+            def log_message(self, *args):
+                pass
+
+            def do_POST(self):
+                body = self.rfile.read(int(self.headers["Content-Length"]))
+                requests.append([element.text for element in ET.fromstring(body).iter()
+                                 if element.tag.split("}")[-1] == "Key"])
+                status, payload = replies.popleft()
+                self.send_response(status)
+                self.send_header("Content-Type", "application/xml")
+                self.send_header("Content-Length", str(len(payload.encode())))
+                self.send_header("x-amz-request-id", "http-test")
+                self.end_headers()
+                self.wfile.write(payload.encode())
+
+        with tempfile.TemporaryDirectory() as folder:
+            server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+            thread = threading.Thread(target=server.serve_forever, daemon=True)
+            thread.start()
+            try:
+                with patch.dict("os.environ", WARP_ACCESS_KEY="test", WARP_SECRET_KEY="test-secret"):
+                    session = Session(f"http://127.0.0.1:{server.server_port}", Path(folder) / "events.jsonl", 10)
+                    with patch("scripts.performance_s3.time.sleep"):
+                        pending, errors = delete_batch(session, "test-bucket", ["a", "b"], 3)
+                    self.assertEqual((pending, errors), ([], 1))
+                    self.assertEqual(session.http_503, 0)
+                    pending, errors = delete_batch(session, "test-bucket", ["c"], 1)
+                    self.assertEqual((pending, errors), (["c"], 1))
+                    self.assertEqual(session.http_503, 1)
+                    self.assertEqual(session.request_failures, 1)
+                    self.assertEqual(requests, [["a", "b"], ["b"], ["c"]])
+                    self.assertFalse(replies)  # SDK did not retry the 503.
+                    self.assertNotIn("test-secret", session.events.read_text())
+            finally:
+                server.shutdown()
+                server.server_close()
+                thread.join(timeout=2)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/test_rustfs_performance_testing.py b/test_rustfs_performance_testing.py
index cd4c6d4..656a9b5 100644
--- a/test_rustfs_performance_testing.py
+++ b/test_rustfs_performance_testing.py
@@ -2,11 +2,13 @@
 """Benchmark result contracts exercised with a local fake warp executable."""
 
 import csv
+import json
 import os
 from pathlib import Path
 import shutil
 import subprocess
 import tempfile
+import time
 import unittest
 
 
@@ -21,20 +23,26 @@ def setUp(self):
         self.results = self.root / "results"
         self.shims = self.root / "fake-tools.sh"
         self.shims.write_text(r'''warp() {
-  local method="$1" size="" objects=0 mode="${FAKE_WARP_MODE:-success}" omit="${FAKE_WARP_OMIT:-}"
+  local method="$1" size="" bucket="" noclear=0 objects=0 mode="${FAKE_WARP_MODE:-success}" omit="${FAKE_WARP_OMIT:-}"
   shift
   while [[ $# -gt 0 ]]; do
     case "$1" in
       --obj.size) size="$2"; shift;;
       --objects) objects=1; shift;;
+      --bucket) bucket="$2"; shift;;
+      --noclear) noclear=1;;
+      --access-key|--secret-key) return 22;;
     esac
     shift
   done
   printf '%s %s\n' "$method" "$size" >> "$FAKE_WARP_CALLS"
+  printf '%s\n' "$bucket" >> "$FAKE_WARP_BUCKETS"
+  [[ "$noclear" == 1 ]] || return 23
   [[ "$method" == put && "$objects" == 0 || "$method" != put && "$objects" == 1 ]] || return 21
   printf 'diagnostic %s %s\n' "$method" "$size" >&2
   if [[ "$mode" == fail || "$mode" == mixed-failure && "$method" == put ]]; then return 42; fi
   if [[ "$mode" == hang ]]; then sleep 30; fi
+  if [[ "$mode" == brief-hang ]]; then sleep 1; fi
   local sections=() section text
   case "$method" in
     get) sections=(GET);;
@@ -59,11 +67,51 @@ def setUp(self):
   if [[ "$mode" == hang-after-report ]]; then sleep 30; fi
 }
 ''')
+        # The real helper runs here, but its S3 client cannot reach a network.
+        (self.root / "boto3.py").write_text('''import os
+import time
+from pathlib import Path
+class Client:
+    def __init__(self):
+        self.keys = ["leftover"] if os.environ.get("FAKE_S3_MODE") == "cleanup-recovered" else []
+        self.deletes = 0
+    def create_bucket(self, **kw):
+        if os.environ.get("FAKE_S3_MODE") == "setup-failure": raise RuntimeError("blocked")
+        if os.environ.get("FAKE_S3_MODE") == "setup-wait":
+            Path(os.environ["FAKE_S3_SETUP_STARTED"]).touch()
+            deadline = time.monotonic() + 5
+            while not Path(os.environ["FAKE_S3_SETUP_RELEASE"]).exists() and time.monotonic() < deadline:
+                time.sleep(0.01)
+        return {"ResponseMetadata": {"HTTPStatusCode": 200, "RetryAttempts": 0}}
+    def list_objects_v2(self, **kw):
+        return {"Contents": [{"Key": key} for key in self.keys], "IsTruncated": False, "ResponseMetadata": {"HTTPStatusCode": 200}}
+    def delete_objects(self, **kw):
+        self.deletes += 1
+        if self.deletes == 1:
+            return {"Errors": [{"Key": "leftover", "Code": "ServiceUnavailable"}], "ResponseMetadata": {"HTTPStatusCode": 200}}
+        self.keys = []
+        return {"Deleted": [{"Key": "leftover"}], "ResponseMetadata": {"HTTPStatusCode": 200}}
+    def list_multipart_uploads(self, **kw):
+        return {"Uploads": [], "ResponseMetadata": {"HTTPStatusCode": 200}}
+    def delete_bucket(self, **kw):
+        if os.environ.get("FAKE_S3_MODE") == "cleanup-failure": raise RuntimeError("blocked")
+        return {"ResponseMetadata": {"HTTPStatusCode": 204}}
+def client(*args, **kwargs): return Client()
+''')
+        (self.root / "botocore").mkdir()
+        (self.root / "botocore" / "__init__.py").write_text("")
+        (self.root / "botocore" / "config.py").write_text(
+            "class Config:\n    def __init__(self, **kwargs): pass\n")
         self.calls = self.root / "calls"
-        self.env = dict(os.environ, BASH_ENV=str(self.shims), FAKE_WARP_CALLS=str(self.calls), WARP_RESULT_DIR=str(self.results), WARP_HOST="127.0.0.1:1", WARP_ACCESS_KEY="test-access", WARP_SECRET_KEY="test-secret", WARP_METHODS="get,put,mixed", WARP_SIZES="1KiB,4KiB", WARP_SLEEP_BETWEEN_ROUNDS="0", WARP_DURATION="1s")
+        self.buckets = self.root / "buckets"
+        self.env = dict(os.environ, BASH_ENV=str(self.shims), PYTHONPATH=str(self.root), FAKE_WARP_BUCKETS=str(self.buckets), FAKE_WARP_CALLS=str(self.calls), WARP_RESULT_DIR=str(self.results), WARP_HOST="127.0.0.1:1", WARP_ACCESS_KEY="test-access", WARP_SECRET_KEY="test-secret", WARP_METHODS="get,put,mixed", WARP_SIZES="1KiB,4KiB", WARP_SLEEP_BETWEEN_ROUNDS="0", WARP_DURATION="1s")
         self.script = Path(__file__).with_name("rustfs_performance_testing.sh")
 
     def run_benchmark(self, mode="success", **overrides):
+        if self.results.exists():
+            shutil.rmtree(self.results)
+        self.calls.unlink(missing_ok=True)
+        self.buckets.unlink(missing_ok=True)
         result = subprocess.run(["bash", str(self.script)], cwd=self.root, env=dict(self.env, FAKE_WARP_MODE=mode, **overrides), capture_output=True, text=True)
         with (self.results / "summary.tsv").open() as stream:
             rows = list(csv.DictReader(stream, delimiter="\t"))
@@ -83,6 +131,12 @@ def test_success_keeps_selected_matrix_and_optional_ttfb(self):
         self.assertEqual(len(rows), 6)
         self.assert_diagnostics(6)
         self.assertIn("parsed=6", (self.results / "master.log").read_text())
+        buckets = self.buckets.read_text().splitlines()
+        self.assertEqual(len(set(buckets)), 6)
+        self.assertTrue(all(bucket.startswith("warp-benchmark-bucket-") for bucket in buckets))
+        self.assertTrue(all(row["cleanup"] == "PASS" and row["remaining_objects"] == "0"
+                            for row in self.round_evidence()))
+        self.assertNotIn("test-secret", "".join(path.read_text() for path in self.results.iterdir()))
 
     def test_all_failed_rounds_return_failure_after_collecting_every_round(self):
         result, rows = self.run_benchmark("fail")
@@ -143,6 +197,129 @@ def test_zero_exit_with_service_errors_retains_metrics_but_is_not_a_pass(self):
         self.assertEqual(evidence["error_lines"], "1")
         self.assertIn("Average:", (self.results / "get_1KiB.txt").read_text())
 
+    def test_incomplete_cleanup_aborts_without_certifying_or_reusing_the_bucket(self):
+        result, rows = self.run_benchmark(FAKE_S3_MODE="cleanup-failure")
+        self.assertNotEqual(result.returncode, 0)
+        self.assertEqual(rows, [])
+        self.assertEqual(len(self.calls.read_text().splitlines()), 1)
+        evidence = self.round_evidence()
+        self.assertEqual(evidence[0]["cleanup"], "FAIL")
+        self.assertEqual(evidence[0]["reason"], "cleanup_unverified")
+        self.assertTrue(all(row["verdict"] == "NOT_RUN" and row["reason"] == "cleanup_incomplete"
+                            for row in evidence[1:]))
+
+    def test_setup_failure_never_runs_warp(self):
+        result, rows = self.run_benchmark(FAKE_S3_MODE="setup-failure")
+        self.assertNotEqual(result.returncode, 0)
+        self.assertEqual(rows, [])
+        self.assertFalse(self.calls.exists())
+        self.assertEqual(self.round_evidence()[0]["reason"], "setup_failed")
+
+    def test_recovered_cleanup_errors_still_fail_but_do_not_abort_the_matrix(self):
+        result, rows = self.run_benchmark(FAKE_S3_MODE="cleanup-recovered")
+        self.assertNotEqual(result.returncode, 0)
+        self.assertEqual(rows, [])
+        self.assertEqual(len(self.calls.read_text().splitlines()), 6)
+        for row in self.round_evidence():
+            self.assertEqual(row["verdict"], "FAIL")
+            self.assertEqual(row["reason"], "cleanup_errors")
+            self.assertEqual(row["cleanup"], "PASS")
+            self.assertEqual(row["remaining_objects"], "0")
+            self.assertEqual(row["cleanup_error_entries"], "1")
+            self.assertEqual(row["cleanup_http_503_attempts"], "0")
+
+    def test_legacy_output_keeps_unknown_cleanup_status_when_reparsed(self):
+        self.run_benchmark(WARP_METHODS="get", WARP_SIZES="1KiB")
+        output = self.results / "get_1KiB.txt"
+        output.write_text(output.read_text().replace("RustFS lifecycle schema: 1\n", ""))
+        (self.results / "get_1KiB.s3.json").unlink()
+        result = subprocess.run(["bash", str(self.script), "--parse-only", str(self.results)],
+                                env={**self.env, "WARP_METHODS": "get", "WARP_SIZES": "1KiB"},
+                                capture_output=True, text=True)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(self.round_evidence("rounds.reparsed.tsv")[0]["cleanup"], "unknown")
+
+    def test_term_after_bucket_creation_runs_exit_cleanup_and_cannot_be_certified(self):
+        process = subprocess.Popen(["bash", str(self.script)], env={**self.env,
+                                   "FAKE_WARP_MODE": "brief-hang", "WARP_METHODS": "get", "WARP_SIZES": "1KiB"},
+                                   stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+        try:
+            deadline = time.monotonic() + 5
+            while not self.calls.exists() and process.poll() is None and time.monotonic() < deadline:
+                time.sleep(0.01)
+            self.assertTrue(self.calls.exists())
+            process.terminate()
+            output, errors = process.communicate(timeout=10)
+            self.assertEqual(process.returncode, 143, output + errors)
+            receipt = json.loads((self.results / "get_1KiB.s3.json").read_text())
+            self.assertEqual(receipt["cleanup"], "PASS")
+            self.assertEqual(receipt["remaining_objects"], 0)
+            self.assertIn("RECOVERY cleanup_on_exit", output)
+            self.assertNotIn("RustFS round status: warp=0 capture=0", (self.results / "get_1KiB.txt").read_text())
+        finally:
+            if process.poll() is None:
+                process.kill()
+                process.communicate()
+
+    def test_term_during_preparation_cleans_the_bucket_before_warp_can_start(self):
+        started, release = self.root / "setup-started", self.root / "setup-release"
+        process = subprocess.Popen(["bash", str(self.script)], env={**self.env,
+                                   "FAKE_S3_MODE": "setup-wait", "WARP_METHODS": "get", "WARP_SIZES": "1KiB",
+                                   "FAKE_S3_SETUP_STARTED": str(started), "FAKE_S3_SETUP_RELEASE": str(release)},
+                                   stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+        try:
+            deadline = time.monotonic() + 5
+            while not started.exists() and process.poll() is None and time.monotonic() < deadline:
+                time.sleep(0.01)
+            self.assertTrue(started.exists())
+            process.terminate()
+            release.touch()
+            output, errors = process.communicate(timeout=10)
+            self.assertEqual(process.returncode, 143, output + errors)
+            self.assertFalse(self.calls.exists())
+            receipt = json.loads((self.results / "get_1KiB.s3.json").read_text())
+            self.assertEqual(receipt["prepare"], "PASS")
+            self.assertEqual(receipt["cleanup"], "PASS")
+            self.assertEqual(receipt["remaining_objects"], 0)
+            self.assertIn("RECOVERY cleanup_on_exit", output)
+        finally:
+            release.touch()
+            if process.poll() is None:
+                process.kill()
+                process.communicate()
+
+    def test_reparse_requires_the_new_cleanup_receipt_and_preserves_original_files(self):
+        self.run_benchmark(WARP_METHODS="get", WARP_SIZES="1KiB")
+        state = self.results / "get_1KiB.s3.json"
+        for mutation in ("errors", "missing", "residue", "history", "legacy-history", "interrupted", "invalid-history"):
+            with self.subTest(mutation=mutation):
+                document = json.loads(state.read_text()) if state.exists() else original
+                original = dict(document)
+                if mutation == "missing":
+                    state.unlink()
+                else:
+                    changes = {"errors": {"cleanup_error_entries": 1}, "residue": {"remaining_objects": 1},
+                               "history": {"cleanup_had_errors": True},
+                               "legacy-history": {"cleanup_error_code": "InvalidListResponse"},
+                               "interrupted": {"cleanup_in_progress": True},
+                               "invalid-history": {"cleanup_had_errors": "False"}}
+                    document.update(changes[mutation])
+                    if mutation == "legacy-history":
+                        document.pop("cleanup_had_errors", None)
+                    state.write_text(json.dumps(document))
+                parsed = subprocess.run(["bash", str(self.script), "--parse-only", str(self.results)],
+                                        env={**self.env, "WARP_METHODS": "get", "WARP_SIZES": "1KiB"},
+                                        capture_output=True, text=True)
+                self.assertNotEqual(parsed.returncode, 0)
+                state.write_text(json.dumps(original))
+
+    def test_a_new_execution_cannot_overwrite_an_existing_report(self):
+        self.run_benchmark(WARP_METHODS="get", WARP_SIZES="1KiB")
+        original = {path.name: path.read_bytes() for path in self.results.iterdir()}
+        result = subprocess.run(["bash", str(self.script)], env=self.env, capture_output=True, text=True)
+        self.assertNotEqual(result.returncode, 0)
+        self.assertEqual(original, {name: (self.results / name).read_bytes() for name in original})
+
     def test_timeout_before_and_after_report_preserve_distinct_evidence(self):
         for mode, reason in (("hang", "timeout_before_report"), ("hang-after-report", "timeout_after_report")):
             with self.subTest(mode=mode):