ci(validation): dispatch isolated packages through the driver

test(validation): verify isolated package publication failures
test(validation): enforce private candidate manifest schema
2026-10-27 22:37:13 +00:00 · 2026-10-12 12:29:11 +08:00 · 2026-10-12 12:22:53 +08:00 · 2026-10-12 12:18:15 +08:00 · 2026-10-12 12:06:59 +08:00 · 2026-10-12 11:36:29 +08:00
64 changed files with 2819 additions and 929 deletions
@@ -92,6 +92,12 @@ setup = 'ecstore-large-stack'
 filter = 'package(rustfs-ecstore) | package(rustfs-s3select-api) | package(rustfs-scanner) | (package(rustfs) & test(/^(app::multipart_usecase::tests::concurrent_completions_share_durable_bucket_quota_reservations|app::object::delete::tests::compressed_delete_requests_update_observed_usage_without_releasing_quota_floor|app::object::internal_put::tests::internal_multipart_roundtrip_completes_and_abort_leaves_nothing|app::object::restore::tests::execute_restore_object_maps_failures_to_typed_s3_errors|storage::access::tests::(delete_object_access_captures_authorized_bucket_incarnation|copy_operations_reject_recreated_source_bucket_after_authorization|request_slot_keeps_bucket_policy_bound_to_its_store))$/))'
 setup = 'ecstore-base-stack'

+# These handler/store fixtures and the relocated-pool GET compose deep storage
+# futures in libtest threads; on Linux they overflow the default 2 MiB stack.
+[[profile.default.scripts]]
+filter = 'package(rustfs) & (test(/^admin::handlers::table_catalog::tests::/) | test(/^table_catalog::store::tests::/) | test(=app::object::get::tests::execute_get_object_resumes_from_relocated_pool_without_splicing_body))'
+setup = 'ecstore-base-stack'
+
 [[profile.default.scripts]]
 filter = 'binary(lifecycle_integration_test) | (package(rustfs) & test(/^app::lifecycle_transition_api_test::/))'
 setup = 'lifecycle-large-stack'
@@ -337,6 +343,10 @@ setup = 'ecstore-large-stack'
 filter = 'package(rustfs-ecstore) | package(rustfs-s3select-api) | package(rustfs-scanner) | (package(rustfs) & test(/^(app::multipart_usecase::tests::concurrent_completions_share_durable_bucket_quota_reservations|app::object::delete::tests::compressed_delete_requests_update_observed_usage_without_releasing_quota_floor|app::object::internal_put::tests::internal_multipart_roundtrip_completes_and_abort_leaves_nothing|app::object::restore::tests::execute_restore_object_maps_failures_to_typed_s3_errors|storage::access::tests::(delete_object_access_captures_authorized_bucket_incarnation|copy_operations_reject_recreated_source_bucket_after_authorization|request_slot_keeps_bucket_policy_bound_to_its_store))$/))'
 setup = 'ecstore-base-stack'

+[[profile.ci.scripts]]
+filter = 'package(rustfs) & (test(/^admin::handlers::table_catalog::tests::/) | test(/^table_catalog::store::tests::/) | test(=app::object::get::tests::execute_get_object_resumes_from_relocated_pool_without_splicing_body))'
+setup = 'ecstore-base-stack'
+
 [[profile.ci.scripts]]
 filter = 'binary(lifecycle_integration_test) | (package(rustfs) & test(/^app::lifecycle_transition_api_test::/))'
 setup = 'lifecycle-large-stack'
@@ -440,6 +440,17 @@ jobs:
            fi
          done

+      - name: Verify macOS binary has no dynamic liblzma dependency
+        if: matrix.platform == 'macos'
+        shell: bash
+        run: |
+          binary="target/${{ matrix.target }}/release/rustfs"
+          if otool -L "$binary" | grep -E 'liblzma.*\.dylib'; then
+            echo "macOS release binary must not depend on a dynamic liblzma library" >&2
+            exit 1
+          fi
+          "$binary" --version
+
      - name: Generate the CPU symbol catalog
        if: endsWith(matrix.target, '-linux-gnu') && matrix.features == 'pyroscope'
        shell: bash
@@ -20,13 +20,65 @@ on:
  workflow_dispatch:
    inputs:
      build_run_id:
-        description: Successful nightly build run on main
+        description: Successful nightly or validation package run; empty for package-only build
        type: string
-        required: true
+        required: false
+        default: ''
      build_run_attempt:
        description: Exact successful build attempt
        type: string
-        required: true
+        required: false
+        default: ''
+      validation_package_sha:
+        description: Exact source SHA for an isolated package-only build
+        type: string
+        required: false
+        default: ''
+      validation_case:
+        description: Canonical case ID, environment.inspect for read-only resources, or empty for functional chain
+        type: string
+        required: false
+        default: ''
+      auto_testing_sha:
+        description: Exact auto-testing commit for a validation case
+        type: string
+        required: false
+        default: ''
+      validation_tools_only:
+        description: Build candidate-bound validation tools without running a fleet case
+        type: boolean
+        required: false
+        default: false
+      validation_native_tools:
+        description: Include exact-source native regressions in a tools-only build
+        type: boolean
+        required: false
+        default: false
+      validation_shared_tools:
+        description: Include the controlled C10 shared-reference encoder in a tools-only build
+        type: boolean
+        required: false
+        default: false
+      validation_manager_tools:
+        description: Include exact checkpoint/MRF/incarnation regressions in a tools-only build
+        type: boolean
+        required: false
+        default: false
+      validation_cleanup_run_id:
+        description: Failed validation run whose supported retained environment needs evidence-based finalization
+        type: string
+        required: false
+        default: ''
+      validation_tools_run_id:
+        description: Successful tool build run; required by Heal physical metadata checks
+        type: string
+        required: false
+        default: ''
+      validation_tools_run_attempt:
+        description: Exact tool build attempt
+        type: string
+        required: false
+        default: '1'
  workflow_run:
    workflows: ["Nightly GNU Build"]
    types: [completed]
@@ -36,12 +88,47 @@ permissions:
  actions: read

 concurrency:
-  group: rustfs-functional-chain-runs
+  group: ${{ github.event_name == 'workflow_dispatch' && inputs.validation_package_sha != '' && format('rustfs-functional-validation-package-{0}', inputs.validation_package_sha) || github.event_name == 'workflow_dispatch' && inputs.validation_tools_only && 'rustfs-functional-validation-tools' || 'rustfs-functional-chain-runs' }}
  cancel-in-progress: false

 jobs:
+  validation-package-request:
+    if: ${{ github.event_name == 'workflow_dispatch' && inputs.validation_package_sha != '' }}
+    runs-on: ubuntu-latest
+    timeout-minutes: 5
+    steps:
+      - name: Validate exclusive package-only request
+        env:
+          PACKAGE_INPUTS: ${{ toJSON(inputs) }}
+        run: |
+          python3 - <<'PY'
+          import json
+          import os
+          import re
+          value = json.loads(os.environ["PACKAGE_INPUTS"])
+          if re.fullmatch(r"[0-9a-f]{40}", value["validation_package_sha"]) is None:
+              raise ValueError("package build requires an exact source SHA")
+          exclusive = ("build_run_id", "build_run_attempt", "validation_case", "auto_testing_sha",
+                       "validation_tools_only", "validation_native_tools", "validation_shared_tools",
+                       "validation_manager_tools", "validation_cleanup_run_id", "validation_tools_run_id")
+          if any(value.get(name) for name in exclusive):
+              raise ValueError("package-only build cannot select existing candidate, tools, fleet or recovery execution")
+          PY
+
+  validation-package:
+    needs: validation-package-request
+    if: ${{ github.event_name == 'workflow_dispatch' && inputs.validation_package_sha != '' }}
+    uses: ./.github/workflows/rustfs-validation-package.yml
+    with:
+      source_sha: ${{ inputs.validation_package_sha }}
+    secrets:
+      R2_ACCESS_KEY_ID: ${{ secrets.R2_ACCESS_KEY_ID }}
+      R2_SECRET_ACCESS_KEY: ${{ secrets.R2_SECRET_ACCESS_KEY }}
+      R2_ENDPOINT: ${{ secrets.R2_ENDPOINT }}
+      R2_BUCKET: ${{ secrets.R2_BUCKET }}
+
  prepare:
-    if: ${{ github.event_name == 'workflow_dispatch' || github.event.workflow_run.event == 'schedule' }}
+    if: ${{ (github.event_name == 'workflow_dispatch' && inputs.validation_package_sha == '' && inputs.validation_case == '' && !inputs.validation_tools_only && !inputs.validation_native_tools && !inputs.validation_shared_tools && !inputs.validation_manager_tools && inputs.validation_cleanup_run_id == '') || github.event.workflow_run.event == 'schedule' }}
    runs-on: smoke-testing
    timeout-minutes: 10
    outputs:
@@ -89,6 +176,282 @@ jobs:
            echo "performance_ready=offline" >> "$GITHUB_OUTPUT"
          fi

+  validation:
+    if: ${{ github.event_name == 'workflow_dispatch' && inputs.validation_package_sha == '' && (inputs.validation_case != '' || inputs.validation_tools_only || inputs.validation_native_tools || inputs.validation_shared_tools || inputs.validation_manager_tools || inputs.validation_cleanup_run_id != '') }}
+    runs-on: ${{ inputs.validation_tools_only && 'ubuntu-latest' || 'smoke-testing' }}
+    timeout-minutes: 90
+    outputs:
+      source_sha: ${{ steps.candidate.outputs.source_sha }}
+    concurrency:
+      group: ${{ inputs.validation_tools_only && 'rustfs-validation-tools-resolver' || 'rustfs-shared-functional-tests-v2' }}
+      cancel-in-progress: false
+    env:
+      RUSTFS_ACCESS_KEY: ${{ !inputs.validation_tools_only && secrets.RUSTFS_ACCESS_KEY || '' }}
+      RUSTFS_SECRET_KEY: ${{ !inputs.validation_tools_only && secrets.RUSTFS_SECRET_KEY || '' }}
+      RUSTFS_API_ENDPOINT: ${{ !inputs.validation_tools_only && (secrets.RUSTFS_API_ENDPOINT || vars.RUSTFS_API_ENDPOINT || vars.RUSTFS_RC_ENDPOINT) || '' }}
+      RUSTFS_NODES: ${{ !inputs.validation_tools_only && (secrets.RUSTFS_NODES || vars.RUSTFS_NODES) || '' }}
+      RUSTFS_SSH_USER: ${{ !inputs.validation_tools_only && (secrets.RUSTFS_SSH_USER || vars.RUSTFS_SSH_USER) || '' }}
+    steps:
+      - name: Checkout candidate resolver
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+        with:
+          persist-credentials: false
+
+      - name: Initialize validation evidence
+        run: |
+          set -euo pipefail
+          umask 077
+          VALIDATION_ROOT="${RUNNER_TEMP}/rustfs-validation-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
+          mkdir -- "${VALIDATION_ROOT}"
+          printf 'VALIDATION_ROOT=%s\n' "${VALIDATION_ROOT}" >> "${GITHUB_ENV}"
+
+      - name: Resolve immutable candidate
+        id: candidate
+        env:
+          GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
+          BUILD_RUN_ID: ${{ inputs.build_run_id }}
+          BUILD_RUN_ATTEMPT: ${{ inputs.build_run_attempt }}
+          AUTO_TESTING_SHA: ${{ inputs.auto_testing_sha }}
+          CLEANUP_RUN_ID: ${{ inputs.validation_cleanup_run_id }}
+          SELECTED_CASE: ${{ inputs.validation_case }}
+          TOOLS_ONLY: ${{ inputs.validation_tools_only }}
+          NATIVE_TOOLS: ${{ inputs.validation_native_tools }}
+          SHARED_TOOLS: ${{ inputs.validation_shared_tools }}
+          MANAGER_TOOLS: ${{ inputs.validation_manager_tools }}
+        run: |
+          set -euo pipefail
+          python3 - <<'PY'
+          import json
+          import os
+          from pathlib import Path
+          import re
+          from scripts.resolve_functional_candidate import resolve
+
+          revision = os.environ["AUTO_TESTING_SHA"]
+          cleanup = os.environ["CLEANUP_RUN_ID"]
+          if os.environ["TOOLS_ONLY"] == "true" and (cleanup or os.environ["SELECTED_CASE"]):
+              raise ValueError("tools-only execution cannot select a fleet case or cleanup")
+          if os.environ["NATIVE_TOOLS"] == "true" and (os.environ["TOOLS_ONLY"] != "true" or cleanup or os.environ["SELECTED_CASE"]):
+              raise ValueError("native regressions require tools-only execution, without a fleet case or cleanup")
+          if os.environ["SHARED_TOOLS"] == "true" and (os.environ["TOOLS_ONLY"] != "true" or cleanup or os.environ["SELECTED_CASE"]):
+              raise ValueError("shared fixture builds require tools-only execution, without a fleet case or cleanup")
+          if os.environ["MANAGER_TOOLS"] == "true" and (os.environ["TOOLS_ONLY"] != "true" or cleanup or os.environ["SELECTED_CASE"]):
+              raise ValueError("manager regressions require tools-only execution, without a fleet case or cleanup")
+          if cleanup and (not cleanup.isdigit() or int(cleanup) <= 0 or os.environ["SELECTED_CASE"] or os.environ["TOOLS_ONLY"] == "true"):
+              raise ValueError("cleanup requires one exact failed run and is exclusive with case/tool execution")
+          if re.fullmatch(r"[0-9a-f]{40}", revision) is None:
+              raise ValueError("validation requires an exact auto-testing commit SHA")
+          candidate = resolve(int(os.environ["BUILD_RUN_ID"]), int(os.environ["BUILD_RUN_ATTEMPT"]), allow_validation=True)
+          manifest = candidate["manifest"]
+          root = Path(os.environ["VALIDATION_ROOT"])
+          (root / "candidate-source.json").write_text(json.dumps(candidate, sort_keys=True) + "\n")
+          (root / "candidate.json").write_text(json.dumps({
+              "package_url": manifest["package_url"],
+              "package_sha256": manifest["package_sha256"],
+              "source_sha": manifest["source_sha"],
+          }, sort_keys=True) + "\n")
+          with open(os.environ["GITHUB_OUTPUT"], "a") as output:
+              output.write("source_sha=" + manifest["source_sha"] + "\n")
+          PY
+
+      - name: Checkout exact auto-testing revision
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+        with:
+          repository: rustfs/auto-testing
+          ref: ${{ inputs.auto_testing_sha }}
+          token: ${{ secrets.PF_TESTING_GH_TOKEN }}
+          path: auto-testing
+          persist-credentials: false
+
+      - name: Install validation Python client
+        if: ${{ !inputs.validation_tools_only && (startsWith(inputs.validation_case, 'ilm.') || startsWith(inputs.validation_case, 'pool.') || startsWith(inputs.validation_case, 'bucket-replication.') || startsWith(inputs.validation_case, 'site-replication.') || startsWith(inputs.validation_case, 'heal.')) }}
+        run: |
+          set -euo pipefail
+          python3 -m venv "${RUNNER_TEMP}/rustfs-validation-python"
+          "${RUNNER_TEMP}/rustfs-validation-python/bin/python" -m pip install \
+            -r auto-testing/scripts/validation_ilm/requirements.txt
+          printf '%s\n' "${RUNNER_TEMP}/rustfs-validation-python/bin" >> "${GITHUB_PATH}"
+
+      - name: Validate Heal tool build identity
+        if: ${{ !inputs.validation_tools_only && startsWith(inputs.validation_case, 'heal.') }}
+        env:
+          GH_TOKEN: ${{ github.token }}
+          TOOL_RUN_ID: ${{ inputs.validation_tools_run_id }}
+          TOOL_RUN_ATTEMPT: ${{ inputs.validation_tools_run_attempt }}
+        run: |
+          set -euo pipefail
+          [[ "${TOOL_RUN_ID}" =~ ^[1-9][0-9]*$ ]]
+          [[ "${TOOL_RUN_ATTEMPT}" =~ ^[1-9][0-9]*$ ]]
+          gh api "repos/${GITHUB_REPOSITORY}/actions/runs/${TOOL_RUN_ID}/attempts/${TOOL_RUN_ATTEMPT}" \
+            > "${VALIDATION_ROOT}/tool-run.json"
+          python3 - <<'PY'
+          import json
+          import os
+          from pathlib import Path
+          value = json.loads((Path(os.environ["VALIDATION_ROOT"]) / "tool-run.json").read_text())
+          if (value.get("conclusion") != "success" or value.get("status") != "completed"
+                  or value.get("run_attempt") != int(os.environ["TOOL_RUN_ATTEMPT"])
+                  or value.get("path") not in {".github/workflows/rustfs-functional-chain.yml", ".github/workflows/rustfs-validation-tools.yml"}):
+              raise ValueError("Heal tool run is not an exact successful validation build attempt")
+          PY
+
+      - name: Download pinned Heal metadata inspector
+        if: ${{ !inputs.validation_tools_only && startsWith(inputs.validation_case, 'heal.') }}
+        uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
+        with:
+          github-token: ${{ github.token }}
+          run-id: ${{ inputs.validation_tools_run_id }}
+          name: rustfs-validation-tools-${{ inputs.validation_tools_run_id }}-${{ inputs.validation_tools_run_attempt }}
+          path: ${{ runner.temp }}/rustfs-validation-tools-${{ github.run_id }}-${{ github.run_attempt }}
+
+      - name: Verify Heal metadata inspector
+        if: ${{ !inputs.validation_tools_only && startsWith(inputs.validation_case, 'heal.') }}
+        run: |
+          set -euo pipefail
+          TOOL_DIRECTORY="${RUNNER_TEMP}/rustfs-validation-tools-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
+          python3 -B auto-testing/scripts/validation_heal/inspector.py \
+            --directory "${TOOL_DIRECTORY}" --candidate "${VALIDATION_ROOT}/candidate.json" \
+            > "${VALIDATION_ROOT}/tool-manifest.json"
+          printf 'VALIDATION_HEAL_TOOL_DIRECTORY=%s\n' "${TOOL_DIRECTORY}" >> "${GITHUB_ENV}"
+
+      - name: Download pinned C07 native authority evidence
+        if: ${{ inputs.validation_case == 'heal.conflicting-subquorum-authority' }}
+        uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
+        with:
+          github-token: ${{ github.token }}
+          run-id: ${{ inputs.validation_tools_run_id }}
+          name: rustfs-validation-native-${{ inputs.validation_tools_run_id }}-${{ inputs.validation_tools_run_attempt }}
+          path: ${{ runner.temp }}/rustfs-validation-native-${{ github.run_id }}-${{ github.run_attempt }}
+
+      - name: Verify C07 native source and exact test evidence
+        if: ${{ inputs.validation_case == 'heal.conflicting-subquorum-authority' }}
+        run: |
+          set -euo pipefail
+          NATIVE_DIRECTORY="${RUNNER_TEMP}/rustfs-validation-native-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
+          cp "${VALIDATION_ROOT}/tool-run.json" "${NATIVE_DIRECTORY}/tool-run.json"
+          python3 -B auto-testing/scripts/validation_heal/authority.py \
+            --directory "${NATIVE_DIRECTORY}" --candidate "${VALIDATION_ROOT}/candidate.json" \
+            > "${VALIDATION_ROOT}/native-authority-manifest.json"
+          printf 'VALIDATION_HEAL_NATIVE_DIRECTORY=%s\n' "${NATIVE_DIRECTORY}" >> "${GITHUB_ENV}"
+
+      - name: Download pinned manager persistence and incarnation evidence
+        if: ${{ inputs.validation_case == 'heal.checkpoint-and-mrf-corruption' || inputs.validation_case == 'heal.bucket-incarnation-recreation' || inputs.validation_case == 'heal.outcome-counters-and-error-truth' }}
+        uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
+        with:
+          github-token: ${{ github.token }}
+          run-id: ${{ inputs.validation_tools_run_id }}
+          name: rustfs-validation-manager-native-${{ inputs.validation_tools_run_id }}-${{ inputs.validation_tools_run_attempt }}
+          path: ${{ runner.temp }}/rustfs-validation-manager-native-${{ github.run_id }}-${{ github.run_attempt }}
+
+      - name: Verify exact manager native prerequisites before fleet mutation
+        if: ${{ inputs.validation_case == 'heal.checkpoint-and-mrf-corruption' || inputs.validation_case == 'heal.bucket-incarnation-recreation' || inputs.validation_case == 'heal.outcome-counters-and-error-truth' }}
+        env:
+          VALIDATION_CASE: ${{ inputs.validation_case }}
+        run: |
+          set -euo pipefail
+          MANAGER_DIRECTORY="${RUNNER_TEMP}/rustfs-validation-manager-native-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
+          cp "${VALIDATION_ROOT}/tool-run.json" "${MANAGER_DIRECTORY}/tool-run.json"
+          python3 -B auto-testing/scripts/validation_heal/manager_fixture.py \
+            --directory "${MANAGER_DIRECTORY}" --candidate "${VALIDATION_ROOT}/candidate.json" \
+            --case-id "${VALIDATION_CASE}" > "${VALIDATION_ROOT}/manager-prerequisite-manifest.json"
+          printf 'VALIDATION_HEAL_MANAGER_DIRECTORY=%s\n' "${MANAGER_DIRECTORY}" >> "${GITHUB_ENV}"
+
+      - name: Download pinned C10 shared-reference encoder
+        if: ${{ inputs.validation_case == 'heal.dangling-and-orphan-safety' }}
+        uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
+        with:
+          github-token: ${{ github.token }}
+          run-id: ${{ inputs.validation_tools_run_id }}
+          name: rustfs-validation-shared-${{ inputs.validation_tools_run_id }}-${{ inputs.validation_tools_run_attempt }}
+          path: ${{ runner.temp }}/rustfs-validation-shared-${{ github.run_id }}-${{ github.run_attempt }}
+
+      - name: Verify C10 exact-source codec and original tests
+        if: ${{ inputs.validation_case == 'heal.dangling-and-orphan-safety' }}
+        run: |
+          set -euo pipefail
+          SHARED_DIRECTORY="${RUNNER_TEMP}/rustfs-validation-shared-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
+          cp "${VALIDATION_ROOT}/tool-run.json" "${SHARED_DIRECTORY}/tool-run.json"
+          python3 -B auto-testing/scripts/validation_heal/shared_fixture.py \
+            --directory "${SHARED_DIRECTORY}" --candidate "${VALIDATION_ROOT}/candidate.json" \
+            > "${VALIDATION_ROOT}/shared-fixture-manifest.json"
+          printf 'VALIDATION_HEAL_SHARED_DIRECTORY=%s\n' "${SHARED_DIRECTORY}" >> "${GITHUB_ENV}"
+
+      - name: Run selected canonical validation case
+        if: ${{ !inputs.validation_tools_only && inputs.validation_cleanup_run_id == '' && inputs.validation_case != 'environment.inspect' }}
+        env:
+          VALIDATION_CASE: ${{ inputs.validation_case }}
+          AUTO_TESTING_SHA: ${{ inputs.auto_testing_sha }}
+        run: |
+          set -euo pipefail
+          test "$(git -C auto-testing rev-parse HEAD)" = "${AUTO_TESTING_SHA}"
+          ./auto-testing/scripts/run_validation_case.sh \
+            "${VALIDATION_CASE}" "${VALIDATION_ROOT}/candidate.json" \
+            "${AUTO_TESTING_SHA}" "${VALIDATION_ROOT}/run"
+
+      - name: Inspect current Heal fleet resources without mutation
+        if: ${{ !inputs.validation_tools_only && inputs.validation_case == 'environment.inspect' }}
+        env:
+          AUTO_TESTING_SHA: ${{ inputs.auto_testing_sha }}
+        run: |
+          set -euo pipefail
+          test "$(git -C auto-testing rev-parse HEAD)" = "${AUTO_TESTING_SHA}"
+          python3 -B auto-testing/scripts/validation_heal/environment.py \
+            --candidate "${VALIDATION_ROOT}/candidate.json" \
+            --testing-commit "${AUTO_TESTING_SHA}" --evidence "${VALIDATION_ROOT}/environment"
+
+      - name: Audit and finalize retained validation environment
+        if: ${{ inputs.validation_cleanup_run_id != '' }}
+        env:
+          GH_TOKEN: ${{ github.token }}
+          FAILED_RUN_ID: ${{ inputs.validation_cleanup_run_id }}
+          AUTO_TESTING_SHA: ${{ inputs.auto_testing_sha }}
+        run: |
+          set -euo pipefail
+          test "$(git -C auto-testing rev-parse HEAD)" = "${AUTO_TESTING_SHA}"
+          gh run view "${FAILED_RUN_ID}" --repo "${GITHUB_REPOSITORY}" \
+            --json status,conclusion,headBranch,workflowName > "${VALIDATION_ROOT}/failed-run.json"
+          python3 - <<'PY'
+          import json
+          import os
+          from pathlib import Path
+          run = json.loads((Path(os.environ["VALIDATION_ROOT"]) / "failed-run.json").read_text())
+          if (run["status"] != "completed" or run["conclusion"] != "failure"
+              or run["workflowName"] != "RustFS Functional Chain" or run["headBranch"] != os.environ["GITHUB_REF_NAME"]):
+              raise ValueError("cleanup source must be a completed failed Functional Chain run")
+          PY
+          gh run download "${FAILED_RUN_ID}" --repo "${GITHUB_REPOSITORY}" \
+            --pattern "rustfs-validation-${FAILED_RUN_ID}-*" --dir "${VALIDATION_ROOT}/failed-artifact"
+          python3 -B auto-testing/scripts/validation_heal/recover_environment.py \
+            --failed-run-id "${FAILED_RUN_ID}" --artifact "${VALIDATION_ROOT}/failed-artifact" \
+            --candidate "${VALIDATION_ROOT}/candidate.json" --evidence "${VALIDATION_ROOT}/recovery"
+
+      - name: Show validation report
+        if: ${{ always() }}
+        run: |
+          if [ -f "${VALIDATION_ROOT}/run/report.md" ]; then
+            cat "${VALIDATION_ROOT}/run/report.md" >> "${GITHUB_STEP_SUMMARY}"
+          fi
+
+      - name: Upload validation evidence
+        if: ${{ always() }}
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: rustfs-validation-${{ github.run_id }}-${{ github.run_attempt }}
+          path: ${{ runner.temp }}/rustfs-validation-${{ github.run_id }}-${{ github.run_attempt }}
+          if-no-files-found: error
+
+  validation-tools:
+    needs: [validation]
+    if: ${{ github.event_name == 'workflow_dispatch' && inputs.validation_package_sha == '' && inputs.validation_tools_only }}
+    uses: ./.github/workflows/rustfs-validation-tools.yml
+    with:
+      candidate_sha: ${{ needs.validation.outputs.source_sha }}
+      auto_testing_sha: ${{ inputs.auto_testing_sha }}
+      build_native_heal: ${{ inputs.validation_native_tools }}
+      build_shared_heal: ${{ inputs.validation_shared_tools }}
+      build_manager_native: ${{ inputs.validation_manager_tools }}
+    secrets: inherit
+
  upgrade:
    needs: [prepare]
    if: ${{ always() && needs.prepare.result == 'success' }}
