diff --git a/.config/make/tests.mak b/.config/make/tests.mak
index df47452ed..73675f5fc 100644
--- a/.config/make/tests.mak
+++ b/.config/make/tests.mak
@@ -28,6 +28,8 @@ script-tests: ## Run shell script tests
 	./scripts/test_entrypoint_credentials.sh
 	./scripts/test_internode_grpc_ab_bench.sh
 	./scripts/test_object_batch_bench_enhanced.sh
+	./scripts/test_hotpath_warp_ab_gate.sh
+	./scripts/test_hotpath_warp_abba.sh
 	./scripts/test_exact_1mib_handoff_abba.sh
 	./scripts/test_pinned_paired_abba_bench.sh
 	./scripts/test_manual_transition_runbooks.sh
diff --git a/.github/workflows/performance-ab.yml b/.github/workflows/performance-ab.yml
index 8c75c56c8..9cb0a2984 100644
--- a/.github/workflows/performance-ab.yml
+++ b/.github/workflows/performance-ab.yml
@@ -225,8 +225,34 @@ jobs:
           cp target/release/rustfs baseline-bin/rustfs
           echo "built=true" >> "$GITHUB_OUTPUT"
 
+      - name: Build baseline on cache miss (different candidate)
+        id: baseline_build
+        if: >-
+          steps.baseline_cache.outputs.cache-hit != 'true' &&
+          steps.commits.outputs.baseline_sha != steps.commits.outputs.candidate_sha
+        run: |
+          set -euo pipefail
+          baseline_root="$RUNNER_TEMP/rustfs-baseline-${{ github.run_id }}"
+          baseline_target="$RUNNER_TEMP/rustfs-baseline-target-${{ github.run_id }}"
+          git worktree add --detach "$baseline_root" "${{ steps.commits.outputs.baseline_sha }}"
+          cargo build --release --manifest-path "$baseline_root/Cargo.toml" --bin rustfs --target-dir "$baseline_target"
+          mkdir -p baseline-bin
+          cp "$baseline_target/release/rustfs" baseline-bin/rustfs
+          git worktree remove --force "$baseline_root"
+          echo "built=true" >> "$GITHUB_OUTPUT"
+
+      - name: Build candidate binary
+        id: candidate_build
+        if: steps.commits.outputs.baseline_sha != steps.commits.outputs.candidate_sha
+        run: |
+          set -euo pipefail
+          cargo build --release --bin rustfs
+          mkdir -p candidate-bin
+          cp target/release/rustfs candidate-bin/rustfs
+          echo "built=true" >> "$GITHUB_OUTPUT"
+
       - name: Save self-healed baseline to cache
-        if: steps.selfheal.outputs.built == 'true'
+        if: steps.selfheal.outputs.built == 'true' || steps.baseline_build.outputs.built == 'true'
         uses: actions/cache/save@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6
         with:
           path: baseline-bin/rustfs
@@ -238,10 +264,9 @@ jobs:
           INPUT_DURATION: ${{ github.event.inputs.duration }}
         run: |
           set -euo pipefail
-          # Budget note: with perf-3's cached baseline the nightly does no source
-          # build on a cache hit, so the wall-clock is dominated by the short warp
-          # matrix — duration/rounds/cooldown are kept small to fit all 24 cells
-          # (6 workloads x 2 phases x 2 drive-sync) rather than dropping cells.
+          # The formal runner executes A1 baseline -> B1 candidate -> B2 candidate
+          # -> A2 baseline for each workload and drive-sync cell. It requires three
+          # rounds per leg to emit tail latency and error-rate evidence.
           # --health-timeout 180 outlasts the server's own 120s startup-readiness
           # budget, which the rig's previous 60s health poll undershot (the first
           # two nightly failures). perf-6 recalibrates these once the noise study
@@ -251,39 +276,42 @@ jobs:
           candidate_sha="${{ steps.commits.outputs.candidate_sha }}"
           baseline_hit="${{ steps.baseline_cache.outputs.cache-hit }}"
           selfheal_built="${{ steps.selfheal.outputs.built }}"
+          baseline_built="${{ steps.baseline_build.outputs.built }}"
+          candidate_built="${{ steps.candidate_build.outputs.built }}"
 
-          args=(--duration "$duration" --rounds 2 --cooldown 5 --health-timeout 180)
+          args=(--duration "$duration" --rounds 3 --cooldown 5 --health-timeout 180 --baseline-revision "$baseline_sha" --candidate-revision "$candidate_sha")
 
-          if [[ "$baseline_hit" == "true" || "$selfheal_built" == "true" ]]; then
+          if [[ "$baseline_hit" == "true" || "$selfheal_built" == "true" || "$baseline_built" == "true" ]]; then
             chmod +x baseline-bin/rustfs
             base_bin="$PWD/baseline-bin/rustfs"
             args+=(--baseline-bin "$base_bin")
             if [[ "$baseline_hit" == "true" ]]; then
               base_src="actions-cache (rustfs-baseline-$baseline_sha)"
-            else
+            elif [[ "$selfheal_built" == "true" ]]; then
               base_src="source build (cache self-heal, saved as rustfs-baseline-$baseline_sha)"
+            else
+              base_src="isolated origin/main source build (saved as rustfs-baseline-$baseline_sha)"
             fi
             if [[ "$candidate_sha" == "$baseline_sha" ]]; then
               # Nightly on main: the candidate is the same commit as the baseline,
               # so reuse the one binary for both phases and skip all builds.
-              args+=(--candidate-bin "$base_bin" --skip-build)
+              args+=(--candidate-bin "$base_bin")
               cand_src="same binary as baseline (same commit)"
-            else
+            elif [[ "$candidate_built" == "true" ]]; then
+              chmod +x candidate-bin/rustfs
+              args+=(--candidate-bin "$PWD/candidate-bin/rustfs")
               cand_src="source build of the checked-out ref"
+            else
+              echo "::error::candidate binary was not built" >&2
+              exit 2
             fi
           else
-            # Cache miss with candidate != baseline (opt-in PR gate only): fall
-            # back to the source double-build. With the post-#4806 LTO profile
-            # this will overrun the job budget and alert; rerun once the push
-            # cache build for origin/main has completed, or wait for perf-7's
-            # merge-base caching.
-            args+=(--baseline-ref origin/main)
-            base_src="source build of origin/main (cache miss)"
-            cand_src="source build of the checked-out ref"
+            echo "::error::baseline binary was not restored or built" >&2
+            exit 2
           fi
 
-          args+=(--provenance-note "baseline commit: $baseline_sha - $base_src")
-          args+=(--provenance-note "candidate commit: $candidate_sha - $cand_src")
+          echo "baseline binary: $base_src"
+          echo "candidate binary: $cand_src"
 
           if [[ "${{ steps.exempt.outputs.allow_regression }}" == "true" ]]; then
             args+=(--allow-regression --exemption-reason "labeled perf-deliberate-tradeoff / dispatch override")
@@ -291,7 +319,7 @@ jobs:
           # Do not let a gate FAIL abort the job here; capture status and surface
           # it after the PR comment is posted.
           set +e
-          bash scripts/run_hotpath_warp_ab.sh "${args[@]}"
+          bash scripts/run_hotpath_warp_abba.sh "${args[@]}"
           echo "status=$?" >> "$GITHUB_OUTPUT"
           set -e
           # Locate the newest run dir + gate.md for the summary/comment/artifact
@@ -299,10 +327,10 @@ jobs:
           # holds server-logs/ for diagnosis.
           # Run dirs are UTC-timestamp names (no special chars); ls is safe here.
           # shellcheck disable=SC2012
-          run_dir="$(ls -td target/hotpath-ab/*/ 2>/dev/null | head -n1 || true)"
+          run_dir="$(ls -td target/hotpath-abba/*/ 2>/dev/null | head -n1 || true)"
           echo "run_dir=${run_dir%/}" >> "$GITHUB_OUTPUT"
           # shellcheck disable=SC2012
-          gate_md="$(ls -t target/hotpath-ab/*/gate.md 2>/dev/null | head -n1 || true)"
+          gate_md="$(ls -t target/hotpath-abba/*/candidate_gate.md 2>/dev/null | head -n1 || true)"
           echo "gate_md=$gate_md" >> "$GITHUB_OUTPUT"
 
       - name: Upload A/B results
@@ -310,10 +338,10 @@ jobs:
         uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
         with:
           name: hotpath-warp-ab-${{ github.run_number }}
-          # Includes per-cell median_summary.csv / baseline_compare.csv, gate.md,
+          # Includes per-cell median_summary.csv / baseline_compare.csv, both gates,
           # and server-logs/ (rustfs.log + startup env per phase) so a failed run
           # is diagnosable. Short retention: this is churny nightly debug data.
-          path: target/hotpath-ab/
+          path: target/hotpath-abba/
           if-no-files-found: warn
           retention-days: 14
 
