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 14:27:10 +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
47 changed files with 1401 additions and 8849 deletions
@@ -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
@@ -275,15 +275,6 @@ dependencies = [
 "windows-sys 0.61.2",
 ]

-[[package]]
-name = "anyerror"
-version = "0.1.13"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "71add24cc141a1e8326f249b74c41cfd217aeb2a67c9c6cf9134d175469afd49"
-dependencies = [
- "serde",
-]
-
 [[package]]
 name = "anyhow"
 version = "1.0.104"
@@ -1715,30 +1706,6 @@ dependencies = [
 "syn 3.0.6",
 ]

-[[package]]
-name = "borsh"
-version = "1.8.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "553c5d846a6ba5150c65e3b1b8ec073bcf1abc20f9b7220de384a4443ea4e20a"
-dependencies = [
- "borsh-derive",
- "bytes",
- "cfg_aliases",
-]
-
-[[package]]
-name = "borsh-derive"
-version = "1.8.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "12cdfe656708a01f89b451a7d36466e6fe6c414de0aa18fc54f864f6f9ca9f56"
-dependencies = [
- "once_cell",
- "proc-macro-crate",
- "proc-macro2",
- "quote",
- "syn 3.0.6",
-]
-
 [[package]]
 name = "brotli"
 version = "8.0.4"
@@ -1805,18 +1772,6 @@ version = "3.20.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649"

-[[package]]
-name = "byte-unit"
-version = "5.2.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c719d56f7e96194cfc53460976d3ba51c85719747c9c62ed99981847b551152b"
-dependencies = [
- "rust_decimal",
- "schemars 1.2.2",
- "serde",
- "utf8-width",
-]
-
 [[package]]
 name = "bytemuck"
 version = "1.25.2"
@@ -3769,34 +3724,13 @@ dependencies = [
 "syn 2.0.119",
 ]

-[[package]]
-name = "derive_more"
-version = "1.0.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4a9b99b9cbbe49445b21764dc0625032a89b145a2642e67603e1c936f5458d05"
-dependencies = [
- "derive_more-impl 1.0.0",
-]
-
 [[package]]
 name = "derive_more"
 version = "2.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d751e9e49156b02b44f9c1815bcb94b984cdcc4396ecc32521c739452808b134"
 dependencies = [
- "derive_more-impl 2.1.1",
-]
-
-[[package]]
-name = "derive_more-impl"
-version = "1.0.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cb7330aeadfbe296029522e6c40f315320aba36fc43a5b3632f3795348f3bd22"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.119",
- "unicode-xid",
+ "derive_more-impl",
 ]

 [[package]]
