@@ -2330,6 +2330,46 @@ mod tests {
     use insta::assert_snapshot;
     use tokio::sync::Notify;
 
+    #[tokio::test]
+    async fn sort_preparation_waits_for_permit_and_cancels() {
+        let permits = Arc::new(Semaphore::new(1));
+        let held = permits.acquire().await.unwrap();
+        let stream = Box::pin(RecordBatchStreamAdapter::new(
+            Arc::new(Schema::empty()),
+            futures::stream::pending(),
+        ));
+        let mut preparation =
+            ExternalSorter::prepare_sort_run(stream, Arc::clone(&permits)).boxed();
+        assert_is_pending(&mut preparation);
+        assert_eq!(Handle::current().metrics().num_alive_tasks(), 0);
+        drop(held);
+        assert_is_pending(&mut preparation);
+        // The current-thread runtime cannot run the spawned task until we yield.
+        assert_eq!(Handle::current().metrics().num_alive_tasks(), 1);
+        tokio::task::yield_now().await;
+        drop(preparation);
+        let _permit =
+            tokio::time::timeout(std::time::Duration::from_secs(10), permits.acquire())
+                .await
+                .expect("cancelled preparation must release its permit")
+                .unwrap();
+    }
+
+    #[tokio::test]
+    async fn sort_preparation_propagates_errors_and_releases_permit() {
+        let permits = Arc::new(Semaphore::new(1));
+        let stream = Box::pin(RecordBatchStreamAdapter::new(
+            Arc::new(Schema::empty()),
+            futures::stream::iter([Err(internal_datafusion_err!("sort failed"))]),
+        ));
+        let error = ExternalSorter::prepare_sort_run(stream, Arc::clone(&permits))
+            .await
+            .err()
+            .expect("preparation must propagate the stream error");
+        assert!(error.to_string().contains("sort failed"), "{error}");
+        assert_eq!(permits.available_permits(), 1);
+    }
+
     struct PendingWriteTempFileFactory {
         write_started: Arc<Notify>,
         aborted: Arc<Notify>,
 
@@ -234,7 +234,7 @@ async fn assert_released(pool: &AdjustablePool, runtime: &RuntimeEnv) {
     assert_eq!(state.unchecked_over_limit, 0);
 }
 
-#[tokio::test]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn test_spill_preserves_merge_workspace_after_limit_decreases() -> Result<()> {
     let fixture = fixture()?;
     let pool = AdjustablePool::new(fixture.capacity);
@@ -522,6 +522,86 @@ async fn test_chunked_string_view_spill_can_use_workspace() -> Result<()> {
     check_chunked_string_view_workspace(true).await
 }
 
+/// Two sort jobs share a tight memory budget. The first must return borrowed
+/// workspace when it hands off its first batch, so the second can prepare
+/// before the merge starts consuming output. During merging, looking ahead to
+/// the next batch must also return workspace for the merge's cursors.
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+async fn test_sort_preparation_returns_first_batch_workspace() -> Result<()> {
+    let batch_size = 1024;
+    let values: ArrayRef = Arc::new(StringViewArray::from_iter_values(
+        (0..4096)
+            .rev()
+            .map(|i| format!("row-{i:08}-{}", "x".repeat(87))),
+    ));
+    let batch = RecordBatch::try_from_iter([("key", values)])?;
+    let ordering: LexOrdering = [PhysicalSortExpr::new_default(Arc::new(Column::new(
+        "key", 0,
+    )))]
+    .into();
+    let input_bytes = get_reserved_bytes_for_record_batch(&batch)?;
+    let sorted_batches = sort_batch_chunked(&batch, &ordering, batch_size)?;
+    let loan = sorted_batches
+        .iter()
+        .map(get_record_batch_memory_size)
+        .sum::<usize>()
+        - input_bytes;
+    let first_bytes = get_record_batch_memory_size(&sorted_batches[0]);
+    assert!(loan > first_bytes);
+
+    for consume_output in [false, true] {
+        let workspace = if consume_output {
+            // Also allow the merge's cursors, provided each run looks ahead.
+            2 * loan + first_bytes
+        } else {
+            2 * loan - first_bytes
+        };
+        let runtime = RuntimeEnvBuilder::new()
+            .with_memory_pool(Arc::new(GreedyMemoryPool::new(
+                2 * input_bytes + workspace,
+            )))
+            .build_arc()?;
+        let mut sorter = ExternalSorter::new(
+            0,
+            batch.schema(),
+            ordering.clone(),
+            batch_size,
+            workspace,
+            0,
+            SpillCompression::Uncompressed,
+            &ExecutionPlanMetricsSet::new(),
+            Arc::clone(&runtime),
+        )?;
+        sorter.insert_batch(batch.clone()).await?;
+        sorter.insert_batch(batch.clone()).await?;
+        sorter.merge_reservation.free();
+        let stream = sorter.in_mem_sort_stream(false, false).await?;
+        // Buffered producers would leave two tasks waiting for the merge.
+        let alive = tokio::runtime::Handle::current()
+            .metrics()
+            .num_alive_tasks();
+        assert!(alive <= 1, "{alive} sort tasks remain before merging");
+        assert_eq!(
+            sorter.merge_pool.borrow(usize::MAX).size(),
+            workspace - 2 * (loan - first_bytes)
+        );
+        if consume_output {
+            let output: Vec<RecordBatch> = stream.try_collect().await?;
+            let expected = sort_batch(
+                &concat_batches(&batch.schema(), &[batch.clone(), batch.clone()])?,
+                &ordering,
+                None,
+            )?;
+            assert_eq!(concat_batches(&batch.schema(), &output)?, expected);
+        } else {
+            drop(stream);
+        }
+        drop(sorter);
+        assert_eq!(runtime.memory_pool.reserved(), 0);
+    }
+    Ok(())
+}
+
 #[tokio::test]
 async fn test_single_batch_spill_preserves_workspace_after_limit_decreases() -> Result<()>
 {