diff --git a/rust/worker/src/execution/orchestration/async_function_boundary.rs b/rust/worker/src/execution/orchestration/async_function_boundary.rs
index 8bfe997de31..ab2da172b36 100644
--- a/rust/worker/src/execution/orchestration/async_function_boundary.rs
+++ b/rust/worker/src/execution/orchestration/async_function_boundary.rs
@@ -54,26 +54,14 @@ pub(crate) fn resolve_boundary_plan_from_version_file(
         })
         .collect::<Vec<_>>();
 
-    // Walk versions newest -> oldest. Boundaries above the completion offset
-    // are visited furthest-first, so the first one whose window fits
-    // max_compaction_size is the widest eligible target. Tracking the nearest
-    // boundary as well preserves the oversized-window error below when no
-    // boundary fits.
     let mut historical_version = None;
     let mut next_boundary = None;
-    let mut furthest_fitting_boundary = None;
     for (version, log_position) in version_infos.into_iter().rev() {
         if log_position <= completion_offset {
             historical_version = Some((version, log_position));
             break;
         }
 
-        if furthest_fitting_boundary.is_none()
-            && usize::try_from(log_position - completion_offset)
-                .is_ok_and(|window| window <= max_compaction_size)
-        {
-            furthest_fitting_boundary = Some(log_position);
-        }
         next_boundary = Some(log_position);
     }
 
@@ -90,13 +78,7 @@ pub(crate) fn resolve_boundary_plan_from_version_file(
         None => None,
     };
 