@@ -6300,7 +6234,7 @@ dependencies = [
 "bytes",
 "chrono",
 "dashmap",
- "derive_more 2.1.1",
+ "derive_more",
 "futures-util",
 "getrandom 0.4.3",
 "lazy_static",
@@ -6454,12 +6388,6 @@ dependencies = [
 "quote",
 ]

-[[package]]
-name = "maplit"
-version = "1.0.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3e2e65a1a2e43cfcb47a895c4c8b10d1f4a61097f9f254f183aee60cad9c651d"
-
 [[package]]
 name = "matchers"
 version = "0.2.0"
@@ -7401,42 +7329,6 @@ dependencies = [
 "url",
 ]

-[[package]]
-name = "openraft"
-version = "0.9.25"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a97014fb78acb77be3a40ac2da305f6dd3a6b243f3a908ace87d29b3972eaafd"
-dependencies = [
- "anyerror",
- "byte-unit",
- "chrono",
- "clap",
- "derive_more 1.0.0",
- "futures",
- "maplit",
- "openraft-macros",
- "rand 0.8.8",
- "serde",
- "thiserror 1.0.69",
- "tokio",
- "tracing",
- "tracing-futures",
- "validit",
-]
-
-[[package]]
-name = "openraft-macros"
-version = "0.9.25"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "55417d79f46a787dd88c043d831d8b7a0713f7c69d1925f9de51cc42c437250f"
-dependencies = [
- "chrono",
- "proc-macro2",
- "quote",
- "semver",
- "syn 2.0.119",
-]
-
 [[package]]
 name = "openssl-probe"
 version = "0.2.1"
@@ -9066,15 +8958,6 @@ dependencies = [
 "syn 2.0.119",
 ]

-[[package]]
-name = "redb"
-version = "4.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fb338a6c67830a61bed824b78c2bb034ab1ff401a20616bd7957524f4fdc2f22"
-dependencies = [
- "libc",
-]
-
 [[package]]
 name = "redis"
 version = "1.7.1"
@@ -9581,23 +9464,6 @@ dependencies = [
 "walkdir",
 ]

-[[package]]
-name = "rust_decimal"
-version = "1.43.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7653272e75dcac41dc199fbea6f5797633994fafd339943c06c9af16bf29cd3a"
-dependencies = [
- "arrayvec",
- "borsh",
- "bytes",
- "num-traits",
- "rand 0.8.8",
- "rand 0.9.5",
- "serde",
- "serde_json",
- "wasm-bindgen",
-]
-
 [[package]]
 name = "rustc-demangle"
 version = "0.1.28"
@@ -9800,21 +9666,6 @@ dependencies = [
 "url",
 ]

-[[package]]
-name = "rustfs-capture-authority"
-version = "1.0.1"
-dependencies = [
- "openraft",
- "redb",
- "serde",
- "serde_json",
- "sha2 0.11.0",
- "tempfile",
- "thiserror 2.0.21",
- "tokio",
- "url",
-]
-
 [[package]]
 name = "rustfs-checksums"
 version = "1.0.1"
@@ -10622,7 +10473,6 @@ dependencies = [
 "tokio",
 "tracing",
 "tracing-subscriber",
- "uuid",
 ]

 [[package]]
@@ -13196,16 +13046,6 @@ dependencies = [
 "tracing-subscriber",
 ]

-[[package]]
-name = "tracing-futures"
-version = "0.2.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "97d095ae15e245a057c8e8451bab9b3ee1e1f68e9ba2b4fbc18d0ac5237835f2"
-dependencies = [
- "pin-project",
- "tracing",
-]
-
 [[package]]
 name = "tracing-log"
 version = "0.2.0"
@@ -13356,7 +13196,7 @@ checksum = "79acf4b257aa56dd7f00466a81693d6c79c362c55092b3137909d6e76249ad80"
 dependencies = [
 "async-trait",
 "chrono",
- "derive_more 2.1.1",
+ "derive_more",
 "libc",
 "md-5 0.10.6",
 "thiserror 2.0.21",
@@ -13456,12 +13296,6 @@ version = "2.1.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "daf8dba3b7eb870caf1ddeed7bc9d2a049f3cfdfae7cb521b087cc33ae4c49da"

-[[package]]
-name = "utf8-width"
-version = "0.1.9"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "159a7cadce548703edd50d24069bc294c5415ecab0a480e0cd1ca06d112dc94a"
-
 [[package]]
 name = "utf8_iter"
 version = "1.0.4"
@@ -13488,15 +13322,6 @@ dependencies = [
 "wasm-bindgen",
 ]

-[[package]]
-name = "validit"
-version = "0.2.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4417835826b78aa6265724624d424df2b1e6e71a33329227d8b04a88950780dc"
-dependencies = [
- "anyerror",
-]
-
 [[package]]
 name = "valuable"
 version = "0.1.1"
@@ -18,7 +18,6 @@ members = [
    "crates/audit", # Audit target management system with multi-target fan-out
    "crates/checksums", # client checksums
    "crates/common", # Shared utilities and data structures
-    "crates/capture-authority", # Durable capture authority storage
    "crates/data-usage", # Shared data usage models and algorithms
    "crates/config", # Configuration management
    "crates/connect-symbol-catalog", # Release-only CPU profile symbol catalog generator
@@ -315,8 +314,6 @@ ratelimit = "2.0.0"
 rayon = "1.12.0"
 rustfs-erasure-codec = { version = "9.0.0" }
 reed-solomon-simd = "3.1.0"
-openraft = { version = "=0.9.25", default-features = false, features = ["serde"] }
-redb = "=4.3.0"
 regex = { version = "1.13.1" }
 rumqttc = { package = "rumqttc-next", version = "0.34.0" }
 redis = { version = "1.7.1" }
@@ -1,24 +0,0 @@
-[package]
-name = "rustfs-capture-authority"
-version.workspace = true
-edition.workspace = true
-license.workspace = true
-rust-version.workspace = true
-repository.workspace = true
-description = "Durable local storage for capture authority"
-
-[dependencies]
-openraft = { workspace = true }
-redb = { workspace = true }
-sha2 = { workspace = true }
-serde = { workspace = true, features = ["derive"] }
-thiserror = { workspace = true }
-tokio = { workspace = true }
-url = { workspace = true }
-
-[dev-dependencies]
-serde_json = { workspace = true }
-tempfile = { workspace = true }
-
-[lints]
-workspace = true
@@ -1,862 +0,0 @@
-fn read_native_range_in<R: std::ops::RangeBounds<u64>>(
-    tx: &redb::ReadTransaction,
-    range: R,
-) -> Result<Vec<crate::CaptureRaftEntry>, StoreError> {
-    use std::ops::Bound::{Excluded, Included, Unbounded};
-    let logs = tx.open_table(LOGS)?;
-    let start = match range.start_bound() {
-        Included(value) => *value,
-        Excluded(value) => match value.checked_add(1) {
-            Some(value) => value,
-            None => return Ok(Vec::new()),
-        },
-        Unbounded => 0,
-    };
-    let end = match range.end_bound() {
-        Included(value) => *value,
-        Excluded(value) => match value.checked_sub(1) {
-            Some(value) => value,
-            None => return Ok(Vec::new()),
-        },
-        Unbounded => u64::MAX,
-    };
-    if start > end {
-        return Ok(Vec::new());
-    }
-    let mut entries = Vec::new();
-    for row in logs.range(start..=end)? {
-        let (key, value) = row?;
-        let entry = decode_log(value.value())?;
-        if entry.id.index != key.value() {
-            return Err(StoreError::CorruptRecord);
-        }
-        entries.push(decode_native_entry(entry)?);
-    }
-    Ok(entries)
-}
-
-use crate::{
-    LOGS, LogId, META, Store, StoreError, committed::validate_committed, created::*, decode_log, encode_id, encode_log,
-    next_index, number,
-};
-use redb::{ReadTransaction, ReadableDatabase, ReadableTable, ReadableTableMetadata, TableDefinition};
-use std::collections::BTreeMap;
-
-pub(crate) const HEADS: TableDefinition<&[u8], &[u8]> = TableDefinition::new("heads");
-pub(crate) const VERSIONS: TableDefinition<&[u8], &[u8]> = TableDefinition::new("versions");
-pub(crate) const EVENTS: TableDefinition<&[u8], &[u8]> = TableDefinition::new("events");
-pub(crate) const OPERATIONS: TableDefinition<&[u8], &[u8]> = TableDefinition::new("operations");
-pub(crate) const APPLY_RESULTS: TableDefinition<u64, &[u8]> = TableDefinition::new("apply_results");
-
-fn version_key(object: &ObjectIdentity, version: [u8; 16]) -> Result<Vec<u8>, StoreError> {
-    let mut b = object_bytes(object)?;
-    b.extend(version);
-    Ok(b)
-}
-fn decode_version(b: &[u8]) -> Result<CreatedVersion, StoreError> {
-    let mut d = Decoder::new(b);
-    let v = CreatedVersion {
-        prepared: d.prepared()?,
-        decision_log_id: d.id()?,
-        binding_revision: d.u64()?,
-    };
-    d.finish()?;
-    Ok(v)
-}
-fn decode_event(b: &[u8]) -> Result<CreatedEvent, StoreError> {
-    let mut d = Decoder::new(b);
-    let e = CreatedEvent {
-        object: d.object()?,
-        prepared: d.prepared()?,
-        decision_log_id: d.id()?,
-        binding: d.binding()?,
-    };
-    d.finish()?;
-    Ok(e)
-}
-fn decode_operation(b: &[u8]) -> Result<CreatedOperation, StoreError> {
-    let mut d = Decoder::new(b);
-    let len = usize::try_from(d.u64()?).map_err(|_| StoreError::CorruptRecord)?;
-    let command = b
-        .get(8..8usize.checked_add(len).ok_or(StoreError::CorruptRecord)?)
-        .ok_or(StoreError::CorruptRecord)?;
-    DecideCreated::decode(command)?;
-    let mut d = Decoder::new(&b[8 + len..]);
-    let result = d.result()?;
-    d.finish()?;
-    Ok(CreatedOperation {
-        canonical_command: command.to_vec(),
-        result,
-    })
-}
-fn decode_apply(b: &[u8]) -> Result<CreatedApplyResult, StoreError> {
-    let mut d = Decoder::new(b);
-    let r = CreatedApplyResult {
-        log_id: d.id()?,
-        result: d.result()?,
-    };
-    d.finish()?;
-    Ok(r)
-}
-fn read_binding(meta: &impl ReadableTable<&'static str, &'static [u8]>) -> Result<CaptureBinding, StoreError> {
-    let b = meta.get("capture_binding")?.ok_or(StoreError::BindingNotPinned)?;
-    let mut d = Decoder::new(b.value());
-    let binding = d.binding()?;
-    d.finish()?;
-    Ok(binding)
-}
-fn read_applied(meta: &impl ReadableTable<&'static str, &'static [u8]>) -> Result<Option<LogId>, StoreError> {
-    let b = meta.get("last_applied")?.ok_or(StoreError::CorruptRecord)?;
-    let mut d = Decoder::new(b.value());
-    let id = match d.take::<1>()?[0] {
-        0 => None,
-        1 => Some(d.id()?),
-        _ => return Err(StoreError::CorruptRecord),
-    };
-    d.finish()?;
-    Ok(id)
-}
-fn required<T>(value: Option<T>) -> Result<T, StoreError> {
-    value.ok_or(StoreError::CorruptRecord)
-}
-fn encode_native_entry(entry: &crate::CaptureRaftEntry) -> Result<Vec<u8>, StoreError> {
-    let id: LogId = entry.log_id.into();
-    let (kind, body) = match &entry.payload {
-        openraft::EntryPayload::Blank => (0, vec![]),
-        openraft::EntryPayload::Normal(command) => (1, command.created().encode()?),
-        openraft::EntryPayload::Membership(membership) => {
-            let stored = openraft::StoredMembership::new(Some(entry.log_id), membership.clone());
-            let checked = crate::CaptureMembership::try_from(&stored)?;
-            (2, checked.encode()?)
-        }
-    };
-    let mut payload = b"CENR\x01".to_vec();
-    payload.push(kind);
-    payload.extend(
-        u64::try_from(body.len())
-            .map_err(|_| StoreError::CorruptRecord)?
-            .to_be_bytes(),
-    );
-    payload.extend(body);
-    encode_log(&crate::LogEntry { id, payload })
-}
-
-pub(crate) fn decode_native_entry(entry: crate::LogEntry) -> Result<crate::CaptureRaftEntry, StoreError> {
-    let bytes = &entry.payload;
-    if bytes.len() < 14 || &bytes[..5] != b"CENR\x01" {
-        return Err(StoreError::CorruptRecord);
-    }
-    let length = usize::try_from(number(&bytes[6..14])?).map_err(|_| StoreError::CorruptRecord)?;
-    let body = &bytes[14..];
-    if length != body.len() {
-        return Err(StoreError::CorruptRecord);
-    }
-    let payload = match bytes[5] {
-        0 if body.is_empty() => openraft::EntryPayload::Blank,
-        1 => openraft::EntryPayload::Normal(crate::CaptureCommand::try_created(DecideCreated::decode(body)?)?),
-        2 => {
-            let membership = crate::CaptureMembership::decode(body).map_err(|_| StoreError::CorruptRecord)?;
-            if *membership.log_id() != Some(entry.id) {
-                return Err(StoreError::CorruptRecord);
-            }
-            let stored = openraft::StoredMembership::try_from(&membership).map_err(|_| StoreError::CorruptRecord)?;
-            openraft::EntryPayload::Membership(stored.membership().clone())
-        }
-        _ => return Err(StoreError::CorruptRecord),
-    };
-    Ok(crate::CaptureRaftEntry {
-        log_id: entry.id.into(),
-        payload,
-    })
-}
-
-fn apply_created(
-    tx: &redb::WriteTransaction,
-    id: LogId,
-    command: &DecideCreated,
-    canonical: &[u8],
-    binding: &CaptureBinding,
-) -> Result<CreatedResult, StoreError> {
-    let mut operations = tx.open_table(OPERATIONS)?;
-    let existing = operations
-        .get(command.operation_id.as_slice())?
-        .map(|v| decode_operation(v.value()))
-        .transpose()?;
-    let result = if let Some(existing) = existing {
-        if existing.canonical_command == canonical {
-            existing.result
-        } else {
-            CreatedResult::OperationConflict
-        }
-    } else {
-        let mut heads = tx.open_table(HEADS)?;
-        let mut versions = tx.open_table(VERSIONS)?;
-        let object = object_bytes(&command.object)?;
-        let key = version_key(&command.object, command.prepared.version_id)?;
-        let actual = heads
-            .get(object.as_slice())?
-            .map(|v| {
-                let mut d = Decoder::new(v.value());
-                let id = d.uuid()?;
-                d.finish()?;
-                Ok::<_, StoreError>(id)
-            })
-            .transpose()?;
-        let outcome = if command.binding_revision != binding.revision {
-            CreatedResult::BindingMismatch
-        } else if actual != command.expected_head {
-            CreatedResult::HeadMismatch { actual_head: actual }
-        } else if versions.get(key.as_slice())?.is_some() {
-            CreatedResult::VersionConflict
-        } else {
-            let mut version = Vec::new();
-            put_prepared(&mut version, &command.prepared);
-            version.extend(encode_id(id));
-            version.extend(binding.revision.to_be_bytes());
-            versions.insert(key.as_slice(), version.as_slice())?;
-            let mut event = object;
-            put_prepared(&mut event, &command.prepared);
-            event.extend(encode_id(id));
-            event.extend(binding_bytes(&binding)?);
-            tx.open_table(EVENTS)?
-                .insert(command.operation_id.as_slice(), event.as_slice())?;
-            heads.insert(object_bytes(&command.object)?.as_slice(), command.prepared.version_id.as_slice())?;
-            CreatedResult::Created {
-                version_id: command.prepared.version_id,
-                event_id: command.operation_id,
-                decision_log_id: id,
-            }
-        };
-        let mut op = u64::try_from(canonical.len())
-            .map_err(|_| StoreError::CorruptRecord)?
-            .to_be_bytes()
-            .to_vec();
-        op.extend(canonical);
-        op.extend(result_bytes(&outcome));
-        operations.insert(command.operation_id.as_slice(), op.as_slice())?;
-        outcome
-    };
-    Ok(result)
-}
-
-fn validate_response_kind(entry: &crate::CaptureRaftEntry, response: &crate::CaptureResponse) -> Result<(), StoreError> {
-    let kind = match entry.payload {
-        openraft::EntryPayload::Blank => crate::CaptureResponseKind::Blank,
-        openraft::EntryPayload::Normal(_) => crate::CaptureResponseKind::Created,
-        openraft::EntryPayload::Membership(_) => crate::CaptureResponseKind::Membership,
-    };
-    if response.kind() != kind || *response.log_id() != LogId::from(entry.log_id) {
-        return Err(StoreError::CorruptRecord);
-    }
-    Ok(())
-}
-fn encode_response(response: &crate::CaptureResponse) -> Result<Vec<u8>, StoreError> {
-    let (kind, body) = match response.kind() {
-        crate::CaptureResponseKind::Blank => (0, vec![]),
-        crate::CaptureResponseKind::Created => (1, result_bytes(response.created_result().ok_or(StoreError::CorruptRecord)?)),
-        crate::CaptureResponseKind::Membership => (2, vec![]),
-    };
-    let mut bytes = b"CRSP\x01".to_vec();
-    bytes.push(kind);
-    bytes.extend(encode_id(*response.log_id()));
-    bytes.extend(
-        u64::try_from(body.len())
-            .map_err(|_| StoreError::CorruptRecord)?
-            .to_be_bytes(),
-    );
-    bytes.extend(body);
-    Ok(bytes)
-}
-fn decode_response(bytes: &[u8]) -> Result<crate::CaptureResponse, StoreError> {
-    if bytes.len() < 38
-        || &bytes[..5] != b"CRSP\x01"
-        || number(&bytes[30..38])? != u64::try_from(bytes.len() - 38).map_err(|_| StoreError::CorruptRecord)?
-    {
-        return Err(StoreError::CorruptRecord);
-    }
-    let id = crate::decode_id(&bytes[6..30])?;
-    match bytes[5] {
-        0 if bytes.len() == 38 => crate::CaptureResponse::blank(id),
-        1 => {
-            let mut body = encode_id(id);
-            body.extend(&bytes[38..]);
-            crate::CaptureResponse::created(decode_apply(&body)?)
-        }
-        2 if bytes.len() == 38 => crate::CaptureResponse::membership(id),
-        _ => Err(StoreError::CorruptRecord),
-    }
-}
-fn entry_command(entry: &crate::LogEntry, native: bool) -> Result<DecideCreated, StoreError> {
-    if native {
-        match decode_native_entry(entry.clone())?.payload {
-            openraft::EntryPayload::Normal(command) => Ok(command.created().clone()),
-            _ => Err(StoreError::CorruptRecord),
-        }
-    } else {
-        DecideCreated::decode(&entry.payload)
-    }
-}
-
-pub(crate) fn native_application_state(tx: &ReadTransaction) -> Result<(Option<LogId>, crate::CaptureMembership), StoreError> {
-    let meta = tx.open_table(META)?;
-    if number(required(meta.get("schema")?)?.value())? != 4 {
-        return Err(StoreError::UnsupportedSchema);
-    }
-    let last = read_applied(&meta)?;
-    let membership = crate::CaptureMembership::decode(required(meta.get("native_membership")?)?.value())
-        .map_err(|_| StoreError::CorruptRecord)?;
-    if membership
-        .log_id()
-        .is_some_and(|id| last.is_none_or(|last| id.index > last.index))
-    {
-        return Err(StoreError::CorruptRecord);
-    }
-    Ok((last, membership))
-}
-
-impl Store {
-    /// Explicitly opt an empty, bound store into native entry persistence.
-    pub fn enable_native_empty(&mut self) -> Result<(), StoreError> {
-        let tx = self.write_transaction()?;
-        {
-            let mut meta = tx.open_table(META)?;
-            let schema = number(required(meta.get("schema")?)?.value())?;
-            if !matches!(schema, 3 | 4) {
-                return Err(StoreError::UnsupportedSchema);
-            }
-            read_binding(&meta)?;
-            if schema == 4 {
-                let membership = crate::CaptureMembership::decode(required(meta.get("native_membership")?)?.value())
-                    .map_err(|_| StoreError::CorruptRecord)?;
-                let last = read_applied(&meta)?;
-                if membership
-                    .log_id()
-                    .is_some_and(|id| last.is_none_or(|last| id.index > last.index))
-                    || (last.is_none() && (!membership.configs().is_empty() || !membership.nodes().is_empty()))
-                {
-                    return Err(StoreError::CorruptRecord);
-                }
-                return Ok(());
-            }
-            if read_applied(&meta)?.is_some() || !tx.open_table(LOGS)?.is_empty()? {
-                return Err(StoreError::UnsupportedSchema);
-            }
-            for table in [HEADS, VERSIONS, EVENTS, OPERATIONS] {
-                if !tx.open_table(table)?.is_empty()? {
-                    return Err(StoreError::UnsupportedSchema);
-                }
-            }
-            if !tx.open_table(APPLY_RESULTS)?.is_empty()? || meta.get("committed")?.is_some() || meta.get("tail")?.is_some() {
-                return Err(StoreError::UnsupportedSchema);
-            }
-            if meta.get("native_membership")?.is_some() {
-                return Err(StoreError::CorruptRecord);
-            }
-            let membership = crate::CaptureMembership::new(None, vec![], BTreeMap::new())?;
-            meta.insert("native_membership", membership.encode()?.as_slice())?;
-            meta.insert("schema", 4u64.to_be_bytes().as_slice())?;
-        }
-        self.commit(tx)
-    }
-    pub fn append_native(&mut self, entries: &[crate::CaptureRaftEntry]) -> Result<(), StoreError> {
-        let tx = self.write_transaction()?;
-        {
-            let mut meta = tx.open_table(META)?;
-            if number(required(meta.get("schema")?)?.value())? != 4 {
-                return Err(StoreError::UnsupportedSchema);
-            }
-            if entries.is_empty() {
-                return Ok(());
-            }
-            let tail = meta.get("tail")?.map(|value| crate::decode_id(value.value())).transpose()?;
-            let mut next = next_index(tail.map(|id| id.index))?;
-            let mut encoded = Vec::with_capacity(entries.len());
-            for (position, entry) in entries.iter().enumerate() {
-                if entry.log_id.index != next {
-                    return Err(StoreError::InvalidLogSequence);
-                }
-                encoded.push(encode_native_entry(entry)?);
-                if position + 1 < entries.len() {
-                    next = next_index(Some(next))?;
-                }
-            }
-            let mut logs = tx.open_table(LOGS)?;
-            for (entry, bytes) in entries.iter().zip(&encoded) {
-                logs.insert(entry.log_id.index, bytes.as_slice())?;
-            }
-            let last = entries.last().ok_or(StoreError::InvalidLogSequence)?;
-            meta.insert("tail", encode_id(last.log_id.into()).as_slice())?;
-        }
-        self.commit(tx)
-    }
-    pub fn read_native_entry(&self, index: u64) -> Result<Option<crate::CaptureRaftEntry>, StoreError> {
-        let tx = self.database.begin_read()?;
-        if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
-            return Err(StoreError::UnsupportedSchema);
-        }
-        tx.open_table(LOGS)?
-            .get(index)?
-            .map(|value| {
-                let entry = decode_log(value.value())?;
-                if entry.id.index != index {
-                    return Err(StoreError::CorruptRecord);
-                }
-                decode_native_entry(entry)
-            })
-            .transpose()
-    }
-    /// Read typed native entries within the requested bounds.
-    pub fn read_native_range<R: std::ops::RangeBounds<u64>>(&self, range: R) -> Result<Vec<crate::CaptureRaftEntry>, StoreError> {
-        let tx = self.database.begin_read()?;
-        if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
-            return Err(StoreError::UnsupportedSchema);
-        }
-        read_native_range_in(&tx, range)
-    }
-    pub fn read_native_entries(&self, start: u64, end: u64) -> Result<Vec<crate::CaptureRaftEntry>, StoreError> {
-        let tx = self.database.begin_read()?;
-        if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
-            return Err(StoreError::UnsupportedSchema);
-        }
-        if start > end {
-            return Err(StoreError::InvalidLogSequence);
-        }
-        read_native_range_in(&tx, start..end)
-    }
-
-    pub fn apply_next_native(&mut self, id: LogId) -> Result<crate::CaptureResponse, StoreError> {
-        let tx = self.write_transaction()?;
-        let response;
-        {
-            let mut meta = tx.open_table(META)?;
-            if number(required(meta.get("schema")?)?.value())? != 4 {
-                return Err(StoreError::UnsupportedSchema);
-            }
-            let binding = read_binding(&meta)?;
-            let logs = tx.open_table(LOGS)?;
-            let committed = validate_committed(&meta, &logs)?.ok_or(StoreError::InvalidCommittedLog)?;
-            let entry = decode_log(required(logs.get(id.index)?)?.value())?;
-            if entry.id != id || id.index > committed.index {
-                return Err(StoreError::InvalidCommittedLog);
-            }
-            let native = decode_native_entry(entry)?;
-            let last = read_applied(&meta)?;
-            let mut results = tx.open_table(APPLY_RESULTS)?;
-            if last.is_some_and(|last| id.index <= last.index) {
-                let saved = decode_response(required(results.get(id.index)?)?.value())?;
-                if *saved.log_id() != id {
-                    return Err(StoreError::InvalidApplySequence);
-                }
-                validate_response_kind(&native, &saved)?;
-                return Ok(saved);
-            }
-            if id.index != next_index(last.map(|last| last.index))? {
-                return Err(StoreError::InvalidApplySequence);
-            }
-            response = match native.payload {
-                openraft::EntryPayload::Blank => crate::CaptureResponse::blank(id)?,
-                openraft::EntryPayload::Normal(command) => {
-                    let canonical = command.created().encode()?;
-                    let result = apply_created(&tx, id, command.created(), &canonical, &binding)?;
-                    crate::CaptureResponse::created(CreatedApplyResult { log_id: id, result })?
-                }
-                openraft::EntryPayload::Membership(membership) => {
-                    let stored = openraft::StoredMembership::new(Some(id.into()), membership);
-                    let checked = crate::CaptureMembership::try_from(&stored)?;
-                    meta.insert("native_membership", checked.encode()?.as_slice())?;
-                    crate::CaptureResponse::membership(id)?
-                }
-            };
-            results.insert(id.index, encode_response(&response)?.as_slice())?;
-            let mut last = vec![1];
-            last.extend(encode_id(id));
-            meta.insert("last_applied", last.as_slice())?;
-        }
-        self.commit(tx)?;
-        Ok(response)
-    }
-    pub fn native_apply_response(&self, index: u64) -> Result<Option<crate::CaptureResponse>, StoreError> {
-        let tx = self.database.begin_read()?;
-        if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
-            return Err(StoreError::UnsupportedSchema);
-        }
-        let results = tx.open_table(APPLY_RESULTS)?;
-        results.get(index)?.map(|v| decode_response(v.value())).transpose()
-    }
-    pub fn native_application_state(&self) -> Result<(Option<LogId>, crate::CaptureMembership), StoreError> {
-        native_application_state(&self.database.begin_read()?)
-    }
-    /// Bootstrap the immutable capture policy. The caller must be the trusted
-    /// deployment provisioning owner. This crate does not authenticate callers;
-    /// do not expose this public method through an ordinary object request API.
-    /// Prepared identities assert identity only, not physical data durability.
-    pub fn pin_capture_binding(&mut self, binding: CaptureBinding) -> Result<(), StoreError> {
-        let encoded = binding_bytes(&binding)?;
-        let tx = self.write_transaction()?;
-        {
-            let mut meta = tx.open_table(META)?;
-            if matches!(number(required(meta.get("schema")?)?.value())?, 3 | 4) {
-                if read_binding(&meta)? == binding {
-                    return Ok(());
-                }
-                return Err(StoreError::BindingConflict);
-            }
-            if meta.get("capture_binding")?.is_some() || meta.get("last_applied")?.is_some() {
-                return Err(StoreError::CorruptRecord);
-            }
-            for table in [HEADS, VERSIONS, EVENTS, OPERATIONS] {
-                if !tx.open_table(table)?.is_empty()? {
-                    return Err(StoreError::CorruptRecord);
-                }
-            }
-            if !tx.open_table(APPLY_RESULTS)?.is_empty()? {
-                return Err(StoreError::CorruptRecord);
-            }
-            meta.insert("capture_binding", encoded.as_slice())?;
-            meta.insert("last_applied", &[0][..])?;
-            meta.insert("schema", 3u64.to_be_bytes().as_slice())?;
-        }
-        self.commit(tx)
-    }
-    /// Apply the next exact persisted committed Created command atomically.
-    pub fn apply_next_created(&mut self, id: LogId) -> Result<CreatedResult, StoreError> {
-        let tx = self.write_transaction()?;
-        let result;
-        {
-            let mut meta = tx.open_table(META)?;
-            if number(required(meta.get("schema")?)?.value())? != 3 {
-                return Err(StoreError::UnsupportedSchema);
-            }
-            let binding = read_binding(&meta)?;
-            let logs = tx.open_table(LOGS)?;
-            let committed = validate_committed(&meta, &logs)?.ok_or(StoreError::InvalidCommittedLog)?;
-            let entry = decode_log(required(logs.get(id.index)?)?.value())?;
-            if entry.id != id || id.index > committed.index {
-                return Err(StoreError::InvalidCommittedLog);
-            }
-            let last = read_applied(&meta)?;
-            let mut results = tx.open_table(APPLY_RESULTS)?;
-            if last.is_some_and(|last| id.index <= last.index) {
-                let saved = decode_apply(required(results.get(id.index)?)?.value())?;
-                if saved.log_id != id {
-                    return Err(StoreError::InvalidApplySequence);
-                }
-                return Ok(saved.result);
-            }
-            if id.index != next_index(last.map(|last| last.index))? {
-                return Err(StoreError::InvalidApplySequence);
-            }
-            let command = DecideCreated::decode(&entry.payload)?;
-            result = apply_created(&tx, id, &command, &entry.payload, &binding)?;
-            let mut applied = encode_id(id);
-            applied.extend(result_bytes(&result));
-            results.insert(id.index, applied.as_slice())?;
-            let mut last = vec![1];
-            last.extend(encode_id(id));
-            meta.insert("last_applied", last.as_slice())?;
-        }
-        self.commit(tx)?;
-        Ok(result)
-    }
-    pub fn capture_binding(&self) -> Result<Option<CaptureBinding>, StoreError> {
-        let tx = self.database.begin_read()?;
-        let meta = tx.open_table(META)?;
-        if !matches!(number(required(meta.get("schema")?)?.value())?, 3 | 4) {
-            return Ok(None);
-        }
-        Ok(Some(read_binding(&meta)?))
-    }
-    pub fn last_applied(&self) -> Result<Option<LogId>, StoreError> {
-        let tx = self.database.begin_read()?;
-        let meta = tx.open_table(META)?;
-        if !matches!(number(required(meta.get("schema")?)?.value())?, 3 | 4) {
-            return Ok(None);
-        }
-        read_applied(&meta)
-    }
-    pub fn created_head(&self, object: &ObjectIdentity) -> Result<Option<[u8; 16]>, StoreError> {
-        let tx = self.database.begin_read()?;
-        let t = tx.open_table(HEADS)?;
-        t.get(object_bytes(object)?.as_slice())?
-            .map(|v| {
-                let mut d = Decoder::new(v.value());
-                let id = d.uuid()?;
-                d.finish()?;
-                Ok(id)
-            })
-            .transpose()
-    }
-    pub fn created_version(&self, object: &ObjectIdentity, version: [u8; 16]) -> Result<Option<CreatedVersion>, StoreError> {
-        let tx = self.database.begin_read()?;
-        let t = tx.open_table(VERSIONS)?;
-        t.get(version_key(object, version)?.as_slice())?
-            .map(|v| decode_version(v.value()))
-            .transpose()
-    }
-    pub fn created_event(&self, operation: [u8; 16]) -> Result<Option<CreatedEvent>, StoreError> {
-        let tx = self.database.begin_read()?;
-        let t = tx.open_table(EVENTS)?;
-        t.get(operation.as_slice())?.map(|v| decode_event(v.value())).transpose()
-    }
-    pub fn created_operation(&self, operation: [u8; 16]) -> Result<Option<CreatedOperation>, StoreError> {
-        let tx = self.database.begin_read()?;
-        let t = tx.open_table(OPERATIONS)?;
-        t.get(operation.as_slice())?.map(|v| decode_operation(v.value())).transpose()
-    }
-    pub fn created_apply_result(&self, index: u64) -> Result<Option<CreatedApplyResult>, StoreError> {
-        let tx = self.database.begin_read()?;
-        let t = tx.open_table(APPLY_RESULTS)?;
-        if number(required(tx.open_table(META)?.get("schema")?)?.value())? == 4 {
-            t.get(index)?
-                .map(|v| {
-                    decode_response(v.value()).map(|r| {
-                        r.created_result().cloned().map(|result| CreatedApplyResult {
-                            log_id: *r.log_id(),
-                            result,
-                        })
-                    })
-                })
-                .transpose()
-                .map(Option::flatten)
-        } else {
-            t.get(index)?.map(|v| decode_apply(v.value())).transpose()
-        }
-    }
-}
-
-pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreError> {
-    let meta = tx.open_table(META)?;
-    let schema = number(required(meta.get("schema")?)?.value())?;
-    let native = schema == 4;
-    if !matches!(schema, 3 | 4) {
-        return Ok(());
-    }
-    let binding = read_binding(&meta).map_err(|error| match error {
-        StoreError::BindingNotPinned => StoreError::CorruptRecord,
-        other => other,
-    })?;
-    let last = read_applied(&meta)?;
-    let logs = tx.open_table(LOGS)?;
-    let committed = validate_committed(&meta, &logs)?;
-    let heads = tx.open_table(HEADS)?;
-    let versions = tx.open_table(VERSIONS)?;
-    let events = tx.open_table(EVENTS)?;
-    let operations = tx.open_table(OPERATIONS)?;
-    let results = tx.open_table(APPLY_RESULTS)?;
-    if last.is_some_and(|last| committed.is_none_or(|c| last.index > c.index)) {
-        return Err(StoreError::CorruptRecord);
-    }
-    let mut next = Some(0);
-    let mut observed = None;
-    let mut first_operations = BTreeMap::new();
-    let mut expected_membership = crate::CaptureMembership::new(None, vec![], BTreeMap::new())?;
-    let mut replay_operations = BTreeMap::<[u8; 16], (Vec<u8>, CreatedResult)>::new();
-    let mut replay_heads = BTreeMap::<Vec<u8>, [u8; 16]>::new();
-    let mut replay_versions = std::collections::BTreeSet::new();
-    for row in results.iter()? {
-        let (k, v) = row?;
-        let entry = decode_log(required(logs.get(k.value())?)?.value())?;
-        let r = if native {
-            let response = decode_response(v.value())?;
-            let native_entry = decode_native_entry(entry.clone())?;
-            validate_response_kind(&native_entry, &response)?;
-            if next != Some(k.value()) || response.log_id().index != k.value() {
-                return Err(StoreError::CorruptRecord);
-            }
-            observed = Some(*response.log_id());
-            next = k.value().checked_add(1);
-            match native_entry.payload {
-                openraft::EntryPayload::Membership(membership) => {
-                    expected_membership =
-                        crate::CaptureMembership::try_from(&openraft::StoredMembership::new(Some(entry.id.into()), membership))?;
-                    continue;
-                }
-                openraft::EntryPayload::Blank => continue,
-                openraft::EntryPayload::Normal(_) => CreatedApplyResult {
-                    log_id: *response.log_id(),
-                    result: response.created_result().cloned().ok_or(StoreError::CorruptRecord)?,
-                },
-            }
-        } else {
-            decode_apply(v.value())?
-        };
-        if !native && (next != Some(k.value()) || r.log_id.index != k.value()) {
-            return Err(StoreError::CorruptRecord);
-        }
-        if entry.id != r.log_id {
-            return Err(StoreError::CorruptRecord);
-        }
-        let c = entry_command(&entry, native)?;
-        let canonical = c.encode()?;
-        let op = decode_operation(required(operations.get(c.operation_id.as_slice())?)?.value())?;
-        let canonical_matches = op.canonical_command == canonical;
-        let expected = if canonical_matches {
-            op.result
-        } else {
-            CreatedResult::OperationConflict
-        };
-        if native {
-            let replay = if let Some((previous, result)) = replay_operations.get(&c.operation_id) {
-                if *previous == canonical {
-                    result.clone()
-                } else {
-                    CreatedResult::OperationConflict
-                }
-            } else {
-                let object = object_bytes(&c.object)?;
-                let key = version_key(&c.object, c.prepared.version_id)?;
-                let actual = replay_heads.get(&object).copied();
-                let result = if c.binding_revision != binding.revision {
-                    CreatedResult::BindingMismatch
-                } else if actual != c.expected_head {
-                    CreatedResult::HeadMismatch { actual_head: actual }
-                } else if replay_versions.contains(&key) {
-                    CreatedResult::VersionConflict
-                } else {
-                    replay_heads.insert(object, c.prepared.version_id);
-                    replay_versions.insert(key);
-                    CreatedResult::Created {
-                        version_id: c.prepared.version_id,
-                        event_id: c.operation_id,
-                        decision_log_id: r.log_id,
-                    }
-                };
-                replay_operations.insert(c.operation_id, (canonical.clone(), result.clone()));
-                result
-            };
-            if replay != r.result {
-                return Err(StoreError::CorruptRecord);
-            }
-        }
-        if expected != r.result {
-            return Err(StoreError::CorruptRecord);
-        }
-        if canonical_matches {
-            first_operations.entry(c.operation_id).or_insert(r.log_id);
-        }
-        observed = Some(r.log_id);
-        next = k.value().checked_add(1)
-    }
-    if observed != last {
-        return Err(StoreError::CorruptRecord);
-    }
-    if native {
-        if native_application_state(tx)?.1 != expected_membership
-            || heads.len()? != u64::try_from(replay_heads.len()).map_err(|_| StoreError::CorruptRecord)?
-            || versions.len()? != u64::try_from(replay_versions.len()).map_err(|_| StoreError::CorruptRecord)?
-        {
-            return Err(StoreError::CorruptRecord);
-        }
-        for (object, id) in replay_heads {
-            if required(heads.get(object.as_slice())?)?.value() != id {
-                return Err(StoreError::CorruptRecord);
-            }
-        }
-    }
-    for row in operations.iter()? {
-        let (k, v) = row?;
-        let op = decode_operation(v.value())?;
-        let c = DecideCreated::decode(&op.canonical_command)?;
-        if k.value() != c.operation_id {
-            return Err(StoreError::CorruptRecord);
-        }
-        let first = required(first_operations.get(&c.operation_id))?;
-        if let CreatedResult::Created {
-            version_id,
-            event_id,
-            decision_log_id,
-        } = op.result
-        {
-            if version_id != c.prepared.version_id
-                || event_id != c.operation_id
-                || c.binding_revision != binding.revision
-                || decision_log_id != *first
-            {
-                return Err(StoreError::CorruptRecord);
-            }
-            let event = decode_event(required(events.get(event_id.as_slice())?)?.value())?;
-            if event
-                != (CreatedEvent {
-                    object: c.object.clone(),
-                    prepared: c.prepared.clone(),
-                    decision_log_id,
-                    binding: binding.clone(),
-                })
-            {
-                return Err(StoreError::CorruptRecord);
-            }
-        } else if events.get(c.operation_id.as_slice())?.is_some() {
-            return Err(StoreError::CorruptRecord);
-        }
-    }
-    for row in events.iter()? {
-        let (k, v) = row?;
-        let e = decode_event(v.value())?;
-        if e.binding != binding {
-            return Err(StoreError::CorruptRecord);
-        }
-        let op = decode_operation(required(operations.get(k.value())?)?.value())?;
-        let c = DecideCreated::decode(&op.canonical_command)?;
-        if c.object != e.object
-            || c.prepared != e.prepared
-            || op.result
-                != (CreatedResult::Created {
-                    version_id: e.prepared.version_id,
-                    event_id: c.operation_id,
-                    decision_log_id: e.decision_log_id,
-                })
-        {
-            return Err(StoreError::CorruptRecord);
-        }
-        let version =
-            decode_version(required(versions.get(version_key(&e.object, e.prepared.version_id)?.as_slice())?)?.value())?;
-        if version
-            != (CreatedVersion {
-                prepared: e.prepared.clone(),
-                decision_log_id: e.decision_log_id,
-                binding_revision: binding.revision,
-            })
-        {
-            return Err(StoreError::CorruptRecord);
-        }
-    }
-    for row in versions.iter()? {
-        let (k, v) = row?;
-        let version = decode_version(v.value())?;
-        let mut d = Decoder::new(k.value());
-        let object = d.object()?;
-        let id = d.uuid()?;
-        d.finish()?;
-        if id != version.prepared.version_id || version.binding_revision != binding.revision {
-            return Err(StoreError::CorruptRecord);
-        }
-        let entry = decode_log(required(logs.get(version.decision_log_id.index)?)?.value())?;
-        let c = entry_command(&entry, native)?;
-        if entry.id != version.decision_log_id
-            || c.object != object
-            || c.prepared != version.prepared
-            || events.get(c.operation_id.as_slice())?.is_none()
-        {
-            return Err(StoreError::CorruptRecord);
-        }
-        let head = required(heads.get(object_bytes(&object)?.as_slice())?)?;
-        let mut d = Decoder::new(head.value());
-        let head_id = d.uuid()?;
-        d.finish()?;
-        let current = decode_version(required(versions.get(version_key(&object, head_id)?.as_slice())?)?.value())?;
-        if current.decision_log_id.index < version.decision_log_id.index {
-            return Err(StoreError::CorruptRecord);
-        }
-    }
-    for row in heads.iter()? {
-        let (k, v) = row?;
-        let mut d = Decoder::new(k.value());
-        let object = d.object()?;
-        d.finish()?;
-        let mut d = Decoder::new(v.value());
-        let id = d.uuid()?;
-        d.finish()?;
-        decode_version(required(versions.get(version_key(&object, id)?.as_slice())?)?.value())?;
-    }
-    Ok(())
-}
@@ -1,531 +0,0 @@
-use super::*;
-use crate::application::{APPLY_RESULTS, EVENTS, OPERATIONS, VERSIONS};
-use crate::tests::CrashBackend;
-use redb::ReadableTableMetadata;
-use std::{
-    io::{BufRead, BufReader, Write},
-    process::{Command, Stdio},
-    sync::mpsc,
-    time::Duration,
-};
-fn identity() -> NodeIdentity {
-    NodeIdentity {
-        cluster_id: [7; 16],
-        node_id: 0,
-    }
-}
-fn binding() -> CaptureBinding {
-    CaptureBinding {
-        revision: 9,
-        binding_id: [8; 16],
-        target_id: [9; 16],
-    }
-}
-fn command() -> DecideCreated {
-    DecideCreated {
-        operation_id: [1; 16],
-        object: ObjectIdentity {
-            bucket_incarnation: [2; 16],
-            key: vec![0, 255, 3],
-        },
-        expected_head: None,
-        prepared: PreparedIdentity {
-            version_id: [3; 16],
-            preparation_id: [4; 16],
-            content_digest: [5; 32],
-            content_length: 123,
-        },
-        binding_revision: 9,
-    }
-}
-fn id(index: u64) -> LogId {
-    LogId {
-        term: 8,
-        leader_node: 0,
-        index,
-    }
-}
-fn append(store: &mut Store, index: u64, c: &DecideCreated) {
-    store
-        .append(&[LogEntry {
-            id: id(index),
-            payload: c.encode().expect("encode command"),
-        }])
-        .expect("append command");
-}
-fn apply(store: &mut Store, index: u64, c: &DecideCreated) -> CreatedResult {
-    append(store, index, c);
-    store.save_committed(id(index)).expect("commit log");
-    store.apply_next_created(id(index)).expect("apply command")
-}
-fn success(c: &DecideCreated, index: u64) -> CreatedResult {
-    CreatedResult::Created {
-        version_id: c.prepared.version_id,
-        event_id: c.operation_id,
-        decision_log_id: id(index),
-    }
-}
-fn complete(store: &Store) {
-    let c = command();
-    assert_eq!(store.capture_binding().expect("binding"), Some(binding()));
-    assert_eq!(
-        store.created_head(&c.object).expect("head"),
-        Some(c.prepared.version_id),
-        "U11 acknowledged head lost"
-    );
-    assert_eq!(
-        store.created_version(&c.object, c.prepared.version_id).expect("version"),
-        Some(CreatedVersion {
-            prepared: c.prepared.clone(),
-            decision_log_id: id(0),
-            binding_revision: 9
-        })
-    );
-    assert_eq!(
-        store.created_event(c.operation_id).expect("event"),
-        Some(CreatedEvent {
-            object: c.object.clone(),
-            prepared: c.prepared.clone(),
-            decision_log_id: id(0),
-            binding: binding()
-        }),
-        "U11 complete event missing"
-    );
-    assert_eq!(
-        store.created_operation(c.operation_id).expect("operation"),
-        Some(CreatedOperation {
-            canonical_command: c.encode().expect("encode"),
-            result: success(&c, 0)
-        })
-    );
-    assert_eq!(
-        store.created_apply_result(0).expect("result"),
-        Some(CreatedApplyResult {
-            log_id: id(0),
-            result: success(&c, 0)
-        })
-    );
-    assert_eq!(store.last_applied().expect("last applied"), Some(id(0)));
-}
-fn fixture() -> (tempfile::TempDir, Store) {
-    let dir = tempfile::tempdir().expect("tempdir");
-    let mut store = Store::initialize(dir.path().join("db"), identity()).expect("initialize");
-    store.pin_capture_binding(binding()).expect("pin");
-    (dir, store)
-}
-fn reopen(path: &std::path::Path) -> Store {
-    let result = Store::open(path, identity());
-    assert!(result.is_ok(), "U11 complete reopen: {:?}", result.as_ref().err());
-    result.expect("asserted successful reopen")
-}
-#[test]
-fn created_codec_is_canonical_and_strict() {
-    let c = command();
-    let b = c.encode().expect("encode");
-    let mut golden = b"CRTD\x01".to_vec();
-    golden.extend([1; 16]);
-    golden.extend([2; 16]);
-    golden.extend(3u32.to_be_bytes());
-    golden.extend([0, 255, 3]);
-    golden.push(0);
-    golden.extend([3; 16]);
-    golden.extend([4; 16]);
-    golden.extend([5; 32]);
-    golden.extend(123u64.to_be_bytes());
-    golden.extend(9u64.to_be_bytes());
-    assert_eq!(b, golden);
-    assert_eq!(DecideCreated::decode(&b).expect("decode"), c);
-    let mut variants = Vec::new();
-    let mut v = c.clone();
-    v.operation_id = [6; 16];
-    variants.push(v);
-    let mut v = c.clone();
-    v.object.bucket_incarnation = [6; 16];
-    variants.push(v);
-    let mut v = c.clone();
-    v.object.key.push(4);
-    variants.push(v);
-    let mut v = c.clone();
-    v.expected_head = Some([6; 16]);
-    variants.push(v);
-    let mut v = c.clone();
-    v.prepared.version_id = [6; 16];
-    variants.push(v);
-    let mut v = c.clone();
-    v.prepared.preparation_id = [6; 16];
-    variants.push(v);
-    let mut v = c.clone();
-    v.prepared.content_digest = [6; 32];
-    variants.push(v);
-    let mut v = c.clone();
-    v.prepared.content_length += 1;
-    variants.push(v);
-    let mut v = c.clone();
-    v.binding_revision += 1;
-    variants.push(v);
-    for v in variants {
-        assert_ne!(v.encode().expect("variant"), b)
-    }
-    for n in 0..b.len() {
-        assert!(DecideCreated::decode(&b[..n]).is_err())
-    }
-    let mut bad = b.clone();
-    bad[4] = 2;
-    assert!(DecideCreated::decode(&bad).is_err());
-    let mut bad = b.clone();
-    bad[44] = 2;
-    assert!(DecideCreated::decode(&bad).is_err());
-    let mut bad = b.clone();
-    bad.push(0);
-    assert!(DecideCreated::decode(&bad).is_err());
-    let mut bad = b;
-    bad[37..41].copy_from_slice(&4097u32.to_be_bytes());
-    assert!(DecideCreated::decode(&bad).is_err());
-    let mut bad = c.clone();
-    bad.operation_id = [0; 16];
-    assert!(bad.encode().is_err());
-    bad = c;
-    bad.object.key = vec![0; MAX_OBJECT_KEY_LENGTH + 1];
-    assert!(bad.encode().is_err());
-}
-#[test]
-fn created_committed_order_and_identity_are_enforced() {
-    let (_dir, mut s) = fixture();
-    let c = command();
-    append(&mut s, 0, &c);
-    append(&mut s, 1, &c);
-    assert!(s.apply_next_created(id(0)).is_err());
-    assert_eq!(s.last_applied().expect("no apply"), None);
-    s.save_committed(id(1)).expect("committed");
-    assert!(matches!(s.apply_next_created(id(1)), Err(StoreError::InvalidApplySequence)));
-    let mut wrong = id(0);
-    wrong.term += 1;
-    assert!(s.apply_next_created(wrong).is_err());
-    wrong = id(0);
-    wrong.leader_node += 1;
-    assert!(s.apply_next_created(wrong).is_err());
-    assert_eq!(s.apply_next_created(id(0)).expect("first"), success(&c, 0));
-    complete(&s);
-    assert_eq!(s.apply_next_created(id(0)).expect("retry"), success(&c, 0));
-    complete(&s);
-    assert!(s.apply_next_created(wrong).is_err());
-    assert_eq!(s.apply_next_created(id(1)).expect("duplicate log"), success(&c, 0));
-    assert_eq!(s.last_applied().expect("last"), Some(id(1)));
-    assert_eq!(
-        s.created_apply_result(1).expect("result"),
-        Some(CreatedApplyResult {
-            log_id: id(1),
-            result: success(&c, 0)
-        })
-    );
-    let tx = s.database.begin_read().expect("read");
-    assert_eq!(tx.open_table(EVENTS).expect("events").len().expect("len"), 1);
-}
-#[test]
-fn created_head_version_event_and_dedup_reopen_together() {
-    let (dir, mut s) = fixture();
-    let c = command();
-    assert_eq!(apply(&mut s, 0, &c), success(&c, 0));
-    drop(s);
-    let mut s = reopen(&dir.path().join("db"));
-    complete(&s);
-    assert_eq!(apply(&mut s, 1, &c), success(&c, 0));
-    let mut changed = c.clone();
-    changed.prepared.content_digest = [6; 32];
-    assert_eq!(
-        apply(&mut s, 2, &changed),
-        CreatedResult::OperationConflict,
-        "U11 full canonical operation conflict"
-    );
-    assert_eq!(
-        s.created_operation(c.operation_id)
-            .expect("original")
-            .expect("operation")
-            .result,
-        success(&c, 0)
-    );
-    let mut stale = c.clone();
-    stale.operation_id = [6; 16];
-    stale.prepared.version_id = [6; 16];
-    assert_eq!(
-        apply(&mut s, 3, &stale),
-        CreatedResult::HeadMismatch {
-            actual_head: Some(c.prepared.version_id)
-        }
-    );
-    let mut reuse = c.clone();
-    reuse.operation_id = [7; 16];
-    reuse.expected_head = Some(c.prepared.version_id);
-    assert_eq!(apply(&mut s, 4, &reuse), CreatedResult::VersionConflict);
-    let mut replacement = stale.clone();
-    replacement.operation_id = [10; 16];
-    replacement.expected_head = Some(c.prepared.version_id);
-    assert_eq!(apply(&mut s, 5, &replacement), success(&replacement, 5));
-    assert_eq!(s.created_head(&c.object).expect("head"), Some(replacement.prepared.version_id));
-    assert_eq!(
-        s.created_version(&c.object, c.prepared.version_id)
-            .expect("old version")
-            .expect("version")
-            .decision_log_id,
-        id(0)
-    );
-    assert!(s.created_event(c.operation_id).expect("old event").is_some());
-    assert!(s.created_event(replacement.operation_id).expect("new event").is_some());
-    assert!(s.created_event(stale.operation_id).expect("rejection event").is_none());
-    assert_eq!(
-        apply(&mut s, 6, &stale),
-        CreatedResult::HeadMismatch {
-            actual_head: Some(c.prepared.version_id)
-        }
-    );
-    drop(s);
-    let s = reopen(&dir.path().join("db"));
-    assert_eq!(s.last_applied().expect("last"), Some(id(6)));
-    let tx = s.database.begin_read().expect("read");
-    assert_eq!(tx.open_table(EVENTS).expect("events").len().expect("len"), 2);
-}
-#[test]
-fn created_binding_is_pinned_and_not_caller_selected() {
-    let (dir, mut s) = fixture();
-    let mut bad = command();
-    bad.binding_revision = 10;
-    assert_eq!(apply(&mut s, 0, &bad), CreatedResult::BindingMismatch);
-    assert!(s.created_head(&bad.object).expect("head").is_none());
-    assert!(s.created_event(bad.operation_id).expect("event").is_none());
-    assert_eq!(
-        s.created_operation(bad.operation_id)
-            .expect("operation")
-            .expect("saved")
-            .result,
-        CreatedResult::BindingMismatch
-    );
-    s.pin_capture_binding(binding()).expect("idempotent pin");
-    let mut changed = binding();
-    changed.target_id = [10; 16];
-    assert!(matches!(s.pin_capture_binding(changed), Err(StoreError::BindingConflict)));
-    drop(s);
-    let s = reopen(&dir.path().join("db"));
-    assert_eq!(s.capture_binding().expect("binding"), Some(binding()));
-}
-#[test]
-fn created_schema_compatibility_and_corruption_are_checked() {
-    let dir = tempfile::tempdir().expect("dir");
-    let path = dir.path().join("db");
-    let s = Store::initialize(&path, identity()).expect("schema1");
-    assert_eq!(s.capture_binding().expect("unmigrated"), None);
-    drop(s);
-    let mut s = reopen(&path);
-    assert_eq!(s.last_applied().expect("no apply"), None);
-    append(&mut s, 0, &command());
-    s.save_committed(id(0)).expect("schema2");
-    drop(s);
-    let mut s = reopen(&path);
-    assert_eq!(s.capture_binding().expect("schema2 unchanged"), None);
-    s.pin_capture_binding(binding()).expect("explicit upgrade");
-    drop(s);
-    let mut s = reopen(&path);
-    assert_eq!(s.capture_binding().expect("schema3"), Some(binding()));
-    s.apply_next_created(id(0)).expect("apply");
-    drop(s);
-    let db = Database::open(&path).expect("raw db");
-    let tx = db.begin_write().expect("write corruption");
-    tx.open_table(EVENTS)
-        .expect("events")
-        .remove(command().operation_id.as_slice())
-        .expect("remove reference");
-    tx.commit().expect("commit corruption");
-    drop(db);
-    let result = Store::open(&path, identity());
-    assert!(
-        matches!(result, Err(StoreError::CorruptRecord)),
-        "U11 missing event rejected before expect"
-    );
-    let (dir, mut s) = fixture();
-    assert_eq!(s.read_committed().expect("empty committed"), None);
-    append(&mut s, 0, &command());
-    s.save_committed(id(0)).expect("committed schema3");
-    assert_eq!(s.capture_binding().expect("no downgrade"), Some(binding()));
-    drop(s);
-    let path = dir.path().join("db");
-    let db = Database::open(&path).expect("raw db");
-    let tx = db.begin_write().expect("write");
-    let mut marker = vec![1];
-    marker.extend(encode_id(id(0)));
-    tx.open_table(META)
-        .expect("meta")
-        .insert("last_applied", marker.as_slice())
-        .expect("corrupt applied");
-    tx.commit().expect("commit");
-    drop(db);
-    assert!(
-        matches!(Store::open(&path, identity()), Err(StoreError::CorruptRecord)),
-        "U11 last applied missing result rejected"
-    );
-    let (dir, s) = fixture();
-    drop(s);
-    let path = dir.path().join("db");
-    let db = Database::open(&path).expect("raw pinned db");
-    let tx = db.begin_write().expect("missing binding write");
-    tx.open_table(META)
-        .expect("meta")
-        .remove("capture_binding")
-        .expect("remove binding");
-    tx.commit().expect("binding corruption commit");
-    drop(db);
-    assert!(
-        matches!(Store::open(&path, identity()), Err(StoreError::CorruptRecord)),
-        "U11 missing pinned binding rejected"
-    );
-
-    let (dir, mut s) = fixture();
-    let c = command();
-    assert_eq!(apply(&mut s, 0, &c), success(&c, 0));
-    assert_eq!(apply(&mut s, 1, &c), success(&c, 0));
-    drop(s);
-    let path = dir.path().join("db");
-    let db = Database::open(&path).expect("raw duplicate db");
-    let tx = db.begin_write().expect("retarget decision write");
-    let mut key = created::object_bytes(&c.object).expect("object key");
-    key.extend(c.prepared.version_id);
-    let mut version = Vec::new();
-    created::put_prepared(&mut version, &c.prepared);
-    version.extend(encode_id(id(1)));
-    version.extend(binding().revision.to_be_bytes());
-    tx.open_table(VERSIONS)
-        .expect("versions")
-        .insert(key.as_slice(), version.as_slice())
-        .expect("retarget version");
-    let mut event = created::object_bytes(&c.object).expect("event object");
-    created::put_prepared(&mut event, &c.prepared);
-    event.extend(encode_id(id(1)));
-    event.extend(created::binding_bytes(&binding()).expect("binding bytes"));
-    tx.open_table(EVENTS)
-        .expect("events")
-        .insert(c.operation_id.as_slice(), event.as_slice())
-        .expect("retarget event");
-    let canonical = c.encode().expect("canonical");
-    let mut operation = u64::try_from(canonical.len()).expect("length").to_be_bytes().to_vec();
-    operation.extend(canonical);
-    operation.extend(created::result_bytes(&success(&c, 1)));
-    tx.open_table(OPERATIONS)
-        .expect("operations")
-        .insert(c.operation_id.as_slice(), operation.as_slice())
-        .expect("retarget original response");
-    for index in 0..=1 {
-        let mut result = encode_id(id(index));
-        result.extend(created::result_bytes(&success(&c, 1)));
-        tx.open_table(APPLY_RESULTS)
-            .expect("results")
-            .insert(index, result.as_slice())
-            .expect("retarget duplicate response");
-    }
-    tx.commit().expect("coherent retarget commit");
-    drop(db);
-    assert!(
-        matches!(Store::open(&path, identity()), Err(StoreError::CorruptRecord)),
-        "U11 original decision cannot retarget duplicate log"
-    );
-}
-#[test]
-fn created_process_child() {
-    if let Some(path) = std::env::var_os("U11_CHILD_DB") {
-        let mut s = Store::initialize(path, identity()).expect("child init");
-        s.pin_capture_binding(binding()).expect("child pin");
-        assert_eq!(apply(&mut s, 0, &command()), success(&command(), 0));
-        complete(&s);
-        println!("U11_ACK");
-        std::io::stdout().flush().expect("flush ACK");
-        loop {
-            std::thread::park()
-        }
-    }
-    let (_dir, mut s) = fixture();
-    assert_eq!(apply(&mut s, 0, &command()), success(&command(), 0));
-    complete(&s);
-}
-#[test]
-fn created_acknowledged_apply_survives_child_kill() {
-    let dir = tempfile::tempdir().expect("dir");
-    let path = dir.path().join("db");
-    let mut child = Command::new(std::env::current_exe().expect("test executable"))
-        .args(["--exact", "application_tests::created_process_child", "--nocapture"])
-        .env("U11_CHILD_DB", &path)
-        .stdout(Stdio::piped())
-        .stderr(Stdio::inherit())
-        .spawn()
-        .expect("spawn");
-    let stdout = child.stdout.take().expect("stdout");
-    let (tx, rx) = mpsc::channel();
-    let reader = std::thread::spawn(move || {
-        for line in BufReader::new(stdout).lines() {
-            let line = line.expect("child line");
-            if line.ends_with("U11_ACK") {
-                let _ = tx.send(());
-                break;
-            }
-        }
-    });
-    let acknowledged = rx.recv_timeout(Duration::from_secs(20));
-    child.kill().expect("kill child");
-    child.wait().expect("wait child");
-    reader.join().expect("reader cleanup");
-    assert!(acknowledged.is_ok(), "U11 bounded child ACK missing");
-    complete(&reopen(&path));
-}
-#[test]
-fn created_power_cut_before_and_after_apply_flush() {
-    let (dir, mut s) = fixture();
-    append(&mut s, 0, &command());
-    s.save_committed(id(0)).expect("committed");
-    drop(s);
-    let path = dir.path().join("db");
-    let backend = CrashBackend::new(&path);
-    let mut s = Store::from_database(backend.database(), identity()).expect("backend");
-    assert_eq!(s.apply_next_created(id(0)).expect("apply ACK"), success(&command(), 0));
-    backend.cut();
-    drop(s);
-    complete(&reopen(&path));
-    for sync in 1..=2 {
-        for after in [false, true] {
-            let (dir, mut s) = fixture();
-            append(&mut s, 0, &command());
-            s.save_committed(id(0)).expect("committed");
-            drop(s);
-            let path = dir.path().join("db");
-            let backend = CrashBackend::new(&path);
-            let mut s = Store::from_database(backend.database(), identity()).expect("backend");
-            let (entered, resume) = backend.pause_sync(sync, after);
-            let writer = std::thread::spawn(move || {
-                let result = s.apply_next_created(id(0));
-                (s, result)
-            });
-            entered.wait();
-            backend.cut();
-            resume.wait();
-            let (s, result) = writer.join().expect("writer");
-            drop(s);
-            let reopened = Store::open(&path, identity());
-            assert!(
-                reopened.is_ok(),
-                "U11 apply cut must reopen whole old/new state sync={sync} after={after}: {:?}",
-                reopened.as_ref().err()
-            );
-            let s = reopened.expect("asserted reopen");
-            if s.last_applied().expect("last").is_some() {
-                complete(&s)
-            } else {
-                assert!(result.is_err());
-                assert_eq!(s.created_head(&command().object).expect("old head"), None);
-                assert_eq!(
-                    s.created_version(&command().object, command().prepared.version_id)
-                        .expect("old version"),
-                    None
-                );
-                assert_eq!(s.created_event(command().operation_id).expect("old event"), None);
-                assert_eq!(s.created_operation(command().operation_id).expect("old op"), None);
-                assert_eq!(s.created_apply_result(0).expect("old result"), None)
-            }
-            eprintln!("U11 cut boundary sync={sync} after={after}");
-        }
-    }
-}
@@ -1,63 +0,0 @@
-use crate::{LOGS, LogId, META, Store, StoreError, decode_id, decode_log, encode_id, number};
-use redb::{ReadableDatabase, ReadableTable};
-
-pub(crate) fn validate_committed(
-    meta: &impl ReadableTable<&'static str, &'static [u8]>,
-    logs: &impl ReadableTable<u64, &'static [u8]>,
-) -> Result<Option<LogId>, StoreError> {
-    let schema = number(meta.get("schema")?.ok_or(StoreError::CorruptRecord)?.value())?;
-    let marker = meta.get("committed")?;
-    match (schema, marker) {
-        (1 | 3 | 4, None) => {
-            if matches!(schema, 3 | 4) && meta.get("last_applied")?.is_none_or(|v| v.value() != [0]) {
-                return Err(StoreError::CorruptRecord);
-            }
-            Ok(None)
-        }
-        (2 | 3 | 4, Some(marker)) => {
-            let id = decode_id(marker.value())?;
-            let record = logs.get(id.index)?.ok_or(StoreError::CorruptRecord)?;
-            if decode_log(record.value())?.id != id {
-                return Err(StoreError::CorruptRecord);
-            }
-            Ok(Some(id))
-        }
-        (1 | 2, _) => Err(StoreError::CorruptRecord),
-        _ => Err(StoreError::UnsupportedSchema),
-    }
-}
-
-impl Store {
-    /// Read the exact committed boundary, validating it against the stored log.
-    pub fn read_committed(&self) -> Result<Option<LogId>, StoreError> {
-        let tx = self.database.begin_read()?;
-        validate_committed(&tx.open_table(META)?, &tx.open_table(LOGS)?)
-    }
-
-    /// Persist a monotonic committed boundary supplied by the authority owner.
-    pub fn save_committed(&mut self, id: LogId) -> Result<(), StoreError> {
-        let tx = self.write_transaction()?;
-        {
-            let mut meta = tx.open_table(META)?;
-            let logs = tx.open_table(LOGS)?;
-            let previous = validate_committed(&meta, &logs)?;
-            let record = logs.get(id.index)?.ok_or(StoreError::InvalidCommittedLog)?;
-            if decode_log(record.value())?.id != id {
-                return Err(StoreError::InvalidCommittedLog);
-            }
-            if let Some(previous) = previous {
-                if id.index < previous.index {
-                    return Err(StoreError::CommittedRegression);
-                }
-                if id == previous {
-                    return Ok(());
-                }
-            }
-            meta.insert("committed", encode_id(id).as_slice())?;
-            if previous.is_none() && !matches!(number(meta.get("schema")?.ok_or(StoreError::CorruptRecord)?.value())?, 3 | 4) {
-                meta.insert("schema", 2u64.to_be_bytes().as_slice())?;
-            }
-        }
-        self.commit(tx)
-    }
-}
@@ -1,242 +0,0 @@
-use crate::{LogId, StoreError, decode_id, encode_id};
-
-pub const MAX_OBJECT_KEY_LENGTH: usize = 4096;
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct ObjectIdentity {
-    pub bucket_incarnation: [u8; 16],
-    pub key: Vec<u8>,
-}
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct PreparedIdentity {
-    pub version_id: [u8; 16],
-    pub preparation_id: [u8; 16],
-    pub content_digest: [u8; 32],
-    pub content_length: u64,
-}
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct DecideCreated {
-    pub operation_id: [u8; 16],
-    pub object: ObjectIdentity,
-    pub expected_head: Option<[u8; 16]>,
-    pub prepared: PreparedIdentity,
-    pub binding_revision: u64,
-}
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct CaptureBinding {
-    pub revision: u64,
-    pub binding_id: [u8; 16],
-    pub target_id: [u8; 16],
-}
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub enum CreatedResult {
-    Created {
-        version_id: [u8; 16],
-        event_id: [u8; 16],
-        decision_log_id: LogId,
-    },
-    HeadMismatch {
-        actual_head: Option<[u8; 16]>,
-    },
-    OperationConflict,
-    BindingMismatch,
-    VersionConflict,
-}
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct CreatedVersion {
-    pub prepared: PreparedIdentity,
-    pub decision_log_id: LogId,
-    pub binding_revision: u64,
-}
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct CreatedEvent {
-    pub object: ObjectIdentity,
-    pub prepared: PreparedIdentity,
-    pub decision_log_id: LogId,
-    pub binding: CaptureBinding,
-}
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct CreatedOperation {
-    pub canonical_command: Vec<u8>,
-    pub result: CreatedResult,
-}
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct CreatedApplyResult {
-    pub log_id: LogId,
-    pub result: CreatedResult,
-}
-
-pub(crate) struct Decoder<'a> {
-    bytes: &'a [u8],
-}
-impl<'a> Decoder<'a> {
-    pub(crate) fn new(bytes: &'a [u8]) -> Self {
-        Self { bytes }
-    }
-    pub(crate) fn take<const N: usize>(&mut self) -> Result<[u8; N], StoreError> {
-        let a = self
-            .bytes
-            .get(..N)
-            .ok_or(StoreError::CorruptRecord)?
-            .try_into()
-            .map_err(|_| StoreError::CorruptRecord)?;
-        self.bytes = &self.bytes[N..];
-        Ok(a)
-    }
-    pub(crate) fn u64(&mut self) -> Result<u64, StoreError> {
-        Ok(u64::from_be_bytes(self.take()?))
-    }
-    pub(crate) fn uuid(&mut self) -> Result<[u8; 16], StoreError> {
-        let id = self.take()?;
-        if id == [0; 16] {
-            return Err(StoreError::CorruptRecord);
-        }
-        Ok(id)
-    }
-    pub(crate) fn option(&mut self) -> Result<Option<[u8; 16]>, StoreError> {
-        match self.take::<1>()?[0] {
-            0 => Ok(None),
-            1 => Ok(Some(self.uuid()?)),
-            _ => Err(StoreError::CorruptRecord),
-        }
-    }
-    pub(crate) fn object(&mut self) -> Result<ObjectIdentity, StoreError> {
-        let bucket_incarnation = self.uuid()?;
-        let len = usize::try_from(u32::from_be_bytes(self.take()?)).map_err(|_| StoreError::CorruptRecord)?;
-        if len > MAX_OBJECT_KEY_LENGTH {
-            return Err(StoreError::CorruptRecord);
-        }
-        let key = self.bytes.get(..len).ok_or(StoreError::CorruptRecord)?.to_vec();
-        self.bytes = &self.bytes[len..];
-        Ok(ObjectIdentity { bucket_incarnation, key })
-    }
-    pub(crate) fn prepared(&mut self) -> Result<PreparedIdentity, StoreError> {
-        Ok(PreparedIdentity {
-            version_id: self.uuid()?,
-            preparation_id: self.uuid()?,
-            content_digest: self.take()?,
-            content_length: self.u64()?,
-        })
-    }
-    pub(crate) fn id(&mut self) -> Result<LogId, StoreError> {
-        decode_id(&self.take::<24>()?)
-    }
-    pub(crate) fn binding(&mut self) -> Result<CaptureBinding, StoreError> {
-        Ok(CaptureBinding {
-            revision: self.u64()?,
-            binding_id: self.uuid()?,
-            target_id: self.uuid()?,
-        })
-    }
-    pub(crate) fn finish(self) -> Result<(), StoreError> {
-        if self.bytes.is_empty() {
-            Ok(())
-        } else {
-            Err(StoreError::CorruptRecord)
-        }
-    }
-    pub(crate) fn result(&mut self) -> Result<CreatedResult, StoreError> {
-        match self.take::<1>()?[0] {
-            0 => Ok(CreatedResult::Created {
-                version_id: self.uuid()?,
-                event_id: self.uuid()?,
-                decision_log_id: self.id()?,
-            }),
-            1 => Ok(CreatedResult::HeadMismatch {
-                actual_head: self.option()?,
-            }),
-            2 => Ok(CreatedResult::OperationConflict),
-            3 => Ok(CreatedResult::BindingMismatch),
-            4 => Ok(CreatedResult::VersionConflict),
-            _ => Err(StoreError::CorruptRecord),
-        }
-    }
-}
-pub(crate) fn put_option(b: &mut Vec<u8>, id: Option<[u8; 16]>) {
-    b.push(u8::from(id.is_some()));
-    if let Some(id) = id {
-        b.extend(id)
-    }
-}
-pub(crate) fn object_bytes(o: &ObjectIdentity) -> Result<Vec<u8>, StoreError> {
-    if o.bucket_incarnation == [0; 16] || o.key.len() > MAX_OBJECT_KEY_LENGTH {
-        return Err(StoreError::CorruptRecord);
-    }
-    let capacity = 20usize.checked_add(o.key.len()).ok_or(StoreError::CorruptRecord)?;
-    let mut b = Vec::with_capacity(capacity);
-    b.extend(o.bucket_incarnation);
-    b.extend(
-        u32::try_from(o.key.len())
-            .map_err(|_| StoreError::CorruptRecord)?
-            .to_be_bytes(),
-    );
-    b.extend(&o.key);
-    Ok(b)
-}
-pub(crate) fn put_prepared(b: &mut Vec<u8>, p: &PreparedIdentity) {
-    b.extend(p.version_id);
-    b.extend(p.preparation_id);
-    b.extend(p.content_digest);
-    b.extend(p.content_length.to_be_bytes())
-}
-pub(crate) fn binding_bytes(v: &CaptureBinding) -> Result<Vec<u8>, StoreError> {
-    let mut b = v.revision.to_be_bytes().to_vec();
-    b.extend(v.binding_id);
-    b.extend(v.target_id);
-    let mut d = Decoder::new(&b);
-    d.binding()?;
-    d.finish()?;
-    Ok(b)
-}
-pub(crate) fn result_bytes(r: &CreatedResult) -> Vec<u8> {
-    let mut b = Vec::new();
-    match r {
-        CreatedResult::Created {
-            version_id,
-            event_id,
-            decision_log_id,
-        } => {
-            b.push(0);
-            b.extend(version_id);
-            b.extend(event_id);
-            b.extend(encode_id(*decision_log_id))
-        }
-        CreatedResult::HeadMismatch { actual_head } => {
-            b.push(1);
-            put_option(&mut b, *actual_head)
-        }
-        CreatedResult::OperationConflict => b.push(2),
-        CreatedResult::BindingMismatch => b.push(3),
-        CreatedResult::VersionConflict => b.push(4),
-    }
-    b
-}
-impl DecideCreated {
-    pub fn encode(&self) -> Result<Vec<u8>, StoreError> {
-        let object = object_bytes(&self.object)?;
-        let capacity = object.len().checked_add(134).ok_or(StoreError::CorruptRecord)?;
-        let mut b = Vec::with_capacity(capacity);
-        b.extend(b"CRTD\x01");
-        b.extend(self.operation_id);
-        b.extend(object);
-        put_option(&mut b, self.expected_head);
-        put_prepared(&mut b, &self.prepared);
-        b.extend(self.binding_revision.to_be_bytes());
-        Self::decode(&b)?;
-        Ok(b)
-    }
-    pub fn decode(bytes: &[u8]) -> Result<Self, StoreError> {
-        let mut d = Decoder::new(bytes);
-        if d.take::<5>()? != *b"CRTD\x01" {
-            return Err(StoreError::CorruptRecord);
-        }
-        let c = Self {
-            operation_id: d.uuid()?,
-            object: d.object()?,
-            expected_head: d.option()?,
-            prepared: d.prepared()?,
-            binding_revision: d.u64()?,
-        };
-        d.finish()?;
-        Ok(c)
-    }
-}
@@ -1,371 +0,0 @@
-//! Validated capture peer metadata and its independent CNOD v1 codec.
-use serde::{Deserialize, Deserializer, Serialize, Serializer, ser::SerializeStruct};
-
-#[derive(Clone, Debug, Default, PartialEq, Eq)]
-pub struct CaptureNode {
-    rpc_uri: String,
-}
-
-#[derive(Debug, thiserror::Error)]
-pub enum CaptureNodeError {
-    #[error("invalid RPC URI")]
-    InvalidRpcUri,
-    #[error("invalid RPC URI")]
-    InvalidRpcUriParse(#[source] url::ParseError),
-    #[error("RPC URI too long")]
-    RpcUriTooLong,
-    #[error("invalid node magic")]
-    InvalidNodeMagic,
-    #[error("unsupported node version")]
-    UnsupportedNodeVersion,
-    #[error("invalid node length")]
-    InvalidNodeLength,
-    #[error("truncated node record")]
-    TruncatedNodeRecord,
-    #[error("invalid node UTF-8")]
-    InvalidNodeUtf8(#[source] std::str::Utf8Error),
-    #[error("trailing node bytes")]
-    TrailingNodeBytes,
-}
-impl CaptureNodeError {
-    pub fn code(&self) -> &'static str {
-        match self {
-            Self::InvalidRpcUri | Self::InvalidRpcUriParse(_) => "invalid_rpc_uri",
-            Self::RpcUriTooLong => "rpc_uri_too_long",
-            Self::InvalidNodeMagic => "invalid_node_magic",
-            Self::UnsupportedNodeVersion => "unsupported_node_version",
-            Self::InvalidNodeLength => "invalid_node_length",
-            Self::TruncatedNodeRecord => "truncated_node_record",
-            Self::InvalidNodeUtf8(_) => "invalid_node_utf8",
-            Self::TrailingNodeBytes => "trailing_node_bytes",
-        }
-    }
-}
-impl CaptureNode {
-    pub fn new(rpc_uri: String) -> Result<Self, CaptureNodeError> {
-        let node = Self { rpc_uri };
-        node.validate()?;
-        Ok(node)
-    }
-
-    pub fn rpc_uri(&self) -> &str {
-        &self.rpc_uri
-    }
-
-    pub fn validate(&self) -> Result<(), CaptureNodeError> {
-        let uri = &self.rpc_uri;
-        if uri.len() > 4096 {
-            return Err(CaptureNodeError::RpcUriTooLong);
-        }
-        if uri.is_empty() || uri.chars().any(|c| c.is_whitespace() || c.is_control() || c == '\\') {
-            return Err(CaptureNodeError::InvalidRpcUri);
-        }
-        let rest = uri
-            .strip_prefix("http://")
-            .or_else(|| uri.strip_prefix("https://"))
-            .ok_or(CaptureNodeError::InvalidRpcUri)?;
-        if uri.contains('?') || uri.contains('#') {
-            return Err(CaptureNodeError::InvalidRpcUri);
-        }
-        let (authority, path) = rest.split_once('/').map_or((rest, ""), |(a, p)| (a, p));
-        // An empty suffix after the first slash is the only allowed path.
-        if !path.is_empty() || authority.contains('@') {
-            return Err(CaptureNodeError::InvalidRpcUri);
-        }
-        let (host, port) = if authority.starts_with('[') {
-            let end = authority.find(']').ok_or(CaptureNodeError::InvalidRpcUri)?;
-            let port = authority
-                .get(end + 1..)
-                .and_then(|suffix| suffix.strip_prefix(':'))
-                .ok_or(CaptureNodeError::InvalidRpcUri)?;
-            (&authority[..=end], port)
-        } else {
-            authority.split_once(':').ok_or(CaptureNodeError::InvalidRpcUri)?
-        };
-        if host.is_empty() || port.is_empty() || !port.bytes().all(|b| b.is_ascii_digit()) {
-            return Err(CaptureNodeError::InvalidRpcUri);
-        }
-        let port = port.parse::<u16>().map_err(|_| CaptureNodeError::InvalidRpcUri)?;
-        if port == 0 {
-            return Err(CaptureNodeError::InvalidRpcUri);
-        }
-        let parsed_url = url::Url::parse(uri).map_err(CaptureNodeError::InvalidRpcUriParse)?;
-        if parsed_url.host().is_none() {
-            return Err(CaptureNodeError::InvalidRpcUri);
-        }
-        Ok(())
-    }
-
-    pub fn encode(&self) -> Result<Vec<u8>, CaptureNodeError> {
-        self.validate()?;
-        let length = u32::try_from(self.rpc_uri.len()).map_err(|_| CaptureNodeError::InvalidNodeLength)?;
-        let mut bytes = b"CNOD\x01".to_vec();
-        bytes.extend_from_slice(&length.to_be_bytes());
-        bytes.extend_from_slice(self.rpc_uri.as_bytes());
-        Ok(bytes)
-    }
-
-    pub fn decode(bytes: &[u8]) -> Result<Self, CaptureNodeError> {
-        if bytes.len() < 9 {
-            return Err(CaptureNodeError::TruncatedNodeRecord);
-        }
-        if &bytes[..4] != b"CNOD" {
-            return Err(CaptureNodeError::InvalidNodeMagic);
-        }
-        if bytes[4] != 1 {
-            return Err(CaptureNodeError::UnsupportedNodeVersion);
-        }
-        let length = u32::from_be_bytes(bytes[5..9].try_into().map_err(|_| CaptureNodeError::InvalidNodeLength)?);
-        if !(1..=4096).contains(&length) {
-            return Err(CaptureNodeError::InvalidNodeLength);
-        }
-        let length = usize::try_from(length).map_err(|_| CaptureNodeError::InvalidNodeLength)?;
-        let end = 9_usize.checked_add(length).ok_or(CaptureNodeError::InvalidNodeLength)?;
-        if bytes.len() < end {
-            return Err(CaptureNodeError::TruncatedNodeRecord);
-        }
-        if bytes.len() != end {
-            return Err(CaptureNodeError::TrailingNodeBytes);
-        }
-        let uri = std::str::from_utf8(&bytes[9..end]).map_err(CaptureNodeError::InvalidNodeUtf8)?;
-        Self::new(uri.to_owned())
-    }
-}
-impl Serialize for CaptureNode {
-    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
-        self.validate().map_err(serde::ser::Error::custom)?;
-        let mut state = serializer.serialize_struct("CaptureNode", 1)?;
-        state.serialize_field("rpc_uri", &self.rpc_uri)?;
-        state.end()
-    }
-}
-impl<'de> Deserialize<'de> for CaptureNode {
-    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
-        #[derive(Deserialize)]
-        #[serde(deny_unknown_fields)]
-        struct Fields {
-            rpc_uri: String,
-        }
-        let fields = Fields::deserialize(deserializer).map_err(|_| serde::de::Error::custom("invalid capture node fields"))?;
-        Self::new(fields.rpc_uri).map_err(serde::de::Error::custom)
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use std::error::Error;
-
-    fn record(payload: &[u8]) -> Vec<u8> {
-        let mut bytes = b"CNOD\x01".to_vec();
-        bytes.extend_from_slice(&u32::try_from(payload.len()).expect("test payload length").to_be_bytes());
-        bytes.extend_from_slice(payload);
-        bytes
-    }
-    fn reject_uri(uri: &str) {
-        let result = CaptureNode::new(uri.to_owned());
-        assert!(result.is_err(), "URI should reject: {uri:?}");
-        assert_eq!(result.expect_err("rejected URI").code(), "invalid_rpc_uri");
-    }
-    fn reject_record(bytes: &[u8], code: &str) {
-        let result = CaptureNode::decode(bytes);
-        assert!(result.is_err(), "record should reject with {code}");
-        assert_eq!(result.expect_err("rejected record").code(), code);
-    }
-    #[test]
-    fn actual_openraft_node_bound() {
-        fn require_node<T: openraft::Node>() {}
-        require_node::<CaptureNode>();
-        assert_eq!(CaptureNode::default().rpc_uri(), "");
-    }
-    #[test]
-    fn default_and_empty_rejected_at_every_boundary() {
-        let node = CaptureNode::default();
-        assert_eq!(node.rpc_uri(), "");
-        reject_uri("");
-        assert_eq!(node.validate().expect_err("empty validation").code(), "invalid_rpc_uri");
-        assert!(serde_json::to_string(&node).is_err());
-        assert!(serde_json::from_str::<CaptureNode>(r#"{"rpc_uri":""}"#).is_err());
-        assert_eq!(node.encode().expect_err("empty encoding").code(), "invalid_rpc_uri");
-        reject_record(&record(b""), "invalid_node_length");
-    }
-    #[test]
-    fn accepted_uris_preserve_exact_bytes() {
-        for uri in [
-            "http://node:80",
-            "https://node:443",
-            "http://node:1",
-            "https://node:65535/",
-            "http://127.0.0.1:9000",
-            "https://[::1]:443/",
-            "http://[2001:db8::1]:1",
-            "http://NoDe:00080/",
-        ] {
-            let result = CaptureNode::new(uri.to_owned());
-            assert!(result.is_ok(), "valid URI construction: {uri}");
-            let node = result.expect("valid URI");
-            assert_eq!(node.rpc_uri(), uri, "preserve explicit default port and raw URI");
-            assert_eq!(node.encode().expect("encode URI"), record(uri.as_bytes()));
-            assert_eq!(CaptureNode::decode(&record(uri.as_bytes())).expect("decode URI"), node);
-        }
-    }
-    #[test]
-    fn golden_cnod_bytes() {
-        let node = CaptureNode {
-            rpc_uri: "http://node:80".to_owned(),
-        };
-        let golden = b"CNOD\x01\x00\x00\x00\x0ehttp://node:80";
-        assert_eq!(node.encode().ok().as_deref(), Some(golden.as_slice()), "golden CNOD bytes");
-    }
-    #[test]
-    fn valid_cnod_decode() {
-        let golden = b"CNOD\x01\x00\x00\x00\x0ehttp://node:80";
-        let decoded = CaptureNode::decode(golden);
-        assert!(decoded.is_ok(), "golden decode succeeds");
-        assert_eq!(decoded.expect("golden decoded node").rpc_uri(), "http://node:80");
-    }
-    #[test]
-    fn serde_shape_roundtrip_and_strict_fields() {
-        let node = CaptureNode::new("https://node:443/".to_owned()).expect("serde node");
-        let json = r#"{"rpc_uri":"https://node:443/"}"#;
-        assert_eq!(serde_json::to_string(&node).expect("serialize node"), json);
-        assert_eq!(serde_json::from_str::<CaptureNode>(json).expect("deserialize node"), node);
-        for json in [
-            r#"{}"#,
-            r#"{"rpc_uri":1}"#,
-            r#"{"rpc_uri":null}"#,
-            r#"{"rpc_uri":"http://node:1","extra":0}"#,
-            r#"{"rpc_uri":"http://node:1","rpc_uri":"http://node:2"}"#,
-        ] {
-            assert!(serde_json::from_str::<CaptureNode>(json).is_err(), "strict fields: {json}");
-        }
-    }
-    #[test]
-    fn credentials_and_raw_uri_shapes_reject() {
-        for uri in [
-            "http://user@node:1",
-            "http://user:password@node:1",
-            "http://@node:1",
-            "HTTP://node:1",
-            "Https://node:1",
-            "http://node",
-            "http://node:",
-            "http://node:0",
-            "http://node:-1",
-            "http://node:+1",
-            "http://node:65536",
-            "http://node:999999999999999999999",
-            "http://[::1:1",
-            "http://[::1]extra:1",
-            "http://[invalid]:1",
-            "http://::1:1",
-            "http://:1",
-            "/disk/path",
-            "file:///disk/path",
-            "http://node:1/./",
-            "http://node:1/../",
-            "http://node:1/%2e/",
-            "http://node:1/%2E%2e/",
-            "http://node:1/path",
-            "http://node:1\\",
-            " http://node:1",
-            "http://node:1\n",
-            "http://no\tde:1",
-            "http://node:1\u{00a0}",
-            "http://node:1\u{0085}",
-            "http://node:1\0",
-        ] {
-            reject_uri(uri);
-        }
-    }
-    #[test]
-    fn query_and_fragment_markers_reject() {
-        for uri in [
-            "http://node:1?",
-            "http://node:1?x=y",
-            "http://node:1/?",
-            "http://node:1#",
-            "http://node:1#fragment",
-            "http://node:1/#",
-        ] {
-            reject_uri(uri);
-        }
-    }
-    #[test]
-    fn utf8_byte_length_boundaries() {
-        for length in [4095, 4096] {
-            let prefix = "http://node:";
-            let uri = format!("{prefix}{}1", "0".repeat(length - prefix.len() - 1));
-            assert_eq!(uri.len(), length);
-            assert!(url::Url::parse(&uri).is_ok(), "boundary fixture accepted by actual URL parser");
-            let node = CaptureNode::new(uri.clone()).expect("boundary node");
-            assert_eq!(node.rpc_uri(), uri);
-            assert_eq!(
-                CaptureNode::decode(&node.encode().expect("boundary encoding")).expect("boundary decoding"),
-                node
-            );
-        }
-        let uri = format!("http://node:{}1", "0".repeat(4097 - "http://node:".len() - 1));
-        assert_eq!(CaptureNode::new(uri).expect_err("oversize URI").code(), "rpc_uri_too_long");
-        let unicode = format!("http://{}:1", "é".repeat(2044));
-        assert!(unicode.chars().count() < 4096);
-        assert!(unicode.len() > 4096);
-        assert_eq!(CaptureNode::new(unicode).expect_err("UTF-8 byte limit").code(), "rpc_uri_too_long");
-    }
-    #[test]
-    fn cnod_headers_and_lengths_reject_exact_codes() {
-        let valid = record(b"http://node:80");
-        for end in 0..9 {
-            reject_record(&valid[..end], "truncated_node_record");
-        }
-        let mut bad = valid.clone();
-        bad[0] = b'X';
-        reject_record(&bad, "invalid_node_magic");
-        bad = valid.clone();
-        bad[4] = 2;
-        reject_record(&bad, "unsupported_node_version");
-        for length in [0_u32, 4097, u32::MAX] {
-            bad = b"CNOD\x01".to_vec();
-            bad.extend_from_slice(&length.to_be_bytes());
-            reject_record(&bad, "invalid_node_length");
-        }
-        for end in 9..valid.len() {
-            reject_record(&valid[..end], "truncated_node_record");
-        }
-        reject_record(&record(&[0xff]), "invalid_node_utf8");
-        reject_record(&record(&[0xc3, 0x28]), "invalid_node_utf8");
-        reject_record(&record(b"invalid URI"), "invalid_rpc_uri");
-    }
-    #[test]
-    fn trailing_bytes_reject() {
-        for tail in [&b"x"[..], &b"xyz"[..]] {
-            let mut bytes = record(b"http://node:80");
-            bytes.extend_from_slice(tail);
-            reject_record(&bytes, "trailing_node_bytes");
-        }
-    }
-    #[test]
-    fn errors_and_sources_redact_input() {
-        let sentinel = "http://SECRET_USER:SECRET_PASSWORD@node:1";
-        let error = CaptureNode::new(sentinel.to_owned()).expect_err("credentials reject");
-        let mut current: &dyn Error = &error;
-        loop {
-            let rendered = format!("{current} {current:?}");
-            assert!(!rendered.contains("SECRET"));
-            match current.source() {
-                Some(source) => current = source,
-                None => break,
-            }
-        }
-        let json = format!(r#"{{"rpc_uri":"{sentinel}"}}"#);
-        let error = serde_json::from_str::<CaptureNode>(&json).expect_err("serde credentials reject");
-        assert!(!format!("{error} {error:?}").contains("SECRET"));
-        let json = format!(r#"{{"{sentinel}":0}}"#);
-        let error = serde_json::from_str::<CaptureNode>(&json).expect_err("unknown fields reject");
-        assert!(!format!("{error} {error:?}").contains("SECRET"));
-        let error = CaptureNode::new("http://[bad]:1".to_owned()).expect_err("parser rejection");
-        assert!(error.source().is_some(), "retain URL parser source");
-        let error = CaptureNode::decode(&record(&[0xff])).expect_err("UTF-8 rejection");
-        assert!(error.source().is_some(), "retain UTF-8 source");
-    }
-}
@@ -1,43 +0,0 @@
-use crate::{LOGS, META, Store, StoreError, decode_id, decode_log, encode_id};
-use redb::ReadableTable;
-
-impl Store {
-    /// Atomically remove every log entry at or after `from`.
-    pub fn truncate_suffix(&mut self, from: u64) -> Result<(), StoreError> {
-        let tx = self.write_transaction()?;
-        {
-            let mut meta = tx.open_table(META)?;
-            let logs = tx.open_table(LOGS)?;
-            if crate::committed::validate_committed(&meta, &logs)?.is_some_and(|id| from <= id.index) {
-                return Err(StoreError::CommittedLogProtected);
-            }
-            drop(logs);
-            let tail = meta.get("tail")?.map(|v| decode_id(v.value())).transpose()?;
-            if tail.is_none_or(|id| from > id.index) {
-                return Ok(());
-            }
-            let mut logs = tx.open_table(LOGS)?;
-            let new_tail = if from == 0 {
-                None
-            } else {
-                let index = from - 1;
-                let record = logs.get(index)?.ok_or(StoreError::InvalidLogSequence)?;
-                let id = decode_log(record.value())?.id;
-                if id.index != index {
-                    return Err(StoreError::CorruptRecord);
-                }
-                Some(id)
-            };
-            logs.retain(|index, _| index < from)?;
-            match new_tail {
-                Some(id) => {
-                    meta.insert("tail", encode_id(id).as_slice())?;
-                }
-                None => {
-                    meta.remove("tail")?;
-                }
-            }
-        }
-        self.commit(tx)
-    }
-}
@@ -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?;
@@ -371,24 +371,6 @@ pub mod config {
            with_config_object_read_lock, with_config_object_write_lock, with_server_config_read_lock,
            with_server_config_write_lock,
        };
-
-        /// Read config bytes and metadata without treating an empty payload as absent.
-        pub async fn read_config_preserve_empty_with_metadata_opts(
-            api: std::sync::Arc<crate::store::ECStore>,
-            file: &str,
-            opts: &crate::object_api::ObjectOptions,
-        ) -> crate::error::Result<(Vec<u8>, crate::object_api::ObjectInfo)> {
-            crate::config::com::read_config_preserve_empty_with_metadata_opts(api, file, opts).await
-        }
-
-        pub async fn read_config_limited_preserve_empty_with_metadata_opts(
-            api: std::sync::Arc<crate::store::ECStore>,
-            file: &str,
-            opts: &crate::object_api::ObjectOptions,
-            max_bytes: usize,
-        ) -> crate::error::Result<(Vec<u8>, crate::object_api::ObjectInfo)> {
-            crate::config::com::read_config_limited_preserve_empty_with_metadata_opts(api, file, opts, max_bytes).await
-        }
    }

    pub mod storageclass {
@@ -495,22 +495,10 @@ pub(crate) async fn read_config_preserve_empty<S>(api: Arc<S>, file: &str) -> Re
 where
    S: EcstoreObjectIO,
 {
-    let (data, _obj) = read_config_preserve_empty_with_metadata_opts(api, file, &ObjectOptions::default()).await?;
+    let (data, _obj) = read_config_with_metadata_inner(api, file, &ObjectOptions::default(), true, None).await?;
    Ok(data)
 }

-/// Preserve an empty config payload, its metadata, and caller-supplied read options.
-pub(crate) async fn read_config_preserve_empty_with_metadata_opts<S>(
-    api: Arc<S>,
-    file: &str,
-    opts: &ObjectOptions,
-) -> Result<(Vec<u8>, ObjectInfo)>
-where
-    S: EcstoreObjectIO,
-{
-    read_config_with_metadata_inner(api, file, opts, true, None).await
-}
-
 pub async fn read_config_no_lock<S>(api: Arc<S>, file: &str) -> Result<Vec<u8>>
 where
    S: ObjectIO<
@@ -539,13 +527,15 @@ pub(crate) async fn read_config_no_lock_preserve_empty_with_metadata<S>(api: Arc
 where
    S: EcstoreObjectIO,
 {
-    read_config_preserve_empty_with_metadata_opts(
+    read_config_with_metadata_inner(
        api,
        file,
        &ObjectOptions {
            no_lock: true,
            ..Default::default()
        },
+        true,
+        None,
    )
    .await
 }
@@ -2825,10 +2815,9 @@ mod tests {
        config_task_join_error, configs_semantically_equal, decode_server_config_blob, encode_server_config_blob,
        heal_config_descriptor, is_standard_object_server_config, lookup_configs, map_system_metadata_write_error,
        new_and_save_server_config, read_config, read_config_no_lock_preserve_empty_with_metadata, read_config_preserve_empty,
-        read_config_preserve_empty_with_metadata_opts, read_config_with_metadata, read_config_without_migrate,
-        read_server_config_snapshot, save_config_with_opts_inner, save_server_config, save_server_config_snapshot,
-        save_server_config_snapshot_with_generation, server_config_transaction_lock_path, should_warn_ignored_scalar_section,
-        storage_class_kvs_mut,
+        read_config_with_metadata, read_config_without_migrate, read_server_config_snapshot, save_config_with_opts_inner,
+        save_server_config, save_server_config_snapshot, save_server_config_snapshot_with_generation,
+        server_config_transaction_lock_path, should_warn_ignored_scalar_section, storage_class_kvs_mut,
    };
    use crate::config::{audit, heal, notify, oidc, scanner};
    use crate::disk::{RUSTFS_META_BUCKET, endpoint::Endpoint};
@@ -5076,7 +5065,6 @@ mod tests {
        heal_replacement: Option<Vec<u8>>,
        heal_calls: AtomicUsize,
        write_calls: AtomicUsize,
-        last_get_no_lock: AtomicBool,
        last_put_no_lock: AtomicBool,
        last_put_preconditions: Mutex<Option<HTTPPreconditions>>,
        revision: AtomicUsize,
@@ -5092,7 +5080,6 @@ mod tests {
                heal_replacement,
                heal_calls: AtomicUsize::new(0),
                write_calls: AtomicUsize::new(0),
-                last_get_no_lock: AtomicBool::new(false),
                last_put_no_lock: AtomicBool::new(false),
                last_put_preconditions: Mutex::new(None),
                revision: AtomicUsize::new(1),
@@ -5148,9 +5135,8 @@ mod tests {
            _object: &str,
            _range: Option<HTTPRangeSpec>,
            _h: HeaderMap,
-            opts: &ObjectOptions,
+            _opts: &ObjectOptions,
        ) -> Result<GetObjectReader> {
-            self.last_get_no_lock.store(opts.no_lock, Ordering::SeqCst);
            let data = match &*self.state.lock().expect("state lock poisoned") {
                RecoveryReadState::Missing => return Err(Error::ConfigNotFound),
                RecoveryReadState::Blob(data) => data.clone(),
@@ -5445,60 +5431,6 @@ mod tests {
        assert!(data.is_empty());
    }

-    #[tokio::test]
-    async fn read_config_preserve_empty_metadata_keeps_empty_and_options() {
-        let store = Arc::new(RecoveryMockStore::new(RecoveryReadState::Blob(Vec::new()), None));
-        for no_lock in [false, true] {
-            let opts = ObjectOptions {
-                no_lock,
-                ..Default::default()
-            };
-            let (data, info) = read_config_preserve_empty_with_metadata_opts(store.clone(), "config/empty.json", &opts)
-                .await
-                .expect("an existing empty config must retain its payload and metadata");
-            assert!(data.is_empty());
-            assert_eq!(info.size, 0);
-            assert_eq!(info.actual_size, 0);
-            assert_eq!(info.etag.as_deref(), Some("config-1"));
-            assert_eq!(info.data_dir, Some(uuid::Uuid::from_u128(1)));
-            assert_eq!(store.last_get_no_lock.load(Ordering::SeqCst), no_lock);
-        }
-        let err = read_config(store, "config/empty.json")
-            .await
-            .expect_err("the existing reader must still treat empty configs as missing");
-        assert!(matches!(err, Error::ConfigNotFound));
-    }
-
-    #[tokio::test]
-    async fn read_config_preserve_empty_metadata_keeps_payload_and_metadata() {
-        let payload = b"original config bytes".to_vec();
-        let store = Arc::new(RecoveryMockStore::new(RecoveryReadState::Blob(payload.clone()), None));
-        let (data, info) =
-            read_config_preserve_empty_with_metadata_opts(store, "config/nonempty.json", &ObjectOptions::default())
-                .await
-                .expect("a nonempty config must retain its payload and metadata");
-        assert_eq!(data, payload);
-        let size = i64::try_from(payload.len()).expect("test payload size fits i64");
-        assert_eq!(info.size, size);
-        assert_eq!(info.actual_size, size);
-        assert_eq!(info.etag.as_deref(), Some("config-1"));
-        assert_eq!(info.data_dir, Some(uuid::Uuid::from_u128(1)));
-    }
-
-    #[tokio::test]
-    async fn read_config_preserve_empty_metadata_preserves_read_errors() {
-        let missing = Arc::new(RecoveryMockStore::new(RecoveryReadState::Missing, None));
-        let err = read_config_preserve_empty_with_metadata_opts(missing, "config/missing.json", &ObjectOptions::default())
-            .await
-            .expect_err("physical absence must remain ConfigNotFound");
-        assert!(matches!(err, Error::ConfigNotFound));
-        let quorum = Arc::new(RecoveryMockStore::new(RecoveryReadState::QuorumError, None));
-        let err = read_config_preserve_empty_with_metadata_opts(quorum, "config/quorum.json", &ObjectOptions::default())
-            .await
-            .expect_err("a quorum failure must retain its typed read error");
-        assert!(matches!(err, Error::ErasureReadQuorum));
-    }
-
    #[async_trait::async_trait]
    impl StorageAdminApi for RecoveryMockStore {
        type BackendInfo = rustfs_madmin::BackendInfo;
@@ -12477,227 +12477,6 @@ mod test {
        }
    }

-    // SIGKILL tests process death on a healthy filesystem, not power loss.
-    #[cfg(unix)]
-    #[tokio::test]
-    async fn rename_data_process_kill_preserves_publication_boundary() {
-        use std::io::Write as _;
-        use std::os::unix::process::ExitStatusExt;
-        use std::process::{Child, Command, ExitStatus, Stdio};
-        use std::time::{Duration, Instant};
-
-        const TEST: &str = "disk::local::test::rename_data_process_kill_preserves_publication_boundary";
-        const ROLE: &str = "RUSTFS_TEST_PUBLICATION_KILL_ROLE";
-        const ROOT: &str = "RUSTFS_TEST_PUBLICATION_KILL_ROOT";
-        const NONCE: &str = "RUSTFS_TEST_PUBLICATION_KILL_NONCE";
-        const BUCKET: &str = "bucket";
-        const OBJECT: &str = "process-kill-object";
-        const STAGED: &str = "process-kill-staged";
-        const VERSION: Uuid = Uuid::from_u128(0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa);
-        const OLD: Uuid = Uuid::from_u128(0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb);
-        const NEW: Uuid = Uuid::from_u128(0xcccccccccccccccccccccccccccccccc);
-
-        struct ChildGuard(Child);
-        impl Drop for ChildGuard {
-            fn drop(&mut self) {
-                let _ = self.0.kill();
-                let _ = self.0.wait();
-            }
-        }
-
-        async fn wait_exit(child: &mut ChildGuard) -> ExitStatus {
-            let deadline = Instant::now() + Duration::from_secs(60);
-            loop {
-                if child.0.try_wait().expect("poll owned child").is_some() {
-                    return child.0.wait().expect("reap owned child");
-                }
-                assert!(Instant::now() < deadline, "owned child exceeded exit deadline");
-                tokio::time::sleep(Duration::from_millis(20)).await;
-            }
-        }
-
-        fn spawn(root: &std::path::Path, nonce: &str, role: &str) -> ChildGuard {
-            let stdout = std::fs::File::create(root.join(format!("{role}.stdout"))).expect("child stdout file");
-            let stderr = std::fs::File::create(root.join(format!("{role}.stderr"))).expect("child stderr file");
-            ChildGuard(
-                Command::new(std::env::current_exe().expect("test executable"))
-                    .args(["--exact", TEST, "--nocapture"])
-                    .env(ROLE, role)
-                    .env(ROOT, root)
-                    .env(NONCE, nonce)
-                    .stdout(Stdio::from(stdout))
-                    .stderr(Stdio::from(stderr))
-                    .spawn()
-                    .expect("spawn owned publication child"),
-            )
-        }
-
-        fn child_logs(root: &std::path::Path, role: &str) -> String {
-            let stdout = std::fs::read_to_string(root.join(format!("{role}.stdout"))).expect("child stdout");
-            let stderr = std::fs::read_to_string(root.join(format!("{role}.stderr"))).expect("child stderr");
-            format!("{stdout} {stderr}")
-        }
-
-        fn file_info(data_dir: Uuid) -> FileInfo {
-            let mut fi = test_file_info(OBJECT, VERSION, Some(data_dir), None);
-            fi.size = 8;
-            fi.parts[0].size = 8;
-            fi.parts[0].actual_size = 8;
-            fi
-        }
-
-        if let Some(role) = std::env::var_os(ROLE) {
-            let role = role.to_str().expect("child role is UTF-8");
-            assert!(matches!(role, "writer-before" | "writer-after" | "verify-before" | "verify-after"));
-            let root = std::path::PathBuf::from(std::env::var_os(ROOT).expect("private child root"));
-            let nonce = std::env::var(NONCE).expect("private child nonce");
-            Uuid::parse_str(&nonce).expect("nonce is a UUID");
-            assert_eq!(root.file_name().and_then(|name| name.to_str()), Some(nonce.as_str()));
-            assert_eq!(std::fs::read_to_string(root.join("control-nonce")).expect("bound root nonce"), nonce);
-            let endpoint = Endpoint::try_from(root.to_str().expect("private root UTF-8")).expect("private endpoint");
-            let _strict = durability_mode_override::set(DurabilityMode::Strict);
-            let disk = LocalDisk::new(&endpoint, false).await.expect("child opens real disk");
-            let object_dir = root.join(BUCKET).join(OBJECT);
-            if role.starts_with("verify-") {
-                let fi = disk
-                    .read_version("", BUCKET, OBJECT, &VERSION.to_string(), &ReadOptions::default())
-                    .await
-                    .expect("fresh verifier reads actual version");
-                let (expected_dir, expected_bytes) = if role == "verify-before" {
-                    assert_eq!(
-                        fs::read(object_dir.join(STORAGE_FORMAT_FILE))
-                            .await
-                            .expect("read old metadata"),
-                        fs::read(root.join("old-meta")).await.expect("read old fixture"),
-                        "prepublication metadata must remain byte-for-byte old"
-                    );
-                    (OLD, b"old-data")
-                } else {
-                    (NEW, b"new-data")
-                };
-                assert_eq!(fi.data_dir, Some(expected_dir), "{role}: actual reopened version data_dir");
-                let part = object_dir
-                    .join(fi.data_dir.expect("referenced data dir").to_string())
-                    .join("part.1");
-                assert_eq!(fs::read(part).await.expect("read referenced shard"), expected_bytes);
-                return;
-            }
-
-            let staged = root.join(RUSTFS_META_TMP_BUCKET).join(STAGED).join(NEW.to_string());
-            fs::create_dir_all(&staged).await.expect("staged shard directory");
-            fs::write(staged.join("part.1"), b"new-data")
-                .await
-                .expect("stage new shard bytes");
-            let ready = root.join(format!("{role}.ready"));
-            let pause = move || {
-                let pending = ready.with_extension("pending");
-                std::fs::write(&pending, &nonce).expect("write actual writer cut marker");
-                std::fs::rename(pending, &ready).expect("announce complete writer cut marker");
-                let (_sender, receiver) = std::sync::mpsc::channel::<()>();
-                receiver.recv().expect("writer remains paused until SIGKILL");
-            };
-            let _hook = if role == "writer-before" {
-                // Use the executor's exact path spelling, including canonical
-                // macOS roots and Linux mount-pinned I/O roots.
-                let destination = disk
-                    .io_get_object_path(BUCKET, &format!("{OBJECT}/{STORAGE_FORMAT_FILE}"))
-                    .expect("actual metadata publication path");
-                assert_eq!(
-                    std::fs::canonicalize(&destination).expect("canonical publication destination"),
-                    disk.root.join(BUCKET).join(OBJECT).join(STORAGE_FORMAT_FILE),
-                    "hook destination must identify this disk's seeded metadata"
-                );
-                Some(os::prepared_publication_test_hooks::install_at(
-                    os::prepared_publication_test_hooks::Stage::PreparedRename,
-                    &destination,
-                    pause,
-                ))
-            } else {
-                disk.rename_data(RUSTFS_META_TMP_BUCKET, STAGED, file_info(NEW), BUCKET, OBJECT)
-                    .await
-                    .expect("Strict real rename_data completes before cut");
-                pause();
-                return;
-            };
-            disk.rename_data(RUSTFS_META_TMP_BUCKET, STAGED, file_info(NEW), BUCKET, OBJECT)
-                .await
-                .expect("before writer must remain in physical hook");
-            panic!("prepublication hook unexpectedly returned");
-        }
-
-        let temp = tempfile::tempdir().expect("parent owns private disk roots");
-        for phase in ["before", "after"] {
-            let nonce = Uuid::new_v4().to_string();
-            let root = temp.path().join(&nonce);
-            fs::create_dir(&root).await.expect("private scenario root");
-            fs::write(root.join("control-nonce"), &nonce)
-                .await
-                .expect("bind private root");
-            let endpoint = Endpoint::try_from(root.to_str().expect("private root UTF-8")).expect("private endpoint");
-            let disk = LocalDisk::new(&endpoint, false).await.expect("seed real disk");
-            ensure_test_volume(&disk, BUCKET).await;
-            ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await;
-            let object_dir = root.join(BUCKET).join(OBJECT);
-            fs::create_dir_all(object_dir.join(OLD.to_string()))
-                .await
-                .expect("old shard directory");
-            fs::write(object_dir.join(OLD.to_string()).join("part.1"), b"old-data")
-                .await
-                .expect("seed old shard bytes");
-            let old_meta = test_meta(file_info(OLD));
-            fs::write(object_dir.join(STORAGE_FORMAT_FILE), &old_meta)
-                .await
-                .expect("seed old metadata");
-            fs::write(root.join("old-meta"), &old_meta)
-                .await
-                .expect("retain exact old fixture");
-            drop(disk);
-
-            let writer_role = format!("writer-{phase}");
-            let mut writer = spawn(&root, &nonce, &writer_role);
-            let deadline = Instant::now() + Duration::from_secs(60);
-            let ready = root.join(format!("{writer_role}.ready"));
-            loop {
-                if let Some(status) = writer.0.try_wait().expect("poll writer") {
-                    panic!(
-                        "writer exited before cut: {writer_role} status={status}: {}",
-                        child_logs(&root, &writer_role)
-                    );
-                }
-                match fs::read_to_string(&ready).await {
-                    Ok(marker) => {
-                        assert_eq!(marker, nonce, "cut marker belongs to this private writer");
-                        break;
-                    }
-                    Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
-                    Err(error) => panic!("read cut marker: {error}"),
-                }
-                if Instant::now() >= deadline {
-                    panic!("writer cut deadline: {writer_role}: {}", child_logs(&root, &writer_role));
-                }
-                tokio::time::sleep(Duration::from_millis(20)).await;
-            }
-            let pid = writer.0.id();
-            writer.0.kill().expect("SIGKILL owned writer at verified cut");
-            let status = wait_exit(&mut writer).await;
-            assert_eq!(status.signal(), Some(9), "owned writer must actually die by SIGKILL");
-            writeln!(std::io::stdout(), "{writer_role} pid={pid} signal={:?}", status.signal())
-                .expect("record actual writer wait status");
-
-            let verifier_role = format!("verify-{phase}");
-            let mut verifier = spawn(&root, &nonce, &verifier_role);
-            let status = wait_exit(&mut verifier).await;
-            let logs = child_logs(&root, &verifier_role);
-            assert!(status.success(), "{verifier_role} status={status}: {logs}");
-            let stdout = std::fs::read_to_string(root.join(format!("{verifier_role}.stdout"))).expect("verifier stdout");
-            assert!(
-                stdout.contains(TEST) && stdout.contains("1 passed"),
-                "verifier must execute exact test: {logs}"
-            );
-            writeln!(std::io::stdout(), "{verifier_role} status={status}").expect("record actual verifier exit");
-        }
-    }
-
    /// Crash-consistency harness for the rename_data commit sequence
    /// (rustfs/backlog#935 HP-14, test plan rustfs/backlog#896; hard rule from
    /// rustfs/backlog#878: "After partial commit, the object can only be an old or new version; it cannot be mixed").
@@ -9221,7 +9221,7 @@ mod test {
        };
        let snapshot = PersistentListMetadataObject::from_object_info(&object);
        assert!(snapshot.to_object_info("bucket").etag.is_none());
-        let mut old_snapshot = snapshot.clone();
+        let mut old_snapshot = snapshot;
        old_snapshot.etag = object.etag.clone();
        let old_contents = format!(
            "# rustfs-listobjects-key-only-v2\n{}\n",
@@ -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;
@@ -3345,23 +3345,27 @@ mod tests {

    #[test]
    fn test_extract_jwt_claims_basic() {
-        let user_identity = UserIdentity::new(Credentials {
-            access_key: "test-access-key".to_string(),
-            secret_key: "test-secret-key".to_string(),
-            session_token: "invalid-token".to_string(), // Invalid token for testing error handling
-            expiration: None,
-            status: "enabled".to_string(),
-            parent_user: "".to_string(),
-            groups: None,
-            claims: Some({
-                let mut claims = HashMap::new();
-                claims.insert("sub".to_string(), json!("test-user"));
-                claims.insert("aud".to_string(), json!("test-audience"));
-                claims
-            }),
-            name: None,
-            description: None,
-        });
+        let user_identity = UserIdentity {
+            version: 1,
+            credentials: Credentials {
+                access_key: "test-access-key".to_string(),
+                secret_key: "test-secret-key".to_string(),
+                session_token: "invalid-token".to_string(), // Invalid token for testing error handling
+                expiration: None,
+                status: "enabled".to_string(),
+                parent_user: "".to_string(),
+                groups: None,
+                claims: Some({
+                    let mut claims = HashMap::new();
+                    claims.insert("sub".to_string(), json!("test-user"));
+                    claims.insert("aud".to_string(), json!("test-audience"));
+                    claims
+                }),
+                name: None,
+                description: None,
+            },
+            update_at: Some(OffsetDateTime::now_utc()),
+        };

        let result = extract_jwt_claims(&user_identity);
        // In test environment without proper JWT setup, this should fail
@@ -3370,18 +3374,22 @@ mod tests {

    #[test]
    fn test_extract_jwt_claims_no_claims() {
-        let user_identity = UserIdentity::new(Credentials {
-            access_key: "test-access-key".to_string(),
-            secret_key: "test-secret-key".to_string(),
-            session_token: "".to_string(), // Empty token
-            expiration: None,
-            status: "enabled".to_string(),
-            parent_user: "".to_string(),
-            groups: None,
-            claims: None,
-            name: None,
-            description: None,
-        });
+        let user_identity = UserIdentity {
+            version: 1,
+            credentials: Credentials {
+                access_key: "test-access-key".to_string(),
+                secret_key: "test-secret-key".to_string(),
+                session_token: "".to_string(), // Empty token
+                expiration: None,
+                status: "enabled".to_string(),
+                parent_user: "".to_string(),
+                groups: None,
+                claims: None,
+                name: None,
+                description: None,
+            },
+            update_at: Some(OffsetDateTime::now_utc()),
+        };

        let result = extract_jwt_claims(&user_identity);
        // Should fail with empty session token
@@ -3440,7 +3448,11 @@ mod tests {
            description: None,
        };

-        let user_identity = UserIdentity::new(credentials);
+        let user_identity = UserIdentity {
+            version: 1,
+            credentials,
+            update_at: Some(OffsetDateTime::now_utc()),
+        };

        // Test basic structure
        assert_eq!(user_identity.version, 1);
@@ -91,7 +91,6 @@ pub(crate) mod crate_boundary {

 pub(crate) mod object_store {
    pub(crate) use super::storage_contracts::{HTTPPreconditions, ListOperations, ObjectInfoOrErr, ObjectOperations};
-    pub(crate) use rustfs_ecstore::api::config::com::read_config_preserve_empty_with_metadata_opts;
 }

 pub(crate) mod runtime {
@@ -42,7 +42,6 @@ use tracing::{debug, error, warn};

 use crate::storage_api::object_store::{
    HTTPPreconditions, ListOperations, ObjectInfoOrErr as StorageObjectInfoOrErr, ObjectOperations,
-    read_config_preserve_empty_with_metadata_opts,
 };

 pub static IAM_CONFIG_PREFIX: LazyLock<String> = LazyLock::new(|| format!("{IAM_CONFIG_ROOT_PREFIX}/iam"));
@@ -415,14 +414,9 @@ impl ObjectStore {
        &self,
        path: impl AsRef<str> + Send,
        mode: LoadMode,
-        preserve_empty: bool,
    ) -> Result<(Vec<u8>, IamObjectInfo)> {
        let path_ref = path.as_ref();
-        let (data, obj) = if preserve_empty {
-            read_config_preserve_empty_with_metadata_opts(self.object_api.clone(), path_ref, &mode.read_opts()).await?
-        } else {
-            read_iam_config_with_metadata(self.object_api.clone(), path_ref, &mode.read_opts()).await?
-        };
+        let (data, obj) = read_iam_config_with_metadata(self.object_api.clone(), path_ref, &mode.read_opts()).await?;
        let outcome = Self::decrypt_data_with_source(&data)?;
        self.maybe_schedule_lazy_rewrite(path_ref, &outcome, &obj);

@@ -620,9 +614,8 @@ impl ObjectStore {
    /// Parameterized core of [`Store::load_user_identity`].
    async fn load_user_identity_with(&self, name: &str, user_type: UserType, mode: LoadMode) -> Result<UserIdentity> {
        let mut u: UserIdentity = self
-            .load_iamconfig_bytes_with_metadata(get_user_identity_path(name, user_type), mode, true)
+            .load_iam_config_with(get_user_identity_path(name, user_type), mode)
            .await
-            .and_then(|(data, _)| serde_json::from_slice(&data).map_err(Error::from))
            .map_err(|err| {
                if is_err_config_not_found(&err) {
                    debug!(name = %MaskedAccessKey(name), user_type = ?user_type, "IAM user identity missing");
@@ -702,7 +695,7 @@ impl ObjectStore {
    /// Parameterized core of [`Store::load_policy`].
    async fn load_policy_with(&self, name: &str, mode: LoadMode) -> Result<PolicyDoc> {
        let (data, obj) = self
-            .load_iamconfig_bytes_with_metadata(get_policy_doc_path(name), mode, false)
+            .load_iamconfig_bytes_with_metadata(get_policy_doc_path(name), mode)
            .await
            .map_err(|err| {
                if is_err_config_not_found(&err) {
@@ -2667,7 +2667,11 @@ mod tests {
                    name: None,
                    description: None,
                };
-                let parent_identity = UserIdentity::new(creds);
+                let parent_identity = UserIdentity {
+                    version: 1,
+                    credentials: creds,
+                    update_at: Some(OffsetDateTime::now_utc()),
+                };
                let mut users = HashMap::new();
                users.insert(PARENT_USER.to_string(), parent_identity);

@@ -2699,7 +2703,11 @@ mod tests {
                name: None,
                description: None,
            };
-            let parent_identity = UserIdentity::new(creds);
+            let parent_identity = UserIdentity {
+                version: 1,
+                credentials: creds,
+                update_at: Some(OffsetDateTime::now_utc()),
+            };
            let mut users = HashMap::new();
            users.insert(PARENT_USER.to_string(), parent_identity);

@@ -2812,12 +2820,16 @@ mod tests {
    async fn group_writes_stamped_before_the_cache_load_time_still_publish() {
        let iam_sys = test_iam_sys().await;
        let member = "group-stamp-member";
-        let identity = UserIdentity::new(Credentials {
-            access_key: member.to_string(),
-            secret_key: "longenoughsecret".to_string(),
-            status: "on".to_string(),
-            ..Default::default()
-        });
+        let identity = UserIdentity {
+            version: 1,
+            credentials: Credentials {
+                access_key: member.to_string(),
+                secret_key: "longenoughsecret".to_string(),
+                status: "on".to_string(),
+                ..Default::default()
+            },
+            update_at: Some(OffsetDateTime::now_utc()),
+        };
        iam_sys.store.cache.with_write_lock(|cache| {
            cache.add_or_update_user(member, &identity, OffsetDateTime::now_utc());
            // The startup load publishes every entity with the load time.
@@ -22,11 +22,7 @@
 #[allow(unused_imports)]
 pub(crate) mod fixture {
    pub(crate) use rustfs_ecstore::api::bucket::migration::try_migrate_iam_config;
-    pub(crate) use rustfs_ecstore::api::config::com::save_config;
-    pub(crate) use rustfs_ecstore::api::global::set_object_store_resolver;
    pub(crate) use rustfs_ecstore::api::layout::SetupType;
-    pub(crate) use rustfs_ecstore::api::runtime::{InstanceContext, first_cluster_node_is_local, object_store_handle};
-    pub(crate) use rustfs_ecstore::api::storage::ECStore;

    // `update_erasure_type` is a write-side global facade entry. Its use is
    // restricted to reviewed storage_api boundaries; this test-only module is
@@ -33,21 +33,12 @@

 mod ecstore_test_compat;

-use ecstore_test_compat::fixture::{
-    ECStore, InstanceContext, SetupType, first_cluster_node_is_local, object_store_handle, save_config,
-    set_object_store_resolver, update_erasure_type,
-};
-use rustfs_credentials::{Credentials, IAM_POLICY_CLAIM_NAME_SA};
+use ecstore_test_compat::fixture::{SetupType, update_erasure_type};
 use rustfs_iam::cache::Cache;
-use rustfs_iam::error::Error as IamError;
-use rustfs_iam::manager::{IamCache, IamState};
-use rustfs_iam::store::object::{IAM_CONFIG_POLICIES_PREFIX, IAM_CONFIG_USERS_PREFIX, ObjectStore};
-use rustfs_iam::store::{GroupInfo, Store, UserType};
-use rustfs_policy::auth::UserIdentity;
+use rustfs_iam::store::object::ObjectStore;
+use rustfs_iam::store::{GroupInfo, Store};
 use serial_test::serial;
 use std::collections::HashMap;
-use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU8, AtomicU64};
-use std::sync::{Arc, Mutex, OnceLock, Weak};

 const TEST_GROUP: &str = "seq-restart-group";
 const TEST_MEMBERS: [&str; 2] = ["alice", "bob"];
@@ -78,8 +69,7 @@ async fn load_all_bypasses_namespace_lock_quorum() {
            .build()
            .await
            .ecstore;
-        let _route_guard = TestStoreRouteGuard::new(&ecstore);
-        let store = ObjectStore::new(ecstore.clone());
+        let store = ObjectStore::new(ecstore);

        // Seed IAM data while namespace locks still work (single-node mode).
        store
@@ -138,447 +128,3 @@ async fn load_all_bypasses_namespace_lock_quorum() {
    })
    .await;
 }
-
-const IDENTITY_PARENT: &str = "fixture-parent";
-const IDENTITY_CHILD: &str = "fixture-service-child";
-
-static TEST_STORE_ROUTE: Mutex<Option<Weak<ECStore>>> = Mutex::new(None);
-static TEST_STORE_RESOLVER: OnceLock<()> = OnceLock::new();
-
-struct TestStoreRouteGuard;
-
-impl TestStoreRouteGuard {
-    fn new(ecstore: &Arc<ECStore>) -> Self {
-        TEST_STORE_RESOLVER.get_or_init(|| {
-            assert!(
-                set_object_store_resolver(Arc::new(|| {
-                    TEST_STORE_ROUTE
-                        .lock()
-                        .expect("lock test store route")
-                        .as_ref()
-                        .and_then(Weak::upgrade)
-                })),
-                "test binary must own its resolver registration"
-            );
-        });
-        let mut route = TEST_STORE_ROUTE.lock().expect("lock test store route");
-        assert!(route.is_none(), "another test store route is active");
-        *route = Some(Arc::downgrade(ecstore));
-        Self
-    }
-}
-
-impl Drop for TestStoreRouteGuard {
-    fn drop(&mut self) {
-        *TEST_STORE_ROUTE.lock().expect("lock test store route for cleanup") = None;
-    }
-}
-
-struct IdentityStoreFixture {
-    previous_store: Option<Arc<ECStore>>,
-    ecstore: Arc<ECStore>,
-    manager: IamCache<ObjectStore>,
-    _route_guard: TestStoreRouteGuard,
-    _temp_dir: tempfile::TempDir,
-}
-
-impl IdentityStoreFixture {
-    async fn new() -> Self {
-        let temp_dir = tempfile::TempDir::with_prefix("rustfs_iam_identity_fixture_").expect("create identity fixture directory");
-        let ctx = Arc::new(InstanceContext::new());
-        let env = rustfs_test_utils::TestECStoreEnv::builder()
-            .base_dir(temp_dir.path())
-            .init_bucket_metadata(false)
-            .instance_ctx(ctx.clone())
-            .build()
-            .await;
-        ctx.set_endpoints(env.endpoint_pools.clone());
-        let ecstore = env.ecstore;
-        let previous_store = object_store_handle();
-        let route_guard = TestStoreRouteGuard::new(&ecstore);
-        let selected = object_store_handle().expect("fixture route selects its ECStore");
-        assert!(Arc::ptr_eq(&selected, &ecstore), "fixture route must select the isolated ECStore");
-        drop(selected);
-        assert!(first_cluster_node_is_local().await, "isolated fixture has a local first endpoint");
-        temp_env::async_with_vars([("RUSTFS_SKIP_BACKGROUND_TASK", Some("1"))], rustfs_iam::build_iam_sys(ecstore.clone()))
-            .await
-            .expect("initialize IAM metadata through production owner");
-        let store = ObjectStore::new(ecstore.clone());
-        let parent = UserIdentity::new(Credentials {
-            access_key: IDENTITY_PARENT.to_string(),
-            secret_key: "fixture-parent-secret".to_string(),
-            expiration: None,
-            session_token: String::new(),
-            status: "on".to_string(),
-            ..Default::default()
-        });
-        let child = UserIdentity::new(Credentials {
-            access_key: IDENTITY_CHILD.to_string(),
-            secret_key: "fixture-child-secret".to_string(),
-            expiration: None,
-            session_token: String::new(),
-            status: "on".to_string(),
-            parent_user: IDENTITY_PARENT.to_string(),
-            claims: Some(HashMap::from([(
-                IAM_POLICY_CLAIM_NAME_SA.to_string(),
-                serde_json::json!("inherited-policy"),
-            )])),
-            ..Default::default()
-        });
-        assert!(child.credentials.is_service_account(), "child fixture must be a service account");
-        store
-            .save_user_identity(IDENTITY_PARENT, UserType::Reg, parent, None)
-            .await
-            .expect("persist parent identity");
-        store
-            .save_user_identity(IDENTITY_CHILD, UserType::Svc, child, None)
-            .await
-            .expect("persist service child identity");
-        let manager = IamCache {
-            cache: Cache::default(),
-            api: store,
-            state: Arc::new(AtomicU8::new(IamState::Ready as u8)),
-            loading: Arc::new(AtomicBool::new(false)),
-            roles: HashMap::new(),
-            send_chan: tokio::sync::mpsc::channel::<i64>(1).0,
-            last_timestamp: AtomicI64::new(0),
-            sync_failures: AtomicU64::new(0),
-            sync_successes: AtomicU64::new(0),
-            last_sync_duration_millis: AtomicU64::new(0),
-        };
-        manager
-            .load_user(IDENTITY_PARENT)
-            .await
-            .expect("load persisted parent into cache");
-        manager
-            .load_user(IDENTITY_CHILD)
-            .await
-            .expect("load persisted service child into cache");
-        Self {
-            previous_store,
-            ecstore,
-            manager,
-            _route_guard: route_guard,
-            _temp_dir: temp_dir,
-        }
-    }
-
-    async fn assert_parent_and_child(&self) {
-        let parent = self
-            .manager
-            .api
-            .load_user_identity(IDENTITY_PARENT, UserType::Reg)
-            .await
-            .expect("read parent from real storage");
-        assert_eq!(parent.credentials.access_key, IDENTITY_PARENT);
-        assert_eq!(parent.credentials.parent_user, "");
-        let child = self
-            .manager
-            .api
-            .load_user_identity(IDENTITY_CHILD, UserType::Svc)
-            .await
-            .expect("read child from real storage");
-        assert_eq!(child.credentials.access_key, IDENTITY_CHILD);
-        assert_eq!(child.credentials.parent_user, IDENTITY_PARENT);
-        assert!(child.credentials.is_service_account());
-        let cached_parent = self.manager.get_user(IDENTITY_PARENT).await.expect("parent remains in cache");
-        assert_eq!(cached_parent.credentials.access_key, IDENTITY_PARENT);
-        assert_eq!(cached_parent.credentials.parent_user, "");
-        let cached_child = self
-            .manager
-            .get_user(IDENTITY_CHILD)
-            .await
-            .expect("service child remains in cache");
-        assert_eq!(cached_child.credentials.access_key, IDENTITY_CHILD);
-        assert_eq!(cached_child.credentials.parent_user, IDENTITY_PARENT);
-        assert!(cached_child.credentials.is_service_account());
-    }
-
-    async fn overwrite_parent_bytes(&self, bytes: &[u8]) {
-        let path = format!("{}{IDENTITY_PARENT}/identity.json", IAM_CONFIG_USERS_PREFIX.as_str());
-        save_config(self.ecstore.clone(), &path, bytes.to_vec())
-            .await
-            .expect("overwrite real parent identity bytes");
-    }
-
-    async fn assert_child_storage_and_identity_cache(&self) {
-        let child = self
-            .manager
-            .api
-            .load_user_identity(IDENTITY_CHILD, UserType::Svc)
-            .await
-            .expect("notification must preserve persisted service child");
-        assert_eq!(child.credentials.access_key, IDENTITY_CHILD);
-        assert_eq!(child.credentials.parent_user, IDENTITY_PARENT);
-        assert!(child.credentials.is_service_account());
-        let parent = self
-            .manager
-            .get_user(IDENTITY_PARENT)
-            .await
-            .expect("failed notification preserves cached parent");
-        assert_eq!(parent.credentials.access_key, IDENTITY_PARENT);
-        assert_eq!(parent.credentials.parent_user, "");
-        let child = self
-            .manager
-            .get_user(IDENTITY_CHILD)
-            .await
-            .expect("failed notification preserves cached child");
-        assert_eq!(child.credentials.access_key, IDENTITY_CHILD);
-        assert_eq!(child.credentials.parent_user, IDENTITY_PARENT);
-        assert!(child.credentials.is_service_account());
-    }
-}
-
-#[tokio::test(flavor = "multi_thread")]
-#[serial]
-async fn identity_store_fixture_roundtrips_parent_and_service_child() {
-    let fixture = IdentityStoreFixture::new().await;
-    let before = fixture.previous_store.clone();
-    let selected = object_store_handle().expect("fixture route selects its ECStore");
-    assert!(Arc::ptr_eq(&selected, &fixture.ecstore));
-    drop(selected);
-    fixture.assert_parent_and_child().await;
-    drop(fixture);
-    assert!(TEST_STORE_ROUTE.lock().expect("inspect cleared test store route").is_none());
-    match (before, object_store_handle()) {
-        (None, None) => {}
-        (Some(before), Some(after)) => assert!(Arc::ptr_eq(&before, &after), "fixture drop restores prior store identity"),
-        _ => panic!("fixture drop must restore the prior store option"),
-    }
-}
-
-#[tokio::test(flavor = "multi_thread")]
-#[serial]
-async fn identity_notification_valid_parent_preserves_child_storage_and_cache() {
-    let fixture = IdentityStoreFixture::new().await;
-    fixture.assert_parent_and_child().await;
-    fixture
-        .manager
-        .user_notification_handler(IDENTITY_PARENT, UserType::Reg)
-        .await
-        .expect("refresh valid parent notification");
-    fixture.assert_parent_and_child().await;
-}
-
-#[tokio::test(flavor = "multi_thread")]
-#[serial]
-async fn identity_store_fixture_raw_valid_overwrite_is_observed() {
-    let fixture = IdentityStoreFixture::new().await;
-    let parent = fixture
-        .manager
-        .api
-        .load_user_identity(IDENTITY_PARENT, UserType::Reg)
-        .await
-        .expect("load valid parent before raw overwrite");
-    let mut raw = serde_json::to_value(parent).expect("serialize valid parent fixture");
-    raw["credentials"]["name"] = serde_json::json!("updated-name");
-    let path = format!("{}{IDENTITY_PARENT}/identity.json", IAM_CONFIG_USERS_PREFIX.as_str());
-    save_config(
-        fixture.ecstore.clone(),
-        &path,
-        serde_json::to_vec(&raw).expect("encode raw parent fixture"),
-    )
-    .await
-    .expect("overwrite parent through real config storage");
-    fixture
-        .manager
-        .load_user(IDENTITY_PARENT)
-        .await
-        .expect("reload parent after raw overwrite");
-    let stored = fixture
-        .manager
-        .api
-        .load_user_identity(IDENTITY_PARENT, UserType::Reg)
-        .await
-        .expect("read overwritten parent from storage");
-    assert_eq!(stored.credentials.name.as_deref(), Some("updated-name"));
-    let cached = fixture
-        .manager
-        .get_user(IDENTITY_PARENT)
-        .await
-        .expect("read overwritten parent from cache");
-    assert_eq!(cached.credentials.name.as_deref(), Some("updated-name"));
-    fixture.assert_parent_and_child().await;
-}
-
-#[tokio::test(flavor = "multi_thread")]
-#[serial]
-async fn identity_corrupt_reads_preserve_crypto_error() {
-    let fixture = IdentityStoreFixture::new().await;
-    fixture.overwrite_parent_bytes(&[0xff, 0xfe, 0xfd]).await;
-    let locked = fixture
-        .manager
-        .api
-        .load_user_identity(IDENTITY_PARENT, UserType::Reg)
-        .await
-        .expect_err("corrupt stored identity must fail locked read");
-    assert!(matches!(locked, IamError::CryptoError(_)), "preserve crypto error, got {locked:?}");
-    let mut users = HashMap::new();
-    let no_lock = fixture
-        .manager
-        .api
-        .load_user_no_lock(IDENTITY_PARENT, UserType::Reg, &mut users)
-        .await
-        .expect_err("corrupt stored identity must fail no-lock read");
-    assert!(matches!(no_lock, IamError::CryptoError(_)), "preserve crypto error, got {no_lock:?}");
-    assert!(users.is_empty(), "failed no-lock read must not populate users");
-}
-
-#[tokio::test(flavor = "multi_thread")]
-#[serial]
-async fn identity_corrupt_notification_preserves_child_storage_and_cache() {
-    let fixture = IdentityStoreFixture::new().await;
-    fixture.assert_parent_and_child().await;
-    fixture.overwrite_parent_bytes(&[0xff, 0xfe, 0xfd]).await;
-    let notification = fixture
-        .manager
-        .user_notification_handler(IDENTITY_PARENT, UserType::Reg)
-        .await;
-    fixture.assert_child_storage_and_identity_cache().await;
-    let error = notification.expect_err("corrupt parent notification must return an error");
-    assert!(
-        matches!(error, IamError::CryptoError(_)),
-        "preserve notification crypto error, got {error:?}"
-    );
-}
-
-#[tokio::test(flavor = "multi_thread")]
-#[serial]
-async fn identity_invalid_schema_preserves_decode_error() {
-    let fixture = IdentityStoreFixture::new().await;
-    fixture.assert_parent_and_child().await;
-    fixture
-        .overwrite_parent_bytes(br#"{"version":"not-an-integer","credentials":{}}"#)
-        .await;
-    let error = fixture
-        .manager
-        .api
-        .load_user_identity(IDENTITY_PARENT, UserType::Reg)
-        .await
-        .expect_err("invalid identity field type must fail decode");
-    assert_identity_data_error(error);
-    let mut users = HashMap::new();
-    let error = fixture
-        .manager
-        .api
-        .load_user_no_lock(IDENTITY_PARENT, UserType::Reg, &mut users)
-        .await
-        .expect_err("invalid identity field type must fail no-lock decode");
-    assert_identity_data_error(error);
-    assert!(users.is_empty(), "failed decode must not populate users");
-    let notification = fixture
-        .manager
-        .user_notification_handler(IDENTITY_PARENT, UserType::Reg)
-        .await;
-    fixture.assert_child_storage_and_identity_cache().await;
-    assert_identity_data_error(notification.expect_err("invalid parent notification must return decode error"));
-}
-
-fn assert_identity_data_error(error: IamError) {
-    let IamError::Io(inner) = error else {
-        panic!("identity decode must preserve Io serde error, got {error:?}");
-    };
-    let cause = inner
-        .get_ref()
-        .and_then(|cause| cause.downcast_ref::<serde_json::Error>())
-        .expect("identity decode error retains serde source");
-    assert!(cause.is_data(), "identity field type must produce serde Data error");
-}
-
-#[tokio::test(flavor = "multi_thread")]
-#[serial]
-async fn identity_physically_missing_is_no_such_user() {
-    let fixture = IdentityStoreFixture::new().await;
-    let name = "physically-missing-identity";
-    let error = fixture
-        .manager
-        .api
-        .load_user_identity(name, UserType::Reg)
-        .await
-        .expect_err("missing identity must be absent");
-    assert!(
-        matches!(error, IamError::NoSuchUser(ref missing) if missing == name),
-        "preserve absent name, got {error:?}"
-    );
-    let mut users = HashMap::new();
-    let error = fixture
-        .manager
-        .api
-        .load_user_no_lock(name, UserType::Reg, &mut users)
-        .await
-        .expect_err("missing no-lock identity must be absent");
-    assert!(
-        matches!(error, IamError::NoSuchUser(ref missing) if missing == name),
-        "preserve absent name, got {error:?}"
-    );
-    assert!(users.is_empty(), "missing read must not populate users");
-}
-
-#[tokio::test(flavor = "multi_thread")]
-#[serial]
-async fn identity_empty_reads_preserve_crypto_error() {
-    let fixture = IdentityStoreFixture::new().await;
-    fixture.overwrite_parent_bytes(&[]).await;
-    let locked = fixture
-        .manager
-        .api
-        .load_user_identity(IDENTITY_PARENT, UserType::Reg)
-        .await
-        .expect_err("empty stored identity must fail locked read");
-    assert!(matches!(locked, IamError::CryptoError(_)), "preserve crypto error, got {locked:?}");
-    let mut users = HashMap::new();
-    let no_lock = fixture
-        .manager
-        .api
-        .load_user_no_lock(IDENTITY_PARENT, UserType::Reg, &mut users)
-        .await
-        .expect_err("empty stored identity must fail no-lock read");
-    assert!(matches!(no_lock, IamError::CryptoError(_)), "preserve crypto error, got {no_lock:?}");
-    assert!(users.is_empty(), "failed no-lock read must not populate users");
-}
-
-#[tokio::test(flavor = "multi_thread")]
-#[serial]
-async fn identity_empty_notification_preserves_child_storage_and_cache() {
-    let fixture = IdentityStoreFixture::new().await;
-    fixture.assert_parent_and_child().await;
-    fixture.overwrite_parent_bytes(&[]).await;
-    let notification = fixture
-        .manager
-        .user_notification_handler(IDENTITY_PARENT, UserType::Reg)
-        .await;
-    fixture.assert_child_storage_and_identity_cache().await;
-    let error = notification.expect_err("empty parent notification must return an error");
-    assert!(
-        matches!(error, IamError::CryptoError(_)),
-        "preserve notification crypto error, got {error:?}"
-    );
-}
-
-#[tokio::test(flavor = "multi_thread")]
-#[serial]
-async fn policy_empty_payload_retains_no_such_policy() {
-    let fixture = IdentityStoreFixture::new().await;
-    let name = "empty-policy";
-    let path = format!("{}{name}/policy.json", IAM_CONFIG_POLICIES_PREFIX.as_str());
-    save_config(fixture.ecstore.clone(), &path, Vec::new())
-        .await
-        .expect("save real empty policy payload");
-    let locked = fixture
-        .manager
-        .api
-        .load_policy(name)
-        .await
-        .err()
-        .expect("empty policy must retain absent-policy behavior");
-    assert!(matches!(locked, IamError::NoSuchPolicy), "preserve policy error, got {locked:?}");
-    let mut policies = HashMap::new();
-    let no_lock = fixture
-        .manager
-        .api
-        .load_policy_doc_no_lock(name, &mut policies)
-        .await
-        .expect_err("empty no-lock policy must retain absent-policy behavior");
-    assert!(matches!(no_lock, IamError::NoSuchPolicy), "preserve policy error, got {no_lock:?}");
-    assert!(policies.is_empty(), "failed no-lock policy read must not populate policies");
-}
@@ -58,7 +58,6 @@ tokio = { workspace = true, features = ["macros", "rt-multi-thread", "time"] }
 time = { workspace = true, features = ["serde", "parsing", "formatting", "macros"] }
 serde = { workspace = true, features = ["derive", "rc"] }
 serde_json = { workspace = true, features = ["raw_value"] }
-uuid = { workspace = true, features = ["serde"] }
 thiserror.workspace = true
 strum = { workspace = true, features = ["derive"] }
 rustfs-crypto = { workspace = true }
@@ -22,33 +22,6 @@ use serde_json::json;
 use std::collections::HashMap;
 use time::OffsetDateTime;

-/// A nonnil stored principal identifier; parsing it does not establish identity authority.
-#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
-#[serde(try_from = "uuid::Uuid", into = "uuid::Uuid")]
-pub struct PrincipalIncarnation(uuid::Uuid);
-
-#[derive(Debug, thiserror::Error)]
-#[error("principal incarnation must not be nil")]
-pub struct InvalidPrincipalIncarnation;
-
-impl TryFrom<uuid::Uuid> for PrincipalIncarnation {
-    type Error = InvalidPrincipalIncarnation;
-
-    fn try_from(value: uuid::Uuid) -> Result<Self, Self::Error> {
-        if value.is_nil() {
-            Err(InvalidPrincipalIncarnation)
-        } else {
-            Ok(Self(value))
-        }
-    }
-}
-
-impl From<PrincipalIncarnation> for uuid::Uuid {
-    fn from(value: PrincipalIncarnation) -> Self {
-        value.0
-    }
-}
-
 #[derive(Serialize, Deserialize, Clone, Debug, Default)]
 pub struct UserIdentity {
    pub version: i64,
@@ -56,8 +29,6 @@ pub struct UserIdentity {
    /// updatedAt (RFC3339), legacy RustFS: update_at. Serialize as updatedAt
    #[serde(rename = "updatedAt", alias = "update_at", default, with = "crate::serde_datetime::option")]
    pub update_at: Option<OffsetDateTime>,
-    #[serde(default, rename = "principalIncarnation", skip_serializing_if = "Option::is_none")]
-    pub principal_incarnation: Option<PrincipalIncarnation>,
 }

 impl UserIdentity {
@@ -73,7 +44,6 @@ impl UserIdentity {
            version: 1,
            credentials,
            update_at: Some(OffsetDateTime::now_utc()),
-            principal_incarnation: None,
        }
    }

@@ -103,124 +73,13 @@ impl From<Credentials> for UserIdentity {
            version: 1,
            credentials: value,
            update_at: Some(OffsetDateTime::now_utc()),
-            principal_incarnation: None,
        }
    }
 }

 #[cfg(test)]
 mod tests {
-    use super::{PrincipalIncarnation, UserIdentity};
-    use rustfs_credentials::Credentials;
-    use serde_json::json;
-    use uuid::Uuid;
-
-    const INCARNATION: &str = "8f422eba-aab4-4ec3-a7ec-d1717463b520";
-
-    #[test]
-    fn test_user_identity_legacy_fields_absent() {
-        let mut legacy = json!({"version": 1, "credentials": {"accessKey": "ak", "secretKey": "sk12345678"}});
-        let identity: UserIdentity = serde_json::from_value(legacy.clone()).expect("decode legacy identity");
-        assert!(identity.principal_incarnation.is_none());
-        legacy["principalIncarnation"] = serde_json::Value::Null;
-        let identity: UserIdentity = serde_json::from_value(legacy).expect("decode null incarnation");
-        assert!(identity.principal_incarnation.is_none());
-    }
-
-    #[test]
-    fn test_user_identity_legacy_serialization_shape() {
-        let identity = UserIdentity::default();
-        let serialized = serde_json::to_value(&identity).expect("serialize legacy identity");
-        assert_eq!(serialized, json!({"version": 0, "credentials": identity.credentials, "updatedAt": null}));
-        assert!(serialized.get("principalIncarnation").is_none());
-    }
-
-    #[test]
-    fn test_user_identity_constructors_do_not_mint_incarnation() {
-        let default = UserIdentity::default();
-        assert!(default.principal_incarnation.is_none());
-        assert_eq!(default.version, 0);
-        assert!(default.update_at.is_none());
-        let credentials = Credentials {
-            access_key: "ak".to_string(),
-            secret_key: "sk12345678".to_string(),
-            ..Default::default()
-        };
-        let expected_credentials = serde_json::to_value(&credentials).expect("serialize expected credentials");
-        for identity in [UserIdentity::new(credentials.clone()), UserIdentity::from(credentials)] {
-            assert!(identity.principal_incarnation.is_none());
-            assert_eq!(identity.version, 1);
-            assert!(identity.update_at.is_some());
-            assert_eq!(
-                serde_json::to_value(&identity.credentials).expect("serialize constructor credentials"),
-                expected_credentials
-            );
-            assert!(
-                serde_json::to_value(identity)
-                    .expect("serialize constructor identity")
-                    .get("principalIncarnation")
-                    .is_none()
-            );
-        }
-    }
-
-    #[test]
-    fn test_principal_incarnation_rejects_nil_and_malformed() {
-        assert!(PrincipalIncarnation::try_from(Uuid::nil()).is_err());
-        for value in ["00000000-0000-0000-0000-000000000000", "not-a-uuid", ""] {
-            assert!(serde_json::from_value::<PrincipalIncarnation>(json!(value)).is_err(), "reject {value}");
-            let identity = json!({"version": 1, "credentials": {}, "principalIncarnation": value});
-            assert!(serde_json::from_value::<UserIdentity>(identity).is_err(), "identity rejects {value}");
-        }
-    }
-
-    #[test]
-    fn test_principal_incarnation_roundtrip_preserves_uuid() {
-        let uuid = Uuid::parse_str(INCARNATION).expect("parse fixed incarnation");
-        let incarnation = PrincipalIncarnation::try_from(uuid).expect("accept nonnil incarnation");
-        assert_eq!(Uuid::from(incarnation).as_bytes(), uuid.as_bytes());
-        let serialized = serde_json::to_value(incarnation).expect("serialize incarnation");
-        assert_eq!(serialized, json!(INCARNATION));
-        assert_eq!(
-            serde_json::from_value::<PrincipalIncarnation>(serialized).expect("decode incarnation"),
-            incarnation
-        );
-    }
-
-    #[test]
-    fn test_user_identity_live_incarnation_roundtrip() {
-        let mut identity = UserIdentity::new(Credentials::default());
-        identity.principal_incarnation = Some(
-            PrincipalIncarnation::try_from(Uuid::parse_str(INCARNATION).expect("parse fixed incarnation"))
-                .expect("accept fixed incarnation"),
-        );
-        let serialized = serde_json::to_value(&identity).expect("serialize identity with incarnation");
-        assert_eq!(serialized["principalIncarnation"], INCARNATION);
-        let encoded = serde_json::to_string(&identity).expect("encode identity with incarnation");
-        let decoded: UserIdentity = serde_json::from_str(&encoded).expect("decode identity with incarnation");
-        assert_eq!(decoded.principal_incarnation, identity.principal_incarnation);
-        assert_eq!(serde_json::to_value(decoded).expect("reserialize identity"), serialized);
-    }
-
-    #[test]
-    fn test_user_identity_timestamp_aliases_unchanged() {
-        let mut minio = json!({"version": 1, "credentials": {}, "updatedAt": "2025-03-07T12:00:00Z"});
-        let encoded = serde_json::to_string(&minio).expect("encode updatedAt fixture");
-        let identity: UserIdentity = serde_json::from_str(&encoded).expect("decode updatedAt timestamp");
-        minio["update_at"] = minio
-            .as_object_mut()
-            .expect("identity object")
-            .remove("updatedAt")
-            .expect("existing timestamp");
-        let encoded = serde_json::to_string(&minio).expect("encode update_at fixture");
-        let legacy: UserIdentity = serde_json::from_str(&encoded).expect("decode update_at alias");
-        assert_eq!(legacy.update_at, identity.update_at);
-        assert_eq!(legacy.version, 1);
-        let serialized = serde_json::to_value(legacy).expect("serialize legacy timestamp");
-        assert_eq!(serialized["updatedAt"], "2025-03-07T12:00:00Z");
-        assert!(serialized.get("update_at").is_none());
-        assert!(serialized.get("principalIncarnation").is_none());
-    }
+    use super::UserIdentity;

    /// Deserialize UserIdentity from MinIO-style JSON (RFC3339 updatedAt).
    #[test]
@@ -27,10 +27,9 @@ pub(crate) mod fixture {
    pub(crate) use rustfs_ecstore::api::disk::endpoint::Endpoint;
    pub(crate) use rustfs_ecstore::api::layout::{EndpointServerPools, Endpoints, PoolEndpoints};
    pub(crate) use rustfs_ecstore::api::object::{PutObjReader, SelectObjectSnapshot};
-    pub(crate) use rustfs_ecstore::api::runtime::InstanceContext;
    #[cfg(feature = "put-object-commit-barrier")]
    pub(crate) use rustfs_ecstore::api::set_disk as ecstore_set_disk;
-    pub(crate) use rustfs_ecstore::api::storage::{ECStore, init_local_disks, init_local_disks_with_instance_ctx};
+    pub(crate) use rustfs_ecstore::api::storage::{ECStore, init_local_disks};
    pub(crate) use rustfs_storage_api::{BucketOperations, BucketOptions, MakeBucketOptions, ObjectIO};

    #[cfg(test)]
@@ -35,9 +35,8 @@ use std::sync::{Arc, Once};
 #[cfg(feature = "put-object-commit-barrier")]
 use ecstore_test_compat::fixture::ecstore_set_disk;
 use ecstore_test_compat::fixture::{
-    BucketOperations as _, BucketOptions, ECStore, Endpoint, EndpointServerPools, Endpoints, InstanceContext, MakeBucketOptions,
-    ObjectIO as _, PoolEndpoints, PutObjReader, SelectObjectSnapshot, init_bucket_metadata_sys, init_local_disks,
-    init_local_disks_with_instance_ctx,
+    BucketOperations as _, BucketOptions, ECStore, Endpoint, EndpointServerPools, Endpoints, MakeBucketOptions, ObjectIO as _,
+    PoolEndpoints, PutObjReader, SelectObjectSnapshot, init_bucket_metadata_sys, init_local_disks,
 };
 use tokio_util::sync::CancellationToken;

@@ -91,7 +90,7 @@ pub struct TestECStoreEnv {
    pub temp_root: PathBuf,
    /// The per-disk directories (`disk1`..`diskN`) under `temp_root`.
    pub disk_paths: Vec<PathBuf>,
-    /// By default, the store is bootstrapped exactly like the historical test setups:
+    /// The store, bootstrapped exactly like the historical test setups:
    /// `init_local_disks` + `ECStore::new` on `127.0.0.1:0` (random port keeps
    /// nextest's process-per-test parallelism safe).
    pub ecstore: Arc<ECStore>,
@@ -156,7 +155,6 @@ pub struct TestECStoreEnvBuilder {
    prefix: String,
    base_dir: Option<PathBuf>,
    init_bucket_metadata: bool,
-    instance_ctx: Option<Arc<InstanceContext>>,
 }

 impl Default for TestECStoreEnvBuilder {
@@ -166,7 +164,6 @@ impl Default for TestECStoreEnvBuilder {
            prefix: "rustfs_test_utils".to_string(),
            base_dir: None,
            init_bucket_metadata: true,
-            instance_ctx: None,
        }
    }
 }
@@ -199,12 +196,6 @@ impl TestECStoreEnvBuilder {
        self
    }

-    /// Use a caller-owned context for local disks and store construction.
-    pub fn instance_ctx(mut self, instance_ctx: Arc<InstanceContext>) -> Self {
-        self.instance_ctx = Some(instance_ctx);
-        self
-    }
-
    /// Build the environment. Panics on any bootstrap failure — this is test
    /// scaffolding, and a broken environment must fail the test loudly.
    pub async fn build(self) -> TestECStoreEnv {
@@ -246,22 +237,14 @@ impl TestECStoreEnvBuilder {
        };
        let endpoint_pools = EndpointServerPools::from(vec![pool_endpoints]);

+        init_local_disks(endpoint_pools.clone()).await.expect("init local disks");
+
        // Port 0 keeps ECStore-backed integration binaries parallel-safe under
        // nextest: no fixed peer port is ever shared between test processes.
        let server_addr: std::net::SocketAddr = "127.0.0.1:0".parse().expect("parse test addr");
-        let ecstore = if let Some(instance_ctx) = self.instance_ctx {
-            init_local_disks_with_instance_ctx(&instance_ctx, endpoint_pools.clone())
-                .await
-                .expect("init instance local disks");
-            ECStore::new_with_instance_ctx(server_addr, endpoint_pools.clone(), CancellationToken::new(), instance_ctx)
-                .await
-                .expect("build instance test ECStore")
-        } else {
-            init_local_disks(endpoint_pools.clone()).await.expect("init local disks");
-            ECStore::new(server_addr, endpoint_pools.clone(), CancellationToken::new())
-                .await
-                .expect("build test ECStore")
-        };
+        let ecstore = ECStore::new(server_addr, endpoint_pools.clone(), CancellationToken::new())
+            .await
+            .expect("build test ECStore");

        // The production bootstrap only persists pool.bin from the elected
        // first cluster node.  Test stores intentionally have no cluster
@@ -295,86 +278,3 @@ impl TestECStoreEnvBuilder {
        }
    }
 }
-
-#[cfg(test)]
-mod context_tests {
-    use super::{InstanceContext, TestECStoreEnv};
-    use crate::ecstore_test_compat::fixture::{read_config, save_config};
-    use std::sync::Arc;
-
-    #[tokio::test]
-    async fn explicit_context_keeps_stores_isolated() {
-        let root_a = tempfile::tempdir().expect("create context A disk root");
-        let root_b = tempfile::tempdir().expect("create context B disk root");
-        let ctx_a = Arc::new(InstanceContext::new());
-        let ctx_b = Arc::new(InstanceContext::new());
-        let env_a = TestECStoreEnv::builder()
-            .base_dir(root_a.path())
-            .init_bucket_metadata(false)
-            .instance_ctx(ctx_a.clone())
-            .build()
-            .await;
-        let env_b = TestECStoreEnv::builder()
-            .base_dir(root_b.path())
-            .init_bucket_metadata(false)
-            .instance_ctx(ctx_b.clone())
-            .build()
-            .await;
-
-        assert!(env_a.ecstore.instance_endpoints().is_none());
-        assert!(env_b.ecstore.instance_endpoints().is_none());
-        ctx_a.set_endpoints(env_a.endpoint_pools.clone());
-        let endpoints_a = env_a
-            .ecstore
-            .instance_endpoints()
-            .expect("instance A sees its caller-owned context topology");
-        assert_eq!(
-            endpoints_a.0[0]
-                .endpoints
-                .into_ref()
-                .first()
-                .expect("context A first endpoint")
-                .to_string(),
-            env_a.disk_paths[0].to_string_lossy(),
-        );
-        assert!(
-            env_b.ecstore.instance_endpoints().is_none(),
-            "publishing context A must not publish context B"
-        );
-        ctx_b.set_endpoints(env_b.endpoint_pools.clone());
-        let endpoints_b = env_b
-            .ecstore
-            .instance_endpoints()
-            .expect("instance B sees its caller-owned context topology");
-        assert_eq!(
-            endpoints_b.0[0]
-                .endpoints
-                .into_ref()
-                .first()
-                .expect("context B first endpoint")
-                .to_string(),
-            env_b.disk_paths[0].to_string_lossy(),
-        );
-        assert_eq!(
-            env_a.ecstore.instance_endpoints().expect("context A remains published").0[0]
-                .endpoints
-                .into_ref()
-                .first()
-                .expect("context A retained endpoint")
-                .to_string(),
-            env_a.disk_paths[0].to_string_lossy(),
-        );
-
-        let path = "config/test-context-isolation.json";
-        let bytes_a = br#"{"context":"a"}"#.to_vec();
-        let bytes_b = br#"{"context":"b"}"#.to_vec();
-        save_config(env_a.ecstore.clone(), path, bytes_a.clone())
-            .await
-            .expect("write context A config");
-        save_config(env_b.ecstore.clone(), path, bytes_b.clone())
-            .await
-            .expect("write context B config");
-        assert_eq!(read_config(env_a.ecstore.clone(), path).await.expect("read context A config"), bytes_a);
-        assert_eq!(read_config(env_b.ecstore.clone(), path).await.expect("read context B config"), bytes_b);
-    }
-}
@@ -32,7 +32,6 @@ Required headings and strings in these files are asserted by `scripts/check_arch

 | Document | Use this when |
 |---|---|
-| [capture-checkpoint-contract.md](capture-checkpoint-contract.md) | changing Capture checkpoint capture, provenance, installation, recovery, or log cleanup |
 | [local-descriptor-cache.md](local-descriptor-cache.md) | changing descriptor cache keys/open modes or invalidation after local mutations |
 | [erasure-coding.md](erasure-coding.md) | changing anything under `crates/ecstore/src/erasure/`, `crates/filemeta/`, `crates/ecstore/src/set_disk/`, storage-class or layout code, or any decode, quorum, or heal boundary (normative spec) |
 | [placement-repair-invariants.md](placement-repair-invariants.md) | changing anything that resolves an object to a pool, set, or disk, or that admits scanner or heal work |
@@ -1,211 +0,0 @@
-**Use this when:** changing Capture checkpoint capture, provenance, installation, recovery, or log cleanup.
-
-# Capture checkpoint contract
-
-This document defines the requirements for a recoverable Capture checkpoint.
-It is a design contract, not evidence that production capture, installation,
-purge, quorum integration, or S3 behavior implements these requirements.
-
-## Existing evidence and the missing boundary
-
-In `crates/capture-authority/src/application.rs`, `apply_created` inserts one
-canonical command and result on the first decision for an operation ID. An
-identical command returns that stored result; a different command returns
-`OperationConflict` without replacing the canonical record. The first result
-remains authoritative even when later state would produce a different result.
-
-`CreatedOperation` in `crates/capture-authority/src/created.rs` contains command
-bytes and a result. Only `CreatedResult::Created` carries `decision_log_id`.
-Negative results carry no first-decision ID. Heads, versions, events, and
-operations therefore cannot reconstruct the source of every first decision.
-For example, a stored `HeadMismatch` records an observed head, but does not
-prove that the head existed at the original decision cut. Later retries do
-not supply the missing historical evidence.
-
-`validate_application` in `crates/capture-authority/src/application.rs` derives
-first operation IDs from retained log/application records and, for native
-schema 4, replays decisions from the beginning using in-memory maps and sets.
-That intact-prefix validator is evidence for existing behavior; it is not a
-checkpoint-baseline validator or a bounded recovery algorithm.
-
-The `Row` comment and `StagingSink` in
-`crates/capture-authority/src/snapshot.rs` reserve operation provenance for
-caller validation and require isolated staging. Rows remain opaque to the
-framing decoder. A framing digest detects byte corruption; it does not establish
-semantic consistency, historical truth, or authority to install a checkpoint.
-
-## Canonical operation provenance
-
-Checkpoint-capable storage must use an explicitly versioned operation envelope
-containing the full first-decision `LogId` and the existing canonical operation
-bytes, including their stored result. Preserve those bytes and result; do not
-recompute a negative result against checkpoint-time heads or binding state.
-The envelope version is a storage capability boundary, distinct from the
-snapshot framing version. The existing CSNP v1 framing and header must not be
-silently extended. Header capability metadata requires a declared framing
-revision or an authenticated, versioned manifest whose consistency with the
-captured header is validated. Distinct storage and framing versions do not
-permit silently changing the v1 format.
-
-The first insertion must atomically record the envelope, application response,
-applied position, and any successful head/version/event changes in the same
-write transaction. Its first-decision ID must identify that insertion's log
-entry. An identical-command retry returns the original result and leaves both
-bytes and first-decision ID unchanged. A conflicting-command retry records its
-own application response as `OperationConflict`; it cannot overwrite the
-canonical envelope or claim to be the first decision.
-
-```mermaid
-flowchart TD
-    A[Committed command and operation ID] --> B{Canonical operation exists?}
-    B -->|No| C[Evaluate once at this log ID]
-    C --> D[Atomically persist first ID, original bytes and result]
-    B -->|Yes| E{Same canonical command bytes?}
-    E -->|Yes| F[Return original result; preserve first ID]
-    E -->|No| G[Return OperationConflict; preserve canonical record]
-```
-
-## Compatibility and promotion
-
-The existing schema 3 to schema 4 migration remains unchanged. Adding provenance
-requires a separately declared storage capability with explicit readers, writers,
-upgrade, and downgrade rules. A schema 4 label alone must not imply provenance
-support. Unknown envelope versions and stores missing required provenance must
-be rejected by checkpoint installation and purge authorization.
-
-An old schema 4 store may backfill first IDs only from its complete, intact,
-validated prefix: replay and validate the original command/results, identify the
-first canonical insertion for each operation, then atomically persist the
-capability and envelopes. A retry's ID or a successful event's ID cannot stand
-in for missing negative provenance. If the prefix has already been removed or
-cannot be validated, refuse promotion and cleanup; do not invent IDs. Readers
-without the new capability must refuse the upgraded format rather than silently
-interpret it as the old operation encoding. Retaining all history forever is
-not the checkpoint design: validated promotion supplies a recoverable baseline
-that permits bounded suffix retention.
-
-## Capture cut and resource ownership
-
-Capture must read the header, heads, versions, events, and operations through
-one owned read transaction. The view must retain the owning store/database
-handle for the entire transaction, cursor, and output lifetime; an owned
-transaction type alone does not prove database lifetime safety. Keep the store
-`Arc` alive so dropping another handle cannot invalidate the view. These are
-ownership requirements, not a claim about unverified library behavior.
-
-The header must describe that same committed, applied cut with its full `LogId`,
-full effective membership including its decision position and node metadata,
-cluster identity, binding identity/revision, and supported storage/envelope
-capabilities. Membership and binding must come from the captured transaction,
-not from a later live read. Never capture beyond the durable committed position.
-A four-table view without this cut and metadata is not independently installable.
-
-Emit rows incrementally in deterministic section/key order. Bound row size,
-section and total bytes, counts, temporary disk consumption, and concurrent
-capture admission. Use checked arithmetic. Do not collect the whole snapshot
-in RAM. Cancellation, disk exhaustion, or limit failure must leave the live
-store unchanged and release task-owned staging resources.
-
-## Isolated validation and trust
-
-Decode only into isolated disk staging. Check framing, limits, section order,
-unique keys, complete header metadata, capability versions, cluster/binding
-compatibility, canonical command/result encoding, envelope IDs at or before the
-cut, and all head/version/event/operation relationships. A successful operation
-must agree with its version, event, and first decision. Negative operations must
-have no success event or side effect attributable to that decision.
-
-These checks still cannot prove the historical predicate behind a negative
-result. Neither a digest nor a first ID proves that a `HeadMismatch`,
-`BindingMismatch`, or `VersionConflict` was justified when first decided.
-Remote installation therefore requires an explicitly authenticated and
-authorized Raft authority for the claimed cluster, cut, and membership, or a
-complete historical proof validated from a trusted baseline. Reject input that
-has neither. Authentication and authorization must precede publication and
-purge; a caller's assertion of trust is insufficient. This document does not
-claim that this remote Raft authentication boundary is implemented.
-
-Replay validation must use disk-backed staging/indexes and bounded buffers,
-including duplicate-operation and version/head tracking. Putting the input on
-disk while retaining replay maps proportional to history does not meet this
-requirement. The existing `validate_application` RAM replay cannot simply be
-reused to claim bounded checkpoint installation.
-
-## Atomic publication and recovery
-
-After isolated validation, publication must have one durable atomic point that
-selects the complete checkpoint generation, its application state, cut,
-membership, binding, capability, and recovery metadata. A pointer switch is
-acceptable only when all referenced bytes are already durable and recovery
-cannot observe a partial generation. Serialize publication with application so
-no committed suffix write is lost between validation and the switch.
-
-Preserve local node identity and persisted vote/hard state. Remote checkpoint
-metadata cannot overwrite either. Validate cluster identity, binding, full cut
-identity, and membership before replacing state. Reject rollback behind the
-local durable application/checkpoint floor; an equal index with a different
-term/leader identity is a conflict, not an equivalent cut. Retain valid local
-suffix and commit state consistently with the installed baseline, rather than
-resetting them from remote metadata.
-
-On reopen, select only a fully published generation and verify its recoverable
-baseline before replay. The suffix starts exactly at cut.index + 1, with checked
-index arithmetic; the complete cut ID must agree wherever the boundary is
-retained. Validate contiguous suffix indices, entry identities, committed and
-applied bounds, typed application responses, canonical retry behavior, and
-membership transitions relative to checkpoint state. Reject gaps, conflicting
-boundary IDs, stale membership, or a suffix belonging to another cluster or
-binding. Recovery must begin from checkpoint heads and canonical operations,
-not from empty state or the assumption that the retained log starts at zero.
-
-Purge is authorized only after the published checkpoint can be reopened and
-validated as a recovery baseline. Delete only entries/results covered by that
-exact cut, retaining everything needed for the continuous suffix and local
-Raft durability. Re-check the selected generation and cut immediately before
-cleanup. Persist a recoverable purge boundary so interrupted cleanup is
-idempotent. Never delete the only evidence before provenance promotion or
-checkpoint durability succeeds.
-
-```mermaid
-flowchart TD
-    A[One owned transaction and store lifetime] --> B[Stream captured cut to isolated disk staging]
-    B --> C[Validate framing, semantics, provenance and authorized source]
-    C --> D[Make complete generation durable]
-    D --> E[Single atomic publication point]
-    E --> F[Reopen baseline and validate continuous suffix]
-    F --> G[Authorize purge only through that exact cut]
-    G --> H[Idempotent cleanup; preserve recoverable baseline and suffix]
-    B -->|Failure| I[Discard staging; live generation remains usable]
-    C -->|Failure| I
-    D -->|Before publication crash| I
-```
-
-A crash before publication leaves the previous generation recoverable. A crash
-after publication must recover the new complete baseline, including when purge
-has not begun or is only partly complete. Orphan staging is never selected as
-live state. Old generations may be reclaimed only after no active view needs
-them and the selected generation remains recoverable.
-
-## Behavioral verification requirements
-
-The following are required behavior checks, not reports of tests run or gates
-passed. Existing symbols above anchor the semantics; future checkpoint paths
-must demonstrate the additional requirements directly.
-
-| Behavior | Required observation |
-|---|---|
-| First negative result | Original command/result and first full ID survive capture, install, and reopen; later state does not re-evaluate it. |
-| Identical and conflicting retries | `apply_created` semantics hold across the checkpoint boundary: original result/ID retained; conflict cannot overwrite the canonical record. |
-| Old schema 4 promotion | An intact prefix validates and supplies first IDs; missing/corrupt prefix refuses promotion and purge; schema 3 to 4 behavior stays unchanged. |
-| Snapshot cut under concurrent writes | Header, membership, binding, and all rows describe one transaction cut; later writes remain in the suffix. |
-| View lifetime | Dropping other store handles during row emission cannot invalidate the retained transaction or database. |
-| Framing versus authority | Valid digest with forged negative history is insufficient; unauthenticated or unauthorized remote input is rejected. |
-| Cross-record corruption | Altered command, first ID, event, version, head, membership, or binding fails isolated validation before publication. |
-| Recovery boundary | Exact cut plus continuous suffix recovers; gaps, conflicting full IDs, cross-cluster input, and rollback are rejected. |
-| Local Raft state | Installation preserves local identity, vote/hard state, and compatible durable suffix/commit information. |
-| Resource exhaustion | Oversize rows, arithmetic overflow, disk exhaustion, cancellation, and large histories fail safely within declared memory/disk bounds. |
-| Crash publication | Crashes before and after the atomic point select a complete old or new generation, never a mixture. |
-| Interrupted purge | Reopen succeeds before cleanup is authorized; each interruption resumes without deleting the recoverable baseline or required suffix. |
-
-Documentation path and whitespace guards only validate document hygiene. They
-cannot establish these production behaviors or the remote trust boundary.
@@ -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.
@@ -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);
 }
@@ -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())
@@ -4343,15 +4343,8 @@ mod tests {
            (S3Action::DeleteObjectTaggingAction, S3Action::DeleteObjectVersionTaggingAction),
        ] {
            for version in [None, Some("null"), Some("0194e0f1-0000-7000-8000-000000000000")] {
-                let action = versioned_tagging_action(current.clone(), version);
-                assert_eq!(
-                    action,
-                    Action::S3Action(if version.is_some() {
-                        versioned.clone()
-                    } else {
-                        current.clone()
-                    })
-                );
+                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,
@@ -415,7 +415,12 @@ pub(crate) fn object_lock_retention_from_metadata(metadata: &HashMap<String, Str
            date.checked_to_offset(time::UtcOffset::UTC)
                .filter(|date| (0..=9999).contains(&date.year()))
                .map(Timestamp::from)
-                .ok_or_else(|| s3_error!(InternalError, "Stored retention date is outside the supported UTC range"))
+                .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 })
@@ -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()
@@ -1,154 +0,0 @@
-"""Verify the real locked capture graph and leader identity feature contract."""
-
-import hashlib
-import json
-import os
-from pathlib import Path
-import shutil
-import subprocess
-import tempfile
-import re
-import unittest
-
-ROOT = Path(__file__).resolve().parents[2]
-REGISTRY = "registry+https://github.com/rust-lang/crates.io-index"
-
-
-def source_files(root):
-    result = subprocess.run(["git", "ls-files", "-z"], cwd=root, check=True, capture_output=True)
-    paths = {Path(os.fsdecode(p)) for p in result.stdout.split(b"\0") if p}
-    paths.add(Path("scripts/tests/test_capture_dependencies.py"))
-    return sorted(paths)
-
-
-def fingerprint(root, paths):
-    digest = hashlib.sha256()
-    for path in paths:
-        digest.update(os.fsencode(path))
-        digest.update(b"\0")
-        source = root / path
-        if source.is_symlink():
-            digest.update(b"symlink\0")
-            digest.update(os.fsencode(os.readlink(source)))
-        else:
-            digest.update(b"file\0")
-            digest.update(source.read_bytes())
-    return digest.hexdigest()
-
-
-def metadata(root, all_features=False):
-    lock = (root / "Cargo.lock").read_bytes()
-    command = ["cargo", "metadata", "--locked", "--format-version", "1"]
-    if all_features:
-        command.append("--all-features")
-    result = subprocess.run(command, cwd=root, capture_output=True, text=True)
-    if result.returncode:
-        raise RuntimeError(f"{' '.join(command)} exited {result.returncode}: {result.stderr}")
-    if (root / "Cargo.lock").read_bytes() != lock:
-        raise AssertionError("CAPTURE_LOCK_CHANGED")
-    print(f"{' '.join(command)}: exit 0; lock {hashlib.sha256(lock).hexdigest()}")
-    return json.loads(result.stdout)
-
-
-def assert_graph(case, root, graph):
-    packages = {package["id"]: package for package in graph["packages"]}
-    capture = next(p for p in packages.values() if p["name"] == "rustfs-capture-authority")
-    dependencies = {p["name"]: p for p in capture["dependencies"]}
-    case.assertIn("openraft", dependencies, "CAPTURE_OPENRAFT_REQUIRED")
-    dependency = dependencies["openraft"]
-    case.assertEqual(dependency["req"], "=0.9.25", "CAPTURE_OPENRAFT_EXACT_PIN")
-    case.assertFalse(dependency["uses_default_features"], "CAPTURE_OPENRAFT_DEFAULTS_DISABLED")
-    case.assertIn("serde", dependency["features"])
-    case.assertEqual(dependency["source"], REGISTRY)
-    for name, kind in (("serde", None), ("url", None), ("serde_json", "dev")):
-        case.assertIn(name, dependencies)
-        case.assertEqual(dependencies[name]["kind"], kind)
-    case.assertIn("derive", dependencies["serde"]["features"])
-    nodes = {node["id"]: node for node in graph["resolve"]["nodes"]}
-    reachable = set()
-    pending = [capture["id"]]
-    while pending:
-        node_id = pending.pop()
-        if node_id not in reachable:
-            reachable.add(node_id)
-            pending.extend(nodes[node_id]["dependencies"])
-    openraft = [p for p in packages.values() if p["name"] == "openraft"]
-    case.assertEqual(len(openraft), 1, "CAPTURE_UNIQUE_OPENRAFT")
-    package = openraft[0]
-    case.assertIn(package["id"], reachable)
-    case.assertEqual(package["version"], "0.9.25")
-    case.assertEqual(package["source"], REGISTRY)
-    features = set(nodes[package["id"]]["features"])
-    case.assertNotIn("single-term-leader", features, "CAPTURE_ADVANCED_LEADER_REQUIRED")
-    case.assertNotIn("storage-v2", features, "CAPTURE_STORAGE_V1_REQUIRED")
-    case.assertIn("serde", features)
-    case.assertEqual(set(dependency["features"]), {"serde"}, "CAPTURE_OPENRAFT_SERDE_ONLY")
-    for name in ("serde", "serde_derive", "url", "serde_json"):
-        case.assertTrue(any(packages[p]["name"] == name for p in reachable), name)
-    case.assertTrue(any("derive" in nodes[p]["features"] for p in reachable if packages[p]["name"] == "serde"))
-    blocks = (root / "Cargo.lock").read_text().split("[[package]]")
-    locked = [dict(re.findall(r'^([a-z]+) = "([^"\n]*)"$', block, re.MULTILINE))
-              for block in blocks if re.search(r'^name = "openraft"$', block, re.MULTILINE)]
-    case.assertEqual(len(locked), 1)
-    case.assertEqual(locked[0]["version"], "0.9.25")
-    case.assertEqual(locked[0]["source"], REGISTRY)
-    case.assertEqual(locked[0]["checksum"], "a97014fb78acb77be3a40ac2da305f6dd3a6b243f3a908ace87d29b3972eaafd", "CAPTURE_PUBLISHED_OPENRAFT_CHECKSUM")
-
-
-class CaptureDependencies(unittest.TestCase):
-    def test_accepted_dependency_graph(self):
-        for all_features in (False, True):
-            with self.subTest(all_features=all_features):
-                graph = metadata(ROOT, all_features)
-                assert_graph(self, ROOT, graph)
-                package = next(p for p in graph["packages"] if p["name"] == "openraft")
-                node = next(n for n in graph["resolve"]["nodes"] if n["id"] == package["id"])
-                print(f"accepted all_features={all_features}: {package['id']}; features {node['features']}")
-
-    def control(self, old, new, expected):
-        paths = source_files(ROOT)
-        before = fingerprint(ROOT, paths)
-        with tempfile.TemporaryDirectory(prefix="capture-dependencies-") as directory:
-            copied = Path(directory)
-            for path in paths:
-                destination = copied / path
-                destination.parent.mkdir(parents=True, exist_ok=True)
-                source = ROOT / path
-                if source.is_symlink():
-                    destination.symlink_to(os.readlink(source))
-                else:
-                    shutil.copyfile(source, destination)
-            self.assertEqual(fingerprint(copied, paths), before)
-            manifest = copied / "Cargo.toml"
-            original = manifest.read_bytes()
-            self.assertEqual(original.count(old.encode()), 1)
-            try:
-                manifest.write_bytes(original.replace(old.encode(), new.encode(), 1))
-                mutated = fingerprint(copied, paths)
-                self.assertNotEqual(mutated, before)
-                for all_features in (False, True):
-                    graph = metadata(copied, all_features)
-                    versions = {p["version"] for p in graph["packages"] if p["name"] == "openraft"}
-                    self.assertEqual(versions, {"0.9.25"})
-                    with self.assertRaisesRegex(AssertionError, expected):
-                        assert_graph(self, copied, graph)
-                print(f"{expected}: metadata exit 0 (default/all-features); source {before}; control {mutated}")
-            finally:
-                manifest.write_bytes(original)
-                self.assertEqual(fingerprint(copied, paths), before)
-                self.assertEqual(fingerprint(ROOT, paths), before)
-                print(f"{expected}: restored source {before}")
-
-    def test_relaxed_pin_is_rejected(self):
-        self.control('version = "=0.9.25"', 'version = "0.9.25"', "CAPTURE_OPENRAFT_EXACT_PIN")
-
-    def test_single_term_leader_is_rejected(self):
-        self.control(
-            'openraft = { version = "=0.9.25", default-features = false, features = ["serde"] }',
-            'openraft = { version = "=0.9.25", default-features = false, features = ["serde", "single-term-leader"] }',
-            "CAPTURE_ADVANCED_LEADER_REQUIRED",
-        )
-
-
-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