@@ -0,0 +1,187 @@
+# Copyright 2024 RustFS Team
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+
+name: RustFS Validation Package
+
+on:
+  workflow_call:
+    inputs:
+      source_sha:
+        description: Exact RustFS commit SHA to build
+        required: true
+        type: string
+    secrets:
+      R2_ACCESS_KEY_ID:
+        required: true
+      R2_SECRET_ACCESS_KEY:
+        required: true
+      R2_ENDPOINT:
+        required: true
+      R2_BUCKET:
+        required: true
+  workflow_dispatch:
+    inputs:
+      source_sha:
+        description: Exact RustFS commit SHA to build
+        required: true
+        type: string
+
+permissions:
+  contents: read
+
+concurrency:
+  group: rustfs-validation-package-${{ inputs.source_sha }}
+  cancel-in-progress: false
+
+jobs:
+  resolve-source:
+    if: github.repository == 'rustfs/rustfs'
+    name: Resolve exact validation source
+    runs-on: sm-standard-2
+    timeout-minutes: 10
+    outputs:
+      source_sha: ${{ steps.source.outputs.sha }}
+      source_ref: ${{ steps.source.outputs.ref }}
+    steps:
+      - name: Validate requested source
+        id: source
+        env:
+          REQUESTED_SHA: ${{ inputs.source_sha }}
+        run: |
+          set -euo pipefail
+          [[ "${REQUESTED_SHA}" =~ ^[0-9a-f]{40}$ ]]
+          source_directory="$(mktemp -d "${RUNNER_TEMP}/validation-source.XXXXXX")"
+          trap 'rm -rf "${source_directory}"' EXIT
+          git init -q "${source_directory}"
+          git -C "${source_directory}" fetch --no-tags --depth=1 https://github.com/rustfs/rustfs.git "${REQUESTED_SHA}"
+          test "$(git -C "${source_directory}" rev-parse FETCH_HEAD)" = "${REQUESTED_SHA}"
+          echo "sha=${REQUESTED_SHA}" >> "${GITHUB_OUTPUT}"
+          echo "ref=${REQUESTED_SHA}" >> "${GITHUB_OUTPUT}"
+
+  build:
+    needs: resolve-source
+    name: Build immutable validation DEB
+    runs-on: sm-standard-4
+    timeout-minutes: 150
+    env:
+      FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
+      VALIDATION_SOURCE_SHA: ${{ needs.resolve-source.outputs.source_sha }}
+      VALIDATION_SOURCE_REF: ${{ needs.resolve-source.outputs.source_ref }}
+    steps:
+      - name: Checkout exact source
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+        with:
+          persist-credentials: false
+          ref: ${{ needs.resolve-source.outputs.source_sha }}
+
+      - name: Confirm checked out source
+        env:
+          EXPECTED_SHA: ${{ needs.resolve-source.outputs.source_sha }}
+        run: test "$(git rev-parse HEAD)" = "${EXPECTED_SHA}"
+
+      - name: Setup Rust environment
+        uses: ./.github/actions/setup
+        with:
+          cache-shared-key: validation-package-x86_64-unknown-linux-gnu
+          cache-save-if: "false"
+          install-build-packaging-tools: "false"
+          install-test-tools: "false"
+
+      - name: Build RustFS
+        run: cargo build --release --locked --target x86_64-unknown-linux-gnu -p rustfs --bins
+
+      - name: Build validation DEB
+        id: package
+        shell: bash
+        run: |
+          set -euo pipefail
+          command -v fakeroot >/dev/null 2>&1 || sudo apt-get install -y -qq fakeroot
+          package_dir="rustfs-validation-${VALIDATION_SOURCE_SHA}"
+          package_file="rustfs-validation-${VALIDATION_SOURCE_SHA}.deb"
+          binary="target/x86_64-unknown-linux-gnu/release/rustfs"
+          test -x "${binary}"
+          mkdir -p "${package_dir}/DEBIAN" "${package_dir}/usr/bin" "${package_dir}/etc/default" \
+            "${package_dir}/lib/systemd/system" "${package_dir}/usr/share/doc/rustfs"
+          cp "${binary}" "${package_dir}/usr/bin/rustfs"
+          chmod 755 "${package_dir}/usr/bin/rustfs"
+          cp deploy/build/rustfs.service "${package_dir}/lib/systemd/system/"
+          cat > "${package_dir}/etc/default/rustfs" <<'ENVEOF'
+          # RustFS Environment Configuration
+          # RUSTFS_ACCESS_KEY=<changeme>
+          # RUSTFS_SECRET_KEY=<changeme>
+          # RUSTFS_VOLUMES=""
+          # RUSTFS_ADDRESS=":9000"
+          ENVEOF
+          cat > "${package_dir}/DEBIAN/control" <<'CONTROLEOF'
+          Package: rustfs
+          Version: 0.0.0~validation
+          Section: utils
+          Priority: optional
+          Architecture: amd64
+          Depends: libc6 (>= 2.31)
+          Maintainer: RustFS Team <support@rustfs.com>
+          Description: RustFS immutable validation package
+           Candidate package for RustFS validation only.
+          CONTROLEOF
+          printf '%s\n' /etc/default/rustfs > "${package_dir}/DEBIAN/conffiles"
+          ./scripts/release/package_service_scripts.sh after-install > "${package_dir}/DEBIAN/postinst"
+          ./scripts/release/package_service_scripts.sh before-remove > "${package_dir}/DEBIAN/prerm"
+          ./scripts/release/package_service_scripts.sh after-remove > "${package_dir}/DEBIAN/postrm"
+          chmod 755 "${package_dir}/DEBIAN/postinst" "${package_dir}/DEBIAN/prerm" "${package_dir}/DEBIAN/postrm"
+          cp LICENSE README.md "${package_dir}/usr/share/doc/rustfs/"
+          fakeroot dpkg-deb --build "${package_dir}" "${package_file}"
+          test -s "${package_file}"
+          echo "package_file=${package_file}" >> "${GITHUB_OUTPUT}"
+
+      - name: Upload validation package to immutable R2 path
+        id: publish
+        env:
+          PACKAGE_FILE: ${{ steps.package.outputs.package_file }}
+          R2_ACCESS_KEY_ID: ${{ secrets.R2_ACCESS_KEY_ID }}
+          R2_SECRET_ACCESS_KEY: ${{ secrets.R2_SECRET_ACCESS_KEY }}
+          R2_ENDPOINT: ${{ secrets.R2_ENDPOINT }}
+          R2_BUCKET: ${{ secrets.R2_BUCKET }}
+          AWS_EC2_METADATA_DISABLED: true
+        shell: bash
+        run: |
+          set -euo pipefail
+          if [[ -z "${R2_ACCESS_KEY_ID}" || -z "${R2_SECRET_ACCESS_KEY}" || -z "${R2_ENDPOINT}" || -z "${R2_BUCKET}" ]]; then
+            echo "Validation package publication requires complete R2 configuration" >&2
+            exit 1
+          fi
+          test "$(git rev-parse HEAD)" = "${VALIDATION_SOURCE_SHA}"
+          export AWS_ACCESS_KEY_ID="${R2_ACCESS_KEY_ID}" AWS_SECRET_ACCESS_KEY="${R2_SECRET_ACCESS_KEY}" AWS_DEFAULT_REGION=auto
+          package_sha256="$(sha256sum "${PACKAGE_FILE}" | cut -d ' ' -f 1)"
+          key="artifacts/rustfs/packages/validation/runs/${GITHUB_RUN_ID}/${GITHUB_RUN_ATTEMPT}/${package_sha256}/rustfs.deb"
+          url="https://dl.rustfs.com/${key}"
+          AWS_CLI=aws
+          if ! "${AWS_CLI}" s3api put-object --generate-cli-skeleton input | jq -e 'has("IfNoneMatch")' >/dev/null; then
+            sudo apt-get update
+            sudo apt-get install -y -qq python3-venv
+            AWS_CLI_DIR="$(mktemp -d "${RUNNER_TEMP}/validation-awscli.XXXXXX")"
+            trap 'rm -rf "${AWS_CLI_DIR}"' EXIT
+            python3 -m venv "${AWS_CLI_DIR}"
+            "${AWS_CLI_DIR}/bin/python" -m pip install --disable-pip-version-check 'awscli==1.44.79'
+            AWS_CLI="${AWS_CLI_DIR}/bin/aws"
+          fi
+          "${AWS_CLI}" s3api put-object --generate-cli-skeleton input | jq -e 'has("IfNoneMatch")' >/dev/null
+          "${AWS_CLI}" s3api put-object --bucket "${R2_BUCKET}" --key "${key}" --body "${PACKAGE_FILE}" \
+            --if-none-match '*' --endpoint-url "${R2_ENDPOINT}"
+          downloaded_sha256="$(curl -fsSL --retry 3 --connect-timeout 15 --max-time 300 "${url}" | sha256sum | cut -d ' ' -f 1)"
+          test "${downloaded_sha256}" = "${package_sha256}"
+          manifest="${RUNNER_TEMP}/validation-candidate-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}.json"
+          jq -n --arg workflow_sha "${GITHUB_SHA}" --arg source_ref "${VALIDATION_SOURCE_REF}" \
+            --arg source_sha "${VALIDATION_SOURCE_SHA}" --arg channel validation \
+            --argjson build_run_id "${GITHUB_RUN_ID}" --argjson build_run_attempt "${GITHUB_RUN_ATTEMPT}" \
+            --arg package_url "${url}" --arg package_sha256 "${package_sha256}" \
+            '{schema: 3, workflow_sha: $workflow_sha, source_ref: $source_ref, source_sha: $source_sha, channel: $channel, build_run_id: $build_run_id, build_run_attempt: $build_run_attempt, package_url: $package_url, package_sha256: $package_sha256}' > "${manifest}"
+          echo "candidate_file=${manifest}" >> "${GITHUB_OUTPUT}"
+
+      - name: Upload validation candidate manifest
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: validation-candidate-${{ github.run_id }}-${{ github.run_attempt }}
+          path: ${{ steps.publish.outputs.candidate_file }}
+          if-no-files-found: error
@@ -0,0 +1,300 @@
+# Build source-bound helpers separately so live validation needs only artifacts.
+name: RustFS Validation Tools
+
+on:
+  workflow_call:
+    inputs:
+      candidate_sha:
+        type: string
+        required: true
+      auto_testing_sha:
+        type: string
+        required: true
+      build_native_heal:
+        type: boolean
+        required: false
+        default: false
+      build_shared_heal:
+        type: boolean
+        required: false
+        default: false
+      build_manager_native:
+        type: boolean
+        required: false
+        default: false
+    secrets:
+      PF_TESTING_GH_TOKEN:
+        required: true
+  workflow_dispatch:
+    inputs:
+      candidate_sha:
+        description: Exact RustFS source SHA from the validation candidate manifest
+        type: string
+        required: true
+      auto_testing_sha:
+        description: Exact auto-testing SHA containing the Heal metadata inspector
+        type: string
+        required: true
+      build_native_heal:
+        description: Also compile and run separate exact-source native Heal regressions
+        type: boolean
+        required: false
+        default: false
+      build_shared_heal:
+        description: Build the controlled C10 shared-reference fixture encoder
+        type: boolean
+        required: false
+        default: false
+      build_manager_native:
+        description: Build and run exact candidate checkpoint/MRF/incarnation regressions
+        type: boolean
+        required: false
+        default: false
+
+permissions:
+  contents: read
+
+jobs:
+  manager-native-regressions:
+    if: ${{ inputs.build_manager_native }}
+    runs-on: ubuntu-22.04
+    timeout-minutes: 180
+    env:
+      CANDIDATE_SHA: ${{ inputs.candidate_sha }}
+      AUTO_TESTING_SHA: ${{ inputs.auto_testing_sha }}
+    steps:
+      - name: Validate immutable manager native inputs
+        run: |
+          set -euo pipefail
+          [[ "${CANDIDATE_SHA}" =~ ^[0-9a-f]{40}$ ]]
+          [[ "${AUTO_TESTING_SHA}" =~ ^[0-9a-f]{40}$ ]]
+
+      - name: Checkout exact manager candidate source
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+        with:
+          ref: ${{ inputs.candidate_sha }}
+          path: candidate
+          persist-credentials: false
+
+      - name: Checkout exact manager collector source
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+        with:
+          repository: rustfs/auto-testing
+          ref: ${{ inputs.auto_testing_sha }}
+          token: ${{ secrets.PF_TESTING_GH_TOKEN }}
+          path: auto-testing
+          persist-credentials: false
+
+      - name: Install manager native build dependencies
+        run: |
+          set -euo pipefail
+          sudo apt-get update
+          sudo apt-get install -y clang lld cmake pkg-config libssl-dev protobuf-compiler
+          df -h .
+
+      - name: Build exact unmodified candidate manager test executable
+        run: |
+          set -euo pipefail
+          test "$(git -C auto-testing rev-parse HEAD)" = "${AUTO_TESTING_SHA}"
+          python3 -B auto-testing/scripts/validation_heal/manager_native.py build \
+            --source candidate --commit "${CANDIDATE_SHA}" --out manager-tools
+
+      - name: Run exact manager regressions independently
+        run: |
+          set -euo pipefail
+          python3 -B auto-testing/scripts/validation_heal/manager_native.py run \
+            --directory manager-tools --out manager-evidence
+
+      - name: Upload manager executable identities and original diagnostics
+        if: ${{ always() }}
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: rustfs-validation-manager-native-${{ github.run_id }}-${{ github.run_attempt }}
+          path: |
+            manager-tools/heal-manager-native-tests
+            manager-tools/*.json
+            manager-tools/*.jsonl
+            manager-tools/*.stderr
+            manager-tools/*.log
+            manager-evidence/
+          if-no-files-found: warn
+
+  native-heal-regressions:
+    if: ${{ inputs.build_native_heal }}
+    runs-on: ubuntu-22.04
+    timeout-minutes: 180
+    env:
+      CANDIDATE_SHA: ${{ inputs.candidate_sha }}
+      AUTO_TESTING_SHA: ${{ inputs.auto_testing_sha }}
+    steps:
+      - name: Validate immutable native source inputs
+        run: |
+          set -euo pipefail
+          [[ "${CANDIDATE_SHA}" =~ ^[0-9a-f]{40}$ ]]
+          [[ "${AUTO_TESTING_SHA}" =~ ^[0-9a-f]{40}$ ]]
+
+      - name: Checkout exact native candidate source
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+        with:
+          ref: ${{ inputs.candidate_sha }}
+          path: candidate
+          persist-credentials: false
+
+      - name: Checkout exact native collector source
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+        with:
+          repository: rustfs/auto-testing
+          ref: ${{ inputs.auto_testing_sha }}
+          token: ${{ secrets.PF_TESTING_GH_TOKEN }}
+          path: auto-testing
+          persist-credentials: false
+
+      - name: Install native build dependencies
+        run: |
+          set -euo pipefail
+          sudo apt-get update
+          sudo apt-get install -y clang lld cmake pkg-config libssl-dev protobuf-compiler
+          df -h .
+
+      - name: Build and enumerate the unmodified candidate test executable
+        run: |
+          set -euo pipefail
+          test "$(git -C auto-testing rev-parse HEAD)" = "${AUTO_TESTING_SHA}"
+          python3 -B auto-testing/scripts/validation_heal/native_regressions.py build \
+            --source candidate --commit "${CANDIDATE_SHA}" --out native-tools
+
+      - name: Run exact native regressions separately from fleet validation
+        run: |
+          set -euo pipefail
+          python3 -B auto-testing/scripts/validation_heal/native_regressions.py run \
+            --directory native-tools --out native-evidence
+
+      - name: Upload native executable identities and original diagnostics
+        if: ${{ always() }}
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: rustfs-validation-native-${{ github.run_id }}-${{ github.run_attempt }}
+          path: |
+            native-tools/ecstore-native-tests
+            native-tools/*.json
+            native-tools/*.jsonl
+            native-tools/*.stderr
+            native-tools/*.log
+            native-evidence/
+          if-no-files-found: warn
+
+  heal-metadata-inspector:
+    runs-on: ubuntu-22.04
+    timeout-minutes: 30
+    env:
+      CANDIDATE_SHA: ${{ inputs.candidate_sha }}
+      AUTO_TESTING_SHA: ${{ inputs.auto_testing_sha }}
+    steps:
+      - name: Validate immutable source inputs
+        run: |
+          set -euo pipefail
+          [[ "${CANDIDATE_SHA}" =~ ^[0-9a-f]{40}$ ]]
+          [[ "${AUTO_TESTING_SHA}" =~ ^[0-9a-f]{40}$ ]]
+
+      - name: Checkout exact candidate source
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+        with:
+          ref: ${{ inputs.candidate_sha }}
+          path: candidate
+          persist-credentials: false
+
+      - name: Checkout exact inspector source
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+        with:
+          repository: rustfs/auto-testing
+          ref: ${{ inputs.auto_testing_sha }}
+          token: ${{ secrets.PF_TESTING_GH_TOKEN }}
+          path: auto-testing
+          persist-credentials: false
+
+      - name: Build the read-only candidate metadata decoder
+        run: |
+          set -euo pipefail
+          test "$(git -C candidate rev-parse HEAD)" = "${CANDIDATE_SHA}"
+          test "$(git -C auto-testing rev-parse HEAD)" = "${AUTO_TESTING_SHA}"
+          df -h .
+          python3 auto-testing/scripts/validation_heal/prepare_heal_meta_inspector.py \
+            --source candidate --commit "${CANDIDATE_SHA}" --out inspector-build
+          # Seed the helper's separate graph from the candidate's frozen graph.
+          # Cargo adds the helper package; both lock identities are retained.
+          cp candidate/Cargo.lock inspector-build/Cargo.lock
+          cargo build --release --manifest-path inspector-build/Cargo.toml
+          python3 auto-testing/scripts/validation_heal/check_bitrot_tool.py \
+            --binary inspector-build/target/release/heal-meta-inspect --source candidate
+          mkdir validation-tools
+          cp inspector-build/target/release/heal-meta-inspect validation-tools/
+          cp inspector-build/Cargo.lock validation-tools/inspector-Cargo.lock
+          python3 - <<'PY'
+          import hashlib
+          import json
+          import os
+          from pathlib import Path
+          import subprocess
+
+          source = json.loads(Path("inspector-build/source.json").read_text())
+          binary = Path("validation-tools/heal-meta-inspect")
+          source.update(built=True, testingCommit=os.environ["AUTO_TESTING_SHA"],
+                        binarySha256=hashlib.sha256(binary.read_bytes()).hexdigest(),
+                        inspectorLockSha256=hashlib.sha256(Path("validation-tools/inspector-Cargo.lock").read_bytes()).hexdigest(),
+                        toolchain=subprocess.check_output(["rustc", "-vV"], text=True),
+                        target="x86_64-unknown-linux-gnu", runId=os.environ["GITHUB_RUN_ID"],
+                        runAttempt=os.environ["GITHUB_RUN_ATTEMPT"])
+          Path("validation-tools/heal-meta-inspect.json").write_text(json.dumps(source, indent=2) + "\n")
+          PY
+          git -C candidate diff --exit-code
+
+      - name: Build and test the controlled C10 metadata encoder
+        if: ${{ inputs.build_shared_heal }}
+        run: |
+          set -euo pipefail
+          python3 -B auto-testing/scripts/validation_heal/prepare_heal_shared_fixture.py \
+            --source candidate --commit "${CANDIDATE_SHA}" --out shared-build
+          cp candidate/Cargo.lock shared-build/Cargo.lock
+          export CARGO_TARGET_DIR="${GITHUB_WORKSPACE}/inspector-build/target"
+          cargo build --release --manifest-path shared-build/Cargo.toml
+          mkdir shared-tools
+          if ! cargo test --release --manifest-path shared-build/Cargo.toml -- --test-threads=1 \
+            > shared-tools/shared-fixture-tests.log 2>&1; then
+            cat shared-tools/shared-fixture-tests.log
+            exit 1
+          fi
+          cp "${CARGO_TARGET_DIR}/release/heal-shared-fixture" shared-tools/
+          cp shared-build/Cargo.lock shared-tools/shared-Cargo.lock
+          python3 - <<'PY'
+          import hashlib
+          import json
+          import os
+          from pathlib import Path
+          import subprocess
+
+          source = json.loads(Path("shared-build/source.json").read_text())
+          source.update(schemaVersion=1, built=True, sourceUnmodified=True,
+              candidateSourceTree=subprocess.check_output(["git", "-C", "candidate", "rev-parse", "HEAD^{tree}"], text=True).strip(),
+              testingCommit=os.environ["AUTO_TESTING_SHA"],
+              binarySha256=hashlib.sha256(Path("shared-tools/heal-shared-fixture").read_bytes()).hexdigest(),
+              sharedLockSha256=hashlib.sha256(Path("shared-tools/shared-Cargo.lock").read_bytes()).hexdigest(),
+              toolchain=subprocess.check_output(["rustc", "-vV"], text=True), target="x86_64-unknown-linux-gnu",
+              runId=os.environ["GITHUB_RUN_ID"], runAttempt=os.environ["GITHUB_RUN_ATTEMPT"])
+          Path("shared-tools/heal-shared-fixture.json").write_text(json.dumps(source, indent=2) + "\n")
+          PY
+          git -C candidate diff --exit-code
+
+      - name: Upload controlled C10 fixture encoder evidence
+        if: ${{ always() && inputs.build_shared_heal }}
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: rustfs-validation-shared-${{ github.run_id }}-${{ github.run_attempt }}
+          path: shared-tools/
+          if-no-files-found: warn
+
+      - name: Upload immutable validation tools
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: rustfs-validation-tools-${{ github.run_id }}-${{ github.run_attempt }}
+          path: validation-tools/
+          if-no-files-found: error
@@ -10594,6 +10594,7 @@ dependencies = [
 "faster-hex",
 "futures",
 "hex-simd",
+ "hmac 0.13.0",
 "hotpath",
 "http 1.5.0",
 "http-body-util",
@@ -11060,6 +11061,7 @@ dependencies = [
 "async-compression",
 "futures",
 "hotpath",
+ "liblzma",
 "rustfs-rio",
 "rustfs-tokio-tar",
 "serde",
@@ -740,7 +740,7 @@ async fn test_bucket_tags_opa_contract() -> TestResult {
        .version_id(version)
        .send()
        .await?;
-    opa.expect_bucket_tags("s3:GetObject", source, Some("finance")).await?;
+    opa.expect_bucket_tags("s3:GetObjectVersion", source, Some("finance")).await?;
    client.list_objects_v2().bucket(source).send().await?;
    opa.expect_bucket_tags("s3:ListBucket", source, Some("finance")).await?;
    client.list_object_versions().bucket(source).send().await?;
@@ -2939,6 +2939,20 @@ fn resolve_tier_free_version_recovery_enabled(value: Result<String, env::VarErro
 }

 fn spawn_tier_free_version_recovery_once(api: Arc<ECStore>, started: &OnceLock<()>) -> Option<JoinHandle<()>> {
+    spawn_tier_free_version_recovery_once_with_options(
+        api,
+        started,
+        TIER_FREE_VERSION_RECOVERY_LEADER_LOCK.to_string(),
+        tier_free_version_recovery_leader_retry_delay,
+    )
+}
+
+fn spawn_tier_free_version_recovery_once_with_options(
+    api: Arc<ECStore>,
+    started: &OnceLock<()>,
+    leader_lock_name: String,
+    retry_delay: fn() -> StdDuration,
+) -> Option<JoinHandle<()>> {
    if !tier_free_version_recovery_enabled() {
        warn!(
            event = EVENT_LIFECYCLE_WORKER_STATE,
@@ -2956,15 +2970,15 @@ fn spawn_tier_free_version_recovery_once(api: Arc<ECStore>, started: &OnceLock<(
    Some(tokio::spawn(async move {
        let cancel_token = api.ctx.background_cancel_token().unwrap_or_default();
        let acquire_api = Arc::clone(&api);
+        let acquire_lock_name = Arc::<str>::from(leader_lock_name);
        run_tier_free_version_recovery_coordinator(
            cancel_token,
-            tier_free_version_recovery_leader_retry_delay,
+            retry_delay,
            move || {
                let api = Arc::clone(&acquire_api);
+                let lock_name = Arc::clone(&acquire_lock_name);
                async move {
-                    let lock = api
-                        .new_ns_lock(RUSTFS_META_BUCKET, TIER_FREE_VERSION_RECOVERY_LEADER_LOCK)
-                        .await?;
+                    let lock = api.new_ns_lock(RUSTFS_META_BUCKET, &lock_name).await?;
                    match lock.get_write_lock_quiet(TIER_FREE_VERSION_RECOVERY_LOCK_TIMEOUT).await {
                        Ok(guard) => Ok(Some(guard)),
                        Err(LockError::Timeout { .. }) => Ok(None),
@@ -14193,6 +14207,10 @@ mod tests {
        .await;
    }

+    fn fast_tier_free_version_recovery_leader_retry_delay() -> StdDuration {
+        StdDuration::from_millis(10)
+    }
+
    #[tokio::test]
    #[serial]
    async fn tier_free_version_recovery_production_entrypoint_enqueues_seeded_item() {
@@ -14208,8 +14226,13 @@ mod tests {
        seed_recoverable_free_version(&disk_paths, &bucket, object, None, None).await;

        let started = OnceLock::new();
-        let recovery = super::spawn_tier_free_version_recovery_once(Arc::clone(&ecstore), &started)
-            .expect("production recovery entrypoint should start once");
+        let recovery = super::spawn_tier_free_version_recovery_once_with_options(
+            Arc::clone(&ecstore),
+            &started,
+            format!("tier-free-version-recovery/test/{}", Uuid::new_v4()),
+            fast_tier_free_version_recovery_leader_retry_delay,
+        )
+        .expect("production recovery entrypoint should start once");
        let task = tokio::time::timeout(StdDuration::from_secs(30), async { recovery_rx.lock().await.recv().await })
            .await
            .expect("production recovery entrypoint should enqueue the seeded item")
@@ -1292,14 +1292,34 @@ pub(crate) async fn get_object_lock_config_state_in(
 pub(crate) async fn get_object_lock_config_and_incarnation_from_disk_in(
    ctx: &crate::runtime::instance::InstanceContext,
    bucket: &str,
+) -> Result<(ObjectLockConfigState, Uuid, OffsetDateTime)> {
+    get_object_lock_config_and_incarnation_from_disk_with_guard_in(ctx, bucket, None).await
+}
+
+/// Bucket deletion already owns the namespace/lifecycle fences. Reuse its
+/// live guard rather than recursively acquiring the bucket namespace lock.
+/// The caller must also hold the metadata transaction read lock.
+pub(crate) async fn get_object_lock_config_and_incarnation_from_disk_with_guard_in(
+    ctx: &crate::runtime::instance::InstanceContext,
+    bucket: &str,
+    guard: Option<&rustfs_lock::NamespaceLockGuard>,
 ) -> Result<(ObjectLockConfigState, Uuid, OffsetDateTime)> {
    let bucket_meta_sys_lock = bucket_metadata_sys_of(ctx)?;
    let bucket_meta_sys = bucket_meta_sys_lock.read().await.clone();

-    match bucket_meta_sys
-        .read_authoritative_metadata_from_disk_under_transaction_lock(bucket)
-        .await?
-    {
+    let authority = match guard {
+        Some(guard) => {
+            bucket_meta_sys
+                .read_authoritative_metadata_from_disk_under_guard(bucket, guard)
+                .await?
+        }
+        None => {
+            bucket_meta_sys
+                .read_authoritative_metadata_from_disk_under_transaction_lock(bucket)
+                .await?
+        }
+    };
+    match authority {
        BucketMetadataAuthority::Authoritative(metadata)
            if metadata.bucket_incarnation_sidecar && !metadata.bucket_incarnation_id.is_nil() =>
        {
@@ -2911,7 +2911,7 @@ mod tests {

        let timed = wrapper.track_disk_health_with_op(
            "write_all",
-            || std::future::pending::<Result<()>>(),
+            std::future::pending::<Result<()>>,
            SLOW_DISK_OPERATION_THRESHOLD + Duration::from_secs(1),
        );
        tokio::pin!(timed);
@@ -442,12 +442,19 @@ impl WritePlan {
        }

        if let Some(encryption) = self.encryption {
+            let etag_nonce = match encryption.mode {
+                WriteEncryptionMode::Singlepart { base_nonce } | WriteEncryptionMode::MultipartLegacy { base_nonce, .. } => {
+                    base_nonce
+                }
+                WriteEncryptionMode::SinglepartObjectKey | WriteEncryptionMode::MultipartObjectKey { .. } => [0u8; 12],
+            };
+            let protected_reader = rustfs_rio::EtagReader::new_encrypted(reader, encryption.key_bytes, etag_nonce)?;
            reader = match encryption.mode {
                WriteEncryptionMode::SinglepartObjectKey => HashReader::from_reader(
                    #[cfg(feature = "rio-v2")]
-                    EncryptReader::new_with_object_key(reader, encryption.key_bytes),
+                    EncryptReader::new_with_object_key(protected_reader, encryption.key_bytes),
                    #[cfg(not(feature = "rio-v2"))]
-                    EncryptReader::new(reader, encryption.key_bytes, [0u8; 12]),
+                    EncryptReader::new(protected_reader, encryption.key_bytes, [0u8; 12]),
                    HashReader::SIZE_PRESERVE_LAYER,
                    actual_size,
                    None,
@@ -457,12 +464,12 @@ impl WritePlan {
                WriteEncryptionMode::Singlepart { base_nonce } => {
                    #[cfg(not(feature = "rio-v2"))]
                    let encrypt_reader = if encryption_frame_v2_enabled() {
-                        EncryptReader::new_v2(reader, encryption.key_bytes, base_nonce)
+                        EncryptReader::new_v2(protected_reader, encryption.key_bytes, base_nonce)
                    } else {
-                        EncryptReader::new(reader, encryption.key_bytes, base_nonce)
+                        EncryptReader::new(protected_reader, encryption.key_bytes, base_nonce)
                    };
                    #[cfg(feature = "rio-v2")]
-                    let encrypt_reader = EncryptReader::new(reader, encryption.key_bytes, base_nonce);
+                    let encrypt_reader = EncryptReader::new(protected_reader, encryption.key_bytes, base_nonce);
                    HashReader::from_reader(encrypt_reader, HashReader::SIZE_PRESERVE_LAYER, actual_size, None, None, false)?
                }
                WriteEncryptionMode::MultipartLegacy {
@@ -471,20 +478,25 @@ impl WritePlan {
                } => {
                    #[cfg(not(feature = "rio-v2"))]
                    let encrypt_reader = if encryption_frame_v2_enabled() {
-                        EncryptReader::new_multipart_v2(reader, encryption.key_bytes, base_nonce, multipart_part_number)
+                        EncryptReader::new_multipart_v2(protected_reader, encryption.key_bytes, base_nonce, multipart_part_number)
                    } else {
-                        EncryptReader::new_multipart(reader, encryption.key_bytes, base_nonce, multipart_part_number)
+                        EncryptReader::new_multipart(protected_reader, encryption.key_bytes, base_nonce, multipart_part_number)
                    };
                    #[cfg(feature = "rio-v2")]
                    let encrypt_reader =
-                        EncryptReader::new_multipart(reader, encryption.key_bytes, base_nonce, multipart_part_number);
+                        EncryptReader::new_multipart(protected_reader, encryption.key_bytes, base_nonce, multipart_part_number);
                    HashReader::from_reader(encrypt_reader, HashReader::SIZE_PRESERVE_LAYER, actual_size, None, None, false)?
                }
                WriteEncryptionMode::MultipartObjectKey { multipart_part_number } => HashReader::from_reader(
                    #[cfg(feature = "rio-v2")]
-                    EncryptReader::new_multipart_with_object_key(reader, encryption.key_bytes, multipart_part_number),
+                    EncryptReader::new_multipart_with_object_key(protected_reader, encryption.key_bytes, multipart_part_number),
                    #[cfg(not(feature = "rio-v2"))]
-                    EncryptReader::new_multipart(reader, encryption.key_bytes, [0u8; 12], multipart_part_number as usize),
+                    EncryptReader::new_multipart(
+                        protected_reader,
+                        encryption.key_bytes,
+                        [0u8; 12],
+                        multipart_part_number as usize,
+                    ),
                    HashReader::SIZE_PRESERVE_LAYER,
                    actual_size,
                    None,
@@ -497,6 +509,7 @@ impl WritePlan {
        // `ignore_value` deliberately avoids a second hasher over compressed or
        // encrypted bytes. The inner reader still validates the plaintext request
        // checksum while this outer reader exposes the request checksum context.
+        reader.encrypted_etag = encrypted;
        reader.add_non_trailing_checksum(checksum, true)?;
        reader.set_trailer(trailer);

@@ -513,6 +526,36 @@ mod tests {
    use std::io::Cursor;
    use tokio::io::AsyncReadExt;

+    #[tokio::test]
+    async fn write_plan_encryption_persists_protected_etags_for_every_mode() {
+        let data = b"Salary 2026: 85000 EUR";
+        let size = i64::try_from(data.len()).expect("test data size");
+        let mut control =
+            HashReader::from_stream(Cursor::new(data), size, size, None, None, false).expect("create control reader");
+        control.read_to_end(&mut Vec::new()).await.expect("read control plaintext");
+        let md5 = control.try_resolve_etag().expect("control MD5");
+        for encryption in [
+            WriteEncryption::singlepart([1; 32], [1; 12]),
+            WriteEncryption::singlepart_object_key([1; 32]),
+            WriteEncryption::multipart([1; 32], [1; 12], 1),
+            WriteEncryption::multipart_object_key([1; 32], 1),
+        ] {
+            let plaintext = HashReader::from_stream(Cursor::new(data), size, size, Some(md5.clone()), None, false)
+                .expect("create checked reader");
+            let mut reader = WritePlan::new()
+                .with_encryption(encryption)
+                .apply(plaintext, size)
+                .expect("apply encryption");
+            let mut ciphertext = Vec::new();
+            reader.read_to_end(&mut ciphertext).await.expect("stream ciphertext");
+            assert_ne!(ciphertext, data);
+            let tag = reader.try_resolve_etag().expect("resolve persisted encrypted ETag");
+            assert_ne!(tag, md5, "stored ETag must not reveal the plaintext MD5");
+            assert_eq!(tag.len(), 32, "multipart ETags require a 16-byte hexadecimal part tag");
+            assert_eq!(reader.try_resolve_etag().as_ref(), Some(&tag));
+        }
+    }
+
    async fn assert_non_trailing_checksum_survives(plan: WritePlan) {
        let plaintext = b"checksum-context-through-write-plan".repeat(256);
        let actual_size = plaintext.len() as i64;
@@ -3004,6 +3004,10 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
        // etag
        let etag = resolve_complete_etag(opts, &uploaded_parts);

+        // An upload can contain parts from an older writer or a peer. Until
+        // every part's format is attested, keep its ETag out of keyless lists.
+        rustfs_utils::http::metadata_compat::remove_str(&mut fi.metadata, rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG);
+
        fi.metadata.insert("etag".to_owned(), etag);

        let persist_encryption_original_size = should_persist_encryption_original_size(&fi.metadata);
@@ -3847,6 +3847,16 @@ impl SetDisks {
                .map(crate::io_support::rio::compression_index_storage_bytes);

            let mut etag = data.stream.try_resolve_etag().unwrap_or_default();
+            if data.stream.encrypted_etag {
+                if opts.preserve_etag.as_ref().is_none_or(|preserved| preserved == &etag) {
+                    insert_str(&mut user_defined, rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG, "v1".to_owned());
+                } else {
+                    rustfs_utils::http::metadata_compat::remove_str(
+                        &mut user_defined,
+                        rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG,
+                    );
+                }
+            }
            if let Some(ref tag) = opts.preserve_etag {
                etag = tag.clone();
            }
@@ -10610,6 +10620,65 @@ mod object_encryption_resolver_wiring_tests {
    use std::sync::Mutex;
    use std::sync::atomic::{AtomicUsize, Ordering};

+    #[tokio::test]
+    async fn encrypted_put_attests_only_etags_produced_by_its_encryption_reader() {
+        use super::hermetic_set_disks_support::hermetic_set_disks_isolated;
+        use crate::io_support::rio::{WriteEncryption, WritePlan};
+        use crate::storage_api_contracts::object::ObjectIO as _;
+        use tokio::io::AsyncReadExt;
+
+        let (_temp_dirs, disks, set_disks) = hermetic_set_disks_isolated(4).await;
+        let bucket = "encrypted-etag-proof";
+        for disk in &disks {
+            disk.make_volume(bucket).await.expect("create bucket volume");
+        }
+        let plaintext = b"Salary 2026: 85000 EUR";
+        let size = plaintext.len() as i64;
+        let mut control = HashReader::from_stream(Cursor::new(plaintext), size, size, None, None, false).expect("control reader");
+        control.read_to_end(&mut Vec::new()).await.expect("read control plaintext");
+        let plaintext_md5 = control.try_resolve_etag().expect("control MD5");
+        for (object, preserved) in [("new", None), ("preserved", Some(plaintext_md5.clone()))] {
+            let reader =
+                HashReader::from_stream(Cursor::new(plaintext), size, size, None, None, false).expect("plaintext reader");
+            let reader = WritePlan::new()
+                .with_encryption(WriteEncryption::singlepart([1; 32], [2; 12]))
+                .apply(reader, size)
+                .expect("encrypted reader");
+            let mut opts = ObjectOptions {
+                preserve_etag: preserved.clone(),
+                user_defined: HashMap::from([
+                    ("x-amz-server-side-encryption-customer-algorithm".to_owned(), "AES256".to_owned()),
+                    (rustfs_utils::http::SSEC_ORIGINAL_SIZE_HEADER.to_owned(), size.to_string()),
+                ]),
+                ..Default::default()
+            };
+            // A copied producer marker cannot attest to a preserved plaintext tag.
+            insert_str(&mut opts.user_defined, rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG, "v1".to_owned());
+            set_disks
+                .put_object(bucket, object, &mut PutObjReader::new(reader), &opts)
+                .await
+                .expect("store encrypted bytes");
+            let stored = set_disks
+                .get_object_info(bucket, object, &ObjectOptions::default())
+                .await
+                .expect("read persisted metadata");
+            if let Some(preserved) = preserved {
+                assert_eq!(stored.etag.as_deref(), Some(preserved.as_str()));
+                assert!(!rustfs_utils::http::etag_can_be_listed(&stored.user_defined));
+            } else {
+                assert_ne!(stored.etag.as_ref(), Some(&plaintext_md5));
+                assert_eq!(
+                    rustfs_utils::http::get_consistent_str(
+                        &stored.user_defined,
+                        rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG
+                    ),
+                    Some("v1")
+                );
+                assert!(rustfs_utils::http::etag_can_be_listed(&stored.user_defined));
+            }
+        }
+    }
+
    #[derive(Clone, Default)]
    struct CapturedLogs(Arc<Mutex<Vec<u8>>>);

@@ -901,6 +901,7 @@ impl ECStore {
            }
        };
        let confirmed_missing = existing_bucket_info.is_none();
+        let mut pending_creation_metadata = None;
        if confirmed_missing
            && !opts.no_lock
            && !is_meta_bucketname(bucket)
@@ -921,13 +922,25 @@ impl ECStore {
                    ),
                    other => other,
                })?;
-            return Err(StorageError::InvalidArgument(
-                "CreateBucket".to_string(),
-                "bucket".to_string(),
-                format!("orphaned bucket generation {old_incarnation} requires administrator recovery before recreation"),
-            ));
+            let (metadata, persisted) = metadata_sys::get_config_from_disk_with_presence_in(&self.ctx, bucket).await?;
+            let metadata_store = metadata_sys::object_store_in(&self.ctx).await?;
+            let pending_intent = persisted
+                && metadata.bucket_incarnation_id == old_incarnation
+                && metadata.needs_bucket_creation_commit()
+                && !crate::bucket::retirement::is_retired(metadata_store, bucket, old_incarnation).await?;
+            if pending_intent {
+                pending_creation_metadata = Some(metadata);
+            } else {
+                return Err(StorageError::InvalidArgument(
+                    "CreateBucket".to_string(),
+                    "bucket".to_string(),
+                    format!("orphaned bucket generation {old_incarnation} requires administrator recovery before recreation"),
+                ));
+            }
        }
-        let existing_metadata = if opts.force_create && !confirmed_missing && !is_meta_bucketname(bucket) {
+        let existing_metadata = if let Some(metadata) = pending_creation_metadata {
+            Some(metadata)
+        } else if opts.force_create && !confirmed_missing && !is_meta_bucketname(bucket) {
            let (mut metadata, persisted) = metadata_sys::get_config_from_disk_with_presence_in(&self.ctx, bucket).await?;
            if !persisted {
                metadata = BucketMetadata::new(bucket);
@@ -1347,6 +1360,23 @@ impl ECStore {
        } else {
            None
        };
+        // Match metadata migration: lifecycle -> metadata transaction ->
+        // bucket namespace. Holding the transaction while waiting for the
+        // namespace prevents a migration writer and force delete from
+        // acquiring these two locks in opposite orders.
+        let object_lock_metadata_transaction = if opts.force && !is_meta_bucketname(bucket) {
+            Some(
+                await_bucket_namespace_operation(
+                    bucket_lifecycle_guard.as_ref(),
+                    bucket,
+                    "Object Lock metadata transaction acquisition",
+                    crate::bucket::metadata_sys::acquire_bucket_metadata_transaction_read_lock_in(&self.ctx, bucket),
+                )
+                .await?,
+            )
+        } else {
+            None
+        };
        let ns_guard = if !opts.no_lock {
            let ns_lock = self.new_ns_lock(bucket, bucket).await?;
            Some(
@@ -1402,6 +1432,29 @@ impl ECStore {
            }
        }

+        // Keep configuration changes fenced until data deletion finishes;
+        // a force delete cannot bypass any Object Lock protection.
+        let object_lock_metadata_guard = if bucket_exists && let Some(guard) = object_lock_metadata_transaction {
+            let (state, incarnation, _) =
+                crate::bucket::metadata_sys::get_object_lock_config_and_incarnation_from_disk_with_guard_in(
+                    &self.ctx,
+                    bucket,
+                    Some(ns_guard.as_ref().unwrap_or(&guard)),
+                )
+                .await?;
+            match state {
+                crate::bucket::metadata_sys::ObjectLockConfigState::ConfirmedAbsent => Some((guard, incarnation)),
+                crate::bucket::metadata_sys::ObjectLockConfigState::Configured { .. } => {
+                    return Err(StorageError::PrefixAccessDenied(bucket.to_owned(), String::new()));
+                }
+                crate::bucket::metadata_sys::ObjectLockConfigState::Fabricated => {
+                    return Err(StorageError::FileCorrupt);
+                }
+            }
+        } else {
+            None
+        };
+
        if sr_delete && !bucket_exists {
            delete_opts.force_if_empty = true;
        }
@@ -1415,7 +1468,12 @@ impl ECStore {
                ns_guard.as_ref(),
                bucket,
                "bucket delete marker creation",
-                self.mark_bucket_deleted(bucket),
+                await_bucket_namespace_operation(
+                    object_lock_metadata_guard.as_ref().map(|(guard, _)| guard),
+                    bucket,
+                    "Object Lock configuration fence during bucket retirement",
+                    self.mark_bucket_deleted(bucket),
+                ),
            )
            .await?;
        }
@@ -1424,9 +1482,13 @@ impl ECStore {
        // Legacy buckets without a stamp cannot produce retirement authority.
        let retirement = if bucket_exists && bucket_lifecycle_guard.is_some() && !is_meta_bucketname(bucket) {
            if let Some(store) = metadata_sys::object_store_if_initialized_in(&self.ctx).await {
-                crate::bucket::metadata::load_bucket_incarnation(store.clone(), bucket)
-                    .await?
-                    .map(|incarnation| (store, incarnation))
+                if let Some((_, incarnation)) = object_lock_metadata_guard.as_ref() {
+                    Some((store, *incarnation))
+                } else {
+                    crate::bucket::metadata::load_bucket_incarnation(store.clone(), bucket)
+                        .await?
+                        .map(|incarnation| (store, incarnation))
+                }
            } else {
                None
            }
@@ -1438,13 +1500,19 @@ impl ECStore {
            bucket_lifecycle_guard.as_ref(),
            bucket,
            "physical bucket deletion",
-            run_physical_bucket_deletion(ns_guard.as_ref(), bucket, async {
-                self.delete_bucket_on_sets(bucket, &delete_opts)
-                    .await
-                    .map_err(|err| to_object_err(err, vec![bucket]))
-            }),
+            await_bucket_namespace_operation(
+                object_lock_metadata_guard.as_ref().map(|(guard, _)| guard),
+                bucket,
+                "Object Lock configuration fence during bucket deletion",
+                run_physical_bucket_deletion(ns_guard.as_ref(), bucket, async {
+                    self.delete_bucket_on_sets(bucket, &delete_opts)
+                        .await
+                        .map_err(|err| to_object_err(err, vec![bucket]))
+                }),
+            ),
        )
        .await;
+        drop(object_lock_metadata_guard);
        if let Err(err) = delete_result
            && (!sr_delete || !is_err_strict_volume_not_found(&err))
        {
@@ -3689,6 +3757,116 @@ mod tests {
        assert!(!any_disk_path_exists(&disk_paths, bucket_deleted_marker_volume(&bucket)).await);
    }

+    #[tokio::test]
+    #[serial]
+    async fn force_bucket_delete_preserves_object_lock_versions() {
+        let (_temp_dir, store) = setup_bucket_quorum_test_env(&[4], Some(2)).await;
+        metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        let bucket = format!("force-delete-worm-{}", Uuid::new_v4().simple());
+        store
+            .make_bucket(
+                &bucket,
+                &MakeBucketOptions {
+                    lock_enabled: true,
+                    versioning_enabled: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("create Object Lock bucket");
+        let until = (OffsetDateTime::now_utc() + time::Duration::days(365))
+            .format(&time::format_description::well_known::Rfc3339)
+            .expect("format future retention date");
+        let written = store
+            .put_object(
+                &bucket,
+                "retained",
+                &mut PutObjReader::from_vec(b"retained bytes".to_vec()),
+                &ObjectOptions {
+                    versioned: true,
+                    user_defined: HashMap::from([
+                        ("x-amz-object-lock-mode".to_owned(), "COMPLIANCE".to_owned()),
+                        ("x-amz-object-lock-retain-until-date".to_owned(), until),
+                    ]),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("write retained object");
+        for operation in [SRBucketDeleteOp::NoOp, SRBucketDeleteOp::MarkDelete, SRBucketDeleteOp::Purge] {
+            let err = store
+                .delete_bucket(
+                    &bucket,
+                    &DeleteBucketOptions {
+                        force: true,
+                        srdelete_op: operation,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect_err("force deletion must not bypass Object Lock");
+            assert!(matches!(err, StorageError::PrefixAccessDenied(_, _)), "unexpected denial: {err}");
+            let info = store
+                .get_object_info(
+                    &bucket,
+                    "retained",
+                    &ObjectOptions {
+                        version_id: written.version_id.map(|version| version.to_string()),
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("retained version must survive the rejected bucket deletion");
+            assert_eq!(info.etag, written.etag);
+            assert_eq!(info.size, written.size);
+            assert_eq!(info.version_id, written.version_id);
+        }
+        metadata_sys::inject_object_lock_disk_read_error_in(&store.ctx, &bucket)
+            .await
+            .expect("inject authoritative metadata read failure");
+        let error = store
+            .delete_bucket(
+                &bucket,
+                &DeleteBucketOptions {
+                    force: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect_err("an unreadable Object Lock configuration must fail closed");
+        assert!(error.to_string().contains("injected Object Lock metadata disk read failure"));
+        let info = store
+            .get_object_info(&bucket, "retained", &ObjectOptions::default())
+            .await
+            .expect("metadata read failure must leave the retained object intact");
+        assert_eq!(info.version_id, written.version_id);
+
+        let unlocked = format!("force-delete-unlocked-{}", Uuid::new_v4().simple());
+        store
+            .make_bucket(&unlocked, &MakeBucketOptions::default())
+            .await
+            .expect("create unlocked control bucket");
+        store
+            .put_object(
+                &unlocked,
+                "ordinary",
+                &mut PutObjReader::from_vec(b"ordinary bytes".to_vec()),
+                &ObjectOptions::default(),
+            )
+            .await
+            .expect("write ordinary control object");
+        store
+            .delete_bucket(
+                &unlocked,
+                &DeleteBucketOptions {
+                    force: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("force deletion must still work for an unlocked bucket");
+    }
+
    #[tokio::test]
    #[serial]
    async fn bucket_delete_mark_delete_rejects_hidden_object_paths_without_force() {
@@ -18318,7 +18318,7 @@ mod tests {
        let bucket = format!("lock-create-intent-retry-{}", uuid::Uuid::new_v4());
        let mut intent = crate::bucket::metadata::BucketMetadata::new(&bucket);
        intent.lock_enabled = true;
-        crate::bucket::metadata_sys::set_new_bucket_metadata_in(&ctx, intent)
+        crate::bucket::metadata_sys::set_new_bucket_metadata_intent_in(&ctx, intent)
            .await
            .expect("persist pre-visibility lock intent");
        assert!(
@@ -448,8 +448,8 @@ const MAX_LIST_OBJECTS_METADATA_FAST_STALENESS_MS: u64 = 60_000;
 const LIST_OBJECTS_INDEX_PROVIDER_WALKER_KEY_ONLY: &str = "walker_key_only";
 const LIST_OBJECTS_INDEX_PROVIDER_PERSISTENT_KEY_ONLY: &str = "persistent_key_only";
 const LIST_OBJECTS_INDEX_PROVIDER_PERSISTENT_KEY_ONLY_DEFAULT_GENERATION: &str = "persistent-key-only";
-const PERSISTENT_KEY_ONLY_INDEX_FORMAT_VERSION: u8 = 2;
-const PERSISTENT_KEY_ONLY_INDEX_HEADER: &str = "# rustfs-listobjects-key-only-v2";
+const PERSISTENT_KEY_ONLY_INDEX_FORMAT_VERSION: u8 = 3;
+const PERSISTENT_KEY_ONLY_INDEX_HEADER: &str = "# rustfs-listobjects-key-only-v3";
 const PERSISTENT_KEY_ONLY_INDEX_BUCKET_HEADER: &str = "# bucket=";
 const PERSISTENT_KEY_ONLY_INDEX_GENERATION_HEADER: &str = "# generation=";
 const PERSISTENT_KEY_ONLY_INDEX_CHECKPOINT_HEADER: &str = "# checkpoint_high_water_mark=";
@@ -656,7 +656,10 @@ impl PersistentListMetadataObject {
            name: object.name.clone(),
            size: object.size,
            mod_time: object.mod_time,
-            etag: object.etag.clone(),
+            etag: object
+                .etag
+                .clone()
+                .filter(|_| rustfs_utils::http::etag_can_be_listed(&object.user_defined)),
            storage_class: object.storage_class.clone(),
        }
    }
@@ -2361,24 +2364,43 @@ fn list_objects_paginate(
    (objects, prefixes, is_truncated, next_marker, next_version_idmarker)
 }

-fn list_objects_paginate_versions<'a>(
-    mut get_objects: Vec<ObjectInfo>,
-    delimiter: &Option<String>,
+struct VersionedListPagination<'a, I> {
+    delimiter: &'a Option<String>,
    max_keys: i32,
    disk_has_more: bool,
-    cache_id: Option<&str>,
-    marker: Option<&str>,
-    last_scanned_key: Option<&str>,
-    raw_keys: impl Iterator<Item = &'a str>,
-) -> (Vec<ObjectInfo>, Vec<String>, bool, Option<String>, Option<String>) {
-    filter_versions_common_prefixes_after_marker(&mut get_objects, delimiter.as_deref(), marker);
-    let prefix_at_page_boundary =
-        versioned_prefix_page_boundary_key(&get_objects, raw_keys, delimiter.as_deref(), max_keys, disk_has_more);
+    cache_id: Option<&'a str>,
+    marker: Option<&'a str>,
+    last_scanned_key: Option<&'a str>,
+    raw_keys: I,
+}

-    let (objects, prefixes, is_truncated, mut next_marker, mut next_version_idmarker) =
-        list_objects_paginate(get_objects, delimiter, max_keys, disk_has_more, cache_id, true, last_scanned_key);
+fn list_objects_paginate_versions<'a, I>(
+    mut get_objects: Vec<ObjectInfo>,
+    pagination: VersionedListPagination<'a, I>,
+) -> (Vec<ObjectInfo>, Vec<String>, bool, Option<String>, Option<String>)
+where
+    I: Iterator<Item = &'a str>,
+{
+    filter_versions_common_prefixes_after_marker(&mut get_objects, pagination.delimiter.as_deref(), pagination.marker);
+    let prefix_at_page_boundary = versioned_prefix_page_boundary_key(
+        &get_objects,
+        pagination.raw_keys,
+        pagination.delimiter.as_deref(),
+        pagination.max_keys,
+        pagination.disk_has_more,
+    );
+
+    let (objects, prefixes, is_truncated, mut next_marker, mut next_version_idmarker) = list_objects_paginate(
+        get_objects,
+        pagination.delimiter,
+        pagination.max_keys,
+        pagination.disk_has_more,
+        pagination.cache_id,
+        true,
+        pagination.last_scanned_key,
+    );
    if is_truncated && let Some(raw_key) = prefix_at_page_boundary {
-        next_marker = Some(append_list_cache_id_to_marker(raw_key, cache_id));
+        next_marker = Some(append_list_cache_id_to_marker(raw_key, pagination.cache_id));
        next_version_idmarker = None;
    }

@@ -4326,13 +4348,15 @@ impl ECStore {

        let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = list_objects_paginate_versions(
            get_objects,
-            &delimiter,
-            max_keys,
-            disk_has_more,
-            next_cache_id.as_deref(),
-            opts.marker.as_deref(),
-            last_scanned_key.as_deref(),
-            entries.entries().iter().map(|entry| entry.name.as_str()),
+            VersionedListPagination {
+                delimiter: &delimiter,
+                max_keys,
+                disk_has_more,
+                cache_id: next_cache_id.as_deref(),
+                marker: opts.marker.as_deref(),
+                last_scanned_key: last_scanned_key.as_deref(),
+                raw_keys: entries.entries().iter().map(|entry| entry.name.as_str()),
+            },
        );

        Ok(ListObjectVersionsInfo {
@@ -5754,13 +5778,15 @@ impl Sets {

        let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = list_objects_paginate_versions(
            get_objects,
-            &delimiter,
-            max_keys,
-            disk_has_more,
-            next_cache_id.as_deref(),
-            opts.marker.as_deref(),
-            last_scanned_key.as_deref(),
-            entries.entries().iter().map(|entry| entry.name.as_str()),
+            VersionedListPagination {
+                delimiter: &delimiter,
+                max_keys,
+                disk_has_more,
+                cache_id: next_cache_id.as_deref(),
+                marker: opts.marker.as_deref(),
+                last_scanned_key: last_scanned_key.as_deref(),
+                raw_keys: entries.entries().iter().map(|entry| entry.name.as_str()),
+            },
        );

        Ok(ListObjectVersionsInfo {
@@ -6568,13 +6594,15 @@ impl SetDisks {

        let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = list_objects_paginate_versions(
            get_objects,
-            &delimiter,
-            max_keys,
-            disk_has_more,
-            next_cache_id.as_deref(),
-            opts.marker.as_deref(),
-            last_scanned_key.as_deref(),
-            entries.entries().iter().map(|entry| entry.name.as_str()),
+            VersionedListPagination {
+                delimiter: &delimiter,
+                max_keys,
+                disk_has_more,
+                cache_id: next_cache_id.as_deref(),
+                marker: opts.marker.as_deref(),
+                last_scanned_key: last_scanned_key.as_deref(),
+                raw_keys: entries.entries().iter().map(|entry| entry.name.as_str()),
+            },
        );

        Ok(ListObjectVersionsInfo {
@@ -8848,13 +8876,15 @@ mod test {
            let get_objects = fold_delimiter_page(&window, "", "-");
            let (objects, prefixes, is_truncated, next_marker, _v) = super::list_objects_paginate_versions(
                get_objects,
-                &delimiter,
-                max_keys,
-                disk_has_more,
-                None,
-                marker.as_deref(),
-                last_scanned.as_deref(),
-                window.iter().map(String::as_str),
+                super::VersionedListPagination {
+                    delimiter: &delimiter,
+                    max_keys,
+                    disk_has_more,
+                    cache_id: None,
+                    marker: marker.as_deref(),
+                    last_scanned_key: last_scanned.as_deref(),
+                    raw_keys: window.iter().map(String::as_str),
+                },
            );

            assert!(
@@ -8927,13 +8957,15 @@ mod test {
            let get_objects = fold_delimiter_page(&window, "", "-");
            let (objects, prefixes, is_truncated, next_marker, _) = super::list_objects_paginate_versions(
                get_objects,
-                &delimiter,
-                max_keys,
-                disk_has_more,
-                None,
-                marker.as_deref(),
-                last_scanned.as_deref(),
-                window.iter().map(String::as_str),
+                super::VersionedListPagination {
+                    delimiter: &delimiter,
+                    max_keys,
+                    disk_has_more,
+                    cache_id: None,
+                    marker: marker.as_deref(),
+                    last_scanned_key: last_scanned.as_deref(),
+                    raw_keys: window.iter().map(String::as_str),
+                },
            );

            assert_eq!(objects.len() + prefixes.len(), 1, "a page must respect max-keys");
@@ -9176,6 +9208,38 @@ mod test {
        assert_eq!(parsed, object);
    }

+    #[test]
+    fn persistent_listing_snapshots_cannot_restore_legacy_encrypted_fingerprints() {
+        let mut object = ObjectInfo {
+            name: "encrypted".to_owned(),
+            etag: Some("fbfcbcc6d2f035411e6268c8b119593e".to_owned()),
+            user_defined: Arc::new(HashMap::from([(
+                "x-amz-server-side-encryption-customer-algorithm".to_owned(),
+                "AES256".to_owned(),
+            )])),
+            ..Default::default()
+        };
+        let snapshot = PersistentListMetadataObject::from_object_info(&object);
+        assert!(snapshot.to_object_info("bucket").etag.is_none());
+        let mut old_snapshot = snapshot;
+        old_snapshot.etag = object.etag.clone();
+        let old_contents = format!(
+            "# rustfs-listobjects-key-only-v2\n{}\n",
+            encode_persistent_list_metadata_object(&old_snapshot)
+        );
+        let old_index = parse_persistent_key_only_index(&old_contents);
+        assert_ne!(old_index.format_version, PERSISTENT_KEY_ONLY_INDEX_FORMAT_VERSION);
+
+        rustfs_utils::http::metadata_compat::insert_str(
+            Arc::make_mut(&mut object.user_defined),
+            rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG,
+            "v1".to_owned(),
+        );
+        assert_eq!(PersistentListMetadataObject::from_object_info(&object).etag, object.etag);
+        object.user_defined = Arc::new(HashMap::new());
+        assert_eq!(PersistentListMetadataObject::from_object_info(&object).etag, object.etag);
+    }
+
    #[test]
    fn persistent_key_only_index_parses_legacy_keys_and_metadata_snapshots() {
        let object = PersistentListMetadataObject {
@@ -5351,13 +5351,41 @@ impl ECStore {
        tier_journal_api: Option<Arc<ECStore>>,
    ) -> (Vec<DeletedObject>, Vec<Option<Error>>, Vec<Option<DeleteAccounting>>) {
        Box::pin(async move {
+            let count = objects.len();
+            let mut deleted = vec![DeletedObject::default(); count];
+            let mut errors: Vec<Option<Error>> = (0..count).map(|_| None).collect();
+            let mut accounting = vec![None; count];
+            let mut valid_objects = Vec::with_capacity(count);
+            let mut valid_indices = Vec::with_capacity(count);
+            // Validate the literal key before encoding it or taking locks. A
+            // dot-segment key must never reach filesystem path resolution.
+            for (index, object) in objects.into_iter().enumerate() {
+                match check_del_obj_args(bucket, &object.object_name) {
+                    Ok(()) => {
+                        valid_indices.push(index);
+                        valid_objects.push(object);
+                    }
+                    Err(err) => errors[index] = Some(err),
+                }
+            }
+            if valid_objects.is_empty() {
+                return (deleted, errors, accounting);
+            }
            let mut opts = opts;
            let receipt_sink = install_tier_free_version_receipt_sink(&mut opts);
-            let result = self
-                .handle_delete_objects_with_journal_and_accounting_inner(bucket, objects, opts, tier_journal_api)
+            let (valid_deleted, valid_errors, valid_accounting) = self
+                .handle_delete_objects_with_journal_and_accounting_inner(bucket, valid_objects, opts, tier_journal_api)
                .await;
            enqueue_recorded_tier_free_versions(self, receipt_sink).await;
-            result
+            for (index, ((object, error), receipt)) in valid_indices
+                .into_iter()
+                .zip(valid_deleted.into_iter().zip(valid_errors).zip(valid_accounting))
+            {
+                deleted[index] = object;
+                errors[index] = error;
+                accounting[index] = receipt;
+            }
+            (deleted, errors, accounting)
        })
        .await
    }
@@ -9094,6 +9122,58 @@ mod tests {
        (dirs, store)
    }

+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn batch_delete_rejects_dot_segments_and_keeps_valid_result_positions() {
+        let bucket = "batch-delete-key-validation";
+        let (_dirs, store) = multipool_version_test_store(bucket).await;
+        let options = ObjectOptions {
+            versioned: true,
+            ..Default::default()
+        };
+        let victim = store
+            .put_object(bucket, "victim/object", &mut PutObjReader::from_vec(b"victim".to_vec()), &options)
+            .await
+            .expect("write protected-prefix control object");
+        store
+            .put_object(bucket, "allowed/object", &mut PutObjReader::from_vec(b"allowed".to_vec()), &options)
+            .await
+            .expect("write valid deletion control object");
+        let keys = [
+            "allowed/../victim/object",
+            "allowed/object",
+            "a/./b",
+            "../other-bucket/object",
+        ];
+        let objects = keys
+            .iter()
+            .map(|key| ObjectToDelete {
+                object_name: (*key).to_owned(),
+                ..Default::default()
+            })
+            .collect();
+        let (deleted, errors) = store.handle_delete_objects(bucket, objects, options).await;
+        assert_eq!(deleted.len(), keys.len());
+        assert_eq!(errors.len(), keys.len());
+        for index in [0, 2, 3] {
+            assert!(
+                matches!(&errors[index], Some(StorageError::ObjectNameInvalid(_, _))),
+                "invalid key {}: {:?}",
+                keys[index],
+                errors[index]
+            );
+            assert!(!deleted[index].delete_marker);
+        }
+        assert!(errors[1].is_none(), "valid deletion must still succeed: {:?}", errors[1]);
+        assert!(deleted[1].delete_marker);
+        assert_eq!(deleted[1].object_name, "allowed/object");
+        let after = store
+            .get_object_info(bucket, "victim/object", &ObjectOptions::default())
+            .await
+            .expect("the resolved victim path must remain visible");
+        assert_eq!(after.version_id, victim.version_id);
+        assert_eq!(after.etag, victim.etag);
+    }
+
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn retired_marker_converges_after_bucket_recreation_on_sixteen_disks() {
        let bucket = "retired-marker-c11";
@@ -392,7 +392,7 @@ impl PartialWrites {
                    object: anchor.object.clone(),
                    version_id: anchor.version_id,
                    scope: anchor.scope,
-                    delete_marker_purge: anchor.delete_marker_purge.clone(),
+                    delete_marker_purge: anchor.delete_marker_purge,
                    lease: Some(anchor.lease),
                    bucket_incarnation_id: anchor.bucket_incarnation_id,
                    disposition: MrfVerifiedRepairDisposition::AuthoritativelyAbsent,
@@ -274,9 +274,9 @@ impl HealTask {
                    let mut progress = self.progress.write().await;
                    progress.update_stage(3, 3);
                }
-                Err(Error::TaskExecutionFailed {
-                    message: format!("Failed to heal bucket {bucket}: {e}"),
-                })
+                // Keep storage/quorum identity for the cluster retry classifier.
+                // The structured log above already carries the bucket context.
+                Err(e)
            }
        }
    }
@@ -727,6 +727,85 @@ mod canonical_outcome {
        );
    }

+    #[tokio::test(start_paused = true)]
+    async fn cluster_bucket_write_quorum_retries_preserve_budget_and_truthful_coverage() {
+        for (failures, succeeds) in [(1, true), (4, false)] {
+            let storage = Arc::new(MockStorage {
+                listed_buckets: Mutex::new(Some(vec!["bucket-a".to_string(), "bucket-b".to_string()])),
+                typed_bucket_heal_errors: Mutex::new(HashMap::from([(
+                    "bucket-a".to_string(),
+                    (0..failures)
+                        .map(|_| Error::Storage(EcstoreError::ErasureWriteQuorum))
+                        .collect(),
+                )])),
+                ..Default::default()
+            });
+            let task = HealTask::from_request(
+                HealRequest::new(
+                    HealType::Cluster,
+                    HealOptions {
+                        recursive: true,
+                        timeout: None,
+                        ..Default::default()
+                    },
+                    HealPriority::Normal,
+                ),
+                storage.clone(),
+            );
+            let result = task.execute().await;
+            assert_eq!(
+                result.is_ok(),
+                succeeds,
+                "original cluster task must honor its bounded bucket retry budget"
+            );
+            let outcome = task.get_outcome().await;
+            if succeeds {
+                assert_eq!(outcome.execution, HealExecutionOutcome::Completed);
+                assert_eq!(outcome.coverage, HealTraversalCoverage::Complete);
+            } else {
+                assert_eq!(outcome.execution, HealExecutionOutcome::Aborted(HealAbortReason::Untraversable));
+                assert_eq!(outcome.coverage, HealTraversalCoverage::Partial);
+                let failure = task.take_batch_failure().await.expect("exhausted bucket retains its failure");
+                assert_eq!((failure.failed, failure.retryable, failure.permanent), (1, 1, 0));
+            }
+            let calls = storage.bucket_heal_calls.lock().expect("original bucket calls");
+            assert_eq!(
+                calls.iter().filter(|bucket| bucket.as_str() == "bucket-a").count(),
+                if succeeds { 2 } else { 4 }
+            );
+            assert_eq!(calls.last().map(String::as_str), Some("bucket-b"), "later buckets must still be visited");
+        }
+    }
+
+    #[tokio::test]
+    async fn bucket_prepass_preserves_typed_quorum_and_terminal_access_denial() {
+        for error in [
+            Error::Storage(EcstoreError::ErasureWriteQuorum),
+            Error::Disk(DiskError::FileAccessDenied),
+        ] {
+            let retryable = error.is_recoverable_heal();
+            let storage = Arc::new(MockStorage {
+                typed_bucket_heal_errors: Mutex::new(HashMap::from([("bucket-a".to_string(), VecDeque::from([error]))])),
+                ..Default::default()
+            });
+            let task = bucket_task(storage.clone());
+            let observed = task
+                .heal_bucket("bucket-a")
+                .await
+                .expect_err("bucket prepass must retain the storage error");
+            assert!(matches!(
+                observed,
+                Error::Storage(EcstoreError::ErasureWriteQuorum) | Error::Disk(DiskError::FileAccessDenied)
+            ));
+            assert_eq!(observed.is_recoverable_heal(), retryable);
+            assert_eq!(storage.bucket_heal_calls.lock().expect("bucket calls").as_slice(), ["bucket-a"]);
+            assert!(
+                storage.heal_object_calls.lock().expect("object calls").is_empty(),
+                "failed prepass must not heal objects"
+            );
+        }
+    }
+
    #[tokio::test(start_paused = true)]
    async fn cancellation_and_deadline_leave_partial_coverage() {
        for cancel in [false, true] {
@@ -1482,6 +1561,7 @@ struct MockStorage {
    include_object_dir_candidate: Mutex<bool>,
    listed_buckets: Mutex<Option<Vec<String>>>,
    bucket_heal_errors: Mutex<HashMap<String, VecDeque<&'static str>>>,
+    typed_bucket_heal_errors: Mutex<HashMap<String, VecDeque<Error>>>,
    bucket_heal_calls: Mutex<Vec<String>>,
    block_heal_object: Mutex<bool>,
    resume_disk: Mutex<Option<DiskStore>>,
@@ -2247,6 +2327,15 @@ impl HealStorageAPI for MockStorage {
    async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result<HealResultItem> {
        self.bucket_heal_calls.lock().unwrap().push(bucket.to_string());
        self.bucket_heal_opts.lock().unwrap().push(*opts);
+        if let Some(error) = self
+            .typed_bucket_heal_errors
+            .lock()
+            .unwrap()
+            .get_mut(bucket)
+            .and_then(VecDeque::pop_front)
+        {
+            return Err(error);
+        }
        if let Some(message) = self
            .bucket_heal_errors
            .lock()
@@ -563,9 +563,31 @@ fn admin_root_completes_after_unversioned_delete() {
    run_delete_case(DeleteCase::Unversioned);
 }

-#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[test]
 #[serial]
-async fn admin_root_rejects_recreated_bucket_before_successor_repair() {
+fn admin_root_rejects_recreated_bucket_before_successor_repair() {
+    // Incarnation-bound repairs use a spawned owner; retain the server's stack
+    // budget for the whole scenario. block_on alone would poll that future on
+    // libtest's smaller calling-thread stack.
+    let stack_size = 8 * 1024 * 1024;
+    std::thread::Builder::new()
+        .name("heal-concurrent-delete".to_owned())
+        .stack_size(stack_size)
+        .spawn(move || {
+            let runtime = tokio::runtime::Builder::new_multi_thread()
+                .worker_threads(4)
+                .thread_stack_size(stack_size)
+                .enable_all()
+                .build()
+                .expect("concurrent delete runtime");
+            runtime.block_on(recreated_bucket_repair());
+        })
+        .expect("concurrent delete scenario thread starts")
+        .join()
+        .expect("concurrent delete scenario must not panic");
+}
+
+async fn recreated_bucket_repair() {
    let directory = tempfile::tempdir().expect("recreated bucket fixture");
    let env = TestECStoreEnv::builder().base_dir(directory.path()).build().await;
    env.make_bucket(BUCKET, false).await;
@@ -30,6 +30,7 @@ workspace = true

 [features]
 default = []
+test-util = []
 hotpath = [
    "hotpath/hotpath",
    "hotpath/tokio",
@@ -18,7 +18,6 @@ mod mapper;
 mod model;
 pub mod oidc;
 mod provider;
-mod registry;
 mod transaction;

 pub use binding::FederatedSessionBinding;
@@ -27,10 +26,9 @@ pub use mapper::{CoreFederatedAuthorizationMapper, FederatedAuthorizationRules};
 pub(crate) use mapper::{FederatedAuthorizationRule, FederatedAuthorizationRuleRef};
 pub(crate) use model::OpaqueLogoutContinuation;
 pub use model::{
-    FederatedAuthorization, FederatedClaims, FederatedCodeExchange, FederatedLoginSession, FederatedProviderRef,
-    FederatedSession, FederatedSessionTransaction, OIDC_VIRTUAL_PARENT_CLAIM, VerifiedFederatedIdentity,
+    FederatedAuthorization, FederatedClaims, FederatedLoginSession, FederatedProviderRef, FederatedSession,
+    FederatedSessionTransaction, OIDC_VIRTUAL_PARENT_CLAIM, VerifiedFederatedIdentity,
 };
-pub use provider::{FederatedIdentityProvider, FederatedProviderView, FederatedRedirectPolicy};
-pub(crate) use provider::{StandardOidcAuthentication, VerifiedFederatedCodeExchange};
-pub use registry::FederatedIdentityRegistry;
+pub(crate) use provider::{FederatedProviderQuery, StandardOidcAuthentication, VerifiedFederatedCodeExchange};
+pub use provider::{FederatedProviderView, FederatedRedirectPolicy};
 pub use transaction::FederatedIdentityService;
@@ -164,40 +164,6 @@ impl FederatedAuthorization {
    }
 }

-pub struct FederatedCodeExchange {
-    pub authorization: FederatedAuthorization,
-    pub redirect_after: Option<String>,
-    pub id_token: String,
-}
-
-impl FederatedCodeExchange {
-    pub(crate) fn into_parts(self) -> (FederatedAuthorization, Option<String>, OpaqueLogoutContinuation) {
-        let Self {
-            authorization,
-            redirect_after,
-            id_token,
-        } = self;
-        let continuation = OpaqueLogoutContinuation {
-            provider: FederatedProviderRef::new(authorization.provider_id.clone()),
-            id_token,
-        };
-        (authorization, redirect_after, continuation)
-    }
-}
-
-impl fmt::Debug for FederatedCodeExchange {
-    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
-        formatter
-            .debug_struct("FederatedCodeExchange")
-            .field("provider_id", &self.authorization.provider_id)
-            .field("policy_count", &self.authorization.policies.len())
-            .field("group_count", &self.authorization.groups.len())
-            .field("redirect_after_present", &self.redirect_after.is_some())
-            .field("id_token_present", &!self.id_token.is_empty())
-            .finish()
-    }
-}
-
 /// Carries the provider-bound ID token between code exchange and logout token creation.
 pub(crate) struct OpaqueLogoutContinuation {
    provider: FederatedProviderRef,
@@ -397,27 +363,13 @@ mod tests {

    #[test]
    fn logout_continuation_binds_provider_and_redacts_token() {
-        let mut authorization = authorization(Vec::new(), Vec::new());
-        authorization
-            .claims
-            .raw
-            .insert("attribute_poison".to_string(), Value::String("secret-attribute".to_string()));
-        let exchange = FederatedCodeExchange {
-            authorization,
-            redirect_after: Some("/console?poison=redirect-secret".to_string()),
-            id_token: "secret-id-token".to_string(),
-        };
-        let debug = format!("{exchange:?}");
-        let (_, _, continuation) = exchange.into_parts();
+        let continuation =
+            OpaqueLogoutContinuation::new(FederatedProviderRef::new("standard_oidc".to_string()), "secret-id-token".to_string());
        let continuation_debug = format!("{continuation:?}");
        let (provider, id_token) = continuation.into_parts();

        assert_eq!(provider.as_str(), "standard_oidc");
        assert_eq!(id_token, "secret-id-token");
-        assert!(!debug.contains("secret-id-token"));
-        assert!(!debug.contains("secret-attribute"));
-        assert!(!debug.contains("redirect-secret"));
-        assert!(debug.contains("id_token_present: true"));
        assert!(!continuation_debug.contains("secret-id-token"));
        assert!(continuation_debug.contains("[REDACTED]"));
    }
@@ -16,10 +16,9 @@ use super::config::OidcConfigQuery;
 use super::{claims, config, discovery, http};
 use crate::{
    federation::{
-        CoreFederatedAuthorizationMapper, FederatedAuthorization, FederatedAuthorizationRules, FederatedCodeExchange,
-        FederatedIdentityProvider, FederatedIdentityService, FederatedProviderRef, FederatedProviderView,
-        FederatedRedirectPolicy, FederationError, OpaqueLogoutContinuation, Result, StandardOidcAuthentication,
-        VerifiedFederatedCodeExchange,
+        CoreFederatedAuthorizationMapper, FederatedAuthorizationRules, FederatedIdentityService, FederatedProviderQuery,
+        FederatedProviderRef, FederatedProviderView, FederatedRedirectPolicy, FederationError, OpaqueLogoutContinuation, Result,
+        StandardOidcAuthentication, VerifiedFederatedCodeExchange,
    },
    oidc::{OidcConfigSnapshot, OidcSys},
 };
@@ -45,8 +44,8 @@ impl StandardOidcAdapter {

    pub fn into_service(self: Arc<Self>, mapper: CoreFederatedAuthorizationMapper) -> FederatedIdentityService {
        let authentication: Arc<dyn StandardOidcAuthentication> = self.clone();
-        let provider: Arc<dyn FederatedIdentityProvider> = self;
-        FederatedIdentityService::from_standard_oidc_parts(provider, authentication, mapper)
+        let provider_query: Arc<dyn FederatedProviderQuery> = self;
+        FederatedIdentityService::from_standard_oidc_parts(provider_query, authentication, mapper)
    }
 }

@@ -58,6 +57,10 @@ impl OidcConfigQuery for StandardOidcAdapter {

 #[async_trait::async_trait]
 impl StandardOidcAuthentication for StandardOidcAdapter {
+    async fn authorize_url(&self, provider_id: &str, redirect_uri: &str, redirect_after: Option<String>) -> Result<String> {
+        http::authorize_url(&self.oidc, provider_id, redirect_uri, redirect_after).await
+    }
+
    async fn exchange_identity(&self, state: &str, code: &str, redirect_uri: &str) -> Result<VerifiedFederatedCodeExchange> {
        let (oidc_claims, provider_id, session, id_token) = http::exchange_code(&self.oidc, state, code, redirect_uri).await?;
        let provider = FederatedProviderRef::new(provider_id);
@@ -81,10 +84,13 @@ impl StandardOidcAuthentication for StandardOidcAdapter {
        let (provider, id_token) = continuation.into_parts();
        http::create_logout_token(&self.oidc, provider.as_str(), &id_token).await
    }
+
+    async fn build_logout_url(&self, logout_token: &str, post_logout_redirect_uri: &str) -> Result<Option<String>> {
+        http::build_logout_url(&self.oidc, logout_token, post_logout_redirect_uri).await
+    }
 }

-#[async_trait::async_trait]
-impl FederatedIdentityProvider for StandardOidcAdapter {
+impl FederatedProviderQuery for StandardOidcAdapter {
    fn has_providers(&self) -> bool {
        self.oidc.has_providers()
    }
@@ -97,40 +103,7 @@ impl FederatedIdentityProvider for StandardOidcAdapter {
        discovery::list_visible_providers(&self.oidc)
    }

-    fn provider_config(&self, provider_id: &str) -> Option<&crate::oidc::OidcProviderConfig> {
-        self.oidc.get_provider_config(provider_id)
-    }
-
    fn redirect_policy(&self, provider_id: &str) -> Option<FederatedRedirectPolicy> {
        config::redirect_policy(&self.oidc, provider_id)
    }
-
-    async fn authorize_url(&self, provider_id: &str, redirect_uri: &str, redirect_after: Option<String>) -> Result<String> {
-        http::authorize_url(&self.oidc, provider_id, redirect_uri, redirect_after).await
-    }
-
-    async fn exchange_code(&self, state: &str, code: &str, redirect_uri: &str) -> Result<FederatedCodeExchange> {
-        let exchange = StandardOidcAuthentication::exchange_identity(self, state, code, redirect_uri).await?;
-        let (identity, redirect_after, continuation) = exchange.into_parts();
-        let authorization = CoreFederatedAuthorizationMapper::new(self.authorization_rules()).map(identity);
-        let (_, id_token) = continuation.into_parts();
-        Ok(FederatedCodeExchange {
-            authorization,
-            redirect_after,
-            id_token,
-        })
-    }
-
-    async fn verify_web_identity_token(&self, jwt: &str) -> Result<FederatedAuthorization> {
-        let identity = StandardOidcAuthentication::verify_identity(self, jwt).await?;
-        Ok(CoreFederatedAuthorizationMapper::new(self.authorization_rules()).map(identity))
-    }
-
-    async fn create_logout_token(&self, provider_id: &str, id_token: &str) -> Result<String> {
-        http::create_logout_token(&self.oidc, provider_id, id_token).await
-    }
-
-    async fn build_logout_url(&self, logout_token: &str, post_logout_redirect_uri: &str) -> Result<Option<String>> {
-        http::build_logout_url(&self.oidc, logout_token, post_logout_redirect_uri).await
-    }
 }
@@ -12,8 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-use super::{FederatedAuthorization, FederatedCodeExchange, OpaqueLogoutContinuation, Result, VerifiedFederatedIdentity};
-use crate::oidc::OidcProviderConfig;
+use super::{OpaqueLogoutContinuation, Result, VerifiedFederatedIdentity};
 use serde::{Deserialize, Serialize};

 /// Provider data exposed to login discovery consumers.
@@ -56,41 +55,21 @@ impl VerifiedFederatedCodeExchange {

 #[async_trait::async_trait]
 pub(crate) trait StandardOidcAuthentication: Send + Sync {
+    async fn authorize_url(&self, provider_id: &str, redirect_uri: &str, redirect_after: Option<String>) -> Result<String>;
    async fn exchange_identity(&self, state: &str, code: &str, redirect_uri: &str) -> Result<VerifiedFederatedCodeExchange>;
    async fn verify_identity(&self, jwt: &str) -> Result<VerifiedFederatedIdentity>;
    async fn create_logout_token(&self, continuation: OpaqueLogoutContinuation) -> Result<String>;
+    async fn build_logout_url(&self, logout_token: &str, post_logout_redirect_uri: &str) -> Result<Option<String>>;
 }

-#[async_trait::async_trait]
-pub trait FederatedIdentityProvider: Send + Sync {
+pub(crate) trait FederatedProviderQuery: Send + Sync {
    fn has_providers(&self) -> bool;

    fn list_providers(&self) -> Vec<FederatedProviderView>;

    fn list_visible_providers(&self) -> Vec<FederatedProviderView>;

-    /// Compatibility accessor for existing provider implementations.
-    /// Login redirect consumers should use [`Self::redirect_policy`].
-    fn provider_config(&self, _provider_id: &str) -> Option<&OidcProviderConfig> {
-        None
-    }
-
-    fn redirect_policy(&self, provider_id: &str) -> Option<FederatedRedirectPolicy> {
-        self.provider_config(provider_id).map(|config| FederatedRedirectPolicy {
-            redirect_uri: config.redirect_uri.clone(),
-            allow_request_origin: config.redirect_uri_dynamic,
-        })
-    }
-
-    async fn authorize_url(&self, provider_id: &str, redirect_uri: &str, redirect_after: Option<String>) -> Result<String>;
-
-    async fn exchange_code(&self, state: &str, code: &str, redirect_uri: &str) -> Result<FederatedCodeExchange>;
-
-    async fn verify_web_identity_token(&self, jwt: &str) -> Result<FederatedAuthorization>;
-
-    async fn create_logout_token(&self, provider_id: &str, id_token: &str) -> Result<String>;
-
-    async fn build_logout_url(&self, logout_token: &str, post_logout_redirect_uri: &str) -> Result<Option<String>>;
+    fn redirect_policy(&self, provider_id: &str) -> Option<FederatedRedirectPolicy>;
 }

 #[cfg(test)]
@@ -1,32 +0,0 @@
-// Copyright 2024 RustFS Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use super::FederatedIdentityProvider;
-use std::sync::Arc;
-
-/// Compatibility facade for integrations built against the original provider API.
-/// Production standard OIDC startup composes the adapter, mapper, and service directly.
-pub struct FederatedIdentityRegistry {
-    standard_oidc: Arc<dyn FederatedIdentityProvider>,
-}
-
-impl FederatedIdentityRegistry {
-    pub fn new(standard_oidc: Arc<dyn FederatedIdentityProvider>) -> Self {
-        Self { standard_oidc }
-    }
-
-    pub(crate) fn standard_oidc_arc(&self) -> Arc<dyn FederatedIdentityProvider> {
-        Arc::clone(&self.standard_oidc)
-    }
-}
@@ -13,11 +13,10 @@
 // limitations under the License.

 use super::{
-    CoreFederatedAuthorizationMapper, FederatedAuthorization, FederatedIdentityProvider, FederatedIdentityRegistry,
-    FederatedLoginSession, FederatedProviderView, FederatedRedirectPolicy, FederatedSession, FederatedSessionBinding,
-    FederatedSessionTransaction, FederationError, OpaqueLogoutContinuation, Result, StandardOidcAuthentication,
+    CoreFederatedAuthorizationMapper, FederatedAuthorization, FederatedLoginSession, FederatedProviderQuery,
+    FederatedProviderView, FederatedRedirectPolicy, FederatedSession, FederatedSessionBinding, FederatedSessionTransaction,
+    FederationError, OpaqueLogoutContinuation, Result, StandardOidcAuthentication,
 };
-use crate::oidc::OidcProviderConfig;
 use std::sync::Arc;

 const DEFAULT_OIDC_PROVIDER_ID: &str = "default";
@@ -32,11 +31,7 @@ fn sorted_provider_views(mut providers: Vec<FederatedProviderView>) -> Vec<Feder
 }

 pub struct FederatedIdentityService {
-    provider: Arc<dyn FederatedIdentityProvider>,
-    standard_oidc: Option<StandardOidcRuntime>,
-}
-
-struct StandardOidcRuntime {
+    provider_query: Arc<dyn FederatedProviderQuery>,
    authentication: Arc<dyn StandardOidcAuthentication>,
    mapper: CoreFederatedAuthorizationMapper,
 }
@@ -49,91 +44,57 @@ struct RuntimeCodeExchange {

 impl FederatedIdentityService {
    async fn exchange_code(&self, state: &str, code: &str, redirect_uri: &str) -> Result<RuntimeCodeExchange> {
-        match &self.standard_oidc {
-            Some(runtime) => {
-                let exchange = runtime.authentication.exchange_identity(state, code, redirect_uri).await?;
-                let (identity, redirect_after, logout_continuation) = exchange.into_parts();
-                Ok(RuntimeCodeExchange {
-                    authorization: runtime.mapper.map(identity),
-                    redirect_after,
-                    logout_continuation,
-                })
-            }
-            None => {
-                let exchange = self.provider.exchange_code(state, code, redirect_uri).await?;
-                let (authorization, redirect_after, logout_continuation) = exchange.into_parts();
-                Ok(RuntimeCodeExchange {
-                    authorization,
-                    redirect_after,
-                    logout_continuation,
-                })
-            }
-        }
+        let exchange = self.authentication.exchange_identity(state, code, redirect_uri).await?;
+        let (identity, redirect_after, logout_continuation) = exchange.into_parts();
+        Ok(RuntimeCodeExchange {
+            authorization: self.mapper.map(identity),
+            redirect_after,
+            logout_continuation,
+        })
    }

    async fn verify_identity(&self, jwt: &str) -> Result<FederatedAuthorization> {
-        match &self.standard_oidc {
-            Some(runtime) => Ok(runtime.mapper.map(runtime.authentication.verify_identity(jwt).await?)),
-            None => self.provider.verify_web_identity_token(jwt).await,
-        }
+        Ok(self.mapper.map(self.authentication.verify_identity(jwt).await?))
    }

    async fn create_logout_token(&self, continuation: OpaqueLogoutContinuation) -> Result<String> {
-        match &self.standard_oidc {
-            Some(runtime) => runtime.authentication.create_logout_token(continuation).await,
-            None => {
-                let (provider, id_token) = continuation.into_parts();
-                self.provider.create_logout_token(provider.as_str(), &id_token).await
-            }
-        }
-    }
-
-    /// Build the compatibility service around the published provider interface.
-    pub fn new(registry: FederatedIdentityRegistry) -> Self {
-        Self {
-            provider: registry.standard_oidc_arc(),
-            standard_oidc: None,
-        }
+        self.authentication.create_logout_token(continuation).await
    }

    pub(super) fn from_standard_oidc_parts(
-        provider: Arc<dyn FederatedIdentityProvider>,
+        provider_query: Arc<dyn FederatedProviderQuery>,
        authentication: Arc<dyn StandardOidcAuthentication>,
        mapper: CoreFederatedAuthorizationMapper,
    ) -> Self {
        Self {
-            provider,
-            standard_oidc: Some(StandardOidcRuntime { authentication, mapper }),
+            provider_query,
+            authentication,
+            mapper,
        }
    }

    pub fn has_providers(&self) -> bool {
-        self.provider.has_providers()
+        self.provider_query.has_providers()
    }

    pub fn list_providers(&self) -> Vec<FederatedProviderView> {
-        let providers = self.provider.list_providers();
+        let providers = self.provider_query.list_providers();
        sorted_provider_views(providers)
    }

    pub fn list_visible_providers(&self) -> Vec<FederatedProviderView> {
-        let providers = self.provider.list_visible_providers();
+        let providers = self.provider_query.list_visible_providers();
        sorted_provider_views(providers)
    }

    pub fn redirect_policy(&self, provider_id: &str) -> Option<FederatedRedirectPolicy> {
-        self.provider.redirect_policy(provider_id)
-    }
-
-    /// Compatibility accessor for existing federation consumers.
-    /// Login redirects use [`Self::redirect_policy`], while OIDC-specific
-    /// configuration consumers use the OIDC configuration query.
-    pub fn get_provider_config(&self, provider_id: &str) -> Option<&OidcProviderConfig> {
-        self.provider.provider_config(provider_id)
+        self.provider_query.redirect_policy(provider_id)
    }

    pub async fn authorize_url(&self, provider_id: &str, redirect_uri: &str, redirect_after: Option<String>) -> Result<String> {
-        self.provider.authorize_url(provider_id, redirect_uri, redirect_after).await
+        self.authentication
+            .authorize_url(provider_id, redirect_uri, redirect_after)
+            .await
    }

    pub async fn complete_authorization_code(
@@ -205,7 +166,9 @@ impl FederatedIdentityService {
    }

    pub async fn build_logout_url(&self, logout_token: &str, post_logout_redirect_uri: &str) -> Result<Option<String>> {
-        self.provider.build_logout_url(logout_token, post_logout_redirect_uri).await
+        self.authentication
+            .build_logout_url(logout_token, post_logout_redirect_uri)
+            .await
    }
 }

@@ -213,9 +176,8 @@ impl FederatedIdentityService {
 mod tests {
    use super::*;
    use crate::federation::{
-        FederatedAuthorization, FederatedAuthorizationRule, FederatedAuthorizationRules, FederatedClaims, FederatedCodeExchange,
-        FederatedIdentityProvider, FederatedProviderRef, FederatedSessionBindingError, VerifiedFederatedCodeExchange,
-        VerifiedFederatedIdentity,
+        FederatedAuthorizationRule, FederatedAuthorizationRules, FederatedClaims, FederatedProviderQuery, FederatedProviderRef,
+        FederatedSessionBindingError, VerifiedFederatedCodeExchange, VerifiedFederatedIdentity,
    };
    use rustfs_credentials::Credentials;
    use std::sync::{Arc, Mutex};
@@ -235,6 +197,8 @@ mod tests {
        web_provider_id: &'static str,
        failure: ProviderFailure,
        events: Arc<Mutex<Vec<&'static str>>>,
+        authorize_calls: Mutex<Vec<(String, String, Option<String>)>>,
+        logout_url_calls: Mutex<Vec<(String, String)>>,
        expected_logout: (&'static str, &'static str),
        listed_provider_ids: Vec<&'static str>,
        visible_provider_ids: Vec<&'static str>,
@@ -249,6 +213,8 @@ mod tests {
                web_provider_id: DEFAULT_OIDC_PROVIDER_ID,
                failure: ProviderFailure::None,
                events,
+                authorize_calls: Mutex::new(Vec::new()),
+                logout_url_calls: Mutex::new(Vec::new()),
                expected_logout: (DEFAULT_OIDC_PROVIDER_ID, "id-token"),
                listed_provider_ids: Vec::new(),
                visible_provider_ids: Vec::new(),
@@ -259,32 +225,6 @@ mod tests {
            self.events.lock().expect("event log should not be poisoned").push(event);
        }

-        fn authorization(&self, provider_id: &str) -> FederatedAuthorization {
-            FederatedAuthorization {
-                provider_id: provider_id.to_string(),
-                claims: FederatedClaims {
-                    sub: "subject".to_string(),
-                    email: String::new(),
-                    username: "user".to_string(),
-                    groups: vec!["source-group".to_string()],
-                    raw: Default::default(),
-                },
-                policies: if self.with_policy {
-                    vec!["readwrite".to_string()]
-                } else {
-                    Vec::new()
-                },
-                group_claim_policies: Vec::new(),
-                groups: if self.with_group {
-                    vec!["developers".to_string()]
-                } else {
-                    Vec::new()
-                },
-                roles_claim_key: None,
-                roles: Vec::new(),
-            }
-        }
-
        fn identity(&self, provider_id: &str) -> VerifiedFederatedIdentity {
            VerifiedFederatedIdentity::from_claims(
                FederatedProviderRef::new(provider_id.to_string()),
@@ -313,8 +253,7 @@ mod tests {
            .collect()
    }

-    #[async_trait::async_trait]
-    impl FederatedIdentityProvider for TestProvider {
+    impl FederatedProviderQuery for TestProvider {
        fn has_providers(&self) -> bool {
            true
        }
@@ -333,52 +272,18 @@ mod tests {
                allow_request_origin: true,
            })
        }
-
-        async fn authorize_url(
-            &self,
-            _provider_id: &str,
-            _redirect_uri: &str,
-            _redirect_after: Option<String>,
-        ) -> Result<String> {
-            Ok("https://identity.example/authorize".to_string())
-        }
-
-        async fn exchange_code(&self, _state: &str, _code: &str, _redirect_uri: &str) -> Result<FederatedCodeExchange> {
-            self.record("exchange");
-            if self.failure == ProviderFailure::Exchange {
-                return Err(FederationError::CodeExchange("exchange failed".to_string()));
-            }
-            Ok(FederatedCodeExchange {
-                authorization: self.authorization(self.browser_provider_id),
-                redirect_after: Some("/browser".to_string()),
-                id_token: "id-token".to_string(),
-            })
-        }
-
-        async fn verify_web_identity_token(&self, _jwt: &str) -> Result<FederatedAuthorization> {
-            self.record("verify");
-            if self.failure == ProviderFailure::Verification {
-                return Err(FederationError::TokenVerification("verification failed".to_string()));
-            }
-            Ok(self.authorization(self.web_provider_id))
-        }
-
-        async fn create_logout_token(&self, provider_id: &str, id_token: &str) -> Result<String> {
-            self.record("logout");
-            assert_eq!((provider_id, id_token), self.expected_logout);
-            if self.failure == ProviderFailure::Logout {
-                return Err(FederationError::Logout("logout failed".to_string()));
-            }
-            Ok("logout-token".to_string())
-        }
-
-        async fn build_logout_url(&self, _logout_token: &str, _post_logout_redirect_uri: &str) -> Result<Option<String>> {
-            Ok(Some("https://identity.example/logout".to_string()))
-        }
    }

    #[async_trait::async_trait]
    impl StandardOidcAuthentication for TestProvider {
+        async fn authorize_url(&self, provider_id: &str, redirect_uri: &str, redirect_after: Option<String>) -> Result<String> {
+            self.authorize_calls
+                .lock()
+                .expect("authorize call log should not be poisoned")
+                .push((provider_id.to_string(), redirect_uri.to_string(), redirect_after));
+            Ok("https://identity.example/authorize".to_string())
+        }
+
        async fn exchange_identity(
            &self,
            _state: &str,
@@ -414,6 +319,14 @@ mod tests {
            }
            Ok("logout-token".to_string())
        }
+
+        async fn build_logout_url(&self, logout_token: &str, post_logout_redirect_uri: &str) -> Result<Option<String>> {
+            self.logout_url_calls
+                .lock()
+                .expect("logout URL call log should not be poisoned")
+                .push((logout_token.to_string(), post_logout_redirect_uri.to_string()));
+            Ok(Some("https://identity.example/logout".to_string()))
+        }
    }

    fn standard_service(provider: Arc<TestProvider>) -> FederatedIdentityService {
@@ -432,14 +345,12 @@ mod tests {
            )
        }));
        let authentication: Arc<dyn StandardOidcAuthentication> = provider.clone();
-        let provider: Arc<dyn FederatedIdentityProvider> = provider;
-        FederatedIdentityService {
-            provider,
-            standard_oidc: Some(StandardOidcRuntime {
-                authentication,
-                mapper: CoreFederatedAuthorizationMapper::new(rules),
-            }),
-        }
+        let provider_query: Arc<dyn FederatedProviderQuery> = provider;
+        FederatedIdentityService::from_standard_oidc_parts(
+            provider_query,
+            authentication,
+            CoreFederatedAuthorizationMapper::new(rules),
+        )
    }

    struct RecordingBinding {
@@ -492,7 +403,7 @@ mod tests {
        let mut provider = TestProvider::new(events);
        provider.listed_provider_ids = vec!["zeta", "hidden", DEFAULT_OIDC_PROVIDER_ID, "alpha"];
        provider.visible_provider_ids = vec!["zeta", DEFAULT_OIDC_PROVIDER_ID, "alpha"];
-        let service = FederatedIdentityService::new(FederatedIdentityRegistry::new(Arc::new(provider)));
+        let service = standard_service(Arc::new(provider));

        assert_eq!(
            service
@@ -519,6 +430,48 @@ mod tests {
        );
    }

+    #[tokio::test]
+    async fn authorization_and_logout_urls_preserve_service_arguments() {
+        let provider = Arc::new(TestProvider::new(Arc::new(Mutex::new(Vec::new()))));
+        let service = standard_service(provider.clone());
+
+        let authorize_url = service
+            .authorize_url(
+                "corp",
+                "https://console.example.com/oauth/callback",
+                Some("/browser/path?tab=objects".to_string()),
+            )
+            .await
+            .expect("authorization URL should be returned");
+        let logout_url = service
+            .build_logout_url("opaque-logout-token", "https://console.example.com/signed-out")
+            .await
+            .expect("logout URL should be returned");
+
+        assert_eq!(authorize_url, "https://identity.example/authorize");
+        assert_eq!(logout_url.as_deref(), Some("https://identity.example/logout"));
+        assert_eq!(
+            provider
+                .authorize_calls
+                .lock()
+                .expect("authorize call log should not be poisoned")
+                .as_slice(),
+            [(
+                "corp".to_string(),
+                "https://console.example.com/oauth/callback".to_string(),
+                Some("/browser/path?tab=objects".to_string()),
+            )]
+        );
+        assert_eq!(
+            provider
+                .logout_url_calls
+                .lock()
+                .expect("logout URL call log should not be poisoned")
+                .as_slice(),
+            [("opaque-logout-token".to_string(), "https://console.example.com/signed-out".to_string(),)]
+        );
+    }
+
    #[tokio::test]
    async fn callback_and_web_identity_preserve_provider_and_transaction_boundaries() {
        let events = Arc::new(Mutex::new(Vec::new()));
@@ -526,9 +479,8 @@ mod tests {
        provider.browser_provider_id = "corp";
        provider.web_provider_id = "partner";
        provider.expected_logout = ("corp", "id-token");
-        let provider = Arc::new(provider);
        let binding = Arc::new(RecordingBinding::new(events.clone()));
-        let service = FederatedIdentityService::new(FederatedIdentityRegistry::new(provider));
+        let service = standard_service(Arc::new(provider));

        let login = service
            .complete_authorization_code("state", "code", "https://console.example/callback", 3600, binding.as_ref())
@@ -604,76 +556,13 @@ mod tests {
    }

    #[tokio::test]
-    async fn standard_runtime_checks_authorization_before_binding() {
+    async fn web_identity_without_policy_or_group_is_not_bound() {
        let events = Arc::new(Mutex::new(Vec::new()));
        let mut provider = TestProvider::new(events.clone());
        provider.with_policy = false;
        let binding = Arc::new(RecordingBinding::new(events.clone()));
        let service = standard_service(Arc::new(provider));

-        let error = service
-            .assume_role_with_web_identity("jwt", 3600, None, binding.as_ref())
-            .await
-            .expect_err("standard web identity requires mapped authorization");
-
-        assert!(matches!(error, FederationError::NoAuthorizationContext));
-        assert_eq!(events.lock().expect("event log should not be poisoned").as_slice(), ["verify"]);
-    }
-
-    #[tokio::test]
-    async fn standard_runtime_preserves_browser_failure_order() {
-        for (provider_failure, binding_failure, expected_events) in [
-            (ProviderFailure::Exchange, false, vec!["exchange"]),
-            (ProviderFailure::None, true, vec!["exchange", "bind"]),
-            (ProviderFailure::Logout, false, vec!["exchange", "bind", "logout"]),
-        ] {
-            let events = Arc::new(Mutex::new(Vec::new()));
-            let mut provider = TestProvider::new(events.clone());
-            provider.failure = provider_failure;
-            let mut binding = RecordingBinding::new(events.clone());
-            binding.fail = binding_failure;
-            let service = standard_service(Arc::new(provider));
-
-            service
-                .complete_authorization_code("state", "code", "https://console.example/callback", 3600, &binding)
-                .await
-                .expect_err("the configured standard callback failure should be returned");
-
-            assert_eq!(events.lock().expect("event log should not be poisoned").as_slice(), expected_events);
-        }
-    }
-
-    #[tokio::test]
-    async fn standard_runtime_preserves_web_identity_failure_order() {
-        for (provider_failure, binding_failure, expected_events) in [
-            (ProviderFailure::Verification, false, vec!["verify"]),
-            (ProviderFailure::None, true, vec!["verify", "bind"]),
-        ] {
-            let events = Arc::new(Mutex::new(Vec::new()));
-            let mut provider = TestProvider::new(events.clone());
-            provider.failure = provider_failure;
-            let mut binding = RecordingBinding::new(events.clone());
-            binding.fail = binding_failure;
-            let service = standard_service(Arc::new(provider));
-
-            service
-                .assume_role_with_web_identity("jwt", 3600, None, &binding)
-                .await
-                .expect_err("the configured standard web identity failure should be returned");
-
-            assert_eq!(events.lock().expect("event log should not be poisoned").as_slice(), expected_events);
-        }
-    }
-
-    #[tokio::test]
-    async fn web_identity_without_policy_or_group_is_not_bound() {
-        let events = Arc::new(Mutex::new(Vec::new()));
-        let mut provider = TestProvider::new(events.clone());
-        provider.with_policy = false;
-        let provider = Arc::new(provider);
-        let binding = Arc::new(RecordingBinding::new(events.clone()));
-        let service = FederatedIdentityService::new(FederatedIdentityRegistry::new(provider));
-
        let error = service
            .assume_role_with_web_identity("jwt", 3600, None, binding.as_ref())
            .await
@@ -689,16 +578,15 @@ mod tests {
        let mut provider = TestProvider::new(events.clone());
        provider.with_policy = false;
        provider.with_group = true;
-        let provider = Arc::new(provider);
        let binding = Arc::new(RecordingBinding::new(events.clone()));
-        let service = FederatedIdentityService::new(FederatedIdentityRegistry::new(provider));
+        let service = standard_service(Arc::new(provider));

        let session = service
            .assume_role_with_web_identity("jwt", 3600, None, binding.as_ref())
            .await
            .expect("a mapped group is an authorization context");

-        assert!(session.authorization.policies.is_empty());
+        assert_eq!(session.authorization.policies, ["developers"]);
        assert_eq!(session.authorization.groups, ["developers"]);
        assert_eq!(events.lock().expect("event log should not be poisoned").as_slice(), ["verify", "bind"]);
    }
@@ -716,7 +604,7 @@ mod tests {
            let mut binding = RecordingBinding::new(events.clone());
            binding.fail = binding_failure;
            let binding = Arc::new(binding);
-            let service = FederatedIdentityService::new(FederatedIdentityRegistry::new(Arc::new(provider)));
+            let service = standard_service(Arc::new(provider));

            let error = service
                .complete_authorization_code("state", "code", "https://console.example/callback", 3600, binding.as_ref())
@@ -755,7 +643,7 @@ mod tests {
            let mut binding = RecordingBinding::new(events.clone());
            binding.fail = binding_failure;
            let binding = Arc::new(binding);
-            let service = FederatedIdentityService::new(FederatedIdentityRegistry::new(Arc::new(provider)));
+            let service = standard_service(Arc::new(provider));

            let error = service
                .assume_role_with_web_identity("jwt", 3600, None, binding.as_ref())
@@ -2354,8 +2354,9 @@ fn extract_canonical_group_values(
    groups
 }

-#[cfg(test)]
-pub(crate) fn make_test_sys(configs: Vec<OidcProviderConfig>) -> OidcSys {
+#[cfg(any(test, feature = "test-util"))]
+#[doc(hidden)]
+pub fn make_test_sys(configs: Vec<OidcProviderConfig>) -> OidcSys {
    let configs = configs
        .into_iter()
        .map(test_sourced_config)
@@ -2369,7 +2370,7 @@ pub(crate) fn make_test_sys(configs: Vec<OidcProviderConfig>) -> OidcSys {
    }
 }

-#[cfg(test)]
+#[cfg(any(test, feature = "test-util"))]
 fn test_sourced_config(config: OidcProviderConfig) -> SourcedOidcProviderConfig {
    SourcedOidcProviderConfig {
        config,
@@ -2377,8 +2378,9 @@ fn test_sourced_config(config: OidcProviderConfig) -> SourcedOidcProviderConfig
    }
 }

-#[cfg(test)]
-pub(crate) fn test_config(id: &str) -> OidcProviderConfig {
+#[cfg(any(test, feature = "test-util"))]
+#[doc(hidden)]
+pub fn test_config(id: &str) -> OidcProviderConfig {
    OidcProviderConfig {
        id: id.to_string(),
        enabled: true,
@@ -252,6 +252,13 @@ impl PolicyVariableResolver for VariableResolver {
 }

 pub async fn resolve_aws_variables(pattern: &str, resolver: &dyn PolicyVariableResolver) -> Vec<String> {
+    resolve_aws_variables_with_depth(pattern, resolver, 0).await
+}
+
+async fn resolve_aws_variables_with_depth(pattern: &str, resolver: &dyn PolicyVariableResolver, depth: usize) -> Vec<String> {
+    if depth >= 10 {
+        return vec![pattern.to_string()];
+    }
    let mut results = vec![pattern.to_string()];

    let mut changed = true;
@@ -264,7 +271,7 @@ pub async fn resolve_aws_variables(pattern: &str, resolver: &dyn PolicyVariableR

        let mut new_results = Vec::new();
        for result in &results {
-            let resolved = resolve_single_pass(result, resolver).await;
+            let resolved = resolve_single_pass(result, resolver, depth).await;
            if resolved.len() > 1 || (resolved.len() == 1 && &resolved[0] != result) {
                changed = true;
            }
@@ -288,30 +295,31 @@ pub async fn resolve_aws_variables(pattern: &str, resolver: &dyn PolicyVariableR
 fn resolve_aws_variables_boxed<'a>(
    pattern: &'a str,
    resolver: &'a dyn PolicyVariableResolver,
+    depth: usize,
 ) -> std::pin::Pin<Box<dyn Future<Output = Vec<String>> + Send + 'a>> {
-    Box::pin(resolve_aws_variables(pattern, resolver))
+    Box::pin(resolve_aws_variables_with_depth(pattern, resolver, depth))
 }

 /// Single pass resolution of variables in a string
-async fn resolve_single_pass(pattern: &str, resolver: &dyn PolicyVariableResolver) -> Vec<String> {
+async fn resolve_single_pass(pattern: &str, resolver: &dyn PolicyVariableResolver, depth: usize) -> Vec<String> {
    // Find all ${...} format variables
-    let mut results = vec![pattern.to_string()];
+    let mut results = vec![(pattern.to_string(), 0)];

    // Process each result string
    let mut i = 0;
    while i < results.len() {
-        let mut start = 0;
+        let mut start = results[i].1;
        let mut modified = false;

        // Find variables in current string
-        while let Some(pos) = results[i][start..].find("${") {
+        while let Some(pos) = results[i].0[start..].find("${") {
            let actual_pos = start + pos;

            // Find the matching closing brace, taking into account nested braces
            let mut brace_count = 1;
            let mut end_pos = actual_pos + 2; // Start after "${"

-            let bytes = results[i].as_bytes();
+            let bytes = results[i].0.as_bytes();
            while end_pos < bytes.len() && brace_count > 0 {
                match bytes[end_pos] {
                    b'{' => brace_count += 1,
@@ -324,19 +332,23 @@ async fn resolve_single_pass(pattern: &str, resolver: &dyn PolicyVariableResolve
            }

            if brace_count == 0 {
-                let var_name = &results[i][actual_pos + 2..end_pos];
+                let var_name = &results[i].0[actual_pos + 2..end_pos];

                // Check if this is a nested variable (contains ${...} inside)
                if var_name.contains("${") {
                    // For nested variables like ${${a}-${b}}, we need to resolve the inner variables first
                    // Then use the resolved result as a new variable to resolve
-                    let resolved_inner = resolve_aws_variables_boxed(var_name, resolver).await;
+                    let resolved_inner = resolve_aws_variables_boxed(var_name, resolver, depth + 1).await;
+                    if resolved_inner.len() == 1 && resolved_inner[0] == var_name {
+                        start = end_pos + 1;
+                        continue;
+                    }
                    let mut new_results = Vec::new();

                    for resolved_var_name in resolved_inner {
-                        let prefix = &results[i][..actual_pos];
-                        let suffix = &results[i][end_pos + 1..];
-                        new_results.push(format!("{prefix}{resolved_var_name}{suffix}"));
+                        let prefix = &results[i].0[..actual_pos];
+                        let suffix = &results[i].0[end_pos + 1..];
+                        new_results.push((format!("{prefix}{resolved_var_name}{suffix}"), actual_pos + resolved_var_name.len()));
                    }

                    if !new_results.is_empty() {
@@ -354,11 +366,13 @@ async fn resolve_single_pass(pattern: &str, resolver: &dyn PolicyVariableResolve
                        if !values.is_empty() {
                            // If there are multiple values, create a new result for each value
                            let mut new_results = Vec::new();
-                            let prefix = &results[i][..actual_pos];
-                            let suffix = &results[i][end_pos + 1..];
+                            let prefix = &results[i].0[..actual_pos];
+                            let suffix = &results[i].0[end_pos + 1..];

                            for value in values {
-                                new_results.push(format!("{prefix}{value}{suffix}"));
+                                // Resume after the substitution; a generated placeholder
+                                // belongs to the next bounded pass, not this one.
+                                new_results.push((format!("{prefix}{value}{suffix}"), actual_pos + value.len()));
                            }

                            results.splice(i..i + 1, new_results);
@@ -367,9 +381,9 @@ async fn resolve_single_pass(pattern: &str, resolver: &dyn PolicyVariableResolve
                        } else {
                            // Variable resolved to empty, just remove the variable placeholder
                            let mut new_results = Vec::new();
-                            let prefix = &results[i][..actual_pos];
-                            let suffix = &results[i][end_pos + 1..];
-                            new_results.push(format!("{prefix}{suffix}"));
+                            let prefix = &results[i].0[..actual_pos];
+                            let suffix = &results[i].0[end_pos + 1..];
+                            new_results.push((format!("{prefix}{suffix}"), actual_pos));

                            results.splice(i..i + 1, new_results);
                            modified = true;
@@ -391,7 +405,7 @@ async fn resolve_single_pass(pattern: &str, resolver: &dyn PolicyVariableResolve
        }
    }

-    results
+    results.into_iter().map(|(value, _)| value).collect()
 }

 #[cfg(test)]
@@ -399,6 +413,62 @@ mod tests {
    use super::*;
    use serde_json::Value;
    use std::collections::HashMap;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    struct CycleResolver {
+        calls: AtomicUsize,
+    }
+
+    #[async_trait]
+    impl PolicyVariableResolver for CycleResolver {
+        async fn resolve(&self, name: &str) -> Option<String> {
+            // Make the vulnerable implementation terminate so a regression
+            // reports an assertion failure instead of hanging the test runner.
+            if self.calls.fetch_add(1, Ordering::Relaxed) >= 100 {
+                return None;
+            }
+            match name {
+                "aws:AccountId" => Some("$$".to_string()),
+                "aws:username" => Some("{aws:AccountId}{aws:username}".to_string()),
+                _ => None,
+            }
+        }
+
+        fn is_dynamic(&self, _: &str) -> bool {
+            false
+        }
+    }
+
+    #[tokio::test]
+    async fn generated_policy_variable_cycles_are_bounded() {
+        let resolver = CycleResolver {
+            calls: AtomicUsize::new(0),
+        };
+        let result = resolve_aws_variables("${aws:AccountId}{aws:username}", &resolver).await;
+        assert!(resolver.calls.load(Ordering::Relaxed) <= 10);
+        assert_eq!(result, vec!["${aws:AccountId}{aws:username}".to_string()]);
+    }
+
+    #[tokio::test]
+    async fn a_single_pass_resolves_all_original_variables() {
+        let mut context = VariableContext::new();
+        context.username = Some("alice".to_string());
+        let resolver = VariableResolver::new(context);
+        let pattern = "${aws:username}".repeat(20);
+        assert_eq!(resolve_single_pass(&pattern, &resolver, 0).await, vec!["alice".repeat(20)]);
+    }
+
+    #[tokio::test]
+    async fn nested_policy_variables_have_a_depth_limit() {
+        let resolver = CycleResolver {
+            calls: AtomicUsize::new(0),
+        };
+        let pattern = format!("{}aws:username{}", "${".repeat(100), "}".repeat(100));
+        let result = resolve_aws_variables(&pattern, &resolver).await;
+        assert_eq!(result.len(), 1);
+        assert!(result[0].contains("${"));
+        assert!(resolver.calls.load(Ordering::Relaxed) <= 10);
+    }

    #[tokio::test]
    async fn test_resolve_aws_variables_with_username() {
@@ -80,6 +80,7 @@ rustfs-tls-runtime.workspace = true
 rustfs-utils = { workspace = true, features = ["io", "hash", "compress"] }
 serde_json = { workspace = true, features = ["raw_value"] }
 md-5 = { workspace = true }
+hmac.workspace = true
 minlz.workspace = true
 tracing.workspace = true
 thiserror.workspace = true
@@ -14,8 +14,10 @@

 use crate::compress_index::{Index, TryGetIndex};
 use crate::{BadDigest, EtagResolvable, HashReaderDetector, HashReaderMut};
+use hmac::{Hmac, KeyInit, Mac};
 use md5::{Digest, Md5};
 use pin_project_lite::pin_project;
+use sha2::Sha256;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 use tokio::io::{AsyncRead, ReadBuf};
@@ -29,6 +31,7 @@ pin_project! {
        pub finished: bool,
        pub checksum: Option<String>,
        resolved_etag: Option<String>,
+        encrypted_etag_mac: Option<Hmac<Sha256>>,
    }
 }

@@ -40,9 +43,21 @@ impl<R> EtagReader<R> {
            finished: false,
            checksum,
            resolved_etag: None,
+            encrypted_etag_mac: None,
        }
    }

+    /// Protect the content fingerprint with the object's encryption material.
+    /// Plaintext checksum verification remains the inner reader's responsibility.
+    pub fn new_encrypted(inner: R, key: [u8; 32], nonce: [u8; 12]) -> std::io::Result<Self> {
+        let mut mac = <Hmac<Sha256> as KeyInit>::new_from_slice(&key).map_err(std::io::Error::other)?;
+        mac.update(b"RustFS:encrypted-ETag:v1\0");
+        mac.update(&nonce);
+        let mut reader = Self::new(inner, None);
+        reader.encrypted_etag_mac = Some(mac);
+        Ok(reader)
+    }
+
    /// Get the final md5 value (etag) as a hex string, only compute once.
    /// Can be called multiple times, always returns the same result after finished.
    pub fn get_etag(&mut self) -> String {
@@ -108,6 +123,14 @@ impl<R> EtagResolvable for EtagReader<R> {
        true
    }
    fn try_resolve_etag(&mut self) -> Option<String> {
+        if let Some(mut mac) = self.encrypted_etag_mac.clone() {
+            if !self.finished {
+                return None;
+            }
+            mac.update(self.get_etag().as_bytes());
+            let protected = mac.finalize().into_bytes();
+            return Some(hex_simd::encode_to_string(&protected[..16], hex_simd::AsciiCase::Lower));
+        }
        // EtagReader provides its own etag, not delegating to inner
        if let Some(checksum) = &self.checksum {
            Some(checksum.clone())
@@ -148,6 +171,47 @@ mod tests {
    use std::io::Cursor;
    use tokio::io::{AsyncReadExt, BufReader};

+    #[tokio::test]
+    async fn encrypted_etags_hide_plaintext_fingerprints_including_empty_objects() {
+        for data in [b"".as_slice(), b"Salary 2026: 85000 EUR".as_slice()] {
+            let md5 = hex_simd::encode_to_string(Md5::digest(data), hex_simd::AsciiCase::Lower);
+            let mut tags = Vec::new();
+            for (key, nonce) in [([1; 32], [1; 12]), ([2; 32], [1; 12]), ([1; 32], [2; 12])] {
+                let mut reader = EtagReader::new_encrypted(Cursor::new(data), key, nonce).expect("create protected ETag reader");
+                assert!(reader.try_resolve_etag().is_none());
+                let mut output = Vec::new();
+                reader.read_to_end(&mut output).await.expect("read unchanged plaintext");
+                assert_eq!(output, data);
+                let tag = reader.try_resolve_etag().expect("resolve protected ETag after EOF");
+                assert_eq!(tag.len(), 32);
+                assert_ne!(tag, md5);
+                assert_eq!(reader.try_resolve_etag().as_ref(), Some(&tag));
+                tags.push(tag);
+            }
+            assert_ne!(tags[0], tags[1], "different keys must not link identical content");
+            assert_ne!(tags[0], tags[2], "different objects under a legacy key must not link identical content");
+        }
+    }
+
+    #[tokio::test]
+    async fn encrypted_etag_reader_preserves_plaintext_digest_rejection() {
+        let plaintext = crate::HashReader::from_stream(
+            Cursor::new(b"actual plaintext"),
+            16,
+            16,
+            Some("00000000000000000000000000000000".to_owned()),
+            None,
+            false,
+        )
+        .expect("create checked plaintext reader");
+        let mut reader = EtagReader::new_encrypted(plaintext, [1; 32], [2; 12]).expect("create protected ETag reader");
+        let err = reader
+            .read_to_end(&mut Vec::new())
+            .await
+            .expect_err("wrong plaintext MD5 must still be rejected");
+        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
+    }
+
    #[tokio::test]
    async fn test_etag_reader_basic() {
        let data = b"hello world";
@@ -136,6 +136,7 @@ pin_project! {
        content_sha256: Option<String>,
        content_sha256_hasher: Option<Sha256Hasher>,
        checksum_on_finish: bool,
+        pub encrypted_etag: bool,

        // Read only at EOF; see the timing contract in `crate::trailer`.
        trailer: Option<SharedTrailerSource>,
@@ -184,6 +185,7 @@ impl HashReader {
            content_sha256: sha256hex.clone(),
            content_sha256_hasher: sha256hex.map(|_| Sha256Hasher::new()),
            checksum_on_finish: false,
+            encrypted_etag: false,
            trailer: None,
        })
    }
@@ -224,6 +226,7 @@ impl HashReader {
            content_sha256: sha256hex.clone(),
            content_sha256_hasher: sha256hex.map(|_| Sha256Hasher::new()),
            checksum_on_finish: false,
+            encrypted_etag: false,
            trailer: None,
        })
    }
@@ -294,6 +297,7 @@ impl HashReader {
                content_hash,
                content_hasher,
                checksum_on_finish: false,
+                encrypted_etag: false,
                trailer: existing_hash_reader.get_trailer().cloned(),
            })
        } else {
@@ -322,6 +326,7 @@ impl HashReader {
                content_sha256: sha256hex.clone(),
                content_sha256_hasher: sha256hex.map(|_| Sha256Hasher::new()),
                checksum_on_finish: false,
+                encrypted_etag: false,
                trailer: None,
            })
        }
@@ -55,11 +55,9 @@ pub const MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER: &str = "X-Minio-Internal

 /// Reserved RustFS-branded twin of the MinIO-internal SSE key family.
 ///
-/// No RustFS writer emits these keys today — the SSE writer persists the
-/// MinIO-branded `X-Minio-Internal-Server-Side-Encryption-*` keys verbatim for
-/// interoperability — but redaction (`rustfs_filemeta`) and replication
-/// stripping treat the family as sensitive so that a future or third-party
-/// writer cannot leak sealed material through the reserved names.
+/// The SSE writer persists sealed material under the MinIO-branded names for
+/// interoperability. Redaction and replication stripping also cover this twin
+/// family, including the encrypted ETag format marker.
 pub const RUSTFS_INTERNAL_ENCRYPTION_PREFIX: &str = "x-rustfs-internal-server-side-encryption-";

 pub const REPLICATION_SSEC_ALGORITHM_HEADER: &str = "X-Rustfs-Replication-Ssec-Algorithm";
@@ -71,6 +69,15 @@ pub const REPLICATION_SSE_SEAL_ALGORITHM_HEADER: &str = "X-Rustfs-Replication-Se
 pub const REPLICATION_SSE_SEALED_KEY_HEADER: &str = "X-Rustfs-Replication-Server-Side-Encryption-Sealed-Key";
 pub const REPLICATION_ENCRYPTED_MULTIPART_HEADER: &str = "X-Rustfs-Replication-Encrypted-Multipart";

+pub const SUFFIX_OPAQUE_ENCRYPTED_ETAG: &str = "server-side-encryption-opaque-etag";
+
+/// Old encrypted objects can carry a plaintext MD5. Keyless listings expose
+/// an encrypted ETag only when its producer recorded the protected format.
+pub fn etag_can_be_listed(metadata: &HashMap<String, String>) -> bool {
+    !metadata.keys().any(|key| is_replication_stripped_encryption_key(key))
+        || super::metadata_compat::get_consistent_str(metadata, SUFFIX_OPAQUE_ENCRYPTED_ETAG) == Some("v1")
+}
+
 /// Stored SSE-C metadata keys and the wire names they replicate under.
 ///
 /// Source keys must match what `encryption_material_to_metadata` persists; the
@@ -307,6 +314,28 @@ pub fn stored_managed_encryption_key(metadata: &HashMap<String, String>) -> Opti
 mod tests {
    use super::*;

+    #[test]
+    fn keyless_listings_hide_unproven_encrypted_etags() {
+        assert!(etag_can_be_listed(&HashMap::from([("content-type".to_owned(), "text/plain".to_owned())])));
+        for key in [
+            SSEC_ALGORITHM_HEADER,
+            INTERNAL_ENCRYPTION_KEY_HEADER,
+            MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER,
+            MINIO_INTERNAL_ENCRYPTION_KMS_DATA_KEY_HEADER,
+            "X-AMZ-SERVER-SIDE-ENCRYPTION",
+            "x-rustfs-internal-server-side-encryption-sealed-key",
+        ] {
+            let mut metadata = HashMap::from([(key.to_owned(), "encryption-metadata".to_owned())]);
+            assert!(!etag_can_be_listed(&metadata), "legacy encrypted metadata: {key}");
+            super::super::metadata_compat::insert_str(&mut metadata, SUFFIX_OPAQUE_ENCRYPTED_ETAG, "unknown".to_owned());
+            assert!(!etag_can_be_listed(&metadata), "unknown producer format: {key}");
+            super::super::metadata_compat::insert_str(&mut metadata, SUFFIX_OPAQUE_ENCRYPTED_ETAG, "v1".to_owned());
+            assert!(etag_can_be_listed(&metadata), "protected encrypted metadata: {key}");
+            metadata.insert("x-minio-internal-server-side-encryption-opaque-etag".to_owned(), "unknown".to_owned());
+            assert!(!etag_can_be_listed(&metadata), "conflicting format markers: {key}");
+        }
+    }
+
    #[test]
    fn ssec_transport_projection_retains_only_redacted_reader_headers() {
        let mut headers = http::HeaderMap::new();
@@ -49,6 +49,10 @@ rustfs-rio.workspace = true
 tokio = { workspace = true, features = ["io-util", "macros", "rt"] }
 thiserror = { workspace = true }

+[target.'cfg(target_os = "macos")'.dependencies]
+# Keep XZ decoding self-contained in macOS binaries instead of linking Homebrew liblzma.
+liblzma = { version = "0.4.8", features = ["static"] }
+
 [dev-dependencies]
 rustfs-tokio-tar = { workspace = true }
 futures = { workspace = true }
@@ -21,6 +21,9 @@ For bounded local read-backend startup and cancellation, see

 For legacy protection assessment and bounded protected copies, see [Object integrity inventory, audit, and migration](operations/shard-integrity-audit.md).

+For encrypted listing compatibility and older stored fingerprints, see
+[Encrypted object ETags](operations/encrypted-etags.md).
+
 Operational runbooks live under [`operations/`](operations/). Replication
 operators should start with:

@@ -0,0 +1,44 @@
+# Encrypted object ETags
+
+**Use this when:** upgrading encrypted storage or investigating missing ETags
+in object listings after an upgrade.
+
+New encrypted writes use a keyed fingerprint for the ETag rather than the
+plaintext MD5. Single-part and part ETags are 32-character hexadecimal values;
+completed multipart ETags retain their digest-and-part-count form. They stay
+stable for the stored object. Request MD5 and checksum validation still apply
+to the original plaintext. Encryption and decryption formats do not change.
+
+## Listings and older objects
+
+ListObjects, ListObjectsV2, and ListObjectVersions omit the optional ETag field
+for encrypted objects whose metadata does not attest to the protected format.
+This includes encrypted objects written before the upgrade, preserved source
+ETags without that proof, and completed encrypted multipart uploads. New
+single-part encrypted writes carry the internal format marker and expose their
+protected ETag in listings. Unencrypted listings retain their existing ETags.
+Existing persistent listing snapshots are invalidated; the provider falls
+back to verified storage until a new snapshot is built.
+
+Clients that require an ETag for these older or multipart objects must issue an
+authorized HeadObject request, including the SSE-C key when required. GET,
+HEAD, and conditional requests continue to use the object's stored ETag.
+
+## Existing data and migration
+
+The upgrade does not rewrite existing object metadata or erase fingerprints
+already copied to backups, logs, caches, or replicas. Restrict access to those
+copies separately. Rolling back to an older server also restores its listing
+behavior, so keep the upgraded listing protection on every serving node.
+
+To obtain a protected ETag, read the object with its encryption credentials and
+write it through the upgraded encryption path. Verify the resulting plaintext
+and application references before retiring an old copy. A server-side copy
+that preserves the source ETag is insufficient.
+
+In a versioned bucket, a new write creates a new version; historical versions
+keep their original metadata. Object Lock retention and legal holds still
+apply. Do not bypass them or delete retained versions to migrate an ETag.
+There is no automatic in-place migration of retained metadata in this upgrade.
+Unfinished uploads also keep their original part metadata. Restart an upload
+through the upgraded path when its legacy part fingerprints need to be replaced.
@@ -29,6 +29,8 @@ The [scanner cache cost profile](scanner-cache-cost.md) separates clone, subtree

 The [Pool layout compatibility reference](pool-layout-compatibility.md) defines the topology and EC regression matrix for single-drive, single-node multi-drive, and multi-node expansion pools.

+The [immutable validation candidate reference](validation-candidates.md) describes isolated package builds and exact candidate resolution for the current auto-testing fleet.
+
 ## Naming conventions

 ### Reserved test-name substrings (migration gate)
@@ -0,0 +1,52 @@
+# Immutable validation candidates
+
+Canonical validation on the current auto-testing fleet consumes one exact
+package, source commit and candidate-bound tool build. A successful package
+build is provenance evidence, not a canonical test PASS.
+
+The scheduled Functional Chain continues to resolve successful Nightly GNU
+Build attempts from `main`. Optional manual validation can also resolve a
+validation package built without publishing nightly aliases or release assets.
+
+## Build a validation package
+
+Dispatch **RustFS Functional Chain** from the desired driver revision with
+only `validation_package_sha` set to the full 40-character RustFS source SHA.
+Keep candidate, case, helper, tools and cleanup inputs empty or false. The
+request validator rejects mixed package/fleet execution before building.
+
+The package job calls `rustfs-validation-package.yml`, checks out the requested
+source, confirms its Git SHA and builds the ordinary release binary and DEB.
+This job uses build runners and does not acquire or mutate the Heal fleet.
+The same workflow can be dispatched directly with its `source_sha` input when
+it is available on the repository's default branch.
+
+Publication requires complete R2 configuration, conditional `PutObject` with
+`If-None-Match: *`, and a public download whose SHA-256 matches the local DEB.
+Failure emits no candidate manifest. An existing object is never overwritten.
+The only object path is:
+
+```text
+artifacts/rustfs/packages/validation/runs/<run>/<attempt>/<package-sha256>/rustfs.deb
+```
+
+The schema-3 artifact `validation-candidate-<run>-<attempt>` records the
+workflow SHA independently of the exact source SHA and source ref, the
+`validation` channel, build run/attempt, immutable URL and package checksum.
+The resolver verifies the successful manual run, repository/workflow identity,
+exact artifact name and digest, archive members, schema and immutable URL.
+The ordinary scheduled chain does not enable validation package resolution.
+
+## Validate the candidate
+
+After the package run succeeds, dispatch Functional Chain with its exact
+`build_run_id` and `build_run_attempt`, a canonical `validation_case`, and an
+exact `auto_testing_sha`. Heal also requires a new tool build verified against
+that source and helper revision. Reassess source-bound contracts and native
+prerequisites when the source changes; an older tool artifact does not certify
+a new candidate.
+
+The fleet lane retains the shared environment lock, per-case evidence and
+cleanup gate. Record new results separately. A package build, native test,
+diagnostic replay, clean recovery or partial topology result cannot replace
+complete original source assertions or rewrite a retained canonical FAIL.
@@ -389,6 +389,7 @@ symbolic-demangle = { version = "=13.9.0", default-features = false, features =

 [dev-dependencies]
 uuid = { workspace = true, features = ["v4", "v5", "fast-rng", "macro-diagnostics"] }
+rustfs-iam = { workspace = true, features = ["test-util"] }
 serial_test = { workspace = true }
 tempfile = { workspace = true }
 aws-config = { workspace = true }
@@ -17,12 +17,12 @@
 use crate::admin::auth::authorize_admin_request;
 use crate::admin::router::{AdminOperation, Operation, S3Router};
 use crate::admin::runtime_sources::current_object_store_handle;
+use crate::admin::storage_api::s3::{self, Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result};
 use crate::server::ADMIN_PREFIX;
 use crate::site_replication::{site_replication_enabled, with_site_replication_bucket_mutation_lock};
 use hyper::{Method, StatusCode};
 use matchit::Params;
 use rustfs_policy::policy::action::{Action, AdminAction};
-use s3s::{Body, S3Request, S3Response, S3Result, s3_error};
 use serde::{Deserialize, Serialize};
 use tracing::info;
 use uuid::Uuid;
@@ -63,23 +63,23 @@ impl Operation for RecoverOrphanedBucketHandler {
            .credentials
            .as_ref()
            .map(|credentials| credentials.secret_key.clone())
-            .ok_or_else(|| s3_error!(InvalidRequest, "authentication required"))?;
+            .ok_or_else(|| s3::error(S3ErrorCode::InvalidRequest, "authentication required"))?;
        let bucket = params
            .get("bucket")
            .filter(|bucket| !bucket.is_empty())
-            .ok_or_else(|| s3_error!(InvalidRequest, "bucket name is required"))?;
+            .ok_or_else(|| s3::error(S3ErrorCode::InvalidRequest, "bucket name is required"))?;
        let body = req
            .input
            .store_all_limited(rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE)
            .await
-            .map_err(|_| s3_error!(InvalidRequest, "failed to read recovery request body"))?;
+            .map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "failed to read recovery request body"))?;
        let request: RecoverOrphanedBucketRequest = serde_json::from_slice(&body)
-            .map_err(|_| s3_error!(InvalidRequest, "expected JSON with a non-nil expectedIncarnationId"))?;
+            .map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "expected JSON with a non-nil expectedIncarnationId"))?;
        if request.expected_incarnation_id.is_nil() {
-            return Err(s3_error!(InvalidRequest, "expectedIncarnationId must be non-nil"));
+            return Err(s3::error(S3ErrorCode::InvalidRequest, "expectedIncarnationId must be non-nil"));
        }
        let Some(store) = current_object_store_handle() else {
-            return Err(s3_error!(InternalError, "object store is not initialized"));
+            return Err(s3::error(S3ErrorCode::InternalError, "object store is not initialized"));
        };

        let operation_bucket = bucket.to_owned();
@@ -91,21 +91,21 @@ impl Operation for RecoverOrphanedBucketHandler {
                let lock_bucket = operation_bucket.clone();
                with_site_replication_bucket_mutation_lock(lock_store, &lock_bucket, move || async move {
                    if site_replication_enabled().await? {
-                        return Err(s3_error!(
-                            OperationAborted,
-                            "orphaned bucket recovery is local; reconcile every site before recreating this bucket"
+                        return Err(s3::error(
+                            S3ErrorCode::OperationAborted,
+                            "orphaned bucket recovery is local; reconcile every site before recreating this bucket",
                        ));
                    }
                    operation_store
                        .recover_orphaned_bucket(&operation_bucket, expected_incarnation)
                        .await
-                        .map_err(|error| s3s::S3Error::from(crate::error::ApiError::from(error)))
+                        .map_err(|error| S3Error::from(crate::error::ApiError::from(error)))
                })
                .await??;
-                Ok::<(), s3s::S3Error>(())
+                Ok::<(), S3Error>(())
            })
            .await
-            .map_err(|error| s3s::S3Error::from(crate::error::ApiError::from(error)))?;
+            .map_err(|error| S3Error::from(crate::error::ApiError::from(error)))?;
        completed?;

        info!(
@@ -15,7 +15,7 @@
 use crate::admin::auth::authorize_admin_request;
 use crate::admin::router::{AdminOperation, Operation, S3Router};
 use crate::admin::runtime_sources::{
-    current_deployment_id, current_federated_identity_runtime, current_iam_handle, current_object_store_handle, current_region,
+    current_deployment_id, current_iam_handle, current_object_store_handle, current_oidc_config_query, current_region,
    current_replication_pool_handle, current_replication_stats_handle, current_server_config, current_token_signing_key,
    object_store_from_req,
 };
@@ -2086,40 +2086,11 @@ fn open_id_provider_settings(config: &rustfs_iam::oidc::OidcProviderConfig) -> O
    }
 }

-fn open_id_settings_from_service(service: &rustfs_iam::federation::FederatedIdentityService, region: String) -> OpenIDSettings {
-    let providers = service
-        .list_providers()
-        .into_iter()
-        .filter_map(|provider| {
-            let config = service.get_provider_config(&provider.provider_id)?;
-            Some((provider.provider_id, open_id_provider_settings(config)))
-        })
-        .collect();
-
-    open_id_settings(providers, region)
-}
-
-fn open_id_settings_from_runtime(
-    oidc_config_query: Option<&dyn rustfs_iam::federation::oidc::OidcConfigQuery>,
-    service: Option<&rustfs_iam::federation::FederatedIdentityService>,
-    region: String,
-) -> OpenIDSettings {
-    if let Some(oidc_config_query) = oidc_config_query {
-        return open_id_settings_from_snapshot(&oidc_config_query.config_snapshot(), region);
-    }
-
-    service
-        .map(|service| open_id_settings_from_service(service, region))
-        .unwrap_or_default()
-}
-
 fn local_idp_settings() -> IDPSettings {
    let mut settings = IDPSettings::default();
    let region = current_region().map(|region| region.to_string()).unwrap_or_default();
-    settings.open_id = current_federated_identity_runtime()
-        .map(|(service, oidc_config_query)| {
-            open_id_settings_from_runtime(oidc_config_query.as_deref(), Some(service.as_ref()), region)
-        })
+    settings.open_id = current_oidc_config_query()
+        .map(|query| open_id_settings_from_snapshot(&query.config_snapshot(), region))
        .unwrap_or_default();

    let (ldap, ldap_configs) = load_ldap_idp_settings();
@@ -11444,82 +11415,6 @@ mod tests {
        }
    }

-    struct LegacyOidcProvider {
-        configs: Vec<rustfs_iam::oidc::OidcProviderConfig>,
-    }
-
-    struct SnapshotOidcConfigQuery {
-        snapshot: rustfs_iam::oidc::OidcConfigSnapshot,
-    }
-
-    impl rustfs_iam::federation::oidc::OidcConfigQuery for SnapshotOidcConfigQuery {
-        fn config_snapshot(&self) -> rustfs_iam::oidc::OidcConfigSnapshot {
-            self.snapshot.clone()
-        }
-    }
-
-    #[async_trait::async_trait]
-    impl rustfs_iam::federation::FederatedIdentityProvider for LegacyOidcProvider {
-        fn has_providers(&self) -> bool {
-            !self.configs.is_empty()
-        }
-
-        fn list_providers(&self) -> Vec<rustfs_iam::federation::FederatedProviderView> {
-            self.configs
-                .iter()
-                .map(|config| rustfs_iam::federation::FederatedProviderView {
-                    provider_id: config.id.clone(),
-                    display_name: config.display_name.clone(),
-                })
-                .collect()
-        }
-
-        fn list_visible_providers(&self) -> Vec<rustfs_iam::federation::FederatedProviderView> {
-            self.list_providers()
-        }
-
-        fn provider_config(&self, provider_id: &str) -> Option<&rustfs_iam::oidc::OidcProviderConfig> {
-            self.configs.iter().find(|config| config.id == provider_id)
-        }
-
-        async fn authorize_url(
-            &self,
-            _provider_id: &str,
-            _redirect_uri: &str,
-            _redirect_after: Option<String>,
-        ) -> rustfs_iam::federation::Result<String> {
-            unreachable!("site replication compatibility test does not authorize")
-        }
-
-        async fn exchange_code(
-            &self,
-            _state: &str,
-            _code: &str,
-            _redirect_uri: &str,
-        ) -> rustfs_iam::federation::Result<rustfs_iam::federation::FederatedCodeExchange> {
-            unreachable!("site replication compatibility test does not exchange codes")
-        }
-
-        async fn verify_web_identity_token(
-            &self,
-            _jwt: &str,
-        ) -> rustfs_iam::federation::Result<rustfs_iam::federation::FederatedAuthorization> {
-            unreachable!("site replication compatibility test does not verify tokens")
-        }
-
-        async fn create_logout_token(&self, _provider_id: &str, _id_token: &str) -> rustfs_iam::federation::Result<String> {
-            unreachable!("site replication compatibility test does not create logout tokens")
-        }
-
-        async fn build_logout_url(
-            &self,
-            _logout_token: &str,
-            _post_logout_redirect_uri: &str,
-        ) -> rustfs_iam::federation::Result<Option<String>> {
-            unreachable!("site replication compatibility test does not build logout URLs")
-        }
-    }
-
    fn ldap_settings_kvs() -> rustfs_config::server_config::KVS {
        rustfs_config::server_config::KVS(vec![
            rustfs_config::server_config::KV {
@@ -11692,43 +11587,6 @@ mod tests {
        );
    }

-    #[test]
-    fn service_only_oidc_runtime_preserves_site_replication_settings() {
-        let default = oidc_snapshot_provider("default", "client-a");
-        let corp = oidc_snapshot_provider("corp", "client-b");
-        let expected = open_id_settings_from_snapshot(
-            &rustfs_iam::oidc::OidcConfigSnapshot::new(vec![corp.clone(), default.clone()]),
-            "eu-site-1".to_string(),
-        );
-        let service = rustfs_iam::federation::FederatedIdentityService::new(
-            rustfs_iam::federation::FederatedIdentityRegistry::new(Arc::new(LegacyOidcProvider {
-                configs: vec![corp.config, default.config],
-            })),
-        );
-
-        assert_eq!(
-            serde_json::to_value(open_id_settings_from_runtime(None, Some(&service), "eu-site-1".to_string(),))
-                .expect("serialize legacy OpenID settings"),
-            serde_json::to_value(expected).expect("serialize snapshot OpenID settings"),
-        );
-    }
-
-    #[test]
-    fn oidc_query_takes_precedence_over_service_only_compatibility() {
-        let query = SnapshotOidcConfigQuery {
-            snapshot: rustfs_iam::oidc::OidcConfigSnapshot::new(vec![oidc_snapshot_provider("default", "query-client")]),
-        };
-        let service = rustfs_iam::federation::FederatedIdentityService::new(
-            rustfs_iam::federation::FederatedIdentityRegistry::new(Arc::new(LegacyOidcProvider {
-                configs: vec![oidc_snapshot_provider("default", "service-client").config],
-            })),
-        );
-
-        let settings = open_id_settings_from_runtime(Some(&query), Some(&service), "eu-site-1".to_string());
-
-        assert_eq!(settings.claim_provider.client_id, "query-client");
-    }
-
    // A whole provider object present on one side only must not spill its
    // credential-derived leaves into the rejection message.
    #[test]
@@ -23,12 +23,11 @@ use crate::app::object_usecase::DefaultObjectUsecase;
 use crate::runtime_sources as root_runtime_sources;
 pub(crate) use crate::runtime_sources::{
    AppContext, ServerContextSlot, current_action_credentials, current_boot_time, current_bucket_metadata_handle,
-    current_bucket_monitor_handle, current_deployment_id, current_endpoints_handle, current_federated_identity_runtime,
-    current_federated_identity_service, current_iam_handle, current_kms_runtime_service_manager,
-    current_notification_system_for_context, current_object_data_cache_handle_for_context,
-    current_object_store_handle_for_context, current_ready_iam_handle, current_region, current_replication_pool_handle,
-    current_replication_stats_handle, current_replication_stats_handle_for_context, current_server_config_for_context,
-    current_token_signing_key,
+    current_bucket_monitor_handle, current_deployment_id, current_endpoints_handle, current_federated_identity_service,
+    current_iam_handle, current_kms_runtime_service_manager, current_notification_system_for_context,
+    current_object_data_cache_handle_for_context, current_object_store_handle_for_context, current_oidc_config_query,
+    current_ready_iam_handle, current_region, current_replication_pool_handle, current_replication_stats_handle,
+    current_replication_stats_handle_for_context, current_server_config_for_context, current_token_signing_key,
 };
 #[cfg(test)]
 pub(crate) use crate::runtime_sources::{
@@ -1086,7 +1086,11 @@ fn build_list_object_versions_metadata_output(
                        size: Some(object.size),
                        version_id: Some(version_id),
                        is_latest: Some(object.is_latest),
-                        e_tag: object.etag.clone().map(|etag| to_s3s_etag(&etag)),
+                        e_tag: object
+                            .etag
+                            .as_ref()
+                            .filter(|_| rustfs_utils::http::etag_can_be_listed(&object.user_defined))
+                            .map(|etag| to_s3s_etag(etag)),
                        storage_class: Some(ObjectVersionStorageClass::from(
                            object
                                .storage_class
@@ -1173,7 +1177,11 @@ fn build_list_objects_v2_metadata_output(
                    key: Some(encode_list_objects_v2_value(&object.name, encoding_type)),
                    last_modified: object.mod_time.map(Timestamp::from),
                    size: Some(object.get_actual_size_or_physical()),
-                    e_tag: object.etag.clone().map(|etag| to_s3s_etag(&etag)),
+                    e_tag: object
+                        .etag
+                        .as_ref()
+                        .filter(|_| rustfs_utils::http::etag_can_be_listed(&object.user_defined))
+                        .map(|etag| to_s3s_etag(etag)),
                    storage_class: Some(ObjectStorageClass::from(
                        object
                            .storage_class
@@ -4548,6 +4556,89 @@ mod tests {
        assert_eq!(err.code(), &S3ErrorCode::InternalError);
    }

+    #[test]
+    fn keyless_object_and_version_listings_do_not_expose_legacy_encrypted_fingerprints() {
+        let legacy_md5 = "fbfcbcc6d2f035411e6268c8b119593e";
+        let mut protected_metadata =
+            HashMap::from([("x-amz-server-side-encryption-customer-algorithm".to_owned(), "AES256".to_owned())]);
+        rustfs_utils::http::metadata_compat::insert_str(
+            &mut protected_metadata,
+            rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG,
+            "v1".to_owned(),
+        );
+        let objects = vec![
+            ObjectInfo {
+                name: "legacy-encrypted".to_owned(),
+                etag: Some(legacy_md5.to_owned()),
+                user_defined: Arc::new(HashMap::from([(
+                    "x-amz-server-side-encryption-customer-algorithm".to_owned(),
+                    "AES256".to_owned(),
+                )])),
+                ..Default::default()
+            },
+            ObjectInfo {
+                name: "protected-encrypted".to_owned(),
+                etag: Some("0123456789abcdef0123456789abcdef".to_owned()),
+                user_defined: Arc::new(protected_metadata),
+                ..Default::default()
+            },
+            ObjectInfo {
+                name: "plain".to_owned(),
+                etag: Some("9876543210abcdef9876543210abcdef".to_owned()),
+                ..Default::default()
+            },
+        ];
+        let listings = build_list_objects_v2_metadata_output(
+            ListObjectsV2Info {
+                objects: objects.clone(),
+                ..Default::default()
+            },
+            "demo-bucket",
+            &ListObjectsV2Params {
+                prefix: String::new(),
+                max_keys: 1000,
+                delimiter: None,
+                response_start_after: None,
+                start_after_for_query: None,
+                response_continuation_token: None,
+                decoded_continuation_token: None,
+            },
+            None,
+            false,
+            &HashMap::new(),
+        );
+        assert!(listings.contents[0].object.e_tag.is_none());
+        assert!(listings.contents[1].object.e_tag.is_some());
+        assert!(listings.contents[2].object.e_tag.is_some());
+        let xml = String::from_utf8(serialize_config(&listings).expect("serialize objects")).expect("UTF-8 objects");
+        assert!(!xml.contains(legacy_md5));
+
+        let versions = build_list_object_versions_metadata_output(
+            ListObjectVersionsInfo {
+                objects,
+                ..Default::default()
+            },
+            "demo-bucket",
+            &ListObjectVersionsParams {
+                prefix: String::new(),
+                max_keys: 1000,
+                delimiter: None,
+                key_marker: None,
+                version_id_marker: None,
+            },
+            None,
+            &HashMap::new(),
+        );
+        for (index, entry) in versions.entries.iter().enumerate() {
+            let ListObjectVersionMetadataEntry::Version(version, _) = entry else {
+                panic!("expected object version");
+            };
+            assert_eq!(version.e_tag.is_some(), index != 0);
+        }
+        let xml = String::from_utf8(serialize_config(&versions).expect("serialize versions")).expect("UTF-8 versions");
+        assert!(!xml.contains(legacy_md5));
+    }
+
    #[test]
    fn build_list_object_versions_metadata_output_maps_metadata_and_preserves_entry_order() {
        use time::macros::datetime;
@@ -103,11 +103,6 @@ pub fn resolve_oidc_config_query() -> Option<Arc<dyn OidcConfigQuery>> {
    resolve_oidc_config_query_with(get_global_app_context())
 }

-/// Resolve the paired federation service and OIDC configuration query from one runtime observation.
-pub(crate) fn resolve_federated_identity_runtime() -> Option<FederatedIdentityRuntimeSnapshot> {
-    resolve_federated_identity_runtime_with(get_global_app_context())
-}
-
 /// Publish the initialized federation service and OIDC configuration query.
 pub fn publish_federated_identity_runtime(
    service: Arc<FederatedIdentityService>,
@@ -116,11 +111,6 @@ pub fn publish_federated_identity_runtime(
    handles::publish_default_federated_identity_runtime(service, oidc_config_query)
 }

-/// Publish a service-only federation runtime for compatibility integrations.
-pub fn publish_federated_identity_service(service: Arc<FederatedIdentityService>) -> bool {
-    handles::publish_default_federated_identity_service(service)
-}
-
 /// Resolve a ready IAM system handle using AppContext-first precedence.
 pub fn resolve_ready_iam_handle() -> rustfs_iam::error::Result<Arc<IamSys<ObjectStore>>> {
    if let Some(context) = get_global_app_context() {
@@ -360,6 +350,7 @@ fn resolve_oidc_config_query_with(context: Option<Arc<AppContext>>) -> Option<Ar
    context.and_then(|context| context.federated_identity().oidc_config_query())
 }

+#[cfg(test)]
 fn resolve_federated_identity_runtime_with(context: Option<Arc<AppContext>>) -> Option<FederatedIdentityRuntimeSnapshot> {
    context.and_then(|context| context.federated_identity().runtime_snapshot())
 }
@@ -554,7 +545,7 @@ mod tests {
    use async_trait::async_trait;
    use rustfs_iam::{
        federation::{
-            FederatedIdentityRegistry, FederatedIdentityService,
+            CoreFederatedAuthorizationMapper, FederatedIdentityService,
            oidc::{OidcConfigQuery, StandardOidcAdapter},
        },
        oidc::OidcSys,
@@ -617,25 +608,11 @@ mod tests {
    }

    impl FederatedIdentityInterface for TestFederatedIdentityInterface {
-        fn handle(&self) -> Option<Arc<FederatedIdentityService>> {
-            self.runtime
-                .read()
-                .ok()
-                .and_then(|runtime| runtime.as_ref().map(|runtime| Arc::clone(&runtime.service)))
-        }
-
-        fn oidc_config_query(&self) -> Option<Arc<dyn OidcConfigQuery>> {
-            self.runtime
-                .read()
-                .ok()
-                .and_then(|runtime| runtime.as_ref().map(|runtime| Arc::clone(&runtime.oidc_config_query)))
-        }
-
        fn runtime_snapshot(&self) -> Option<FederatedIdentityRuntimeSnapshot> {
            self.runtime.read().ok().and_then(|runtime| {
                runtime
                    .as_ref()
-                    .map(|runtime| (Arc::clone(&runtime.service), Some(Arc::clone(&runtime.oidc_config_query))))
+                    .map(|runtime| (Arc::clone(&runtime.service), Arc::clone(&runtime.oidc_config_query)))
            })
        }

@@ -654,7 +631,8 @@ mod tests {
    fn test_federated_identity_runtime(oidc: OidcSys) -> (Arc<FederatedIdentityService>, Arc<dyn OidcConfigQuery>) {
        let adapter = Arc::new(StandardOidcAdapter::new(Arc::new(oidc)));
        let oidc_config_query = adapter.clone();
-        let service = Arc::new(FederatedIdentityService::new(FederatedIdentityRegistry::new(adapter)));
+        let mapper = CoreFederatedAuthorizationMapper::new(adapter.authorization_rules());
+        let service = Arc::new(adapter.into_service(mapper));
        (service, oidc_config_query)
    }

@@ -1206,11 +1184,7 @@ mod tests {
        let (resolved_oidc, resolved_oidc_query) =
            resolve_federated_identity_runtime_with(Some(context.clone())).expect("context federated identity runtime");
        assert!(Arc::ptr_eq(&resolved_oidc, &context_oidc));
-        assert!(
-            resolved_oidc_query
-                .as_ref()
-                .is_some_and(|query| Arc::ptr_eq(query, &context_oidc_query))
-        );
+        assert!(Arc::ptr_eq(&resolved_oidc_query, &context_oidc_query));
        assert!(publish_federated_identity_runtime_with(
            Some(context.clone()),
            fallback_oidc.clone(),
@@ -169,11 +169,6 @@ impl AppContext {
        self.federated_identity.publish_runtime(service, oidc_config_query)
    }

-    /// Publish a service-only federation runtime for compatibility integrations.
-    pub fn publish_federated_identity_service(&self, service: Arc<FederatedIdentityService>) -> bool {
-        self.federated_identity.publish_handle(service)
-    }
-
    pub fn kms(&self) -> Arc<dyn KmsInterface> {
        self.kms.clone()
    }
@@ -87,24 +87,15 @@ pub struct FederatedIdentityHandle {

 struct FederatedIdentityRuntime {
    service: Arc<FederatedIdentityService>,
-    oidc_config_query: Option<Arc<dyn OidcConfigQuery>>,
+    oidc_config_query: Arc<dyn OidcConfigQuery>,
 }

 impl FederatedIdentityHandle {
-    pub fn new(service: Option<Arc<FederatedIdentityService>>) -> Self {
-        Self {
-            runtime: StdRwLock::new(service.map(|service| FederatedIdentityRuntime {
-                service,
-                oidc_config_query: None,
-            })),
-        }
-    }
-
    pub fn with_runtime(runtime: Option<(Arc<FederatedIdentityService>, Arc<dyn OidcConfigQuery>)>) -> Self {
        Self {
            runtime: StdRwLock::new(runtime.map(|(service, oidc_config_query)| FederatedIdentityRuntime {
                service,
-                oidc_config_query: Some(oidc_config_query),
+                oidc_config_query,
            })),
        }
    }
@@ -112,45 +103,26 @@ impl FederatedIdentityHandle {

 impl Default for FederatedIdentityHandle {
    fn default() -> Self {
-        Self::new(None)
+        Self::with_runtime(None)
    }
 }

 impl FederatedIdentityInterface for FederatedIdentityHandle {
-    fn handle(&self) -> Option<Arc<FederatedIdentityService>> {
-        self.runtime_snapshot().map(|(service, _)| service)
-    }
-
-    fn oidc_config_query(&self) -> Option<Arc<dyn OidcConfigQuery>> {
-        self.runtime_snapshot().and_then(|(_, oidc_config_query)| oidc_config_query)
-    }
-
    fn runtime_snapshot(&self) -> Option<FederatedIdentityRuntimeSnapshot> {
        self.runtime.read().ok().and_then(|runtime| {
            runtime
                .as_ref()
-                .map(|runtime| (Arc::clone(&runtime.service), runtime.oidc_config_query.as_ref().map(Arc::clone)))
+                .map(|runtime| (Arc::clone(&runtime.service), Arc::clone(&runtime.oidc_config_query)))
        })
    }

-    fn publish_handle(&self, service: Arc<FederatedIdentityService>) -> bool {
-        let Ok(mut published_runtime) = self.runtime.write() else {
-            return false;
-        };
-        *published_runtime = Some(FederatedIdentityRuntime {
-            service,
-            oidc_config_query: None,
-        });
-        true
-    }
-
    fn publish_runtime(&self, service: Arc<FederatedIdentityService>, oidc_config_query: Arc<dyn OidcConfigQuery>) -> bool {
        let Ok(mut published_runtime) = self.runtime.write() else {
            return false;
        };
        *published_runtime = Some(FederatedIdentityRuntime {
            service,
-            oidc_config_query: Some(oidc_config_query),
+            oidc_config_query,
        });
        true
    }
@@ -639,15 +611,6 @@ pub fn publish_default_federated_identity_runtime(
    default_federated_identity_handle().publish_runtime(service, oidc_config_query)
 }

-/// Publish a service-only federation runtime for compatibility integrations.
-pub fn publish_default_federated_identity_service(service: Arc<FederatedIdentityService>) -> bool {
-    default_federated_identity_handle().publish_handle(service)
-}
-
-pub fn federated_identity_interface(service: Option<Arc<FederatedIdentityService>>) -> Arc<dyn FederatedIdentityInterface> {
-    Arc::new(FederatedIdentityHandle::new(service))
-}
-
 pub fn federated_identity_runtime_interface(
    runtime: Option<(Arc<FederatedIdentityService>, Arc<dyn OidcConfigQuery>)>,
 ) -> Arc<dyn FederatedIdentityInterface> {
@@ -686,14 +649,13 @@ pub fn default_buffer_config_interface() -> Arc<dyn BufferConfigInterface> {
 mod tests {
    use super::{
        KmsRuntimeHandle, KmsServiceManager, ServerConfigHandle, default_federated_identity_interface,
-        federated_identity_interface, federated_identity_runtime_interface, publish_default_federated_identity_runtime,
-        runtime_sources,
+        federated_identity_runtime_interface, publish_default_federated_identity_runtime, runtime_sources,
    };
    use crate::app::context::interfaces::{KmsRuntimeInterface, ServerConfigInterface};
    use rustfs_config::server_config::Config;
    use rustfs_iam::{
        federation::{
-            FederatedIdentityRegistry, FederatedIdentityService,
+            CoreFederatedAuthorizationMapper, FederatedIdentityService,
            oidc::{OidcConfigQuery, StandardOidcAdapter},
        },
        oidc::OidcSys,
@@ -705,7 +667,8 @@ mod tests {
        let oidc = OidcSys::empty().expect("empty OIDC configuration should be valid");
        let adapter = Arc::new(StandardOidcAdapter::new(Arc::new(oidc)));
        let oidc_config_query = adapter.clone();
-        let service = Arc::new(FederatedIdentityService::new(FederatedIdentityRegistry::new(adapter)));
+        let mapper = CoreFederatedAuthorizationMapper::new(adapter.authorization_rules());
+        let service = Arc::new(adapter.into_service(mapper));
        (service, oidc_config_query)
    }

@@ -726,7 +689,7 @@ mod tests {
        assert!(Arc::ptr_eq(&resolved_query, &first_query));
        let (resolved_service, resolved_query) = interface.runtime_snapshot().expect("initial federated identity runtime");
        assert!(Arc::ptr_eq(&resolved_service, &first));
-        assert!(resolved_query.as_ref().is_some_and(|query| Arc::ptr_eq(query, &first_query)));
+        assert!(Arc::ptr_eq(&resolved_query, &first_query));

        assert!(interface.publish_runtime(second.clone(), second_query.clone()));
        assert!(
@@ -738,36 +701,11 @@ mod tests {
        assert!(interface.oidc_config_query().is_some());
        let (resolved_service, resolved_query) = interface.runtime_snapshot().expect("replacement federated identity runtime");
        assert!(Arc::ptr_eq(&resolved_service, &second));
-        assert!(resolved_query.as_ref().is_some_and(|query| Arc::ptr_eq(query, &second_query)));
-    }
-
-    #[test]
-    fn federated_identity_handle_preserves_service_only_compatibility() {
-        let (service, _) = test_federated_identity_runtime();
-        let (replacement, _) = test_federated_identity_runtime();
-        let interface = federated_identity_interface(Some(service.clone()));
-
-        assert!(
-            interface
-                .handle()
-                .as_ref()
-                .is_some_and(|resolved| Arc::ptr_eq(resolved, &service))
-        );
-        assert!(interface.oidc_config_query().is_none());
-        assert!(interface.publish_handle(replacement.clone()));
-        assert!(
-            interface
-                .handle()
-                .as_ref()
-                .is_some_and(|resolved| Arc::ptr_eq(resolved, &replacement))
-        );
-        assert!(interface.oidc_config_query().is_none());
-        let (resolved_service, resolved_query) = interface.runtime_snapshot().expect("service-only federated identity runtime");
-        assert!(Arc::ptr_eq(&resolved_service, &replacement));
-        assert!(resolved_query.is_none());
+        assert!(Arc::ptr_eq(&resolved_query, &second_query));
    }

    #[test]
+    #[serial_test::serial(federated_identity_runtime)]
    fn default_federated_identity_handle_shares_early_publish_and_replacement() {
        let (first, first_query) = test_federated_identity_runtime();
        let (second, second_query) = test_federated_identity_runtime();
@@ -48,25 +48,22 @@ pub trait IamInterface: Send + Sync {
 }

 /// Published federation runtime handles for authentication and OIDC configuration consumers.
-pub type FederatedIdentityRuntimeSnapshot = (Arc<FederatedIdentityService>, Option<Arc<dyn OidcConfigQuery>>);
+pub type FederatedIdentityRuntimeSnapshot = (Arc<FederatedIdentityService>, Arc<dyn OidcConfigQuery>);

 pub trait FederatedIdentityInterface: Send + Sync {
-    fn handle(&self) -> Option<Arc<FederatedIdentityService>>;
+    fn runtime_snapshot(&self) -> Option<FederatedIdentityRuntimeSnapshot>;
+
+    fn handle(&self) -> Option<Arc<FederatedIdentityService>> {
+        self.runtime_snapshot().map(|(service, _)| service)
+    }
+
    fn oidc_config_query(&self) -> Option<Arc<dyn OidcConfigQuery>> {
-        None
+        self.runtime_snapshot().map(|(_, oidc_config_query)| oidc_config_query)
    }
-    fn runtime_snapshot(&self) -> Option<FederatedIdentityRuntimeSnapshot> {
-        self.handle().map(|service| (service, self.oidc_config_query()))
-    }
-    /// Compatibility publication hook for service-only implementations.
-    /// Standard OIDC runtimes publish the paired query through
-    /// [`Self::publish_runtime`].
-    fn publish_handle(&self, _service: Arc<FederatedIdentityService>) -> bool {
+
+    fn publish_runtime(&self, _service: Arc<FederatedIdentityService>, _oidc_config_query: Arc<dyn OidcConfigQuery>) -> bool {
        false
    }
-    fn publish_runtime(&self, service: Arc<FederatedIdentityService>, _oidc_config_query: Arc<dyn OidcConfigQuery>) -> bool {
-        self.publish_handle(service)
-    }
 }

 /// KMS interface for application-layer use-cases.
@@ -964,7 +964,11 @@ async fn copy_object_allows_new_version_for_locked_destination_but_blocks_explic
        .await
        .expect_err("explicit CopyObject overwrite of a locked version should be blocked");

-    assert_eq!(err.code(), &s3s::S3ErrorCode::AccessDenied);
+    assert_eq!(err.code(), &s3s::S3ErrorCode::InvalidArgument);
+    assert_eq!(
+        err.message(),
+        Some("Invalid argument: A destination versionId requires replication authorization")
+    );
    assert_eq!(read_object_bytes(&ecstore, bucket.as_str(), dst_object).await, source_payload);
    assert_eq!(live_object_version_count(&ecstore, bucket.as_str(), dst_object).await, 2);
 }
@@ -511,6 +511,15 @@ impl DefaultObjectUsecase {
        // cannot flood the log.
        let mut bulk_denial_logged = false;
        for (idx, obj_id) in delete.objects.iter().enumerate() {
+            if let Err(err) = validate_object_key(&obj_id.key, "DeleteObjects") {
+                delete_results[idx].error = Some(s3s::dto::Error {
+                    code: Some(err.code().as_str().to_string()),
+                    key: Some(obj_id.key.clone()),
+                    message: Some(err.to_string()),
+                    version_id: obj_id.version_id.clone(),
+                });
+                continue;
+            }
            let raw_version_id = obj_id.version_id.clone();
            let (version_id, version_uuid) = match normalize_delete_objects_version_id(raw_version_id.clone()) {
                Ok(parsed) => parsed,
@@ -3002,7 +3002,7 @@ impl DefaultObjectUsecase {
            // resolver; only the response headers, authorization and audit
            // summary are derived here, without a second KMS unwrap.
            Some(headers) => (
-                Some(headers.server_side_encryption),
+                Some(headers.server_side_encryption).filter(|_| headers.sse_customer_algorithm.is_none()),
                headers.sse_customer_algorithm,
                headers.sse_customer_key_md5,
                headers.ssekms_key_id,
@@ -492,12 +492,13 @@ impl DefaultObjectUsecase {
        })?;

        let metadata_map = info.user_defined.clone();
-        let server_side_encryption = metadata_map
-            .get("x-amz-server-side-encryption")
-            .map(|v| ServerSideEncryption::from(v.clone()));
        let sse_customer_algorithm = metadata_map
            .get("x-amz-server-side-encryption-customer-algorithm")
            .map(|v| SSECustomerAlgorithm::from(v.clone()));
+        let server_side_encryption = metadata_map
+            .get("x-amz-server-side-encryption")
+            .filter(|_| sse_customer_algorithm.is_none())
+            .map(|v| ServerSideEncryption::from(v.clone()));
        let sse_customer_key_md5 = metadata_map.get("x-amz-server-side-encryption-customer-key-md5").cloned();
        let sse_kms_key_id = metadata_map.get("x-amz-server-side-encryption-aws-kms-key-id").cloned();
        let storage_class = response_storage_class(&info, &metadata_map);
@@ -1408,7 +1408,10 @@ impl DefaultObjectUsecase {

        let output = PutObjectOutput {
            e_tag,
-            server_side_encryption: committed.effective_sse.clone(),
+            server_side_encryption: committed
+                .effective_sse
+                .clone()
+                .filter(|_| committed.sse_customer_algorithm.is_none()),
            sse_customer_algorithm: committed.sse_customer_algorithm.clone(),
            sse_customer_key_md5: committed.sse_customer_key_md5.clone(),
            ssekms_key_id: committed.effective_kms_key_id.clone(),
@@ -3688,6 +3691,12 @@ mod tests {
    async fn sse_c_put_is_accepted_on_a_bucket_with_default_encryption() {
        use md5::{Digest as _, Md5};

+        let context = crate::storage::storage_api::bootstrap_instance_ctx();
+        assert!(
+            context.object_encryption_resolver().is_some(),
+            "real SSE-C reads need the production encryption resolver"
+        );
+
        for (algorithm, kms_key_id, prefix) in [
            (ServerSideEncryption::AES256, None, "ssec-over-aes256-default"),
            (ServerSideEncryption::AWS_KMS, Some("bucket-key"), "ssec-over-kms-default"),
@@ -3709,14 +3718,69 @@ mod tests {
                .content_length(Some(i64::try_from(payload.len()).expect("test payload length must fit i64")))
                .sse_customer_algorithm(Some("AES256".to_string()))
                .sse_customer_key(Some(base64_simd::STANDARD.encode_to_string(customer_key)))
-                .sse_customer_key_md5(Some(key_md5))
+                .sse_customer_key_md5(Some(key_md5.clone()))
                .build()
                .expect("SSE-C PUT input must build");

-            DefaultObjectUsecase::from_global()
+            let usecase = DefaultObjectUsecase::from_global();
+            let response = usecase
                .execute_put_object(&FS::new(), build_request(input, Method::PUT))
                .await
                .unwrap_or_else(|err| panic!("a {algorithm} default bucket must accept an SSE-C PUT: {err:?}"));
+            assert_eq!(response.output.server_side_encryption, None);
+            assert_eq!(response.output.sse_customer_algorithm.as_deref(), Some("AES256"));
+            assert_eq!(response.output.sse_customer_key_md5.as_deref(), Some(key_md5.as_str()));
+
+            let encoded_key = base64_simd::STANDARD.encode_to_string(customer_key);
+            let mut read_headers = HeaderMap::new();
+            read_headers.insert("x-amz-server-side-encryption-customer-algorithm", HeaderValue::from_static("AES256"));
+            read_headers.insert(
+                "x-amz-server-side-encryption-customer-key",
+                HeaderValue::from_str(&encoded_key).expect("encoded customer key is a valid header"),
+            );
+            read_headers.insert(
+                "x-amz-server-side-encryption-customer-key-md5",
+                HeaderValue::from_str(&key_md5).expect("customer key MD5 is a valid header"),
+            );
+            let head_input = HeadObjectInput::builder()
+                .bucket(bucket.clone())
+                .key("ledger.csv".to_string())
+                .sse_customer_algorithm(Some("AES256".to_string()))
+                .sse_customer_key(Some(encoded_key.clone()))
+                .sse_customer_key_md5(Some(key_md5.clone()))
+                .build()
+                .expect("SSE-C HEAD input must build");
+            let mut head_request = build_request(head_input, Method::HEAD);
+            head_request.headers = read_headers.clone();
+            let head = Box::pin(usecase.execute_head_object(head_request))
+                .await
+                .expect("matching SSE-C key must authorize HEAD");
+            assert_eq!(head.output.server_side_encryption, None);
+            assert_eq!(head.output.sse_customer_algorithm.as_deref(), Some("AES256"));
+            assert_eq!(head.output.sse_customer_key_md5.as_deref(), Some(key_md5.as_str()));
+
+            let get_input = GetObjectInput::builder()
+                .bucket(bucket.clone())
+                .key("ledger.csv".to_string())
+                .sse_customer_algorithm(Some("AES256".to_string()))
+                .sse_customer_key(Some(encoded_key))
+                .sse_customer_key_md5(Some(key_md5.clone()))
+                .build()
+                .expect("SSE-C GET input must build");
+            let mut get_request = build_request(get_input, Method::GET);
+            get_request.headers = read_headers;
+            let get = Box::pin(usecase.execute_get_object(get_request))
+                .await
+                .expect("matching SSE-C key must authorize GET");
+            assert_eq!(get.output.server_side_encryption, None);
+            assert_eq!(get.output.sse_customer_algorithm.as_deref(), Some("AES256"));
+            assert_eq!(get.output.sse_customer_key_md5.as_deref(), Some(key_md5.as_str()));
+            let mut body = get.output.body.expect("SSE-C GET must include the decrypted body");
+            let mut plaintext = Vec::new();
+            while let Some(chunk) = body.next().await {
+                plaintext.extend_from_slice(&chunk.expect("SSE-C body must decrypt completely"));
+            }
+            assert_eq!(plaintext, payload, "header classification must preserve the complete decrypted payload");

            let stored = store
                .get_object_info(&bucket, "ledger.csv", &ObjectOptions::default())
@@ -35,7 +35,6 @@ pub(crate) use context::{
    resolve_bucket_monitor_handle as current_bucket_monitor_handle, resolve_buffer_config as current_buffer_config,
    resolve_daily_tier_stats as current_daily_tier_stats, resolve_deployment_id as current_deployment_id,
    resolve_encryption_service as current_encryption_service, resolve_endpoints_handle as current_endpoints_handle,
-    resolve_federated_identity_runtime as current_federated_identity_runtime,
    resolve_federated_identity_service as current_federated_identity_service, resolve_iam_handle as current_iam_handle,
    resolve_iam_ready as current_iam_ready, resolve_internode_metrics as current_internode_metrics,
    resolve_kms_runtime_service_manager as current_kms_runtime_service_manager,
@@ -47,6 +46,7 @@ pub(crate) use context::{
    resolve_object_data_cache_handle_for_context as current_object_data_cache_handle_for_context,
    resolve_object_store_handle as current_object_store_handle,
    resolve_object_store_handle_for_context as current_object_store_handle_for_context,
+    resolve_oidc_config_query as current_oidc_config_query,
    resolve_or_init_kms_runtime_service_manager as current_or_init_kms_runtime_service_manager,
    resolve_outbound_tls_generation as current_outbound_tls_generation, resolve_outbound_tls_state as current_outbound_tls_state,
    resolve_performance_metrics as current_performance_metrics, resolve_ready_iam_handle as current_ready_iam_handle,
@@ -63,13 +63,13 @@ pub(crate) fn publish_federated_identity_runtime(
    service: Arc<FederatedIdentityService>,
    oidc_config_query: Arc<dyn OidcConfigQuery>,
 ) -> bool {
-    warn_on_browser_redirect_fallback(&service);
+    warn_on_browser_redirect_fallback(service.has_providers());
    context::publish_federated_identity_runtime(service, oidc_config_query)
 }

-fn warn_on_browser_redirect_fallback(service: &FederatedIdentityService) {
+fn warn_on_browser_redirect_fallback(has_providers: bool) {
    let browser_redirect_url = rustfs_utils::get_env_opt_str(ENV_RUSTFS_BROWSER_REDIRECT_URL);
-    if service.has_providers() && browser_redirect_url.as_deref().is_none_or(|value| value.trim().is_empty()) {
+    if has_providers && browser_redirect_url.as_deref().is_none_or(|value| value.trim().is_empty()) {
        warn!(
            event = EVENT_OIDC_BROWSER_REDIRECT_FALLBACK,
            component = LOG_COMPONENT_MAIN,
@@ -103,76 +103,19 @@ pub(crate) fn current_app_context() -> Option<Arc<AppContext>> {
 #[cfg(test)]
 mod tests {
    use super::*;
-    use rustfs_iam::federation::oidc::OidcConfigQuery;
-    use rustfs_iam::federation::{
-        FederatedAuthorization, FederatedCodeExchange, FederatedIdentityProvider, FederatedIdentityRegistry,
-        FederatedProviderView, FederatedRedirectPolicy, Result as FederationResult,
+    use rustfs_iam::{
+        federation::{
+            CoreFederatedAuthorizationMapper,
+            oidc::{OidcConfigQuery, StandardOidcAdapter},
+        },
+        oidc::{OidcSys, make_test_sys, test_config},
    };
-    use rustfs_iam::oidc::OidcConfigSnapshot;
    use std::{
        io::{self, Write},
        sync::Mutex,
    };
    use tracing_subscriber::{Registry, fmt::MakeWriter, layer::SubscriberExt};

-    struct TestProvider {
-        has_providers: bool,
-    }
-
-    #[async_trait::async_trait]
-    impl FederatedIdentityProvider for TestProvider {
-        fn has_providers(&self) -> bool {
-            self.has_providers
-        }
-
-        fn list_providers(&self) -> Vec<FederatedProviderView> {
-            Vec::new()
-        }
-
-        fn list_visible_providers(&self) -> Vec<FederatedProviderView> {
-            Vec::new()
-        }
-
-        fn redirect_policy(&self, _provider_id: &str) -> Option<FederatedRedirectPolicy> {
-            None
-        }
-
-        async fn authorize_url(
-            &self,
-            _provider_id: &str,
-            _redirect_uri: &str,
-            _redirect_after: Option<String>,
-        ) -> FederationResult<String> {
-            unreachable!("startup publication test does not authorize")
-        }
-
-        async fn exchange_code(&self, _state: &str, _code: &str, _redirect_uri: &str) -> FederationResult<FederatedCodeExchange> {
-            unreachable!("startup publication test does not exchange codes")
-        }
-
-        async fn verify_web_identity_token(&self, _jwt: &str) -> FederationResult<FederatedAuthorization> {
-            unreachable!("startup publication test does not verify tokens")
-        }
-
-        async fn create_logout_token(&self, _provider_id: &str, _id_token: &str) -> FederationResult<String> {
-            unreachable!("startup publication test does not create logout tokens")
-        }
-
-        async fn build_logout_url(
-            &self,
-            _logout_token: &str,
-            _post_logout_redirect_uri: &str,
-        ) -> FederationResult<Option<String>> {
-            unreachable!("startup publication test does not build logout URLs")
-        }
-    }
-
-    impl OidcConfigQuery for TestProvider {
-        fn config_snapshot(&self) -> OidcConfigSnapshot {
-            OidcConfigSnapshot::new(Vec::new())
-        }
-    }
-
    #[derive(Clone, Default)]
    struct CapturedLogs(Arc<Mutex<Vec<u8>>>);

@@ -197,14 +140,14 @@ mod tests {
        }
    }

-    fn test_runtime(has_providers: bool) -> (Arc<FederatedIdentityService>, Arc<dyn OidcConfigQuery>) {
-        let provider = Arc::new(TestProvider { has_providers });
-        let service = Arc::new(FederatedIdentityService::new(FederatedIdentityRegistry::new(provider.clone())));
-        (service, provider)
-    }
-
-    fn capture_startup_publication(has_providers: bool, browser_redirect_url: Option<&str>) -> String {
+    fn capture_startup_publication(oidc: OidcSys, browser_redirect_url: Option<&str>) -> String {
        temp_env::with_var(ENV_RUSTFS_BROWSER_REDIRECT_URL, browser_redirect_url, || {
+            let adapter = Arc::new(StandardOidcAdapter::new(Arc::new(oidc)));
+            let oidc_config_query: Arc<dyn OidcConfigQuery> = adapter.clone();
+            let mapper = CoreFederatedAuthorizationMapper::new(adapter.authorization_rules());
+            let service = Arc::new(adapter.into_service(mapper));
+            let expected_service = service.clone();
+            let expected_oidc_config_query = oidc_config_query.clone();
            let logs = CapturedLogs::default();
            let captured = logs.0.clone();
            let subscriber = Registry::default().with(
@@ -217,8 +160,12 @@ mod tests {
            );

            tracing::subscriber::with_default(subscriber, || {
-                let (service, oidc_config_query) = test_runtime(has_providers);
                assert!(publish_federated_identity_runtime(service, oidc_config_query));
+                let (published_service, published_oidc_config_query) = context::default_federated_identity_interface()
+                    .runtime_snapshot()
+                    .expect("published federated identity runtime should resolve");
+                assert!(Arc::ptr_eq(&published_service, &expected_service));
+                assert!(Arc::ptr_eq(&published_oidc_config_query, &expected_oidc_config_query));
            });

            String::from_utf8(captured.lock().expect("captured log lock").clone()).expect("captured logs must be UTF-8")
@@ -226,16 +173,20 @@ mod tests {
    }

    #[test]
+    #[serial_test::serial(federated_identity_runtime)]
    fn oidc_startup_publication_warns_only_for_request_header_fallback() {
-        let output = capture_startup_publication(true, None);
+        let output = capture_startup_publication(make_test_sys(vec![test_config("default")]), None);

        assert!(output.contains("WARN"), "{output}");
        assert!(output.contains(EVENT_OIDC_BROWSER_REDIRECT_FALLBACK), "{output}");
        assert!(output.contains(ENV_RUSTFS_BROWSER_REDIRECT_URL), "{output}");
        assert!(output.contains("request_host_and_forwarded_proto"), "{output}");
        assert!(output.contains("trusted_console_origin_not_configured"), "{output}");
-        assert!(capture_startup_publication(false, None).is_empty());
-        assert!(capture_startup_publication(true, Some("https://console.example.com")).is_empty());
-        assert!(!capture_startup_publication(true, Some("  ")).is_empty());
+        assert!(capture_startup_publication(OidcSys::empty().expect("empty OIDC system should initialize"), None).is_empty());
+        assert!(
+            capture_startup_publication(make_test_sys(vec![test_config("default")]), Some("https://console.example.com"),)
+                .is_empty()
+        );
+        assert!(!capture_startup_publication(make_test_sys(vec![test_config("default")]), Some("  ")).is_empty());
    }
 }
@@ -140,6 +140,9 @@ struct InternalObjectAuthorization;
 #[derive(Clone, Debug)]
 struct RequestedObjectLockRetainUntil(OffsetDateTime);

+#[derive(Clone, Debug)]
+struct RequestedObjectLockRetention(ObjectLockRetention);
+
 #[derive(Clone, Debug)]
 struct StagedMultipartPartAuthorization;

@@ -885,6 +888,21 @@ fn authorization_conditions<T>(
    }
    merge_list_bucket_query_conditions(action, req.uri.query(), &mut conditions);
    merge_request_object_tag_conditions(action, &req.headers, &mut conditions)?;
+    if let Some(RequestedObjectLockRetention(retention)) = req.extensions.get::<RequestedObjectLockRetention>() {
+        // Standalone retention writes apply the XML body, so headers must not
+        // supply a different value to the policy decision.
+        conditions.remove("object-lock-mode");
+        conditions.remove("object-lock-retain-until-date");
+        if let Some(mode) = &retention.mode {
+            conditions.insert("object-lock-mode".to_string(), vec![mode.as_str().to_string()]);
+        }
+        if let Some(date) = &retention.retain_until_date {
+            let date = OffsetDateTime::from(date.clone())
+                .format(&time::format_description::well_known::Rfc3339)
+                .map_err(|_| S3Error::with_message(S3ErrorCode::InvalidArgument, "Invalid retain-until date"))?;
+            conditions.insert("object-lock-retain-until-date".to_string(), vec![date]);
+        }
+    }
    merge_object_lock_remaining_retention_days_condition(
        req.extensions.get::<RequestedObjectLockRetainUntil>(),
        OffsetDateTime::now_utc(),
@@ -908,6 +926,12 @@ pub(crate) fn set_requested_object_lock_retain_until<T>(req: &mut S3Request<T>,
    }
 }

+fn set_requested_object_lock_retention(req: &mut S3Request<PutObjectRetentionInput>) {
+    let retention = req.input.retention.clone().unwrap_or_default();
+    set_requested_object_lock_retain_until(req, retention.retain_until_date.as_ref());
+    req.extensions.insert(RequestedObjectLockRetention(retention));
+}
+
 fn merge_object_lock_remaining_retention_days_condition(
    requested: Option<&RequestedObjectLockRetainUntil>,
    now: OffsetDateTime,
@@ -1123,11 +1147,8 @@ pub(crate) fn delete_object_authorize_action(version_id: Option<&str>) -> Action
 /// conflated at the authorization boundary.
 ///
 /// Note: `ActionSet::is_match` still maps a `s3:GetObjectVersion` grant onto a
-/// `s3:GetObject` request. That mapping is intentionally left in place until the
-/// remaining version-aware read paths (HeadObject, GetObjectAcl, tagging) get the
-/// same treatment — see the GHSA-3ppv follow-up audit. It does not re-open this
-/// disclosure: it only broadens a Version grant toward current reads, never the
-/// reverse.
+/// `s3:GetObject` request. It only broadens a Version grant toward current
+/// reads, never the reverse.
 fn versioned_read_action(version_id: Option<&str>) -> Action {
    if version_id.is_some() {
        Action::S3Action(S3Action::GetObjectVersionAction)
@@ -1136,6 +1157,20 @@ fn versioned_read_action(version_id: Option<&str>) -> Action {
    }
 }

+fn versioned_tagging_action(action: S3Action, version_id: Option<&str>) -> Action {
+    let action = if version_id.is_some() {
+        match action {
+            S3Action::GetObjectTaggingAction => S3Action::GetObjectVersionTaggingAction,
+            S3Action::PutObjectTaggingAction => S3Action::PutObjectVersionTaggingAction,
+            S3Action::DeleteObjectTaggingAction => S3Action::DeleteObjectVersionTaggingAction,
+            action => action,
+        }
+    } else {
+        action
+    };
+    Action::S3Action(action)
+}
+
 async fn get_or_fetch_object_tag_conditions<T>(
    req: &mut S3Request<T>,
    store: &ECStore,
@@ -2560,7 +2595,11 @@ impl S3Access for FS {
        req_info.object = Some(req.input.key.clone());
        req_info.version_id = req.input.version_id.clone();

-        authorize_request(req, Action::S3Action(S3Action::DeleteObjectTaggingAction)).await
+        authorize_request(
+            req,
+            versioned_tagging_action(S3Action::DeleteObjectTaggingAction, req.input.version_id.as_deref()),
+        )
+        .await
    }

    /// Checks whether the DeleteObjects request has accesses to the resources.
@@ -2881,7 +2920,11 @@ impl S3Access for FS {
        req_info.object = Some(req.input.key.clone());
        req_info.version_id = req.input.version_id.clone();

-        authorize_request(req, Action::S3Action(S3Action::GetObjectTaggingAction)).await
+        authorize_request(
+            req,
+            versioned_tagging_action(S3Action::GetObjectTaggingAction, req.input.version_id.as_deref()),
+        )
+        .await
    }

    /// Checks whether the GetObjectTorrent request has accesses to the resources.
@@ -2933,7 +2976,7 @@ impl S3Access for FS {
            return Ok(());
        }

-        authorize_request(req, Action::S3Action(S3Action::GetObjectAction)).await?;
+        authorize_request(req, versioned_read_action(req.input.version_id.as_deref())).await?;
        req.extensions.insert(source_generation);
        Ok(())
    }
@@ -3336,12 +3379,7 @@ impl S3Access for FS {

        let bucket = req.input.bucket.clone();
        let bucket_generation = load_bucket_generation(self, req, &bucket).await;
-        let requested_retain_until = req
-            .input
-            .retention
-            .as_ref()
-            .and_then(|retention| retention.retain_until_date.clone());
-        set_requested_object_lock_retain_until(req, requested_retain_until.as_ref());
+        set_requested_object_lock_retention(req);
        authorize_request(req, Action::S3Action(S3Action::PutObjectRetentionAction)).await?;

        // S3 Standard: When bypass_governance header is set, must have s3:BypassGovernanceRetention permission
@@ -3362,7 +3400,11 @@ impl S3Access for FS {
        req_info.object = Some(req.input.key.clone());
        req_info.version_id = req.input.version_id.clone();

-        authorize_request(req, Action::S3Action(S3Action::PutObjectTaggingAction)).await
+        authorize_request(
+            req,
+            versioned_tagging_action(S3Action::PutObjectTaggingAction, req.input.version_id.as_deref()),
+        )
+        .await
    }

    /// Checks whether the PutPublicAccessBlock request has accesses to the resources.
@@ -3481,9 +3523,9 @@ mod tests {
        owner_can_bypass_policy_deny, post_object_authorize_action, put_bucket_policy_authorize_action, request_context_from_req,
        request_object_store, require_owned_reserved_table_object, retention_write_requested, table_data_plane_admin_action,
        table_data_plane_content_mutation, table_data_plane_resource_for_request, table_publication_guard_error,
-        validate_post_object_success_controls, versioned_read_action,
+        validate_post_object_success_controls, versioned_read_action, versioned_tagging_action,
    };
-    use super::{remaining_retention_days, set_requested_object_lock_retain_until};
+    use super::{remaining_retention_days, set_requested_object_lock_retain_until, set_requested_object_lock_retention};
    use crate::auth::OBJECT_LOCK_REMAINING_RETENTION_DAYS_CONDITION;
    use crate::error::ApiError;
    use crate::storage::storage_api::contract::bucket::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions};
@@ -4285,11 +4327,109 @@ mod tests {
        );
    }

-    fn remaining_retention_days_conditions(req: &S3Request<()>, action: Action) -> HashMap<String, Vec<String>> {
+    #[tokio::test]
+    async fn versioned_tagging_requires_version_permissions() {
+        use rustfs_policy::policy::{Args, Policy};
+        let policy = Policy::parse_config(br#"{
+            "Version":"2012-10-17",
+            "Statement":[{"Effect":"Allow","Action":["s3:GetObjectTagging","s3:PutObjectTagging","s3:DeleteObjectTagging"],"Resource":"arn:aws:s3:::bucket/*"}]
+        }"#).expect("current-object tagging policy should parse");
+        let conditions = HashMap::new();
+        let claims = HashMap::new();
+        let groups = None;
+        for (current, versioned) in [
+            (S3Action::GetObjectTaggingAction, S3Action::GetObjectVersionTaggingAction),
+            (S3Action::PutObjectTaggingAction, S3Action::PutObjectVersionTaggingAction),
+            (S3Action::DeleteObjectTaggingAction, S3Action::DeleteObjectVersionTaggingAction),
+        ] {
+            for version in [None, Some("null"), Some("0194e0f1-0000-7000-8000-000000000000")] {
+                let action = versioned_tagging_action(current, version);
+                assert_eq!(action, Action::S3Action(if version.is_some() { versioned } else { current }));
+                let args = Args {
+                    account: "writer",
+                    groups: &groups,
+                    action,
+                    bucket: "bucket",
+                    conditions: &conditions,
+                    is_owner: false,
+                    object: "object",
+                    claims: &claims,
+                    deny_only: false,
+                };
+                assert_eq!(policy.is_allowed(&args).await, version.is_none(), "version selector {version:?}");
+            }
+        }
+    }
+
+    fn remaining_retention_days_conditions<T>(req: &S3Request<T>, action: Action) -> HashMap<String, Vec<String>> {
        let credentials = rustfs_credentials::Credentials::default();
        authorization_conditions(req, &credentials, None, None, None, None, action).expect("conditions should build")
    }

+    #[tokio::test]
+    async fn retention_policy_checks_body_instead_of_conflicting_headers() {
+        use rustfs_policy::policy::{Args, Policy};
+        let mut req = build_request(
+            PutObjectRetentionInput {
+                retention: Some(ObjectLockRetention {
+                    mode: Some(ObjectLockRetentionMode::from_static("COMPLIANCE")),
+                    retain_until_date: Some(
+                        Timestamp::parse(TimestampFormat::DateTime, "2035-01-02T00:00:00+23:59").expect("body date should parse"),
+                    ),
+                }),
+                ..Default::default()
+            },
+            Method::PUT,
+        );
+        req.headers
+            .insert("x-amz-object-lock-mode", HeaderValue::from_static("GOVERNANCE"));
+        req.headers
+            .insert("x-amz-object-lock-retain-until-date", HeaderValue::from_static("2029-01-01T00:00:00Z"));
+        set_requested_object_lock_retention(&mut req);
+        let action = Action::S3Action(S3Action::PutObjectRetentionAction);
+        let conditions = remaining_retention_days_conditions(&req, action);
+        assert_eq!(conditions.get("object-lock-mode"), Some(&vec!["COMPLIANCE".to_string()]));
+        let applied = OffsetDateTime::from(
+            req.input
+                .retention
+                .as_ref()
+                .expect("body retention")
+                .retain_until_date
+                .clone()
+                .expect("body date"),
+        );
+        let authorized = OffsetDateTime::parse(
+            &conditions["object-lock-retain-until-date"][0],
+            &time::format_description::well_known::Rfc3339,
+        )
+        .expect("condition date should parse");
+        assert_eq!(authorized.unix_timestamp(), applied.unix_timestamp());
+        let policy = Policy::parse_config(br#"{
+            "Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":"s3:PutObjectRetention","Resource":"arn:aws:s3:::bucket/*",
+            "Condition":{"DateLessThan":{"s3:object-lock-retain-until-date":"2030-01-01T00:00:00Z"}}}]
+        }"#).expect("bounded retention policy should parse");
+        let args = Args {
+            account: "writer",
+            groups: &None,
+            action,
+            bucket: "bucket",
+            conditions: &conditions,
+            is_owner: false,
+            object: "object",
+            claims: &HashMap::new(),
+            deny_only: false,
+        };
+        assert!(
+            !policy.is_allowed(&args).await,
+            "an in-range header cannot authorize an out-of-range body"
+        );
+        req.input.retention = None;
+        set_requested_object_lock_retention(&mut req);
+        let conditions = remaining_retention_days_conditions(&req, action);
+        assert!(!conditions.contains_key("object-lock-mode"));
+        assert!(!conditions.contains_key("object-lock-retain-until-date"));
+    }
+
    fn request_retaining_for(retain_for: time::Duration) -> S3Request<()> {
        let mut req = build_request((), Method::PUT);
        let retain_until = OffsetDateTime::now_utc() + retain_for;
@@ -402,6 +402,30 @@ async fn validate_table_catalog_object_mutation(bucket: &str, key: &str) -> S3Re
 const MAXIMUM_RETENTION_DAYS: i32 = 36_500;
 const MAXIMUM_RETENTION_YEARS: i32 = 100;

+pub(crate) fn object_lock_retention_from_metadata(metadata: &HashMap<String, String>) -> S3Result<ObjectLockRetention> {
+    let mode = metadata
+        .get(AMZ_OBJECT_LOCK_MODE_LOWER)
+        .map(|value| ObjectLockRetentionMode::from(value.as_str().to_string()));
+    let retain_until_date = metadata
+        .get(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER)
+        .and_then(|value| OffsetDateTime::parse(value, &Rfc3339).ok())
+        // The S3 timestamp serializer emits a literal Z, so its wall clock
+        // must be UTC before it is serialized.
+        .map(|date| {
+            date.checked_to_offset(time::UtcOffset::UTC)
+                .filter(|date| (0..=9999).contains(&date.year()))
+                .map(Timestamp::from)
+                .ok_or_else(|| {
+                    S3Error::with_message(
+                        S3ErrorCode::InternalError,
+                        "Stored retention date is outside the supported UTC range".to_string(),
+                    )
+                })
+        })
+        .transpose()?;
+    Ok(ObjectLockRetention { mode, retain_until_date })
+}
+
 fn invalid_object_lock_configuration(message: impl Into<String>) -> S3Error {
    S3Error::with_message(S3ErrorCode::MalformedXML, message.into())
 }
@@ -1112,19 +1136,8 @@ impl S3 for FS {
            s3_error!(InternalError, "{}", e.to_string())
        })?;

-        let mode = object_info
-            .user_defined
-            .get(AMZ_OBJECT_LOCK_MODE_LOWER)
-            .map(|v| ObjectLockRetentionMode::from(v.as_str().to_string()));
-
-        let retain_until_date = object_info
-            .user_defined
-            .get(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER)
-            .and_then(|v| OffsetDateTime::parse(v.as_str(), &Rfc3339).ok())
-            .map(Timestamp::from);
-
        let output = GetObjectRetentionOutput {
-            retention: Some(ObjectLockRetention { mode, retain_until_date }),
+            retention: Some(object_lock_retention_from_metadata(&object_info.user_defined)?),
        };
        let version_id = req.input.version_id.clone().unwrap_or_default();
        helper = helper.object(object_info).version_id(version_id);
@@ -340,6 +340,10 @@ pub(crate) fn validate_object_key(key: &str, operation: &str) -> S3Result<()> {
        ));
    }

+    if key.is_empty() || !crate::storage::storage_api::ecstore_bucket::utils::is_valid_object_prefix(key) {
+        return Err(S3Error::with_message(S3ErrorCode::InvalidArgument, "Invalid object key"));
+    }
+
    // Log debug info for keys with special characters to help diagnose encoding issues
    if key.contains([' ', '+', '%']) {
        debug!(
@@ -18,7 +18,8 @@ mod tests {
    use crate::server::cors;
    use crate::storage::StorageError;
    use crate::storage::ecfs::{
-        FS, parse_object_version_id, propagate_object_lock_peer_reload, validate_object_lock_configuration_input,
+        FS, object_lock_retention_from_metadata, parse_object_version_id, propagate_object_lock_peer_reload,
+        validate_object_lock_configuration_input,
    };
    use crate::storage::ecfs_extend::{apply_bucket_default_lock_retention, map_bucket_object_lock_config_state};
    use crate::storage::s3_api::common::{rustfs_initiator, rustfs_owner};
@@ -988,6 +989,44 @@ mod tests {
        assert!(validate_list_object_unordered_with_delimiter(delimiter_some, complex_query_without_unordered).is_ok());
    }

+    #[test]
+    fn retention_response_serialization_preserves_offset_metadata_instant() {
+        for value in [
+            "2035-01-02T00:00:00+23:59",
+            "2035-01-02T00:00:00-23:59",
+            "2035-01-02T00:00:00Z",
+        ] {
+            let metadata = HashMap::from([
+                (AMZ_OBJECT_LOCK_MODE_LOWER.to_owned(), "COMPLIANCE".to_owned()),
+                (AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER.to_owned(), value.to_owned()),
+            ]);
+            let retention = object_lock_retention_from_metadata(&metadata).expect("representable retention instant");
+            assert_eq!(retention.mode.as_ref().expect("mode").as_str(), "COMPLIANCE");
+            let mut serialized = Vec::new();
+            retention
+                .retain_until_date
+                .expect("stored retention date")
+                .format(s3s::dto::TimestampFormat::DateTime, &mut serialized)
+                .expect("serialize the response timestamp");
+            let serialized = String::from_utf8(serialized).expect("UTF-8 timestamp");
+            let returned = OffsetDateTime::parse(&serialized, &Rfc3339).expect("parse wire timestamp");
+            let stored = OffsetDateTime::parse(value, &Rfc3339).expect("parse stored timestamp");
+            assert_eq!(returned.unix_timestamp(), stored.unix_timestamp(), "stored date {value}");
+        }
+    }
+
+    #[test]
+    fn retention_response_rejects_instants_outside_the_utc_wire_range() {
+        for value in ["9999-12-31T23:59:59-23:59", "0000-01-01T00:00:00+23:59"] {
+            let metadata = HashMap::from([
+                (AMZ_OBJECT_LOCK_MODE_LOWER.to_owned(), "COMPLIANCE".to_owned()),
+                (AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER.to_owned(), value.to_owned()),
+            ]);
+            let error = object_lock_retention_from_metadata(&metadata).expect_err("unrepresentable UTC timestamp");
+            assert_eq!(error.code(), &S3ErrorCode::InternalError);
+        }
+    }
+
    #[test]
    fn test_parse_object_lock_retention() {
        use time::macros::datetime;
@@ -368,6 +368,13 @@ pub async fn put_opts_with_replication_authorization(
    metadata: HashMap<String, String>,
    replication_request_authorized: bool,
 ) -> Result<ObjectOptions> {
+    if vid.is_some() && !replication_request_authorized {
+        return Err(StorageError::InvalidArgument(
+            bucket.to_owned(),
+            object.to_owned(),
+            "A destination versionId requires replication authorization".to_owned(),
+        ));
+    }
    let versioning_cfg = bucket_versioning_config_for_write(bucket).await?;
    let versioned = versioning_cfg.prefix_enabled(object);
    let version_suspended = versioning_cfg.prefix_suspended(object);
@@ -1532,7 +1539,15 @@ mod tests {
        let headers = create_test_headers();
        let invalid_uuid = "invalid-uuid".to_string();

-        let result = put_opts("test-bucket", "test-object", Some(invalid_uuid), &headers, HashMap::new()).await;
+        let result = put_opts_with_replication_authorization(
+            "test-bucket",
+            "test-object",
+            Some(invalid_uuid),
+            &headers,
+            HashMap::new(),
+            true,
+        )
+        .await;

        assert!(result.is_err());
        if let Err(err) = result {
@@ -1556,13 +1571,35 @@ mod tests {
        // get_opts / del_opts already do.
        let headers = create_test_headers();

-        let opts = put_opts("test-bucket", "test-object", Some("null".to_string()), &headers, HashMap::new())
-            .await
-            .expect("PUT with versionId=null must be accepted as the null version");
+        let opts = put_opts_with_replication_authorization(
+            "test-bucket",
+            "test-object",
+            Some("null".to_string()),
+            &headers,
+            HashMap::new(),
+            true,
+        )
+        .await
+        .expect("PUT with versionId=null must be accepted as the null version");

        assert_eq!(opts.version_id, Some(Uuid::nil().to_string()));
    }

+    #[tokio::test]
+    async fn ordinary_writes_cannot_choose_a_destination_version() {
+        let headers = create_test_headers();
+        for version in ["null", "", "0194e0f1-0000-7000-8000-000000000000"] {
+            let err = put_opts("test-bucket", "test-object", Some(version.to_owned()), &headers, HashMap::new())
+                .await
+                .expect_err("ordinary PUT must not overwrite a selected version");
+            assert!(matches!(err, StorageError::InvalidArgument(_, _, _)));
+            let err = copy_dst_opts("test-bucket", "test-object", Some(version.to_owned()), &headers, HashMap::new())
+                .await
+                .expect_err("ordinary COPY must not overwrite a selected version");
+            assert!(matches!(err, StorageError::InvalidArgument(_, _, _)));
+        }
+    }
+
    #[tokio::test]
    async fn test_copy_dst_opts() {
        let headers = create_test_headers();
@@ -226,7 +226,11 @@ pub(crate) fn build_list_object_versions_output(
            size: Some(v.size),
            version_id: Some(list_versions_response_version_id(v.version_id)),
            is_latest: Some(v.is_latest),
-            e_tag: v.etag.clone().map(|etag| to_s3s_etag(&etag)),
+            e_tag: v
+                .etag
+                .as_ref()
+                .filter(|_| rustfs_utils::http::etag_can_be_listed(&v.user_defined))
+                .map(|etag| to_s3s_etag(etag)),
            storage_class: v.storage_class.clone().map(ObjectVersionStorageClass::from),
            ..Default::default()
        })
@@ -311,7 +315,11 @@ pub(crate) fn build_list_objects_v2_output(
                // logical-size sentinel; never expose that internal value in
                // an S3 response.
                size: Some(v.get_actual_size_or_physical()),
-                e_tag: v.etag.clone().map(|etag| to_s3s_etag(&etag)),
+                e_tag: v
+                    .etag
+                    .as_ref()
+                    .filter(|_| rustfs_utils::http::etag_can_be_listed(&v.user_defined))
+                    .map(|etag| to_s3s_etag(etag)),
                storage_class: v.storage_class.clone().map(ObjectStorageClass::from),
                ..Default::default()
            };
@@ -709,6 +717,83 @@ mod tests {
        );
    }

+    #[test]
+    fn standard_list_responses_hide_legacy_encrypted_fingerprints() {
+        let mut protected_metadata = std::collections::HashMap::from([(
+            "x-amz-server-side-encryption-customer-algorithm".to_owned(),
+            "AES256".to_owned(),
+        )]);
+        rustfs_utils::http::metadata_compat::insert_str(
+            &mut protected_metadata,
+            rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG,
+            "v1".to_owned(),
+        );
+        let objects: Vec<ObjectInfo> = [
+            std::collections::HashMap::from([(
+                "x-amz-server-side-encryption-customer-algorithm".to_owned(),
+                "AES256".to_owned(),
+            )]),
+            std::collections::HashMap::from([("x-amz-server-side-encryption".to_owned(), "aws:kms".to_owned())]),
+            protected_metadata,
+            std::collections::HashMap::new(),
+        ]
+        .into_iter()
+        .enumerate()
+        .map(|(index, metadata)| ObjectInfo {
+            name: format!("object-{index}"),
+            etag: Some("0123456789abcdef0123456789abcdef".to_owned()),
+            user_defined: std::sync::Arc::new(metadata),
+            size: 42,
+            ..Default::default()
+        })
+        .collect();
+        let v2 = build_list_objects_v2_output(
+            ListObjectsV2Info {
+                objects: objects.clone(),
+                ..Default::default()
+            },
+            false,
+            1000,
+            "bucket".to_owned(),
+            String::new(),
+            None,
+            None,
+            None,
+            None,
+        );
+        let v1 = build_list_objects_output(v2.clone(), None);
+        for listing in [&v1.contents, &v2.contents] {
+            let entries = listing.as_ref().expect("listed objects");
+            assert_eq!(entries.len(), 4);
+            for (index, entry) in entries.iter().enumerate() {
+                assert_eq!(entry.e_tag.is_some(), index >= 2);
+                assert_eq!(entry.key.as_deref(), Some(format!("object-{index}").as_str()));
+                assert_eq!(entry.size, Some(42));
+            }
+        }
+        let versions = build_list_object_versions_output(
+            ListObjectVersionsInfo {
+                objects,
+                ..Default::default()
+            },
+            "bucket".to_owned(),
+            &ListObjectVersionsParams {
+                prefix: String::new(),
+                delimiter: None,
+                key_marker: None,
+                version_id_marker: None,
+                max_keys: 1000,
+            },
+            None,
+        );
+        let versions = versions.versions.expect("listed versions");
+        assert_eq!(versions.len(), 4);
+        for (index, entry) in versions.iter().enumerate() {
+            assert_eq!(entry.e_tag.is_some(), index >= 2);
+            assert_eq!(entry.size, Some(42));
+        }
+    }
+
    #[test]
    fn list_responses_report_standard_for_legacy_label_only_file_metadata() {
        let version_id = Uuid::parse_str("11111111-2222-3333-4444-555555555555").expect("fixture version ID should be valid");
@@ -4127,6 +4127,7 @@ pub fn strip_managed_encryption_metadata(metadata: &mut HashMap<String, String>)
    for key in KEYS.iter() {
        metadata.remove(*key);
    }
+    rustfs_utils::http::metadata_compat::remove_str(metadata, rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG);
 }

 pub fn mark_encrypted_multipart_metadata(metadata: &mut HashMap<String, String>) {
@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""Resolve a nightly manifest from its exact GitHub build attempt and artifact."""
+"""Resolve a candidate manifest from its exact GitHub build attempt and artifact."""
 from __future__ import annotations

 import hashlib
@@ -15,6 +15,7 @@ import zipfile
 REPOSITORY = "rustfs/rustfs"
 ROOT = Path(__file__).resolve().parents[1]
 MAX_ARCHIVE = 1024 * 1024
+VALIDATION_BUILD_WORKFLOWS = {".github/workflows/rustfs-validation-package.yml", ".github/workflows/rustfs-functional-chain.yml"}


 def require(condition, message):
@@ -39,8 +40,11 @@ def validate_manifest(manifest, run):
    require(isinstance(manifest, dict), "manifest must be an object")
    common = {"schema", "source_sha", "build_run_id", "build_run_attempt", "package_url", "package_sha256"}
    version = manifest.get("schema")
-    require(type(version) is int and version in (1, 2), "unsupported candidate schema")
-    require(set(manifest) == (common if version == 1 else common | {"workflow_sha", "source_ref"}), "unexpected candidate fields")
+    require(type(version) is int and version in (1, 2, 3), "unsupported candidate schema")
+    expected_fields = common if version == 1 else common | {"workflow_sha", "source_ref"}
+    if version == 3:
+        expected_fields |= {"channel"}
+    require(set(manifest) == expected_fields, "unexpected candidate fields")
    require(positive(manifest["build_run_id"]) and positive(manifest["build_run_attempt"]), "invalid build identity")
    require((manifest["build_run_id"], manifest["build_run_attempt"]) == (run["id"], run["run_attempt"]), "candidate belongs to another build attempt")
    require(sha(manifest["source_sha"]) and sha(manifest["package_sha256"], 64), "invalid candidate hash")
@@ -49,7 +53,11 @@ def validate_manifest(manifest, run):
    else:
        require(manifest["workflow_sha"] == run["head_sha"] and sha(manifest["workflow_sha"]), "candidate workflow SHA differs from artifact provenance")
        require(isinstance(manifest["source_ref"], str) and bool(re.fullmatch(r"[A-Za-z0-9_./-]{1,200}", manifest["source_ref"])), "invalid build source ref")
-    expected = (f"https://dl.rustfs.com/artifacts/rustfs/packages/nightly/runs/{run['id']}/"
+        if version == 3:
+            require(manifest["channel"] == "validation", "validation candidate has the wrong publication channel")
+            require(manifest["source_ref"] == manifest["source_sha"], "validation candidate must bind an exact source SHA")
+    channel = "validation" if version == 3 else "nightly"
+    expected = (f"https://dl.rustfs.com/artifacts/rustfs/packages/{channel}/runs/{run['id']}/"
                f"{run['run_attempt']}/{manifest['package_sha256']}/rustfs.deb")
    require(manifest["package_url"] == expected, "package URL does not bind the run, attempt and checksum")
    return manifest
@@ -72,19 +80,31 @@ def read_json_artifact(archive, artifact, run, expected_name, member, max_json=1


 def read_manifest(archive, artifact, run):
-    name = f"nightly-candidate-{run['id']}-{run['run_attempt']}"
-    return validate_manifest(read_json_artifact(archive, artifact, run, name, name + ".json"), run)
+    prefix = "validation-candidate" if run.get("path") in VALIDATION_BUILD_WORKFLOWS else "nightly-candidate"
+    name = f"{prefix}-{run['id']}-{run['run_attempt']}"
+    manifest = validate_manifest(read_json_artifact(archive, artifact, run, name, name + ".json"), run)
+    if prefix == "validation-candidate":
+        require(manifest["schema"] == 3 and manifest["channel"] == "validation",
+                "validation workflow requires a schema-3 validation candidate")
+    else:
+        require(manifest["schema"] != 3, "nightly workflow cannot publish a validation candidate")
+    return manifest


-def resolve(run_id, attempt):
+def resolve(run_id, attempt, *, allow_validation=False):
    require(positive(run_id) and positive(attempt), "build run and attempt are required positive integers")
    endpoint = f"repos/{REPOSITORY}/actions/runs/{run_id}"
    run = api(f"{endpoint}/attempts/{attempt}")
    require(run.get("id") == run_id and run.get("run_attempt") == attempt, "GitHub returned a different build attempt")
-    require(run.get("path") == ".github/workflows/nightly-gnu.yml" and run.get("head_branch") == "main", "candidate must come from nightly-gnu on main")
+    path = run.get("path")
+    validation = allow_validation and path in VALIDATION_BUILD_WORKFLOWS
+    require(path == ".github/workflows/nightly-gnu.yml" and run.get("head_branch") == "main" or validation,
+            "candidate must come from nightly-gnu on main or an explicitly allowed validation build")
    require((run.get("head_repository") or {}).get("full_name") == REPOSITORY, "candidate came from another repository")
-    require(run.get("event") in ("schedule", "workflow_dispatch") and run.get("status") == "completed" and run.get("conclusion") == "success", "nightly attempt has not completed successfully")
-    name = f"nightly-candidate-{run_id}-{attempt}"
+    require(run.get("event") in ("schedule", "workflow_dispatch") and run.get("status") == "completed" and run.get("conclusion") == "success", "candidate attempt has not completed successfully")
+    if validation:
+        require(run.get("event") == "workflow_dispatch", "validation candidate must be manually dispatched")
+    name = f"{'validation' if validation else 'nightly'}-candidate-{run_id}-{attempt}"
    artifacts = []
    for page in range(1, 11):
        batch = api(f"{endpoint}/artifacts?per_page=100&page={page}")["artifacts"]
@@ -76,6 +76,98 @@ class CandidateTests(unittest.TestCase):
        self.assertEqual(result["artifact_id"], 789)
        self.assertEqual(result["manifest"]["source_sha"], "b" * 40)

+    def test_validation_manifest_binds_requested_source_and_private_channel(self):
+        manifest = {**self.manifest, "schema": 3, "channel": "validation", "source_ref": self.manifest["source_sha"],
+                    "package_url": self.manifest["package_url"].replace("/packages/nightly/", "/packages/validation/")}
+        run = {**self.run, "path": ".github/workflows/rustfs-validation-package.yml"}
+        payload = io.BytesIO()
+        with zipfile.ZipFile(payload, "w") as archive:
+            archive.writestr("validation-candidate-123-2.json", json.dumps(manifest))
+        raw = payload.getvalue()
+        artifact = {"id": 789, "name": "validation-candidate-123-2", "expired": False,
+                    "size_in_bytes": len(raw), "digest": "sha256:" + hashlib.sha256(raw).hexdigest(),
+                    "workflow_run": {"id": 123, "head_sha": "a" * 40}}
+        result = candidate.read_manifest(raw, artifact, run)
+        self.assertEqual(result["channel"], "validation")
+        self.assertIn("/packages/validation/runs/123/2/", result["package_url"])
+
+    def test_validation_resolver_requires_manual_workflow_dispatch(self):
+        payload, artifact = self.archive()
+        run = {"id": 123, "run_attempt": 2, "head_sha": "a" * 40,
+               "path": ".github/workflows/rustfs-validation-package.yml", "head_branch": "main",
+               "head_repository": {"full_name": "rustfs/rustfs"}, "event": "schedule",
+               "status": "completed", "conclusion": "success", "run_started_at": "2026-10-12T00:00:00Z"}
+        with mock.patch.object(candidate, "api", side_effect=[run]):
+            with self.assertRaisesRegex(ValueError, "manually dispatched"):
+                candidate.resolve(123, 2, allow_validation=True)
+
+    def test_validation_workflow_rejects_legacy_manifest_schema(self):
+        manifest = {**self.manifest}
+        payload_buffer = io.BytesIO()
+        with zipfile.ZipFile(payload_buffer, "w") as archive:
+            archive.writestr("validation-candidate-123-2.json", json.dumps(manifest))
+        payload = payload_buffer.getvalue()
+        artifact = {"id": 789, "name": "validation-candidate-123-2", "expired": False,
+                    "size_in_bytes": len(payload), "digest": "sha256:" + hashlib.sha256(payload).hexdigest(),
+                    "workflow_run": {"id": 123, "head_sha": "a" * 40}}
+        run = {**self.run, "path": ".github/workflows/rustfs-validation-package.yml"}
+        with self.assertRaisesRegex(ValueError, "schema-3"):
+            candidate.read_manifest(payload, artifact, run)
+
+    def test_private_candidate_resolution_is_explicit_and_attempt_bound(self):
+        self.manifest.update(schema=3, channel="validation", source_ref=self.manifest["source_sha"],
+                             package_url=self.manifest["package_url"].replace("/packages/nightly/", "/packages/validation/"))
+        payload, artifact = self.archive(["validation-candidate-123-2.json"])
+        artifact["name"] = "validation-candidate-123-2"
+        base = {**self.run, "head_branch": "cxymds/validation-runner", "head_repository": {"full_name": "rustfs/rustfs"},
+                "event": "workflow_dispatch", "status": "completed", "conclusion": "success",
+                "run_started_at": "2026-10-12T00:00:00Z"}
+        for path in candidate.VALIDATION_BUILD_WORKFLOWS:
+            run = {**base, "path": path}
+            with self.subTest(path=path), mock.patch.object(candidate, "api", side_effect=[run, {"artifacts": [artifact]}, payload]):
+                resolved = candidate.resolve(123, 2, allow_validation=True)
+                self.assertEqual(resolved["manifest"]["source_sha"], "b" * 40)
+                self.assertEqual(resolved["workflow_sha"], "a" * 40)
+            with mock.patch.object(candidate, "api", return_value=run), self.assertRaises(ValueError):
+                candidate.resolve(123, 2)
+        for field, value in (("id", 124), ("run_attempt", 1), ("event", "push"), ("status", "in_progress"),
+                             ("conclusion", "failure"), ("path", ".github/workflows/other.yml"),
+                             ("head_repository", {"full_name": "untrusted/rustfs"})):
+            run = {**base, "path": ".github/workflows/rustfs-functional-chain.yml", field: value}
+            with self.subTest(field=field), mock.patch.object(candidate, "api", return_value=run), self.assertRaises(ValueError):
+                candidate.resolve(123, 2, allow_validation=True)
+
+    def test_private_manifest_rejects_alias_channel_source_and_schema_substitution(self):
+        valid = {**self.manifest, "schema": 3, "channel": "validation", "source_ref": self.manifest["source_sha"],
+                 "package_url": self.manifest["package_url"].replace("/packages/nightly/", "/packages/validation/")}
+        for key, value in (("channel", "nightly"), ("source_ref", "main"), ("source_sha", "e" * 40),
+                           ("package_url", self.manifest["package_url"]), ("workflow_sha", "e" * 40),
+                           ("build_run_attempt", 3), ("schema", True)):
+            with self.subTest(key=key), self.assertRaises(ValueError):
+                candidate.validate_manifest({**valid, key: value}, self.run)
+
+    def test_package_only_mode_rejects_conflicting_fleet_and_tool_requests(self):
+        from check_test_wiring import yaml_block
+        workflow = Path(__file__).resolve().parents[1] / ".github/workflows/rustfs-functional-chain.yml"
+        job = yaml_block(workflow.read_text().splitlines(), "validation-package-request", 2)
+        begin = job.index("        run: |") + 1
+        script = "\n".join(line[10:] for line in job[begin:] if not line.strip() or line.startswith("          "))
+        import os
+        inputs = {"validation_package_sha": "a" * 40}
+        env = {**os.environ, "PACKAGE_INPUTS": json.dumps(inputs)}
+        result = subprocess.run(["bash", "-e", "-c", script], env=env, capture_output=True)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        for field in ("build_run_id", "build_run_attempt", "validation_case", "auto_testing_sha", "validation_tools_only",
+                      "validation_native_tools", "validation_shared_tools", "validation_manager_tools",
+                      "validation_cleanup_run_id", "validation_tools_run_id"):
+            with self.subTest(field=field):
+                env["PACKAGE_INPUTS"] = json.dumps({**inputs, field: "not-empty"})
+                result = subprocess.run(["bash", "-e", "-c", script], env=env, capture_output=True)
+                self.assertNotEqual(result.returncode, 0)
+        for value in ("main", "a" * 39, "a" * 40 + "\nFORGED=value"):
+            env["PACKAGE_INPUTS"] = json.dumps({"validation_package_sha": value})
+            self.assertNotEqual(subprocess.run(["bash", "-e", "-c", script], env=env, capture_output=True).returncode, 0)
+

 class EvidenceTests(unittest.TestCase):
    def setUp(self):
@@ -221,5 +221,82 @@ SH
        self.assertNotIn("          overwrite: true", upload)


+class ValidationPackageTests(unittest.TestCase):
+    def setUp(self):
+        # Exercise the real upload shell with the same conditional object-store
+        # fixture used by the nightly publication tests.
+        self.fixture = NightlyCandidateTests("test_success_binds_actual_package_checkout_and_attempt")
+        self.fixture.setUp()
+        self.addCleanup(self.fixture.temp.cleanup)
+        self.workflow = ROOT / ".github/workflows/rustfs-validation-package.yml"
+        job = yaml_block(self.workflow.read_text().splitlines(), "build", 2)
+        start = job.index("      - name: Upload validation package to immutable R2 path")
+        end = job.index("      - name: Upload validation candidate manifest")
+        upload = job[start:end]
+        begin = upload.index("        run: |") + 1
+        self.fixture.shell = "\n".join(line[10:] for line in upload[begin:] if not line.strip() or line.startswith("          "))
+        self.fixture.shims.write_text(self.fixture.shims.read_text().replace("/packages/nightly/runs/*", "/packages/validation/runs/*"))
+        self.fixture.env.update(PACKAGE_FILE=self.fixture.package.name, VALIDATION_SOURCE_SHA=self.fixture.sha,
+                                VALIDATION_SOURCE_REF=self.fixture.sha, GITHUB_SHA="f" * 40)
+
+    def test_success_uploads_only_one_immutable_object_and_exact_source(self):
+        result = self.fixture.run_publish()
+        self.assertEqual(result.returncode, 0, result.stderr)
+        manifest = self.fixture.manifest()
+        self.assertEqual(manifest["schema"], 3)
+        self.assertEqual(manifest["channel"], "validation")
+        self.assertEqual(manifest["source_sha"], self.fixture.sha)
+        self.assertEqual(manifest["source_ref"], self.fixture.sha)
+        self.assertEqual(manifest["workflow_sha"], "f" * 40)
+        validate_manifest(manifest, {"id": 12345, "run_attempt": 1, "head_sha": "f" * 40})
+        objects = [path for path in self.fixture.store.rglob("*") if path.is_file()]
+        self.assertEqual(len(objects), 1)
+        self.assertEqual(objects[0].read_bytes(), self.fixture.package.read_bytes())
+        self.assertEqual(str(objects[0].relative_to(self.fixture.store)),
+                         f"artifacts/rustfs/packages/validation/runs/12345/1/{self.fixture.digest}/rustfs.deb")
+        self.assertNotIn("s3 cp", (self.fixture.root / "aws.log").read_text())
+
+    def test_failure_or_missing_credentials_cannot_certify_a_candidate(self):
+        for index, mode in enumerate(("upload-failure", "missing-public-url", "wrong-public-bytes", "incomplete-download",
+                                      "bootstrap-failure", "install-failure", "broken-install")):
+            with self.subTest(mode=mode):
+                result = self.fixture.run_publish(FAKE_MODE=mode, GITHUB_RUN_ID=str(20000 + index))
+                self.assertNotEqual(result.returncode, 0)
+                self.assertFalse(self.fixture.output.exists())
+                self.assertEqual(list(self.fixture.root.glob("validation-candidate-*.json")), [])
+        for key in ("R2_ACCESS_KEY_ID", "R2_SECRET_ACCESS_KEY", "R2_ENDPOINT", "R2_BUCKET"):
+            with self.subTest(key=key):
+                self.assertNotEqual(self.fixture.run_publish(**{key: ""}).returncode, 0)
+                self.assertFalse(self.fixture.output.exists())
+
+    def test_stale_source_and_duplicate_objects_are_rejected(self):
+        result = self.fixture.run_publish(VALIDATION_SOURCE_SHA="e" * 40)
+        self.assertNotEqual(result.returncode, 0)
+        self.assertFalse(self.fixture.store.exists())
+        result = self.fixture.run_publish()
+        self.assertEqual(result.returncode, 0, result.stderr)
+        manifest = self.fixture.manifest()
+        key = manifest["package_url"].removeprefix("https://dl.rustfs.com/")
+        stored = self.fixture.store / key
+        stored.write_bytes(b"preexisting conflicting object")
+        (self.fixture.root / "validation-candidate-12345-1.json").unlink()
+        result = self.fixture.run_publish()
+        self.assertNotEqual(result.returncode, 0)
+        self.assertEqual(stored.read_bytes(), b"preexisting conflicting object")
+        self.assertFalse(self.fixture.output.exists())
+
+    def test_exact_source_and_workflow_identity_are_preserved_separately(self):
+        lines = self.workflow.read_text().splitlines()
+        build = "\n".join(yaml_block(lines, "build", 2))
+        source = "\n".join(yaml_block(lines, "resolve-source", 2))
+        self.assertIn('[[ "${REQUESTED_SHA}" =~ ^[0-9a-f]{40}$ ]]', source)
+        self.assertIn('test "$(git -C "${source_directory}" rev-parse FETCH_HEAD)" = "${REQUESTED_SHA}"', source)
+        self.assertIn("ref: ${{ needs.resolve-source.outputs.source_sha }}", build)
+        self.assertIn('--arg source_sha "${VALIDATION_SOURCE_SHA}"', build)
+        self.assertIn('--arg workflow_sha "${GITHUB_SHA}"', build)
+        self.assertNotIn("PF_TESTING_GH_TOKEN", build)
+        self.assertNotIn("publish_nightly_assets", build)
+
+
 if __name__ == "__main__":
    unittest.main()
Author	SHA1	Message	Date
cxymds	16cf19817e	ci(validation): dispatch isolated packages through the driver	2026-10-12 12:29:11 +08:00
cxymds	32fd1e704a	test(validation): verify isolated package publication failures	2026-10-12 12:22:53 +08:00
cxymds	ac197ea655	test(validation): enforce private candidate manifest schema	2026-10-12 12:18:15 +08:00
cxymds	e18ed73942	ci(validation): publish isolated immutable candidate packages	2026-10-12 12:06:59 +08:00
cxymds	3c131f2b2e	fix(heal): preserve typed errors and SSE-C response headers	2026-10-12 11:36:29 +08:00
cxymds	c5f4b5a4f7	fix(heal): preserve typed bucket errors for bounded retries	2026-10-12 10:45:27 +08:00
cxymds	03fb3d78a0	ci(validation): consume pinned manager native prerequisites	2026-10-11 11:55:36 +08:00
cxymds	b8126e50f7	test(ci): expose manager regressions through tools-only dispatch	2026-10-11 11:21:35 +08:00
cxymds	59fb72aa82	test(ci): add isolated manager native validation builds	2026-10-11 11:20:46 +08:00
cxymds	9de01c9170	docs(ci): describe supported validation recovery	2026-10-11 11:07:20 +08:00
cxymds	2ec26fa4b9	feat(ci): expose read-only validation fleet inventory	2026-10-11 10:42:48 +08:00
cxymds	f4507178a3	test(ci): bind C10 execution to the verified shared codec	2026-10-11 10:09:26 +08:00
cxymds	9b4cc40abd	test(validation): build optional C10 fixture codec on hosted Linux	2026-10-11 09:44:09 +08:00
cxymds	b14ff4eeb1	test(validation): bind C07 to pinned native authority evidence	2026-10-11 09:27:59 +08:00
cxymds	ae9b5cbabf	Run validation tool resolution on hosted Linux without acquiring the fleet lock	2026-10-11 07:49:26 +08:00
cxymds	6c78b0d996	test(validation): route native tools through the existing driver	2026-10-11 06:06:17 +08:00
cxymds	0066e34764	test(validation): add optional native Heal regression tooling	2026-10-11 06:03:43 +08:00
cxymds	fbddcc2965	ci: audit retained validation environments before finalization	2026-10-10 23:28:39 +08:00
cxymds	b1793b55a0	test(validation): verify Heal tools against released shard captures	2026-10-10 21:11:43 +08:00
cxymds	835bd3f52d	test(validation): verify pinned Heal tools before fleet execution	2026-10-10 20:46:40 +08:00
cxymds	cb32d55878	test(validation): expose tool artifacts through the existing manual lane	2026-10-10 20:35:14 +08:00
cxymds	864e58592c	test(validation): build the candidate-bound Heal inspector artifact	2026-10-10 20:32:34 +08:00
cxymds	3016209330	test(validation): provide the Python client for Heal cases	2026-10-10 20:30:10 +08:00
cxymds	64e0eeba91	Install validation client for replication cases	2026-10-10 18:52:42 +08:00
cxymds	35fa7488a5	Prepare Python client for pool validation cases	2026-10-10 18:41:48 +08:00
cxymds	1b77846f8e	Install pinned ILM client for canonical validation lane	2026-10-10 18:26:49 +08:00
cxymds	b0c861841e	test(validation): add pinned auto-testing manual lane	2026-10-10 17:34:37 +08:00
Chris	072b57b375	fix: auto-repair breaking changes from `446479791` (#8451 ) * fix: repair s3s footprint ratchet violation, clippy warnings, and stack overflow - Replace s3_error! macro with S3Error::with_message in ecfs.rs to keep s3s footprint ratchet at baseline (1586) after retention validation refactor - Remove redundant clone on Copy type in list_objects.rs (clippy::redundant_clone) - Remove clone_on_copy calls in access.rs test (clippy::clone_on_copy) - Use manual thread with 8MB stack for heal_concurrent_delete_test to fix stack overflow on macOS (consistent with sibling tests in same file) * test: align CI assertions with version authorization rules	2026-10-09 17:01:48 -07:00
GatewayJ	446479791e	refactor(iam): remove legacy OIDC federation paths (#8433 ) Co-authored-by: Hauser <housemecn@gmail.com>	2026-10-09 07:11:50 -07:00
Chris	99af59a914	Merge commit from fork	2026-10-09 21:43:53 +08:00
Chris	b77bd26874	Merge commit from fork	2026-10-09 21:43:08 +08:00
Chris	956c1cf027	Merge commit from fork	2026-10-09 21:40:59 +08:00
Chris	8b9d44110e	Merge commit from fork	2026-10-09 21:40:08 +08:00
Chris	44ce8d114f	Merge commit from fork	2026-10-09 21:39:18 +08:00
Chris	769b0d7578	Merge commit from fork	2026-10-09 21:28:19 +08:00
Chris	03e775947d	Merge commit from fork	2026-10-09 21:18:27 +08:00
Hauser	25e73d3bd6	fix(zip): statically link liblzma on macOS (#8434 ) * fix(zip): statically link liblzma on macOS Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(zip): pin macOS liblzma minimum to 0.4.8 Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(admin): route bucket recovery errors through facade Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(ecstore): group version pagination markers Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(heal): avoid cloning copy purge identity Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com>	2026-10-09 16:25:24 +08:00
Hauser	a464b45367	fix(rustfs): keep recovery handler within S3 gateway (#8429 ) * fix(rustfs): route recovery errors through S3 gateway Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(ecstore): satisfy strict Clippy pagination checks Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(heal): avoid cloning Copy purge identity Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * test: restore one-second drive request budgets * fix(ecstore): resume pending lock bucket creation Allow retries to reuse an uncommitted Object Lock bucket intent when the same incarnation remains unretired and all physical volumes are absent. Isolate the recovery entrypoint test from the shared production leader lock. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * test: provision stack for deep catalog test futures Apply the existing 4 MiB test-thread stack setup to table-catalog handler/store tests and the relocated-pool GET regression in both local and CI nextest profiles. These cases overflowed libtest's 2 MiB default stack on Linux. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: zhi22915 <qiuzgang@gmail.com> Co-authored-by: overtrue <anzhengchao@gmail.com>	2026-10-09 14:08:12 +08:00