diff --git a/crates/ecstore/src/erasure/coding/encode.rs b/crates/ecstore/src/erasure/coding/encode.rs
index db960c3ac..545be864f 100644
--- a/crates/ecstore/src/erasure/coding/encode.rs
+++ b/crates/ecstore/src/erasure/coding/encode.rs
@@ -641,6 +641,7 @@ impl Erasure {
                             let res = self.clone().encode_block_bytes_mut(encode_buf, n).await?;
                             buf = BytesMut::with_capacity(ingest_capacity);
                             let queued_bytes = queued_block_bytes(&res);
+                            let _producer_stage = rustfs_io_metrics::track_ec_encode_producer_bytes(queued_bytes);
                             let send_wait_stage_start = stage_timer_if_enabled();
                             if let Err(err) = send_queued(&tx, res, queued_bytes).await {
                                 return Err(std::io::Error::other(format!("Failed to send encoded data : {err}")));
@@ -670,6 +671,7 @@ impl Erasure {
                             let (res, returned_buf) = self.clone().encode_block(encode_buf, n).await?;
                             buf = returned_buf;
                             let queued_bytes = queued_block_bytes(&res);
+                            let _producer_stage = rustfs_io_metrics::track_ec_encode_producer_bytes(queued_bytes);
                             let send_wait_stage_start = stage_timer_if_enabled();
                             if let Err(err) = send_queued(&tx, res, queued_bytes).await {
                                 return Err(std::io::Error::other(format!("Failed to send encoded data : {err}")));
@@ -712,6 +714,7 @@ impl Erasure {
             if block.is_empty() {
                 break;
             }
+            let _writer_stage = rustfs_io_metrics::track_ec_encode_writer_bytes(queued_block_bytes(&block));
             let write_stage_start = stage_timer_if_enabled();
             if let Err(err) = writers.write(block).await {
                 write_err = Some(err);
@@ -768,6 +771,7 @@ impl Erasure {
             let mut buf = vec![0u8; block_size];
             let mut pending_batch = Vec::with_capacity(batch_blocks);
             let mut pending_batch_bytes = 0usize;
+            let mut pending_batch_stage = None;
             loop {
                 match rustfs_utils::read_full_or_eof(&mut reader, &mut buf).await {
                     Ok(Some(n)) => {
@@ -779,6 +783,8 @@ impl Erasure {
                         let queued_bytes = queued_block_bytes(&res);
                         pending_batch_bytes = pending_batch_bytes.saturating_add(queued_bytes);
                         pending_batch.push(res);
+                        drop(pending_batch_stage.take());
+                        pending_batch_stage = Some(rustfs_io_metrics::track_ec_encode_producer_bytes(pending_batch_bytes));
 
                         if pending_batch.len() >= batch_blocks {
                             let send_wait_stage_start = stage_timer_if_enabled();
@@ -786,6 +792,7 @@ impl Erasure {
                                 return Err(std::io::Error::other(format!("Failed to send encoded data : {err}")));
                             }
                             record_internal_stage_if_enabled("erasure_encode_batched_send_wait", send_wait_stage_start);
+                            drop(pending_batch_stage.take());
                             pending_batch = Vec::with_capacity(batch_blocks);
                             pending_batch_bytes = 0;
                         }
@@ -813,6 +820,7 @@ impl Erasure {
                     return Err(std::io::Error::other(format!("Failed to send encoded data : {err}")));
                 }
                 record_internal_stage_if_enabled("erasure_encode_batched_send_wait", send_wait_stage_start);
+                drop(pending_batch_stage);
             }
 
             Ok((reader, total))
@@ -828,6 +836,7 @@ impl Erasure {
             };
             record_internal_stage_if_enabled("erasure_encode_batched_recv_wait", recv_wait_stage_start);
             let batch = batch.into_inner();
+            let _writer_stage = rustfs_io_metrics::track_ec_encode_writer_bytes(queued_batch_bytes(&batch));
             let write_stage_start = stage_timer_if_enabled();
             for block in batch {
                 if let Err(err) = writers.write(block).await {
@@ -1302,24 +1311,36 @@ mod tests {
     }
 
     async fn writer_error_aborts_blocked_producer(pipeline: EncodePipeline) {
-        const BLOCK_SIZE: usize = 16;
-
         let gauge_baseline = rustfs_io_metrics::current_ec_encode_inflight_bytes();
+        let producer_baseline = rustfs_io_metrics::current_ec_encode_producer_bytes();
+        let writer_baseline = rustfs_io_metrics::current_ec_encode_writer_bytes();
+        let producer_peak_before = rustfs_io_metrics::current_ec_encode_producer_bytes_peak();
+        let writer_peak_before = rustfs_io_metrics::current_ec_encode_writer_bytes_peak();
+        let block_size = match pipeline {
+            EncodePipeline::Batched => usize::try_from(producer_peak_before.max(writer_peak_before).saturating_add(1))
+                .expect("stage peak fits the test address space"),
+            EncodePipeline::Vec | EncodePipeline::BytesMut => 16,
+        };
+        rustfs_io_metrics::set_put_stage_metrics_enabled(true);
+        let erasure = Arc::new(Erasure::new(1, 0, block_size));
+        let batch_blocks = encode_batch_block_count().min(encode_channel_capacity(
+            erasure.shard_size().saturating_mul(erasure.total_shard_count()),
+            erasure_encode_max_inflight_bytes(),
+        ));
         let blocks_before_pending = match pipeline {
-            EncodePipeline::Batched => encode_batch_block_count(),
+            EncodePipeline::Batched => batch_blocks,
             EncodePipeline::Vec | EncodePipeline::BytesMut => 1,
         };
         let (reader, reader_blocked, reader_dropped, _final_block) =
-            BlocksThenPendingReader::new(blocks_before_pending, BLOCK_SIZE);
+            BlocksThenPendingReader::new(blocks_before_pending, block_size);
         let writes = Arc::new(std::sync::atomic::AtomicUsize::new(0));
         let mut writers = vec![Some(bitrot_writer(
             FailAfterReaderBlocksWriter {
                 reader_blocked,
                 writes: writes.clone(),
             },
-            BLOCK_SIZE,
+            block_size,
         ))];
-        let erasure = Arc::new(Erasure::new(1, 0, BLOCK_SIZE));
 
         let result = match pipeline {
             EncodePipeline::Vec => erasure.encode_with_ingest_mode(reader, &mut writers, 1, false).await,
@@ -1346,12 +1367,40 @@ mod tests {
             gauge_baseline,
             "writer failure must settle all queued and pending encoded bytes"
         );
+        assert_eq!(
+            rustfs_io_metrics::current_ec_encode_producer_bytes(),
+            producer_baseline,
+            "writer failure must settle producer stage bytes for every ingest pipeline"
+        );
+        assert_eq!(
+            rustfs_io_metrics::current_ec_encode_writer_bytes(),
+            writer_baseline,
+            "writer failure must settle writer stage bytes for every ingest pipeline"
+        );
+        if matches!(pipeline, EncodePipeline::Batched) {
+            let expected_batch_bytes = u64::try_from(block_size)
+                .expect("block size fits the stage gauge")
+                .checked_mul(u64::try_from(batch_blocks).expect("batch block count fits the stage gauge"))
+                .expect("test batch payload fits the stage gauge");
+            assert!(
+                rustfs_io_metrics::current_ec_encode_producer_bytes_peak() >= producer_peak_before.max(expected_batch_bytes),
+                "batched producer must expose its full pending batch before writer failure"
+            );
+            assert!(
+                rustfs_io_metrics::current_ec_encode_writer_bytes_peak() >= writer_peak_before.max(expected_batch_bytes),
+                "batched writer must expose its full batch before writer failure"
+            );
+        }
+        rustfs_io_metrics::set_put_stage_metrics_enabled(false);
     }
 
     async fn aborting_full_queue_settles_pending_send() {
         const BLOCK_SIZE: usize = 16;
 
         let gauge_baseline = rustfs_io_metrics::current_ec_encode_inflight_bytes();
+        let producer_baseline = rustfs_io_metrics::current_ec_encode_producer_bytes();
+        let writer_baseline = rustfs_io_metrics::current_ec_encode_writer_bytes();
+        rustfs_io_metrics::set_put_stage_metrics_enabled(true);
         let erasure = Arc::new(Erasure::new(1, 0, BLOCK_SIZE));
         let inflight_blocks = encode_channel_capacity(
             erasure.shard_size().saturating_mul(erasure.total_shard_count()),
@@ -1389,6 +1438,15 @@ mod tests {
         })
         .await
         .expect("producer should account for the pending send after the queue fills");
+        tokio::time::timeout(Duration::from_secs(1), async {
+            while rustfs_io_metrics::current_ec_encode_producer_bytes() == producer_baseline
+                || rustfs_io_metrics::current_ec_encode_writer_bytes() == writer_baseline
+            {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("full queue must retain producer and writer stage ownership before cancellation");
 
         encode.abort();
         assert!(matches!(encode.await, Err(err) if err.is_cancelled()), "encode task should be cancelled");
@@ -1406,6 +1464,17 @@ mod tests {
             gauge_baseline,
             "cancelling a full queue must settle queued and pending bytes"
         );
+        assert_eq!(
+            rustfs_io_metrics::current_ec_encode_producer_bytes(),
+            producer_baseline,
+            "cancelling a full queue must settle the pending producer stage"
+        );
+        assert_eq!(
+            rustfs_io_metrics::current_ec_encode_writer_bytes(),
+            writer_baseline,
+            "cancelling a full queue must settle the stalled writer stage"
+        );
+        rustfs_io_metrics::set_put_stage_metrics_enabled(false);
     }
 
     #[tokio::test]
@@ -1872,6 +1941,61 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn bytesmut_streaming_encode_observes_all_payload_stage_peaks() {
+        let queue_baseline = rustfs_io_metrics::current_ec_encode_inflight_bytes();
+        let producer_peak_before = rustfs_io_metrics::current_ec_encode_producer_bytes_peak();
+        let queue_peak_before = rustfs_io_metrics::current_ec_encode_queue_bytes_peak();
+        let writer_peak_before = rustfs_io_metrics::current_ec_encode_writer_bytes_peak();
+        let prior_peak = producer_peak_before.max(queue_peak_before).max(writer_peak_before);
+        let block_size = usize::try_from(prior_peak.saturating_add(1)).expect("stage peak fits the test address space");
+        let erasure = Arc::new(Erasure::new(1, 0, block_size));
+        let encoded_block_bytes = erasure.shard_size() * erasure.total_shard_count();
+        let committed = Arc::new(Mutex::new(Vec::new()));
+        let mut writers = vec![Some(bitrot_writer(DeferredCommitWriter::new(committed.clone()), block_size))];
+        let reader = tokio::io::BufReader::new(Cursor::new(vec![0x5a; block_size * 2]));
+
+        rustfs_io_metrics::set_put_stage_metrics_enabled(true);
+        let result = erasure.encode_with_ingest_mode(reader, &mut writers, 1, true).await;
+        rustfs_io_metrics::set_put_stage_metrics_enabled(false);
+
+        let (_reader, written) = result.expect("bytesmut streaming encode should complete");
+        assert_eq!(written, block_size * 2);
+        assert_eq!(
+            rustfs_io_metrics::current_ec_encode_inflight_bytes(),
+            queue_baseline,
+            "completed streaming encode must not retain queue bytes"
+        );
+        assert_eq!(
+            rustfs_io_metrics::current_ec_encode_producer_bytes(),
+            0,
+            "completed streaming encode must not retain producer bytes"
+        );
+        assert_eq!(
+            rustfs_io_metrics::current_ec_encode_writer_bytes(),
+            0,
+            "completed streaming encode must not retain writer bytes"
+        );
+        let expected_peak = u64::try_from(encoded_block_bytes).expect("encoded block bytes fit the gauge");
+        assert!(
+            rustfs_io_metrics::current_ec_encode_producer_bytes_peak() >= producer_peak_before.max(expected_peak),
+            "producer peak must observe encoded bytes before queue hand-off"
+        );
+        assert!(
+            rustfs_io_metrics::current_ec_encode_queue_bytes_peak() >= queue_peak_before.max(expected_peak),
+            "queue peak must observe encoded bytes pending shard writers"
+        );
+        assert!(
+            rustfs_io_metrics::current_ec_encode_writer_bytes_peak() >= writer_peak_before.max(expected_peak),
+            "writer peak must observe encoded bytes after queue hand-off"
+        );
+        assert!(
+            !committed.lock().expect("committed buffer should be lockable").is_empty(),
+            "stage peak observation must not change writer commit behavior"
+        );
+    }
+
     #[tokio::test]
     async fn encode_streaming_write_quorum_failure_aborts_and_reports_error() {
         const DATA_SHARDS: usize = 2;
diff --git a/crates/ecstore/src/io_support/bitrot.rs b/crates/ecstore/src/io_support/bitrot.rs
index f48635f1a..aabd8004c 100644
--- a/crates/ecstore/src/io_support/bitrot.rs
+++ b/crates/ecstore/src/io_support/bitrot.rs
@@ -768,10 +768,18 @@ mod tests {
     #[derive(Debug, Clone, Default)]
     struct TestRemoteDataTransport {
         bytes: Arc<Mutex<Vec<u8>>>,
+        write_error: Option<io::ErrorKind>,
     }
 
     #[cfg(feature = "hotpath")]
     impl TestRemoteDataTransport {
+        fn with_write_error(kind: io::ErrorKind) -> Self {
+            Self {
+                bytes: Arc::default(),
+                write_error: Some(kind),
+            }
+        }
+
         fn bytes(&self) -> Vec<u8> {
             self.bytes
                 .lock()
@@ -784,11 +792,15 @@ mod tests {
     #[derive(Debug)]
     struct TestRemoteWriter {
         bytes: Arc<Mutex<Vec<u8>>>,
+        write_error: Option<io::ErrorKind>,
     }
 
     #[cfg(feature = "hotpath")]
     impl tokio::io::AsyncWrite for TestRemoteWriter {
         fn poll_write(self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
+            if let Some(kind) = self.write_error {
+                return Poll::Ready(Err(io::Error::from(kind)));
+            }
             self.bytes
                 .lock()
                 .expect("test remote transport bytes lock should not be poisoned")
@@ -815,6 +827,7 @@ mod tests {
         async fn open_write(&self, _request: WriteStreamRequest) -> Result<FileWriter> {
             Ok(Box::new(TestRemoteWriter {
                 bytes: Arc::clone(&self.bytes),
+                write_error: self.write_error,
             }))
         }
 
@@ -855,7 +868,7 @@ mod tests {
     }
 
     #[cfg(feature = "hotpath")]
-    async fn remote_test_disk() -> (DiskStore, TestRemoteDataTransport) {
+    async fn remote_test_disk(transport: TestRemoteDataTransport) -> (DiskStore, TestRemoteDataTransport) {
         use crate::disk::endpoint::Endpoint;
 
         let endpoint = Endpoint {
@@ -865,7 +878,6 @@ mod tests {
             set_idx: 0,
             disk_idx: 0,
         };
-        let transport = TestRemoteDataTransport::default();
         let remote = RemoteDisk::new(
             &endpoint,
             &DiskOption {
@@ -1021,7 +1033,7 @@ mod tests {
             .expect("mmap-copy fallback stream should preserve bytes");
         assert_eq!(fallback_read, fallback_payload);
 
-        let (remote_disk, remote_transport) = remote_test_disk().await;
+        let (remote_disk, remote_transport) = remote_test_disk(TestRemoteDataTransport::default()).await;
         let remote_payload = b"remote shard bytes";
         let mut remote_writer = create_bitrot_writer(
             false,
@@ -1074,6 +1086,43 @@ mod tests {
         }
         assert_eq!(remote_read, remote_payload);
 
+        let (failing_remote_disk, _) = remote_test_disk(TestRemoteDataTransport::with_write_error(io::ErrorKind::Other)).await;
+        let mut failing_writer = create_bitrot_writer(
+            false,
+            Some(&failing_remote_disk),
+            bucket,
+            "obj/hotpath-remote-write-error-part.1",
+            4,
+            shard_size,
+            HashAlgorithm::None,
+        )
+        .await
+        .expect("failing remote bitrot writer should open before its first write");
+        let write_error = failing_writer
+            .write(b"fail")
+            .await
+            .expect_err("raw shard writer failures must remain visible through the bitrot writer");
+        assert_eq!(write_error.kind(), io::ErrorKind::Other);
+
+        let (would_block_remote_disk, _) =
+            remote_test_disk(TestRemoteDataTransport::with_write_error(io::ErrorKind::WouldBlock)).await;
+        let mut would_block_writer = create_bitrot_writer(
+            false,
+            Some(&would_block_remote_disk),
+            bucket,
+            "obj/hotpath-remote-write-would-block-part.1",
+            4,
+            shard_size,
+            HashAlgorithm::None,
+        )
+        .await
+        .expect("would-block remote bitrot writer should open before its first write");
+        let would_block = would_block_writer
+            .write(b"wait")
+            .await
+            .expect_err("would-block must remain visible to the caller");
+        assert_eq!(would_block.kind(), io::ErrorKind::WouldBlock);
+
         drop(guard);
         let report = std::fs::read_to_string(&report_path).expect("HotPath I/O report should be written");
         let report: serde_json::Value = serde_json::from_str(&report).expect("HotPath I/O report should be valid JSON");
@@ -1089,6 +1138,13 @@ mod tests {
             assert!(byte_count > 0, "report must include fixed label {label}");
             byte_count
         };
+        let io_errors = |label: &str, direction: &str| {
+            entries
+                .iter()
+                .filter(|entry| entry["label"].as_str().is_some_and(|entry_label| entry_label == label))
+                .filter_map(|entry| entry[direction]["errors"].as_u64())
+                .sum::<u64>()
+        };
         let payload_len = u64::try_from(payload.len()).expect("test payload length should fit u64");
         assert_eq!(
             io_bytes(RAW_SHARD_READ_LOCAL_LABEL, "read"),
@@ -1103,6 +1159,11 @@ mod tests {
             io_bytes(RAW_SHARD_WRITE_REMOTE_LABEL, "write"),
             u64::try_from(remote_payload.len()).expect("remote payload length should fit u64")
         );
+        assert_eq!(
+            io_errors(RAW_SHARD_WRITE_REMOTE_LABEL, "write"),
+            1,
+            "the wrapper must record the real remote writer failure without changing it, while excluding WouldBlock"
+        );
         for label in [
             RAW_SHARD_READ_LOCAL_LABEL,
             RAW_SHARD_READ_REMOTE_LABEL,
diff --git a/crates/io-metrics/src/lib.rs b/crates/io-metrics/src/lib.rs
index 85e7519bf..68b4be072 100644
--- a/crates/io-metrics/src/lib.rs
+++ b/crates/io-metrics/src/lib.rs
@@ -49,7 +49,10 @@
 #[macro_use]
 extern crate metrics;
 
-use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::{
+    Mutex,
+    atomic::{AtomicBool, AtomicU64, Ordering},
+};
 
 /// Global switch for detailed per-stage PUT metrics (path label, stage durations).
 /// When `false`, `record_put_object_path` and `record_put_object_stage_duration`
@@ -269,6 +272,12 @@ pub use collector::MetricsCollector;
 pub use performance::PerformanceMetrics;
 
 static EC_ENCODE_INFLIGHT_BYTES: AtomicU64 = AtomicU64::new(0);
+static EC_ENCODE_PRODUCER_BYTES_CURRENT: AtomicU64 = AtomicU64::new(0);
+static EC_ENCODE_PRODUCER_BYTES_PEAK: AtomicU64 = AtomicU64::new(0);
+static EC_ENCODE_QUEUE_BYTES_PEAK: AtomicU64 = AtomicU64::new(0);
+static EC_ENCODE_WRITER_BYTES_CURRENT: AtomicU64 = AtomicU64::new(0);
+static EC_ENCODE_WRITER_BYTES_PEAK: AtomicU64 = AtomicU64::new(0);
+static EC_ENCODE_PEAK_PUBLISH_LOCK: Mutex<()> = Mutex::new(());
 static GET_OBJECT_BUFFERED_BYTES: AtomicU64 = AtomicU64::new(0);
 const SHARD_READ_COST_LOCAL: &str = "local";
 const SHARD_READ_COST_REMOTE: &str = "remote";
@@ -313,6 +322,49 @@ fn saturating_sub_atomic(counter: &AtomicU64, bytes: u64) -> u64 {
     }
 }
 
+#[inline(always)]
+fn update_peak_atomic(counter: &AtomicU64, value: u64) -> Option<u64> {
+    let mut peak = counter.load(Ordering::Relaxed);
+    loop {
+        if value <= peak {
+            return None;
+        }
+        match counter.compare_exchange_weak(peak, value, Ordering::Relaxed, Ordering::Relaxed) {
+            Ok(_) => return Some(value),
+            Err(actual) => peak = actual,
+        }
+    }
+}
+
+#[derive(Clone, Copy)]
+enum EcEncodePeakMetric {
+    Producer,
+    Queue,
+    Writer,
+}
+
+fn publish_ec_encode_peak_with(counter: &AtomicU64, value: u64, before_publish_lock: impl FnOnce(), set_gauge: impl FnOnce(f64)) {
+    if update_peak_atomic(counter, value).is_none() {
+        return;
+    }
+    before_publish_lock();
+    let _guard = EC_ENCODE_PEAK_PUBLISH_LOCK.lock().unwrap_or_else(|error| error.into_inner());
+    set_gauge(counter.load(Ordering::Relaxed) as f64);
+}
+
+fn publish_ec_encode_peak(counter: &AtomicU64, metric: EcEncodePeakMetric, value: u64) {
+    publish_ec_encode_peak_with(
+        counter,
+        value,
+        || {},
+        |peak| match metric {
+            EcEncodePeakMetric::Producer => gauge_set_cached!("rustfs_ec_encode_producer_bytes_peak", peak),
+            EcEncodePeakMetric::Queue => gauge_set_cached!("rustfs_ec_encode_queue_bytes_peak", peak),
+            EcEncodePeakMetric::Writer => gauge_set_cached!("rustfs_ec_encode_writer_bytes_peak", peak),
+        },
+    );
+}
+
 #[inline(always)]
 fn usize_to_f64(value: usize) -> f64 {
     value as f64
@@ -2204,6 +2256,9 @@ pub fn record_allocator_memory_observation(backend: &'static str, observation: A
 pub fn add_ec_encode_inflight_bytes(bytes: usize) {
     let next = EC_ENCODE_INFLIGHT_BYTES.fetch_add(bytes as u64, Ordering::Relaxed) + bytes as u64;
     gauge_set_cached!("rustfs_ec_encode_inflight_bytes_current", next as f64);
+    if put_stage_metrics_enabled() {
+        publish_ec_encode_peak(&EC_ENCODE_QUEUE_BYTES_PEAK, EcEncodePeakMetric::Queue, next);
+    }
 }
 
 /// Remove encoded bytes from the tracked erasure encode in-flight gauge.
@@ -2219,6 +2274,109 @@ pub fn current_ec_encode_inflight_bytes() -> u64 {
     EC_ENCODE_INFLIGHT_BYTES.load(Ordering::Relaxed)
 }
 
+/// Tracks encoded payload bytes held before queue hand-off or during shard writes.
+///
+/// Each guard contributes to a process-wide stage total until it is dropped. The
+/// reported peak therefore includes concurrent PUTs, but excludes reader,
+/// allocator, and transport buffers; it is not a per-PUT or process-RSS limit.
+pub struct EcEncodePayloadStageGuard {
+    counter: &'static AtomicU64,
+    bytes: u64,
+    enabled: bool,
+    current_metric: EcEncodePeakMetric,
+}
+
+impl Drop for EcEncodePayloadStageGuard {
+    fn drop(&mut self) {
+        if !self.enabled {
+            return;
+        }
+        let next = saturating_sub_atomic(self.counter, self.bytes);
+        match self.current_metric {
+            EcEncodePeakMetric::Producer => gauge_set_cached!("rustfs_ec_encode_producer_bytes_current", next as f64),
+            EcEncodePeakMetric::Queue => unreachable!("queue bytes use their own ownership guard"),
+            EcEncodePeakMetric::Writer => gauge_set_cached!("rustfs_ec_encode_writer_bytes_current", next as f64),
+        }
+    }
+}
+
+fn track_ec_encode_payload_stage(
+    bytes: usize,
+    counter: &'static AtomicU64,
+    peak: &'static AtomicU64,
+    metric: EcEncodePeakMetric,
+) -> EcEncodePayloadStageGuard {
+    let enabled = put_stage_metrics_enabled();
+    let bytes = bytes as u64;
+    if enabled {
+        let next = counter.fetch_add(bytes, Ordering::Relaxed) + bytes;
+        match metric {
+            EcEncodePeakMetric::Producer => gauge_set_cached!("rustfs_ec_encode_producer_bytes_current", next as f64),
+            EcEncodePeakMetric::Queue => unreachable!("queue bytes use their own ownership guard"),
+            EcEncodePeakMetric::Writer => gauge_set_cached!("rustfs_ec_encode_writer_bytes_current", next as f64),
+        }
+        publish_ec_encode_peak(peak, metric, next);
+    }
+    EcEncodePayloadStageGuard {
+        counter,
+        bytes,
+        enabled,
+        current_metric: metric,
+    }
+}
+
+/// Track encoded producer payload bytes until queue hand-off completes.
+#[inline(always)]
+pub fn track_ec_encode_producer_bytes(bytes: usize) -> EcEncodePayloadStageGuard {
+    track_ec_encode_payload_stage(
+        bytes,
+        &EC_ENCODE_PRODUCER_BYTES_CURRENT,
+        &EC_ENCODE_PRODUCER_BYTES_PEAK,
+        EcEncodePeakMetric::Producer,
+    )
+}
+
+/// Track encoded payload bytes while shard writers own the batch.
+#[inline(always)]
+pub fn track_ec_encode_writer_bytes(bytes: usize) -> EcEncodePayloadStageGuard {
+    track_ec_encode_payload_stage(
+        bytes,
+        &EC_ENCODE_WRITER_BYTES_CURRENT,
+        &EC_ENCODE_WRITER_BYTES_PEAK,
+        EcEncodePeakMetric::Writer,
+    )
+}
+
+/// Return the process-lifetime high-water mark of encoded producer payload bytes.
+#[inline(always)]
+pub fn current_ec_encode_producer_bytes_peak() -> u64 {
+    EC_ENCODE_PRODUCER_BYTES_PEAK.load(Ordering::Relaxed)
+}
+
+/// Return the current process-wide encoded producer payload bytes.
+#[inline(always)]
+pub fn current_ec_encode_producer_bytes() -> u64 {
+    EC_ENCODE_PRODUCER_BYTES_CURRENT.load(Ordering::Relaxed)
+}
+
+/// Return the process-lifetime high-water mark of encoded queue payload bytes.
+#[inline(always)]
+pub fn current_ec_encode_queue_bytes_peak() -> u64 {
+    EC_ENCODE_QUEUE_BYTES_PEAK.load(Ordering::Relaxed)
+}
+
+/// Return the process-lifetime high-water mark of encoded writer payload bytes.
+#[inline(always)]
+pub fn current_ec_encode_writer_bytes_peak() -> u64 {
+    EC_ENCODE_WRITER_BYTES_PEAK.load(Ordering::Relaxed)
+}
+
+/// Return the current process-wide encoded writer payload bytes.
+#[inline(always)]
+pub fn current_ec_encode_writer_bytes() -> u64 {
+    EC_ENCODE_WRITER_BYTES_CURRENT.load(Ordering::Relaxed)
+}
+
 /// Track whole-object buffering on the GET path.
 #[inline(always)]
 pub fn track_get_object_buffered_bytes(bytes: usize) -> Option<MemoryGaugeGuard> {
@@ -2385,7 +2543,9 @@ pub fn record_io_latency_p99(latency_ms: f64) {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use std::sync::Mutex;
+    use metrics_util::MetricKind;
+    use metrics_util::debugging::{DebugValue, DebuggingRecorder};
+    use std::sync::{Arc, Barrier, Mutex};
 
     // Serialize tests that mutate the process-global PUT_STAGE_METRICS_ENABLED flag.
     static METRICS_FLAG_LOCK: Mutex<()> = Mutex::new(());
@@ -2838,6 +2998,9 @@ mod tests {
 
     #[test]
     fn test_ec_encode_inflight_bytes_tracking() {
+        let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+        set_put_stage_metrics_enabled(true);
+        let queue_peak_before = current_ec_encode_queue_bytes_peak();
         EC_ENCODE_INFLIGHT_BYTES.store(0, Ordering::Relaxed);
         add_ec_encode_inflight_bytes(1024);
         add_ec_encode_inflight_bytes(2048);
@@ -2845,6 +3008,104 @@ mod tests {
         remove_ec_encode_inflight_bytes(2048);
         remove_ec_encode_inflight_bytes(4096);
         assert_eq!(current_ec_encode_inflight_bytes(), 0);
+        assert!(
+            current_ec_encode_queue_bytes_peak() >= queue_peak_before.max(3072),
+            "queue peak must retain the largest observed queue occupancy"
+        );
+        set_put_stage_metrics_enabled(false);
+    }
+
+    #[test]
+    fn test_ec_encode_producer_and_writer_stage_guards_aggregate_and_settle() {
+        let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+        set_put_stage_metrics_enabled(true);
+
+        let producer_bytes = 1024;
+        let producer_first = track_ec_encode_producer_bytes(producer_bytes);
+        let producer_second = track_ec_encode_producer_bytes(producer_bytes);
+        assert_eq!(current_ec_encode_producer_bytes(), 2048);
+        assert!(
+            current_ec_encode_producer_bytes_peak() >= 2048,
+            "producer peak must include simultaneous stage ownership"
+        );
+        drop((producer_first, producer_second));
+        assert_eq!(current_ec_encode_producer_bytes(), 0);
+
+        let writer_bytes = 2048;
+        let writer_first = track_ec_encode_writer_bytes(writer_bytes);
+        let writer_second = track_ec_encode_writer_bytes(writer_bytes);
+        assert_eq!(current_ec_encode_writer_bytes(), 4096);
+        assert!(
+            current_ec_encode_writer_bytes_peak() >= 4096,
+            "writer peak must include simultaneous stage ownership"
+        );
+        drop((writer_first, writer_second));
+        assert_eq!(current_ec_encode_writer_bytes(), 0);
+
+        set_put_stage_metrics_enabled(false);
+    }
+
+    #[test]
+    fn test_ec_encode_producer_peak_exports_the_high_water_mark() {
+        let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+        assert_eq!(current_ec_encode_producer_bytes(), 0, "test must start without producer stage ownership");
+        let previous_peak = EC_ENCODE_PRODUCER_BYTES_PEAK.swap(0, Ordering::Relaxed);
+        let recorder = DebuggingRecorder::new();
+        let snapshotter = recorder.snapshotter();
+
+        metrics::with_local_recorder(&recorder, || {
+            set_put_stage_metrics_enabled(true);
+            let first = track_ec_encode_producer_bytes(1024);
+            let second = track_ec_encode_producer_bytes(2048);
+            drop((first, second));
+            set_put_stage_metrics_enabled(false);
+        });
+
+        let exported_peak = snapshotter
+            .snapshot()
+            .into_vec()
+            .into_iter()
+            .find_map(|(composite, _, _, value)| {
+                (composite.kind() == MetricKind::Gauge && composite.key().name() == "rustfs_ec_encode_producer_bytes_peak")
+                    .then_some(value)
+            });
+        assert!(
+            matches!(exported_peak, Some(DebugValue::Gauge(value)) if value.0 == 3072.0),
+            "exported producer peak must retain the aggregate high-water mark"
+        );
+        EC_ENCODE_PRODUCER_BYTES_PEAK.fetch_max(previous_peak, Ordering::Relaxed);
+    }
+
+    #[test]
+    fn test_ec_encode_peak_publish_does_not_regress_after_out_of_order_cas() {
+        let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+        let peak = Arc::new(AtomicU64::new(0));
+        let exported = Arc::new(AtomicU64::new(0));
+        let first_ready = Arc::new(Barrier::new(2));
+        let release_first = Arc::new(Barrier::new(2));
+        let first_peak = Arc::clone(&peak);
+        let first_exported = Arc::clone(&exported);
+        let first_ready_for_thread = Arc::clone(&first_ready);
+        let release_first_for_thread = Arc::clone(&release_first);
+
+        let first = std::thread::spawn(move || {
+            publish_ec_encode_peak_with(
+                &first_peak,
+                10,
+                || {
+                    first_ready_for_thread.wait();
+                    release_first_for_thread.wait();
+                },
+                |value| first_exported.store(value as u64, Ordering::Relaxed),
+            );
+        });
+        first_ready.wait();
+        publish_ec_encode_peak_with(&peak, 20, || {}, |value| exported.store(value as u64, Ordering::Relaxed));
+        release_first.wait();
+        first.join().expect("first publisher should not panic");
+
+        assert_eq!(peak.load(Ordering::Relaxed), 20);
+        assert_eq!(exported.load(Ordering::Relaxed), 20, "late publisher must reload the high-water mark");
     }
 
     #[test]
diff --git a/scripts/hotpath_warp_ab_gate.sh b/scripts/hotpath_warp_ab_gate.sh
index 430d2790d..d184661d6 100755
--- a/scripts/hotpath_warp_ab_gate.sh
+++ b/scripts/hotpath_warp_ab_gate.sh
@@ -26,6 +26,7 @@ set -euo pipefail
 FAIL_PCT=10
 WARN_PCT=5
 ALLOW_REGRESSION="false"
+REQUIRE_TAIL_ERROR="false"
 MARKDOWN_OUT=""
 EXEMPTION_REASON="deliberate correctness tradeoff"
 declare -a COMPARE_CSVS=()
@@ -39,6 +40,8 @@ Usage: hotpath_warp_ab_gate.sh --compare-csv <file> [--compare-csv <file> ...] [
   --warn-pct <n>         Regression budget that warns (default 5).
   --allow-regression     Downgrade every FAIL to an exempted WARN (deliberate
                          correctness tradeoff); the gate then always exits 0.
+  --require-tail-error   Require the extended p90/p99/error-rate schema and
+                         reject missing tail/error or zero-success evidence.
   --exemption-reason <s> Reason recorded when --allow-regression is set.
   --markdown <file>      Also write the result table as Markdown to <file>.
   -h, --help             Show this help.
@@ -51,6 +54,7 @@ while [[ $# -gt 0 ]]; do
     --fail-pct) FAIL_PCT="$2"; shift 2 ;;
     --warn-pct) WARN_PCT="$2"; shift 2 ;;
     --allow-regression) ALLOW_REGRESSION="true"; shift ;;
+    --require-tail-error) REQUIRE_TAIL_ERROR="true"; shift ;;
     --exemption-reason) EXEMPTION_REASON="$2"; shift 2 ;;
     --markdown) MARKDOWN_OUT="$2"; shift 2 ;;
     -h|--help) usage; exit 0 ;;
@@ -67,10 +71,24 @@ for csv in "${COMPARE_CSVS[@]}"; do
   [[ -f "$csv" ]] || { echo "error: compare CSV not found: $csv" >&2; exit 2; }
 done
 
+if [[ "$REQUIRE_TAIL_ERROR" == "true" ]]; then
+  strict_header='size,tool,concurrency,new_median_reqps,baseline_median_reqps,delta_reqps_pct,new_median_latency_ms,baseline_median_latency_ms,delta_latency_pct,new_median_throughput_bps,baseline_median_throughput_bps,delta_throughput_pct,new_median_p90_latency_ms,baseline_median_p90_latency_ms,delta_p90_latency_pct,new_median_p99_latency_ms,baseline_median_p99_latency_ms,delta_p99_latency_pct,new_ok_rounds,baseline_ok_rounds,new_failed_rounds,baseline_failed_rounds,new_error_rate_pct,baseline_error_rate_pct,delta_error_rate_pct'
+  for csv in "${COMPARE_CSVS[@]}"; do
+    [[ "$(head -n1 "$csv")" == "$strict_header" ]] || {
+      echo "error: strict mode requires the current tail/error compare schema: $csv" >&2
+      exit 2
+    }
+    awk 'NR == 2 { found = 1 } END { exit found ? 0 : 1 }' "$csv" || {
+      echo "error: strict mode requires at least one comparison row: $csv" >&2
+      exit 2
+    }
+  done
+fi
+
 # One awk pass over all CSVs. Emits TSV rows "verdict\tworkload\tmetric\tdelta"
 # on stdout and a final "OVERALL\t<verdict>" line. Verdict is PASS/WARN/FAIL.
 gate_output="$(
-  awk -v fail_pct="$FAIL_PCT" -v warn_pct="$WARN_PCT" '
+  awk -v fail_pct="$FAIL_PCT" -v warn_pct="$WARN_PCT" -v strict="$REQUIRE_TAIL_ERROR" '
     function classify(delta, higher_is_better,   regress) {
       if (delta == "" || delta == "N/A") return "SKIP"
       # Signed regression magnitude: positive means "worse than baseline".
@@ -85,15 +103,57 @@ gate_output="$(
       rank = (v == "FAIL") ? 3 : (v == "WARN") ? 2 : 1
       if (rank > worst) worst = rank
     }
+    function contract_failure(workload, metric) {
+      print "FAIL\t" workload "\t" metric "\tinvalid evidence"
+      if (3 > worst) worst = 3
+    }
+    function decimal(value) {
+      return value ~ /^-?[0-9]+([.][0-9]+)?$/
+    }
+    function count(value) {
+      return value ~ /^[0-9]+$/
+    }
+    function error_rate(value) {
+      return decimal(value) && value + 0 >= 0 && value + 0 <= 100
+    }
+    function error_delta(value) {
+      return decimal(value) && value + 0 >= -100 && value + 0 <= 100
+    }
+    function abs(value) {
+      return value < 0 ? -value : value
+    }
     BEGIN { worst = 1 }
     FNR == 1 { next }                      # skip each file header
     {
       n = split($0, f, ",")
-      if (n < 12) next
+      if (n < 12) {
+        if (strict == "true") contract_failure(FILENAME, "compare-schema")
+        next
+      }
       workload = f[1] "/" f[2] "@" f[3]    # size/tool@concurrency
       emit(classify(f[6],  1), workload, "reqps",      f[6])
       emit(classify(f[9],  0), workload, "latency",    f[9])
       emit(classify(f[12], 1), workload, "throughput", f[12])
+      if (strict == "true") {
+        valid = n == 25
+        for (i = 4; i <= 18; i++) {
+          if (!decimal(f[i])) valid = 0
+        }
+        for (i = 19; i <= 22; i++) {
+          if (!count(f[i])) valid = 0
+        }
+        if (f[19] + 0 == 0 || f[20] + 0 == 0 || !error_rate(f[23]) || !error_rate(f[24]) || !error_delta(f[25])) valid = 0
+        new_error_rate = f[21] / (f[19] + f[21]) * 100
+        baseline_error_rate = f[22] / (f[20] + f[22]) * 100
+        if (abs(f[23] - new_error_rate) > 0.005 || abs(f[24] - baseline_error_rate) > 0.005 || abs(f[25] - (new_error_rate - baseline_error_rate)) > 0.005) valid = 0
+        if (!valid) {
+          contract_failure(workload, "tail-error-evidence")
+          next
+        }
+        emit(classify(f[15], 0), workload, "p90-latency", f[15])
+        emit(classify(f[18], 0), workload, "p99-latency", f[18])
+        emit(classify(f[25], 0), workload, "error-rate", f[25])
+      }
     }
     END {
       print "OVERALL\t" (worst == 3 ? "FAIL" : worst == 2 ? "WARN" : "PASS")
diff --git a/scripts/run_hotpath_warp_abba.sh b/scripts/run_hotpath_warp_abba.sh
index ccaff476d..480161b72 100755
--- a/scripts/run_hotpath_warp_abba.sh
+++ b/scripts/run_hotpath_warp_abba.sh
@@ -18,6 +18,8 @@ GATE="${PROJECT_ROOT}/scripts/hotpath_warp_ab_gate.sh"
 
 BASELINE_BIN=""
 CANDIDATE_BIN=""
+BASELINE_REVISION="unknown"
+CANDIDATE_REVISION="unknown"
 ENDPOINT=""
 DEPLOY_HOOK=""
 HEALTH_PATH="/health"
@@ -32,6 +34,7 @@ CONCURRENCY=8
 DURATION="60s"
 ROUNDS=3
 COOLDOWN_SECS=20
+DATASET_SETUP_DURATION="10s"
 HEALTH_TIMEOUT_SECS=180
 FAIL_PCT=10
 WARN_PCT=5
@@ -39,6 +42,7 @@ ALLOW_REGRESSION=false
 EXEMPTION_REASON="deliberate correctness tradeoff"
 OUT_DIR="${PROJECT_ROOT}/target/hotpath-abba/$(date -u +%Y%m%dT%H%M%SZ 2>/dev/null || echo run)"
 DRY_RUN=false
+ALLOW_UNMANAGED_EXTERNAL=false
 
 WORKLOADS=(
   "put-4kib|put|4KiB"
@@ -61,11 +65,15 @@ and drive-sync cell.
 Required:
   --baseline-bin <path>     Baseline RustFS binary.
   --candidate-bin <path>    Candidate RustFS binary.
+  --baseline-revision <id>  Source revision for the baseline binary.
+  --candidate-revision <id> Source revision for the candidate binary.
 
 Local Linux runner mode:
   --address <host:port>     Local RustFS address (default 127.0.0.1:9000).
   --disks <n>               Throwaway local disks per node (default 4).
   --data-root <path>        Local disk root (default /tmp/rustfs-hotpath-abba).
+                            A reused run namespace is rejected; this runner
+                            never deletes existing benchmark disks.
 
 Production / cluster mode:
   --endpoint <host:port>    Existing cluster endpoint. Enables external mode.
@@ -74,12 +82,27 @@ Production / cluster mode:
                               HOTPATH_ABBA_PHASE=baseline|candidate
                               HOTPATH_ABBA_BINARY=<baseline/candidate binary>
                               HOTPATH_ABBA_DRIVE_SYNC=true|false
+                              HOTPATH_ABBA_WORKLOAD=<workload name>
+                              HOTPATH_ABBA_MODE=put|get|mixed
+                              HOTPATH_ABBA_SIZE=<object size>
+                              HOTPATH_ABBA_CELL_ID=<sync/workload/leg>
+                              HOTPATH_ABBA_DATASET_NAMESPACE=<run namespace>
+                              HOTPATH_ABBA_BUCKET=<per-leg benchmark bucket>
+                              HOTPATH_ABBA_DEPLOY_EVIDENCE_FILE=<path>
+                              The hook must deploy the selected binary, reset or
+                              provision HOTPATH_ABBA_BUCKET for the requested
+                              mode, then write a non-empty evidence file.
+  --allow-unmanaged-external Preserve legacy external mode without a deploy
+                              hook. Its output is not formal ABBA evidence.
   --health-path <path>      Readiness path (default /health).
 
 Benchmark:
   --duration <dur>          warp duration per cell (default 60s).
   --rounds <n>              rounds per cell; must be >= 3 (default 3).
   --cooldown <n>            cooldown seconds between rounds/sizes (default 20).
+  --dataset-setup-duration <dur>
+                            isolated Warp PUT warm-up for get/mixed legs
+                            (default 10s; not included in the measurement).
   --concurrency <n>         warp concurrency (default 8).
   --warp-bin <path>         warp binary (default warp).
 
@@ -134,6 +157,8 @@ while [[ $# -gt 0 ]]; do
   case "$1" in
     --baseline-bin) BASELINE_BIN="$2"; shift 2 ;;
     --candidate-bin) CANDIDATE_BIN="$2"; shift 2 ;;
+    --baseline-revision) BASELINE_REVISION="$2"; shift 2 ;;
+    --candidate-revision) CANDIDATE_REVISION="$2"; shift 2 ;;
     --endpoint) ENDPOINT="$2"; shift 2 ;;
     --deploy-hook) DEPLOY_HOOK="$2"; shift 2 ;;
     --health-path) HEALTH_PATH="$2"; shift 2 ;;
@@ -148,6 +173,7 @@ while [[ $# -gt 0 ]]; do
     --duration) DURATION="$2"; shift 2 ;;
     --rounds) ROUNDS="$2"; shift 2 ;;
     --cooldown) COOLDOWN_SECS="$2"; shift 2 ;;
+    --dataset-setup-duration) DATASET_SETUP_DURATION="$2"; shift 2 ;;
     --health-timeout) HEALTH_TIMEOUT_SECS="$2"; shift 2 ;;
     --fail-pct) FAIL_PCT="$2"; shift 2 ;;
     --warn-pct) WARN_PCT="$2"; shift 2 ;;
@@ -155,6 +181,7 @@ while [[ $# -gt 0 ]]; do
     --exemption-reason) EXEMPTION_REASON="$2"; shift 2 ;;
     --out-dir) OUT_DIR="$2"; shift 2 ;;
     --dry-run) DRY_RUN=true; shift ;;
+    --allow-unmanaged-external) ALLOW_UNMANAGED_EXTERNAL=true; shift ;;
     -h|--help) usage; exit 0 ;;
     *) die "unknown argument: $1" ;;
   esac
@@ -170,6 +197,9 @@ validate_positive_int "$WARN_PCT" "--warn-pct"
 [[ "$ROUNDS" -ge 3 ]] || die "--rounds must be >= 3 for formal ABBA evidence"
 [[ -n "$BASELINE_BIN" ]] || die "--baseline-bin is required"
 [[ -n "$CANDIDATE_BIN" ]] || die "--candidate-bin is required"
+if [[ "$DRY_RUN" != "true" ]] && { [[ -z "$BASELINE_REVISION" || "$BASELINE_REVISION" == "unknown" || -z "$CANDIDATE_REVISION" || "$CANDIDATE_REVISION" == "unknown" ]]; }; then
+  die "formal runs require non-unknown --baseline-revision and --candidate-revision"
+fi
 [[ "$DRY_RUN" == "true" || -x "$BASELINE_BIN" ]] || die "baseline binary is not executable: $BASELINE_BIN"
 [[ "$DRY_RUN" == "true" || -x "$CANDIDATE_BIN" ]] || die "candidate binary is not executable: $CANDIDATE_BIN"
 [[ -x "$ENHANCED_BENCH" ]] || die "missing load driver: $ENHANCED_BENCH"
@@ -182,12 +212,24 @@ EXTERNAL=false
 if [[ -n "$ENDPOINT" ]]; then
   EXTERNAL=true
   ADDRESS="$ENDPOINT"
-  [[ -n "$DEPLOY_HOOK" ]] || log "warning: external mode without --deploy-hook; binaries must be swapped out of band"
+  if [[ -z "$DEPLOY_HOOK" && "$ALLOW_UNMANAGED_EXTERNAL" != "true" ]]; then
+    die "external mode requires --deploy-hook to establish each leg's binary and fresh dataset namespace; use --allow-unmanaged-external only for non-formal legacy runs"
+  fi
+  [[ -n "$DEPLOY_HOOK" ]] || log "warning: unmanaged external mode does not establish binary or data isolation and is not formal ABBA evidence"
 fi
 
 mkdir -p "$OUT_DIR"
 SERVER_LOG_DIR="$OUT_DIR/server-logs"
+DEPLOY_EVIDENCE_DIR="$OUT_DIR/deploy-evidence"
+DATASET_NAMESPACE="$(basename "$OUT_DIR" | tr '[:upper:]' '[:lower:]' | tr -c 'a-z0-9-' '-' | sed -E 's/^-+//; s/-+$//' | cut -c1-24)"
+[[ -n "$DATASET_NAMESPACE" ]] || DATASET_NAMESPACE="run"
+RUN_DATA_ROOT="$DATA_ROOT/$DATASET_NAMESPACE"
+if [[ "$EXTERNAL" != "true" && "$DRY_RUN" != "true" ]]; then
+  mkdir -p "$DATA_ROOT"
+  mkdir "$RUN_DATA_ROOT" || die "local run namespace already exists: $RUN_DATA_ROOT; choose a new --out-dir or --data-root instead of reusing benchmark disks"
+fi
 run mkdir -p "$SERVER_LOG_DIR"
+run mkdir -p "$DEPLOY_EVIDENCE_DIR"
 
 SERVER_PID=""
 SERVER_LOG=""
@@ -241,29 +283,47 @@ phase_for_leg() {
   esac
 }
 
+bucket_for_leg() {
+  local sync_label="$1" workload="$2" leg="$3" bucket_leg
+  bucket_leg="$(printf '%s' "$leg" | tr '[:upper:]' '[:lower:]')"
+  printf 'rustfs-abba-%s-%s-%s-%s' "$DATASET_NAMESPACE" "$sync_label" "$workload" "$bucket_leg"
+}
+
 bring_up() {
-  local leg="$1" drive_sync="$2"
+  local leg="$1" drive_sync="$2" workload="$3" mode="$4" size="$5" sync_label="$6" bucket="$7"
   local phase bin
   phase="$(phase_for_leg "$leg")"
   bin="$(binary_for_leg "$leg")"
 
   if [[ "$EXTERNAL" == "true" ]]; then
     if [[ -n "$DEPLOY_HOOK" ]]; then
-      log "deploy hook: leg=$leg phase=$phase drive_sync=$drive_sync"
-      HOTPATH_ABBA_LEG="$leg" HOTPATH_ABBA_PHASE="$phase" HOTPATH_ABBA_BINARY="$bin" HOTPATH_ABBA_DRIVE_SYNC="$drive_sync" \
+      log "deploy hook: leg=$leg phase=$phase drive_sync=$drive_sync workload=$workload"
+      local evidence_file="$DEPLOY_EVIDENCE_DIR/$sync_label-$workload-$leg.env"
+      local binary_sha
+      binary_sha="$(sha256_file "$bin")"
+      [[ "$binary_sha" != "unknown" ]] || die "cannot attest external binary without a SHA-256 tool"
+      HOTPATH_ABBA_LEG="$leg" HOTPATH_ABBA_PHASE="$phase" HOTPATH_ABBA_BINARY="$bin" HOTPATH_ABBA_BINARY_SHA256="$binary_sha" HOTPATH_ABBA_DRIVE_SYNC="$drive_sync" HOTPATH_ABBA_WORKLOAD="$workload" HOTPATH_ABBA_MODE="$mode" HOTPATH_ABBA_SIZE="$size" HOTPATH_ABBA_CELL_ID="$sync_label/$workload/$leg" HOTPATH_ABBA_DATASET_NAMESPACE="$DATASET_NAMESPACE" HOTPATH_ABBA_BUCKET="$bucket" HOTPATH_ABBA_DEPLOY_EVIDENCE_FILE="$evidence_file" \
         run bash -c "$DEPLOY_HOOK"
+      if [[ "$DRY_RUN" != "true" ]]; then
+        [[ -s "$evidence_file" ]] || die "deploy hook did not write evidence for $sync_label/$workload/$leg: $evidence_file"
+        [[ "$(awk -F= '$1=="leg" {print substr($0, length($1) + 2)}' "$evidence_file")" == "$leg" ]] || die "deploy evidence leg does not match $sync_label/$workload/$leg"
+        [[ "$(awk -F= '$1=="phase" {print substr($0, length($1) + 2)}' "$evidence_file")" == "$phase" ]] || die "deploy evidence phase does not match $sync_label/$workload/$leg"
+        [[ "$(awk -F= '$1=="bucket" {print substr($0, length($1) + 2)}' "$evidence_file")" == "$bucket" ]] || die "deploy evidence bucket does not match $sync_label/$workload/$leg"
+        [[ "$(awk -F= '$1=="binary_sha256" {print substr($0, length($1) + 2)}' "$evidence_file")" == "$binary_sha" ]] || die "deploy evidence binary SHA-256 does not match $sync_label/$workload/$leg"
+        [[ "$(awk -F= '$1=="dataset_reset" {print substr($0, length($1) + 2)}' "$evidence_file")" == "true" ]] || die "deploy evidence must attest dataset_reset=true for $sync_label/$workload/$leg"
+      fi
     fi
     wait_health
     return 0
   fi
 
-  local node_dir="$DATA_ROOT/$leg-sync-$drive_sync"
+  local node_dir="$RUN_DATA_ROOT/$workload-$leg-sync-$drive_sync"
   local disks=() d
   for ((d = 1; d <= DISKS; d++)); do
     disks+=("$node_dir/d$d")
   done
   run mkdir -p "${disks[@]}"
-  SERVER_LOG="$SERVER_LOG_DIR/$leg-sync-$drive_sync.log"
+  SERVER_LOG="$SERVER_LOG_DIR/$workload-$leg-sync-$drive_sync.log"
 
   if [[ "$DRY_RUN" == "true" ]]; then
     { printf 'DRY-RUN: RUSTFS_DRIVE_SYNC_ENABLE=%s %q server' "$drive_sync" "$bin"
@@ -272,7 +332,7 @@ bring_up() {
     return 0
   fi
 
-  cat >"$SERVER_LOG_DIR/$leg-sync-$drive_sync.env" <<EOF
+  cat >"$SERVER_LOG_DIR/$workload-$leg-sync-$drive_sync.env" <<EOF
 leg=$leg
 phase=$phase
 drive_sync=$drive_sync
@@ -297,33 +357,78 @@ EOF
 }
 
 measure() {
-  local leg="$1" workload="$2" mode="$3" size="$4" sync_label="$5" baseline_csv="${6:-}"
+  local leg="$1" workload="$2" mode="$3" size="$4" sync_label="$5" bucket="$6" baseline_csv="${7:-}"
   local cell="$OUT_DIR/$workload/$sync_label/$leg"
   local args=(
     --tool warp --warp-bin "$WARP_BIN" --warp-mode "$mode"
     --endpoint "$ADDRESS" --access-key "$ACCESS_KEY" --secret-key "$SECRET_KEY"
-    --region "$REGION" --sizes "$size" --concurrency "$CONCURRENCY"
+    --region "$REGION" --bucket "$bucket" --sizes "$size" --concurrency "$CONCURRENCY"
     --duration "$DURATION" --rounds "$ROUNDS" --cooldown-secs "$COOLDOWN_SECS"
     --out-dir "$cell"
   )
+  [[ "$mode" == "put" ]] || args+=(--extra-args "--noclear")
   [[ -n "$baseline_csv" ]] && args+=(--baseline-csv "$baseline_csv")
   run "$ENHANCED_BENCH" "${args[@]}" >&2
   echo "$cell"
 }
 
+prepare_dataset() {
+  local leg="$1" workload="$2" mode="$3" size="$4" sync_label="$5" bucket="$6"
+  [[ "$mode" != "put" ]] || return 0
+
+  local setup_cell="$OUT_DIR/$workload/$sync_label/$leg/dataset-setup"
+  local args=(
+    --tool warp --warp-bin "$WARP_BIN" --warp-mode put
+    --endpoint "$ADDRESS" --access-key "$ACCESS_KEY" --secret-key "$SECRET_KEY"
+    --region "$REGION" --bucket "$bucket" --sizes "$size" --concurrency "$CONCURRENCY"
+    --duration "$DATASET_SETUP_DURATION" --rounds 1 --cooldown-secs 0
+    --extra-args "--noclear"
+    --out-dir "$setup_cell"
+  )
+  log "preparing isolated dataset: $sync_label/$workload/$leg bucket=$bucket"
+  run "$ENHANCED_BENCH" "${args[@]}" >&2
+}
+
 write_schedule_header() {
-  echo "sync_label,drive_sync,workload,mode,size,leg,phase,binary,out_dir" >"$OUT_DIR/abba_schedule.csv"
+  echo "sync_label,drive_sync,workload,mode,size,leg,phase,binary,out_dir,bucket,dataset_setup" >"$OUT_DIR/abba_schedule.csv"
 }
 
 append_schedule() {
-  local sync_label="$1" drive_sync="$2" workload="$3" mode="$4" size="$5" leg="$6"
+  local sync_label="$1" drive_sync="$2" workload="$3" mode="$4" size="$5" leg="$6" bucket="$7"
   local phase bin
   phase="$(phase_for_leg "$leg")"
   bin="$(binary_for_leg "$leg")"
-  echo "$sync_label,$drive_sync,$workload,$mode,$size,$leg,$phase,$bin,$OUT_DIR/$workload/$sync_label/$leg" >>"$OUT_DIR/abba_schedule.csv"
+  local dataset_setup="none"
+  [[ "$mode" == "put" ]] || dataset_setup="warp-put"
+  echo "$sync_label,$drive_sync,$workload,$mode,$size,$leg,$phase,$bin,$OUT_DIR/$workload/$sync_label/$leg,$bucket,$dataset_setup" >>"$OUT_DIR/abba_schedule.csv"
+}
+
+sha256_file() {
+  if command -v sha256sum >/dev/null 2>&1; then
+    sha256sum "$1" | awk '{print $1}'
+  elif command -v shasum >/dev/null 2>&1; then
+    shasum -a 256 "$1" | awk '{print $1}'
+  else
+    echo unknown
+  fi
+}
+
+isolation_mode() {
+  if [[ "$EXTERNAL" != "true" ]]; then
+    echo local-per-cell-disks
+  elif [[ -n "$DEPLOY_HOOK" ]]; then
+    echo deploy-hook-attested
+  else
+    echo unmanaged
+  fi
 }
 
 write_manifest() {
+  local baseline_sha candidate_sha workloads matrix
+  baseline_sha="$(sha256_file "$BASELINE_BIN" 2>/dev/null || echo unknown)"
+  candidate_sha="$(sha256_file "$CANDIDATE_BIN" 2>/dev/null || echo unknown)"
+  workloads="$(IFS=';'; echo "${WORKLOADS[*]}")"
+  matrix="$(IFS=';'; echo "${DRIVE_SYNC_MATRIX[*]}")"
   cat >"$OUT_DIR/manifest.env" <<EOF
 generated_at_utc=$(date -u +%Y-%m-%dT%H:%M:%SZ 2>/dev/null || echo unknown)
 runner=$(uname -srm 2>/dev/null || echo unknown)
@@ -334,7 +439,19 @@ cooldown_secs=$COOLDOWN_SECS
 concurrency=$CONCURRENCY
 baseline_bin=$BASELINE_BIN
 candidate_bin=$CANDIDATE_BIN
+baseline_revision=$BASELINE_REVISION
+candidate_revision=$CANDIDATE_REVISION
+baseline_binary_sha256=$baseline_sha
+candidate_binary_sha256=$candidate_sha
+workloads=$workloads
+drive_sync_matrix=$matrix
 external=$EXTERNAL
+external_isolation=$(isolation_mode)
+dataset_namespace=$DATASET_NAMESPACE
+local_run_data_root=$RUN_DATA_ROOT
+bucket_isolation=per-leg
+bucket_prefix=rustfs-abba-$DATASET_NAMESPACE
+dataset_setup=get-and-mixed-via-warp-put
 endpoint=$ADDRESS
 warp_version=$("$WARP_BIN" --version 2>/dev/null | head -n1 || echo unknown)
 EOF
@@ -349,37 +466,37 @@ write_schedule_header
 for ds_spec in "${DRIVE_SYNC_MATRIX[@]}"; do
   IFS='|' read -r sync_label drive_sync <<<"$ds_spec"
 
-  for leg in A1 B1 B2 A2; do
-    log "=== $sync_label leg $leg ($(phase_for_leg "$leg")) ==="
-    bring_up "$leg" "$drive_sync"
-
-    for wl_spec in "${WORKLOADS[@]}"; do
-      IFS='|' read -r workload mode size <<<"$wl_spec"
-      append_schedule "$sync_label" "$drive_sync" "$workload" "$mode" "$size" "$leg"
+  for wl_spec in "${WORKLOADS[@]}"; do
+    IFS='|' read -r workload mode size <<<"$wl_spec"
+    for leg in A1 B1 B2 A2; do
+      log "=== $sync_label $workload leg $leg ($(phase_for_leg "$leg")) ==="
+      bucket="$(bucket_for_leg "$sync_label" "$workload" "$leg")"
+      bring_up "$leg" "$drive_sync" "$workload" "$mode" "$size" "$sync_label" "$bucket"
+      prepare_dataset "$leg" "$workload" "$mode" "$size" "$sync_label" "$bucket"
+      append_schedule "$sync_label" "$drive_sync" "$workload" "$mode" "$size" "$leg" "$bucket"
 
       baseline_csv=""
       if [[ "$leg" != "A1" ]]; then
         baseline_csv="$OUT_DIR/$workload/$sync_label/A1/median_summary.csv"
       fi
 
-      cell="$(measure "$leg" "$workload" "$mode" "$size" "$sync_label" "$baseline_csv")"
+      cell="$(measure "$leg" "$workload" "$mode" "$size" "$sync_label" "$bucket" "$baseline_csv")"
       case "$leg" in
         B1|B2) CANDIDATE_COMPARE_CSVS+=("$cell/baseline_compare.csv") ;;
         A2) DRIFT_COMPARE_CSVS+=("$cell/baseline_compare.csv") ;;
       esac
+      tear_down
     done
-
-    tear_down
   done
 done
 
-gate_args=(--fail-pct "$FAIL_PCT" --warn-pct "$WARN_PCT" --markdown "$OUT_DIR/candidate_gate.md")
+gate_args=(--fail-pct "$FAIL_PCT" --warn-pct "$WARN_PCT" --require-tail-error --markdown "$OUT_DIR/candidate_gate.md")
 for csv in "${CANDIDATE_COMPARE_CSVS[@]}"; do
   gate_args+=(--compare-csv "$csv")
 done
 [[ "$ALLOW_REGRESSION" == "true" ]] && gate_args+=(--allow-regression --exemption-reason "$EXEMPTION_REASON")
 
-drift_gate_args=(--fail-pct "$FAIL_PCT" --warn-pct "$WARN_PCT" --markdown "$OUT_DIR/baseline_drift_gate.md")
+drift_gate_args=(--fail-pct "$FAIL_PCT" --warn-pct "$WARN_PCT" --require-tail-error --markdown "$OUT_DIR/baseline_drift_gate.md")
 for csv in "${DRIFT_COMPARE_CSVS[@]}"; do
   drift_gate_args+=(--compare-csv "$csv")
 done
diff --git a/scripts/run_object_batch_bench_enhanced.sh b/scripts/run_object_batch_bench_enhanced.sh
index 1dd1378af..d0f22853d 100755
--- a/scripts/run_object_batch_bench_enhanced.sh
+++ b/scripts/run_object_batch_bench_enhanced.sh
@@ -1289,7 +1289,7 @@ compare_baseline() {
     return
   fi
 
-  echo "size,tool,concurrency,new_median_reqps,baseline_median_reqps,delta_reqps_pct,new_median_latency_ms,baseline_median_latency_ms,delta_latency_pct,new_median_throughput_bps,baseline_median_throughput_bps,delta_throughput_pct" > "$COMPARE_CSV"
+  echo "size,tool,concurrency,new_median_reqps,baseline_median_reqps,delta_reqps_pct,new_median_latency_ms,baseline_median_latency_ms,delta_latency_pct,new_median_throughput_bps,baseline_median_throughput_bps,delta_throughput_pct,new_median_p90_latency_ms,baseline_median_p90_latency_ms,delta_p90_latency_pct,new_median_p99_latency_ms,baseline_median_p99_latency_ms,delta_p99_latency_pct,new_ok_rounds,baseline_ok_rounds,new_failed_rounds,baseline_failed_rounds,new_error_rate_pct,baseline_error_rate_pct,delta_error_rate_pct" > "$COMPARE_CSV"
 
   awk -F',' '
     NR==FNR {
@@ -1298,22 +1298,35 @@ compare_baseline() {
       b_req[key]=$7
       b_lat[key]=$8
       b_thr[key]=$6
+      b_p90[key]=$9
+      b_p99[key]=$10
+      b_ok[key]=$4
+      b_fail[key]=$5
       next
     }
     FNR==1 {next}
     {
       key=$1
-      n_thr=$6; n_req=$7; n_lat=$8
+      n_thr=$6; n_req=$7; n_lat=$8; n_p90=$9; n_p99=$10; n_ok=$4; n_fail=$5
       br=(key in b_req)?b_req[key]:"N/A"
       bl=(key in b_lat)?b_lat[key]:"N/A"
       bt=(key in b_thr)?b_thr[key]:"N/A"
+      bp90=(key in b_p90)?b_p90[key]:"N/A"
+      bp99=(key in b_p99)?b_p99[key]:"N/A"
+      bok=(key in b_ok)?b_ok[key]:"N/A"
+      bfail=(key in b_fail)?b_fail[key]:"N/A"
 
-      dr="N/A"; dl="N/A"; dt="N/A"
+      dr="N/A"; dl="N/A"; dt="N/A"; dp90="N/A"; dp99="N/A"; ne="N/A"; be="N/A"; de="N/A"
       if (br!="N/A" && n_req!="N/A" && br+0!=0) dr=sprintf("%.2f", ((n_req-br)/br)*100)
       if (bl!="N/A" && n_lat!="N/A" && bl+0!=0) dl=sprintf("%.2f", ((n_lat-bl)/bl)*100)
       if (bt!="N/A" && n_thr!="N/A" && bt+0!=0) dt=sprintf("%.2f", ((n_thr-bt)/bt)*100)
+      if (bp90!="N/A" && n_p90!="N/A" && bp90+0!=0) dp90=sprintf("%.2f", ((n_p90-bp90)/bp90)*100)
+      if (bp99!="N/A" && n_p99!="N/A" && bp99+0!=0) dp99=sprintf("%.2f", ((n_p99-bp99)/bp99)*100)
+      if (n_ok!="N/A" && n_fail!="N/A" && n_ok+n_fail>0) ne=sprintf("%.2f", (n_fail/(n_ok+n_fail))*100)
+      if (bok!="N/A" && bfail!="N/A" && bok+bfail>0) be=sprintf("%.2f", (bfail/(bok+bfail))*100)
+      if (ne!="N/A" && be!="N/A") de=sprintf("%.2f", ne-be)
 
-      print key "," $2 "," $3 "," n_req "," br "," dr "," n_lat "," bl "," dl "," n_thr "," bt "," dt
+      print key "," $2 "," $3 "," n_req "," br "," dr "," n_lat "," bl "," dl "," n_thr "," bt "," dt "," n_p90 "," bp90 "," dp90 "," n_p99 "," bp99 "," dp99 "," n_ok "," bok "," n_fail "," bfail "," ne "," be "," de
     }
   ' "$BASELINE_CSV" "$MEDIAN_CSV" >> "$COMPARE_CSV"
 }
diff --git a/scripts/test_hotpath_warp_ab_gate.sh b/scripts/test_hotpath_warp_ab_gate.sh
new file mode 100755
index 000000000..62a6e16f3
--- /dev/null
+++ b/scripts/test_hotpath_warp_ab_gate.sh
@@ -0,0 +1,83 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+GATE="${SCRIPT_DIR}/hotpath_warp_ab_gate.sh"
+TMP_DIR="$(mktemp -d)"
+
+cleanup() {
+  rm -rf "$TMP_DIR"
+}
+trap cleanup EXIT
+
+header='size,tool,concurrency,new_median_reqps,baseline_median_reqps,delta_reqps_pct,new_median_latency_ms,baseline_median_latency_ms,delta_latency_pct,new_median_throughput_bps,baseline_median_throughput_bps,delta_throughput_pct,new_median_p90_latency_ms,baseline_median_p90_latency_ms,delta_p90_latency_pct,new_median_p99_latency_ms,baseline_median_p99_latency_ms,delta_p99_latency_pct,new_ok_rounds,baseline_ok_rounds,new_failed_rounds,baseline_failed_rounds,new_error_rate_pct,baseline_error_rate_pct,delta_error_rate_pct'
+
+printf '%s\n' "$header" >"$TMP_DIR/equal-budget.csv"
+printf '%s\n' '1MiB,warp,1,100,100,0.00,1,1,0.00,100,100,0.00,2,2,0.00,3,3,0.00,9,10,1,0,10.00,0.00,10.00' >>"$TMP_DIR/equal-budget.csv"
+"$GATE" --require-tail-error --fail-pct 10 --warn-pct 5 --compare-csv "$TMP_DIR/equal-budget.csv" >/dev/null
+
+printf '%s\n' "$header" >"$TMP_DIR/rounded-error.csv"
+printf '%s\n' '1MiB,warp,1,100,100,0.00,1,1,0.00,100,100,0.00,2,2,0.00,3,3,0.00,2,5,1,1,33.33,16.67,16.67' >>"$TMP_DIR/rounded-error.csv"
+if "$GATE" --require-tail-error --fail-pct 20 --warn-pct 10 --compare-csv "$TMP_DIR/rounded-error.csv" >/dev/null; then
+  :
+else
+  echo "expected strict gate to accept error-rate deltas rounded from raw fractions" >&2
+  exit 1
+fi
+
+printf '%s\n' "$header" >"$TMP_DIR/p99-regression.csv"
+printf '%s\n' '1MiB,warp,1,100,100,0.00,1,1,0.00,100,100,0.00,2,2,0.00,4,3,33.33,3,3,0,0,0.00,0.00,0.00' >>"$TMP_DIR/p99-regression.csv"
+if "$GATE" --require-tail-error --compare-csv "$TMP_DIR/p99-regression.csv" >/dev/null 2>&1; then
+  echo "expected p99-only regression to fail the strict gate" >&2
+  exit 1
+fi
+
+printf '%s\n' "$header" >"$TMP_DIR/missing-evidence.csv"
+printf '%s\n' '1MiB,warp,1,100,100,0.00,1,1,0.00,100,100,0.00,N/A,N/A,N/A,N/A,N/A,N/A,0,3,3,0,N/A,0.00,N/A' >>"$TMP_DIR/missing-evidence.csv"
+if "$GATE" --require-tail-error --compare-csv "$TMP_DIR/missing-evidence.csv" >/dev/null 2>&1; then
+  echo "expected missing tail/error evidence or zero successes to fail the strict gate" >&2
+  exit 1
+fi
+
+printf '%s\n' 'size,tool,concurrency,new_median_reqps,baseline_median_reqps,delta_reqps_pct' >"$TMP_DIR/old-schema.csv"
+printf '%s\n' '1MiB,warp,1,100,100,0.00' >>"$TMP_DIR/old-schema.csv"
+if "$GATE" --require-tail-error --compare-csv "$TMP_DIR/old-schema.csv" >/dev/null 2>&1; then
+  echo "expected strict gate to reject the old comparison schema" >&2
+  exit 1
+fi
+
+printf '%s\n' "$header" >"$TMP_DIR/header-only.csv"
+if "$GATE" --require-tail-error --compare-csv "$TMP_DIR/header-only.csv" >/dev/null 2>&1; then
+  echo "expected strict gate to reject a comparison without evidence rows" >&2
+  exit 1
+fi
+
+printf '%s\n' "$header" >"$TMP_DIR/malformed-number.csv"
+printf '%s\n' '1MiB,warp,1,100,100,0.00,1,1,0.00,100,100,0.00,2,2,0.00,3,3,0.00,3,3,0,0,0x0,0.00,0.00' >>"$TMP_DIR/malformed-number.csv"
+if "$GATE" --require-tail-error --compare-csv "$TMP_DIR/malformed-number.csv" >/dev/null 2>&1; then
+  echo "expected strict gate to reject malformed numeric evidence" >&2
+  exit 1
+fi
+
+printf '%s\n' "$header" >"$TMP_DIR/out-of-range-error.csv"
+printf '%s\n' '1MiB,warp,1,100,100,0.00,1,1,0.00,100,100,0.00,2,2,0.00,3,3,0.00,3,3,0,0,101.00,0.00,101.00' >>"$TMP_DIR/out-of-range-error.csv"
+if "$GATE" --require-tail-error --compare-csv "$TMP_DIR/out-of-range-error.csv" >/dev/null 2>&1; then
+  echo "expected strict gate to reject out-of-range error evidence" >&2
+  exit 1
+fi
+
+printf '%s\n' "$header" >"$TMP_DIR/inconsistent-error.csv"
+printf '%s\n' '1MiB,warp,1,100,100,0.00,1,1,0.00,100,100,0.00,2,2,0.00,3,3,0.00,3,3,0,0,100.00,0.00,0.00' >>"$TMP_DIR/inconsistent-error.csv"
+if "$GATE" --require-tail-error --compare-csv "$TMP_DIR/inconsistent-error.csv" >/dev/null 2>&1; then
+  echo "expected strict gate to reject error rates inconsistent with round counts" >&2
+  exit 1
+fi
+
+printf '%s\n' "$header,unexpected" >"$TMP_DIR/extra-column.csv"
+printf '%s\n' '1MiB,warp,1,100,100,0.00,1,1,0.00,100,100,0.00,2,2,0.00,3,3,0.00,3,3,0,0,0.00,0.00,0.00,unexpected' >>"$TMP_DIR/extra-column.csv"
+if "$GATE" --require-tail-error --compare-csv "$TMP_DIR/extra-column.csv" >/dev/null 2>&1; then
+  echo "expected strict gate to reject extra columns" >&2
+  exit 1
+fi
+
+echo "hotpath warp A/B gate tests passed"
diff --git a/scripts/test_hotpath_warp_abba.sh b/scripts/test_hotpath_warp_abba.sh
new file mode 100755
index 000000000..5a24f97de
--- /dev/null
+++ b/scripts/test_hotpath_warp_abba.sh
@@ -0,0 +1,113 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+RUNNER="${SCRIPT_DIR}/run_hotpath_warp_abba.sh"
+TMP_DIR="$(mktemp -d)"
+OUT_DIR="${TMP_DIR}/abba"
+TRACE_FILE="${TMP_DIR}/abba-trace.log"
+
+cleanup() {
+  rm -rf "$TMP_DIR"
+}
+trap cleanup EXIT
+
+"$RUNNER" \
+  --baseline-bin /usr/bin/true \
+  --candidate-bin /usr/bin/true \
+  --baseline-revision baseline-test \
+  --candidate-revision candidate-test \
+  --warp-bin /usr/bin/true \
+  --rounds 3 \
+  --out-dir "$OUT_DIR" \
+  --dry-run >"$TRACE_FILE" 2>&1
+
+awk -F',' '
+  NR == 1 {
+    if ($1 != "sync_label" || $6 != "leg" || $7 != "phase" || $10 != "bucket" || $11 != "dataset_setup") exit 1
+    next
+  }
+  {
+    key = $1 SUBSEP $3
+    seen[key]++
+    leg[key, seen[key]] = $6
+    phase[key, seen[key]] = $7
+    bucket[key, seen[key]] = $10
+  }
+  END {
+    for (key in seen) {
+      if (seen[key] != 4) exit 1
+      if (leg[key, 1] != "A1" || leg[key, 2] != "B1" || leg[key, 3] != "B2" || leg[key, 4] != "A2") exit 1
+      if (phase[key, 1] != "baseline" || phase[key, 2] != "candidate" || phase[key, 3] != "candidate" || phase[key, 4] != "baseline") exit 1
+      for (slot = 1; slot <= 4; slot++) {
+        if (length(bucket[key, slot]) > 63 || bucket[key, slot] !~ /^rustfs-abba-[a-z0-9-]+$/ || bucket[key, slot] in all_buckets) exit 1
+        all_buckets[bucket[key, slot]] = 1
+      }
+      cells++
+    }
+    exit cells == 12 ? 0 : 1
+  }
+' "$OUT_DIR/abba_schedule.csv"
+
+rg -qx 'schedule=ABBA' "$OUT_DIR/manifest.env"
+rg -qx 'rounds=3' "$OUT_DIR/manifest.env"
+rg -qx 'baseline_revision=baseline-test' "$OUT_DIR/manifest.env"
+rg -qx 'candidate_revision=candidate-test' "$OUT_DIR/manifest.env"
+rg -qx 'external_isolation=local-per-cell-disks' "$OUT_DIR/manifest.env"
+rg -qx 'bucket_isolation=per-leg' "$OUT_DIR/manifest.env"
+rg -qx 'dataset_setup=get-and-mixed-via-warp-put' "$OUT_DIR/manifest.env"
+[[ "$(rg -c -- '--extra-args --noclear' "$TRACE_FILE")" == "64" ]]
+! rg -q -- 'rustfs-bench' "$TRACE_FILE"
+
+if "$RUNNER" \
+  --baseline-bin /usr/bin/true \
+  --candidate-bin /usr/bin/true \
+  --warp-bin /usr/bin/true \
+  --rounds 3 \
+  --out-dir "${TMP_DIR}/missing-revisions" >/dev/null 2>&1; then
+  echo "expected formal mode to reject unknown binary revisions" >&2
+  exit 1
+fi
+
+mkdir -p "${TMP_DIR}/data/reused/put-4kib-A1-sync-true"
+printf '%s\n' sentinel >"${TMP_DIR}/data/reused/put-4kib-A1-sync-true/sentinel"
+if "$RUNNER" \
+  --baseline-bin /usr/bin/true \
+  --candidate-bin /usr/bin/true \
+  --baseline-revision baseline-test \
+  --candidate-revision candidate-test \
+  --warp-bin /usr/bin/true \
+  --rounds 3 \
+  --data-root "${TMP_DIR}/data" \
+  --out-dir "${TMP_DIR}/reused" >/dev/null 2>&1; then
+  echo "expected local mode to reject a reused run namespace" >&2
+  exit 1
+fi
+
+if "$RUNNER" \
+  --baseline-bin /usr/bin/true \
+  --candidate-bin /usr/bin/true \
+  --endpoint 127.0.0.1:9000 \
+  --warp-bin /usr/bin/true \
+  --rounds 3 \
+  --out-dir "${TMP_DIR}/external" \
+  --dry-run >/dev/null 2>&1; then
+  echo "expected formal external mode to require a deploy hook" >&2
+  exit 1
+fi
+
+if "$RUNNER" \
+  --baseline-bin /usr/bin/true \
+  --candidate-bin /usr/bin/true \
+  --baseline-revision baseline-test \
+  --candidate-revision candidate-test \
+  --endpoint 127.0.0.1:9000 \
+  --deploy-hook ':' \
+  --warp-bin /usr/bin/true \
+  --rounds 3 \
+  --out-dir "${TMP_DIR}/missing-deploy-evidence" >/dev/null 2>&1; then
+  echo "expected a formal external deploy hook to write evidence" >&2
+  exit 1
+fi
+
+echo "hotpath warp ABBA tests passed"
diff --git a/scripts/test_object_batch_bench_enhanced.sh b/scripts/test_object_batch_bench_enhanced.sh
index 6163da2f8..6722f723f 100755
--- a/scripts/test_object_batch_bench_enhanced.sh
+++ b/scripts/test_object_batch_bench_enhanced.sh
@@ -102,3 +102,34 @@ chmod +x "$FAKE_WARP"
   --require-server-provenance >/dev/null 2>&1
 
 rg -q '^32767B,warp,1,1,128,ok,0,[^,]+,[^,]+,653.90 MiB/s,685663846.400000,20925.58,3.5 ms,3.500000,[^,]+,3.6 ms,3.600000,24.1 ms,24.100000$' "${TMP_DIR}/fake-warp-run/round_results.csv"
+
+"$RUNNER" \
+  --tool warp \
+  --endpoint http://127.0.0.1:9000 \
+  --access-key test-access \
+  --secret-key test-secret \
+  --sizes 32767B \
+  --rounds 1 \
+  --retry-per-round 1 \
+  --cooldown-secs 0 \
+  --duration 1s \
+  --out-dir "${TMP_DIR}/fake-warp-candidate" \
+  --warp-bin "$FAKE_WARP" \
+  --baseline-csv "${TMP_DIR}/fake-warp-run/median_summary.csv" \
+  --server-image-ref rustfs/rustfs:bench \
+  --server-image-digest sha256:0123456789abcdef \
+  --server-revision 9f61bad94 \
+  --require-server-provenance >/dev/null 2>&1
+
+awk -F',' '
+  NR == 1 {
+    if (NF != 25 || $15 != "delta_p90_latency_pct" || $18 != "delta_p99_latency_pct" || $25 != "delta_error_rate_pct") exit 1
+  }
+  NR == 2 {
+    if ($15 != "0.00" || $18 != "0.00" || $23 != "0.00" || $25 != "0.00") exit 1
+    found = 1
+  }
+  END { exit found ? 0 : 1 }
+' "${TMP_DIR}/fake-warp-candidate/baseline_compare.csv"
+
+echo "object batch benchmark enhanced tests passed"