-    // Prefer the furthest boundary that fits: one run then covers every
-    // compaction between the completion offset and that boundary, instead of
-    // draining a backlog one small compaction window at a time. Skipped
-    // intermediate boundaries are safe — the completion offset advances to a
-    // real boundary either way, and the work queue retires stale entries by
-    // offset comparison.
-    let target_log_position = furthest_fitting_boundary.or(next_boundary).ok_or_else(|| {
+    let target_log_position = next_boundary.ok_or_else(|| {
         format!(
             "async fn completion offset {} has no next compaction boundary",
             completion_offset
@@ -211,7 +193,7 @@ mod tests {
     }
 
     #[test]
-    fn completion_offset_zero_uses_empty_state_and_widest_fitting_boundary() {
+    fn completion_offset_zero_uses_empty_state_and_first_boundary() {
         let record_segment = test_record_segment();
         let version_file = CollectionVersionFile {
             version_history: Some(CollectionVersionHistory {
@@ -227,122 +209,13 @@ mod tests {
             resolve_boundary_plan_from_version_file(Some(&version_file), 0, 1024, &record_segment)
                 .unwrap();
 
-        assert_eq!(plan.target_log_position, 150);
+        assert_eq!(plan.target_log_position, 100);
         assert!(
             plan.historical_record_segment.is_none(),
             "completion offset zero should use the empty pre-compaction state"
         );
     }
 
-    #[test]
-    fn picks_furthest_boundary_that_fits_max_compaction_size() {
-        let record_segment = test_record_segment();
-        let version_file = CollectionVersionFile {
-            version_history: Some(CollectionVersionHistory {
-                versions: vec![
-                    version_info(1, 100, record_segment.id, "record/v100"),
-                    version_info(2, 150, record_segment.id, "record/v150"),
-                    version_info(3, 200, record_segment.id, "record/v200"),
-                ],
-            }),
-            ..Default::default()
-        };
-
-        let plan = resolve_boundary_plan_from_version_file(
-            Some(&version_file),
-            100,
-            1024,
-            &record_segment,
-        )
-        .unwrap();
-
-        assert_eq!(plan.target_log_position, 200);
-        assert_eq!(
-            plan.historical_record_segment.unwrap().file_path["offset_id_to_data"],
-            vec!["record/v100".to_string()]
-        );
-    }
-
-    #[test]
-    fn skips_boundaries_wider_than_max_compaction_size() {
-        let record_segment = test_record_segment();
-        let version_file = CollectionVersionFile {
-            version_history: Some(CollectionVersionHistory {
-                versions: vec![
-                    version_info(1, 100, record_segment.id, "record/v100"),
-                    version_info(2, 150, record_segment.id, "record/v150"),
-                    version_info(3, 200, record_segment.id, "record/v200"),
-                    version_info(4, 5000, record_segment.id, "record/v5000"),
-                ],
-            }),
-            ..Default::default()
-        };
-
-        let plan = resolve_boundary_plan_from_version_file(
-            Some(&version_file),
-            100,
-            1000,
-            &record_segment,
-        )
-        .unwrap();
-
-        assert_eq!(plan.target_log_position, 200);
-    }
-
-    #[test]
-    fn errors_when_no_boundary_fits_max_compaction_size() {
-        let record_segment = test_record_segment();
-        let version_file = CollectionVersionFile {
-            version_history: Some(CollectionVersionHistory {
-                versions: vec![
-                    version_info(1, 100, record_segment.id, "record/v100"),
-                    version_info(2, 5000, record_segment.id, "record/v5000"),
-                ],
-            }),
-            ..Default::default()
-        };
-
-        let err = resolve_boundary_plan_from_version_file(
-            Some(&version_file),
-            100,
-            1000,
-            &record_segment,
-        )
-        .unwrap_err();
-
-        assert!(err.contains("exceeds max_compaction_size"));
-    }
-
-    #[test]
-    fn deleted_versions_are_not_widened_targets() {
-        let record_segment = test_record_segment();
-        let mut deleted_version = version_info(2, 150, record_segment.id, "record/v150");
-        deleted_version.marked_for_deletion = true;
-
-        let version_file = CollectionVersionFile {
-            version_history: Some(CollectionVersionHistory {
-                versions: vec![
-                    version_info(1, 100, record_segment.id, "record/v100"),
-                    deleted_version,
-                    version_info(3, 5000, record_segment.id, "record/v5000"),
-                ],
-            }),
-            ..Default::default()
-        };
-
-        // The only live boundary above the offset (5000) does not fit, and the
-        // deleted 150 boundary must not be picked in its place.
-        let err = resolve_boundary_plan_from_version_file(
-            Some(&version_file),
-            100,
-            1000,
-            &record_segment,
-        )
-        .unwrap_err();
-
-        assert!(err.contains("exceeds max_compaction_size"));
-    }
-
     #[test]
     fn rejects_non_boundary_completion_offsets_after_first_compaction() {
         let record_segment = test_record_segment();
diff --git a/rust/worker/src/execution/orchestration/compact.rs b/rust/worker/src/execution/orchestration/compact.rs
index 47020c0e8de..207456d66b3 100644
--- a/rust/worker/src/execution/orchestration/compact.rs
+++ b/rust/worker/src/execution/orchestration/compact.rs
@@ -5096,7 +5096,7 @@ mod tests {
 
         // Run three regular compactions to create boundaries at 99, 199, and 299.
         // The async fn-consumer will later read from the historical version at 99
-        // and fetch through the furthest boundary that fits max_compaction_size.
+        // and fetch only through the next boundary at 199.
         for expected_log_position in [99, 199, 299] {
             let compact_result = Box::pin(compact(
                 system.clone(),
@@ -5138,11 +5138,10 @@ mod tests {
             );
         }
 
-        // Now simulate fn-consumer execution from the first boundary. With a
-        // max_compaction_size (1000) wider than the whole backlog, this must:
+        // Now simulate fn-consumer execution from the first boundary. This must:
         // 1. read against the historical version compacted through offset 99
-        // 2. fetch through the furthest committed boundary (299) in one window
-        // 3. materialize exactly the 200 records in (99, 299]
+        // 2. stop fetching at the next committed boundary (199)
+        // 3. materialize exactly the 100 records in (99, 199]
         let mut fn_consumer_context = CompactionContext::new_with_log_offset(
             None,
             50,
@@ -5178,9 +5177,8 @@ mod tests {
         match fetch_response {
             LogFetchOrchestratorResponse::Success(success) => {
                 assert_eq!(
-                    success.collection_info.pulled_log_offset, 299,
-                    "fn-consumer should fetch through the furthest committed boundary \
-                     that fits max_compaction_size"
+                    success.collection_info.pulled_log_offset, 199,
+                    "fn-consumer should only fetch through the next committed boundary"
                 );
 
                 let total_materialized_records: usize = success
@@ -5189,8 +5187,8 @@ mod tests {
                     .map(|batch| batch.result.len())
                     .sum();
                 assert_eq!(
-                    total_materialized_records, 200,
-                    "fn-consumer should materialize every record in the widened window"
+                    total_materialized_records, 100,
+                    "fn-consumer should materialize exactly the records between boundaries"
                 );
 
                 for batch in &success.materialized {
@@ -5210,14 +5208,13 @@ mod tests {
 
         Box::pin(fn_consumer_context.cleanup()).await;
 
-        // Repeat from the first boundary with a max_compaction_size of exactly
-        // one window (100) to prove the cap forces the fetch back to the
-        // nearest committed boundary when the furthest one does not fit.
+        // Repeat from the second boundary to prove fn-consumer continues to hop
+        // exactly one committed compaction window at a time.
         let mut second_window_context = CompactionContext::new_with_log_offset(
             None,
             50,
             10,
-            100,
+            1000,
             50,
             log.clone(),
             sysdb.clone(),
@@ -5231,7 +5228,7 @@ mod tests {
             None,
             None,
             None,
-            99,
+            199,
         );
 
         let second_fetch_response = second_window_context
@@ -5248,9 +5245,8 @@ mod tests {
         match second_fetch_response {
             LogFetchOrchestratorResponse::Success(success) => {
                 assert_eq!(
-                    success.collection_info.pulled_log_offset, 199,
-                    "a tight max_compaction_size must stop the fetch at the nearest \
-                     fitting boundary"
+                    success.collection_info.pulled_log_offset, 299,
+                    "fn-consumer should continue to the next committed boundary on later runs"
                 );
 
                 let total_materialized_records: usize = success
@@ -5260,8 +5256,7 @@ mod tests {
                     .sum();
                 assert_eq!(
                     total_materialized_records, 100,
-                    "a capped fn-consumer run should materialize exactly one \
-                     compaction window"
+                    "each fn-consumer run should materialize exactly one compaction window"
                 );
 
                 for batch in &success.materialized {