diff --git a/doc/user/data/metrics.yml b/doc/user/data/metrics.yml
index 276de9f8288df..c3bc557d5321c 100644
--- a/doc/user/data/metrics.yml
+++ b/doc/user/data/metrics.yml
@@ -1032,18 +1032,18 @@ metrics:
   help: Time checking trace frontiers.
   source: src/compute/src/metrics.rs
   visibility: internal
-- name: mz_index_peek_result_sort_seconds_bucket
-  help: Time sorting intermediate results during peek collection.
+- name: mz_index_peek_result_thinning_seconds_bucket
+  help: Time thinning intermediate peek results down to the rows the finishing can need, summed over the activations an index peek took and reported once it is done. Peeks that are cancelled or dropped before finishing report nothing.
   labels:
   - le
   source: src/compute/src/metrics.rs
   visibility: internal
-- name: mz_index_peek_result_sort_seconds_count
-  help: Time sorting intermediate results during peek collection.
+- name: mz_index_peek_result_thinning_seconds_count
+  help: Time thinning intermediate peek results down to the rows the finishing can need, summed over the activations an index peek took and reported once it is done. Peeks that are cancelled or dropped before finishing report nothing.
   source: src/compute/src/metrics.rs
   visibility: internal
-- name: mz_index_peek_result_sort_seconds_sum
-  help: Time sorting intermediate results during peek collection.
+- name: mz_index_peek_result_thinning_seconds_sum
+  help: Time thinning intermediate peek results down to the rows the finishing can need, summed over the activations an index peek took and reported once it is done. Peeks that are cancelled or dropped before finishing report nothing.
   source: src/compute/src/metrics.rs
   visibility: internal
 - name: mz_index_peek_row_collection_seconds_bucket
@@ -1061,45 +1061,45 @@ metrics:
   source: src/compute/src/metrics.rs
   visibility: internal
 - name: mz_index_peek_row_iteration_seconds_bucket
-  help: Time iterating rows and evaluating MFP.
+  help: Time iterating rows and evaluating MFP, summed over the activations an index peek took and reported once it is done. Peeks that are cancelled or dropped before finishing report nothing.
   labels:
   - le
   source: src/compute/src/metrics.rs
   visibility: internal
 - name: mz_index_peek_row_iteration_seconds_count
-  help: Time iterating rows and evaluating MFP.
+  help: Time iterating rows and evaluating MFP, summed over the activations an index peek took and reported once it is done. Peeks that are cancelled or dropped before finishing report nothing.
   source: src/compute/src/metrics.rs
   visibility: internal
 - name: mz_index_peek_row_iteration_seconds_sum
-  help: Time iterating rows and evaluating MFP.
+  help: Time iterating rows and evaluating MFP, summed over the activations an index peek took and reported once it is done. Peeks that are cancelled or dropped before finishing report nothing.
   source: src/compute/src/metrics.rs
   visibility: internal
 - name: mz_index_peek_seek_fulfillment_seconds_bucket
-  help: Time in seek_fulfillment method including frontier checks and data collection.
+  help: Worker time spent in seek_fulfillment for an index peek, including every not-yet-ready frontier check, summed over activations and reported once the peek is done.
   labels:
   - le
   source: src/compute/src/metrics.rs
   visibility: internal
 - name: mz_index_peek_seek_fulfillment_seconds_count
-  help: Time in seek_fulfillment method including frontier checks and data collection.
+  help: Worker time spent in seek_fulfillment for an index peek, including every not-yet-ready frontier check, summed over activations and reported once the peek is done.
   source: src/compute/src/metrics.rs
   visibility: internal
 - name: mz_index_peek_seek_fulfillment_seconds_sum
-  help: Time in seek_fulfillment method including frontier checks and data collection.
+  help: Worker time spent in seek_fulfillment for an index peek, including every not-yet-ready frontier check, summed over activations and reported once the peek is done.
   source: src/compute/src/metrics.rs
   visibility: internal
 - name: mz_index_peek_total_seconds_bucket
-  help: Total time processing index peeks, from process_peek entry to response. Excluding peeks that use the peek response stash.
+  help: Worker time spent serving an index peek, summed over the activations it took and reported once it is done. Peeks that are cancelled or dropped before finishing report nothing.
   labels:
   - le
   source: src/compute/src/metrics.rs
   visibility: internal
 - name: mz_index_peek_total_seconds_count
-  help: Total time processing index peeks, from process_peek entry to response. Excluding peeks that use the peek response stash.
+  help: Worker time spent serving an index peek, summed over the activations it took and reported once it is done. Peeks that are cancelled or dropped before finishing report nothing.
   source: src/compute/src/metrics.rs
   visibility: internal
 - name: mz_index_peek_total_seconds_sum
-  help: Total time processing index peeks, from process_peek entry to response. Excluding peeks that use the peek response stash.
+  help: Worker time spent serving an index peek, summed over the activations it took and reported once it is done. Peeks that are cancelled or dropped before finishing report nothing.
   source: src/compute/src/metrics.rs
   visibility: internal
 - name: mz_kafka_partition_offset_max
@@ -1527,6 +1527,20 @@ metrics:
   help: The time it takes to parse a SQL statement. (Works for both Simple Queries and the Extended Query protocol.)
   source: src/adapter/src/metrics.rs
   visibility: internal
+- name: mz_peek_processing_seconds_bucket
+  help: Time a worker spent serving pending peeks in one activation, before returning to scheduling dataflows and handling commands. This is the latency the peek yielding budgets exist to bound. Only recorded for activations that had at least one pending peek.
+  labels:
+  - le
+  source: src/compute/src/metrics.rs
+  visibility: internal
+- name: mz_peek_processing_seconds_count
+  help: Time a worker spent serving pending peeks in one activation, before returning to scheduling dataflows and handling commands. This is the latency the peek yielding budgets exist to bound. Only recorded for activations that had at least one pending peek.
+  source: src/compute/src/metrics.rs
+  visibility: internal
+- name: mz_peek_processing_seconds_sum
+  help: Time a worker spent serving pending peeks in one activation, before returning to scheduling dataflows and handling commands. This is the latency the peek yielding budgets exist to bound. Only recorded for activations that had at least one pending peek.
+  source: src/compute/src/metrics.rs
+  visibility: internal
 - name: mz_persist_*_bytes
   help: total encoded size of * batches written
   source: src/persist-client/src/internal/metrics.rs
diff --git a/misc/python/materialize/mzcompose/__init__.py b/misc/python/materialize/mzcompose/__init__.py
index 9fb80b0a1b169..3088ae517486c 100644
--- a/misc/python/materialize/mzcompose/__init__.py
+++ b/misc/python/materialize/mzcompose/__init__.py
@@ -48,7 +48,15 @@
 # a new feature causes benchmarks to become flaky, consider that this can also
 # impact customers' experience and try to find a solution other than disabling
 # the feature here!
-ADDITIONAL_BENCHMARKING_SYSTEM_PARAMETERS = {}
+ADDITIONAL_BENCHMARKING_SYSTEM_PARAMETERS = {
+    # Benchmarks measure against an older Materialize image that does not know
+    # these parameters and therefore ignores them, so a non-production value
+    # here would only slow down one side of the comparison. Pin both to their
+    # production defaults, so peek-heavy scenarios measure the same thing on
+    # both sides.
+    "peek_yielding": "work:100000,time:10",
+    "peek_yielding_total": "work:1000000,time:100",
+}
 
 
 def get_minimal_system_parameters(
@@ -345,6 +353,27 @@ def get_variable_system_parameters(
         VariableSystemParameter(
             "persist_encoding_enable_dictionary", "true", ["true", "false"]
         ),
+        # A work budget well below production, so that any index peek over more
+        # than ~1000 rows resumes at least once and CI covers the resumable scan
+        # path rather than only the completes-in-one-slice path. Peeks over tiny
+        # relations, which is most of sqllogictest, still finish in one slice;
+        # that case is covered directly by
+        # test/clusterd-test-driver/scripts/peek_yielding.spec.
+        #
+        # Kept within an order of magnitude of production on purpose. The value
+        # reaches every mzcompose composition, and a very small budget buys
+        # little extra coverage while multiplying the timely steps a large peek
+        # needs. The randomized runs go lower.
+        VariableSystemParameter(
+            "peek_yielding",
+            "work:1024,time:10",
+            ["work:16,time:10", "work:1024,time:10", "work:100000,time:10"],
+        ),
+        VariableSystemParameter(
+            "peek_yielding_total",
+            "work:8192,time:100",
+            ["work:128,time:100", "work:8192,time:100", "work:1000000,time:100"],
+        ),
         VariableSystemParameter(
             "persist_fast_path_limit",
             "1000",
diff --git a/misc/python/materialize/parallel_workload/action.py b/misc/python/materialize/parallel_workload/action.py
index 44d6f9887ab6c..17fe072936f7e 100644
--- a/misc/python/materialize/parallel_workload/action.py
+++ b/misc/python/materialize/parallel_workload/action.py
@@ -1853,6 +1853,18 @@ def __init__(
             BOOLEAN_FLAG_VALUES
         )
         self.flags_with_values["enable_eager_delta_joins"] = BOOLEAN_FLAG_VALUES
+        # Small work budgets force index peeks to yield and resume, which is
+        # where the interesting state lives.
+        self.flags_with_values["peek_yielding"] = [
+            "'work:16,time:10'",
+            "'work:64,time:10'",
+            "'work:100000,time:10'",
+        ]
+        self.flags_with_values["peek_yielding_total"] = [
+            "'work:128,time:100'",
+            "'work:1024,time:100'",
+            "'work:1000000,time:100'",
+        ]
         self.flags_with_values["enable_public_metrics_endpoint"] = BOOLEAN_FLAG_VALUES
         self.flags_with_values["enable_scoped_system_parameters"] = BOOLEAN_FLAG_VALUES
         self.flags_with_values["persist_batch_structured_key_lower_len"] = [
diff --git a/src/clusterd-test-driver/src/driver.rs b/src/clusterd-test-driver/src/driver.rs
index 68ea7ec5c78c5..de9d93b885002 100644
--- a/src/clusterd-test-driver/src/driver.rs
+++ b/src/clusterd-test-driver/src/driver.rs
@@ -183,11 +183,18 @@ impl Driver {
     /// materialized-view sink's output shard, which is how `SELECT * FROM mv` reads.
     /// A persist peek blocks (async-friendly) until the shard seals through `ts`, so
     /// it doubles as a wait for the writing sink to catch up.
+    /// Peeks `target` at `ts`, optionally restricted to `literal_constraints`.
+    ///
+    /// Each literal is a row of the index's key columns, the same shape the
+    /// optimizer produces for `WHERE key IN (..)`. Only valid against an index
+    /// peek whose key the literals match. The replica sorts them itself, so they
+    /// may be passed in any order. `None` scans the whole arrangement.
     pub async fn peek(
         &self,
         target: PeekTarget,
         result_desc: RelationDesc,
         ts: Timestamp,
+        literal_constraints: Option<Vec<Row>>,
     ) -> anyhow::Result<Vec<Row>> {
         let uuid = uuid::Uuid::new_v4();
         let rx = self.responses.register_peek(uuid);
@@ -201,7 +208,7 @@ impl Driver {
         let peek = Peek {
             target,
             result_desc: result_desc.clone(),
-            literal_constraints: None,
+            literal_constraints,
             uuid,
             timestamp: ts,
             finishing: RowSetFinishing::trivial(arity),
@@ -232,8 +239,12 @@ impl Driver {
         target: PeekTarget,
         result_desc: RelationDesc,
         ts: Timestamp,
+        literal_constraints: Option<Vec<Row>>,
     ) -> anyhow::Result<usize> {
-        Ok(self.peek(target, result_desc, ts).await?.len())
+        Ok(self
+            .peek(target, result_desc, ts, literal_constraints)
+            .await?
+            .len())
     }
 
     /// Registers a subscribe-sink buffer for `id`, so the response pump accumulates
diff --git a/src/clusterd-test-driver/src/script.rs b/src/clusterd-test-driver/src/script.rs
index 5266ff13002e8..5e8122306a46a 100644
--- a/src/clusterd-test-driver/src/script.rs
+++ b/src/clusterd-test-driver/src/script.rs
@@ -45,8 +45,8 @@ use mz_expr_parser::{TestCatalog, try_parse_mir};
 use mz_persist_client::PersistClient;
 use mz_persist_types::{PersistLocation, ShardId};
 use mz_repr::{
-    GlobalId, RelationDesc, ReprRelationType, Row, SqlColumnType, SqlRelationType, SqlScalarType,
-    Timestamp, strconv,
+    Datum, GlobalId, RelationDesc, ReprRelationType, Row, SqlColumnType, SqlRelationType,
+    SqlScalarType, Timestamp, strconv,
 };
 use mz_storage_types::controller::CollectionMetadata;
 use serde::{Deserialize, Serialize};
@@ -462,6 +462,24 @@ pub enum Command {
         /// The timestamp to count at.
         ts: u64,
     },
+    /// Peek `id` directly and emit only how many rows came back.
+    ///
+    /// Unlike [`Command::Count`], which tallies through an ephemeral reduce
+    /// dataflow and then peeks its single-row result, this peeks `id` itself. So
+    /// the peek walks `id`'s whole arrangement, which is what makes it useful
+    /// for exercising the peek scan over many rows without a golden that has to
+    /// spell every one of them out.
+    PeekCount {
+        /// The index's global id.
+        id: u64,
+        /// The timestamp to peek at.
+        ts: u64,
+        /// Restrict the peek to these key values, as `WHERE key IN (..)` would.
+        ///
+        /// Only meaningful for a single-column `Int64` key, which is what the
+        /// sample schema's first column is. `None` scans the whole arrangement.
+        literals: Option<Vec<i64>>,
+    },
     /// Submit (without scheduling) a dataflow built from generic MIR — the
     /// abstraction behind index / materialized-view / subscribe / copy-to.
     ///
@@ -664,6 +682,7 @@ impl ScriptState {
                 PeekTarget::Index { id: out_index_id },
                 count_desc,
                 Timestamp::from(ts),
+                None,
             )
             .await?;
         match rows.as_slice() {
@@ -1049,9 +1068,29 @@ impl ScriptState {
                         id: GlobalId::User(id),
                     },
                 };
-                let rows = self.driver.peek(target, desc, Timestamp::from(ts)).await?;
+                let rows = self
+                    .driver
+                    .peek(target, desc, Timestamp::from(ts), None)
+                    .await?;
                 Ok(render_rows(&rows))
             }
+            Command::PeekCount { id, ts, literals } => {
+                let desc = self.resolve_schema(&None)?;
+                let target = PeekTarget::Index {
+                    id: GlobalId::User(id),
+                };
+                let literal_constraints = literals.map(|literals| {
+                    literals
+                        .into_iter()
+                        .map(|literal| Row::pack_slice(&[Datum::Int64(literal)]))
+                        .collect()
+                });
+                let count = self
+                    .driver
+                    .peek_count(target, desc, Timestamp::from(ts), literal_constraints)
+                    .await?;
+                Ok(count.to_string())
+            }
             Command::AwaitSubscribe {
                 id,
                 up_to,
diff --git a/src/clusterd-test-driver/src/text.rs b/src/clusterd-test-driver/src/text.rs
index aa4faa0b571c7..5e5acd63c387e 100644
--- a/src/clusterd-test-driver/src/text.rs
+++ b/src/clusterd-test-driver/src/text.rs
@@ -279,6 +279,24 @@ fn opt_string(args: &BTreeMap<String, String>, key: &str) -> Option<String> {
 }
 
 /// Parse a `[a,b,c]` list of `usize`s (`[]` is empty).
+fn parse_i64_list(s: &str) -> anyhow::Result<Vec<i64>> {
+    let inner = s
+        .strip_prefix('[')
+        .and_then(|s| s.strip_suffix(']'))
+        .ok_or_else(|| anyhow!("expected a list like `[0,1]`, got `{s}`"))?;
+    if inner.trim().is_empty() {
+        return Ok(vec![]);
+    }
+    inner
+        .split(',')
+        .map(|part| {
+            part.trim()
+                .parse()
+                .with_context(|| format!("bad list element `{part}`"))
+        })
+        .collect()
+}
+
 fn parse_usize_list(s: &str) -> anyhow::Result<Vec<usize>> {
     let inner = s
         .strip_prefix('[')
@@ -490,6 +508,14 @@ fn parse_command(input: &str) -> anyhow::Result<Command> {
             schema: opt_string(&args, "schema"),
             ts: req_u64(&args, "ts")?,
         },
+        "peek-count" => Command::PeekCount {
+            id: req_u64(&args, "id")?,
+            ts: req_u64(&args, "ts")?,
+            literals: opt_string(&args, "literals")
+                .as_deref()
+                .map(parse_i64_list)
+                .transpose()?,
+        },
         "await-subscribe" => Command::AwaitSubscribe {
             id: req_u64(&args, "id")?,
             up_to: req_u64(&args, "up-to")?,
diff --git a/src/clusterd-test-driver/tests/index_smoke.rs b/src/clusterd-test-driver/tests/index_smoke.rs
index b85a867fbde3a..8cec286689d94 100644
--- a/src/clusterd-test-driver/tests/index_smoke.rs
+++ b/src/clusterd-test-driver/tests/index_smoke.rs
@@ -94,7 +94,12 @@ async fn index_over_small_shard() {
         .await
         .expect("frontier");
     let n = driver
-        .peek_count(PeekTarget::Index { id: index_id }, desc, Timestamp::from(0))
+        .peek_count(
+            PeekTarget::Index { id: index_id },
+            desc,
+            Timestamp::from(0),
+            None,
+        )
         .await
         .expect("peek");
     assert_eq!(n, 10_000);
diff --git a/src/compute-types/src/dyncfgs.rs b/src/compute-types/src/dyncfgs.rs
index 210dd347c927f..156d1b85ffa6e 100644
--- a/src/compute-types/src/dyncfgs.rs
+++ b/src/compute-types/src/dyncfgs.rs
@@ -236,6 +236,28 @@ pub const LINEAR_JOIN_YIELDING: Config<&str> = Config::new(
      work, respectively, rather than falling back to some default.",
 );
 
+/// The yielding behavior with which a single index peek should be processed.
+pub const PEEK_YIELDING: Config<&str> = Config::new(
+    "peek_yielding",
+    "work:100000,time:10",
+    "How much work a compute worker may spend on one index peek before moving on to the \
+     next pending peek. Either 'work:<cursor steps>' or 'time:<milliseconds>' or \
+     'work:<cursor steps>,time:<milliseconds>'. Note that omitting one of 'work' or 'time' \
+     will entirely disable peek yielding by time or work, respectively, rather than falling \
+     back to some default.",
+);
+
+/// The yielding behavior with which index peeks as a whole should be processed.
+pub const PEEK_YIELDING_TOTAL: Config<&str> = Config::new(
+    "peek_yielding_total",
+    "work:1000000,time:100",
+    "How much work a compute worker may spend on index peeks in one worker activation, \
+     across all of them. Same format as 'peek_yielding'. Each pending peek gets at most one \
+     'peek_yielding' turn per activation, so this only binds once more than \
+     'peek_yielding_total' / 'peek_yielding' peeks are scanning at once. Peeks that don't get \
+     a turn are served first on the next activation.",
+);
+
 /// Enable lgalloc.
 pub const ENABLE_LGALLOC: Config<bool> =
     Config::new("enable_lgalloc", true, "Enable lgalloc.").scoped(ParameterScope::Replica);
@@ -556,6 +578,8 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&ENABLE_COMPUTE_TEMPORAL_BUCKETING)
         .add(&TEMPORAL_BUCKETING_SUMMARY)
         .add(&LINEAR_JOIN_YIELDING)
+        .add(&PEEK_YIELDING)
+        .add(&PEEK_YIELDING_TOTAL)
         .add(&ENABLE_LGALLOC)
         .add(&LGALLOC_BACKGROUND_INTERVAL)
         .add(&LGALLOC_FILE_GROWTH_DAMPENER)
diff --git a/src/compute/src/compute_state.rs b/src/compute/src/compute_state.rs
index 2b5f44b8e917a..91af88260a0b3 100644
--- a/src/compute/src/compute_state.rs
+++ b/src/compute/src/compute_state.rs
@@ -17,9 +17,7 @@ use std::time::{Duration, Instant};
 use bytesize::ByteSize;
 use differential_dataflow::Hashable;
 use differential_dataflow::lattice::Lattice;
-use differential_dataflow::trace::cursor::BatchCursor;
-use differential_dataflow::trace::implementations::BatchContainer;
-use differential_dataflow::trace::{Cursor, Navigable, TraceReader};
+use differential_dataflow::trace::{Cursor, TraceReader};
 use mz_compute_client::logging::LoggingConfig;
 use mz_compute_client::protocol::command::{
     ComputeCommand, ComputeParameters, InstanceConfig, Peek, PeekTarget,
@@ -34,9 +32,9 @@ use mz_compute_types::dyncfgs::{
     PEEK_RESPONSE_STASH_THRESHOLD_BYTES, PEEK_STASH_BATCH_SIZE, PEEK_STASH_NUM_BATCHES,
 };
 use mz_compute_types::plan::render_plan::RenderPlan;
-use mz_dyncfg::ConfigSet;
+use mz_dyncfg::{Config, ConfigSet};
+use mz_expr::SafeMfpPlan;
 use mz_expr::row::RowCollection;
-use mz_expr::{RowComparator, SafeMfpPlan};
 use mz_ore::cast::{CastFrom, CastLossy};
 use mz_ore::collections::CollectionExt;
 use mz_ore::metrics::{MetricsRegistry, UIntGauge};
@@ -50,7 +48,6 @@ use mz_persist_client::cfg::USE_CRITICAL_SINCE_SNAPSHOT;
 use mz_persist_client::read::ReadHandle;
 use mz_persist_types::PersistLocation;
 use mz_persist_types::codec_impls::UnitSchema;
-use mz_repr::fixed_length::ExtendDatums;
 use mz_repr::{DatumVec, Diff, GlobalId, Row, RowArena, Timestamp};
 use mz_storage_operators::stats::StatsCursor;
 use mz_storage_types::StorageDiff;
@@ -69,14 +66,17 @@ use tracing::{Level, debug, error, info, span, trace, warn};
 use uuid::Uuid;
 
 use crate::arrangement::manager::{TraceBundle, TraceManager};
+use crate::compute_state::peek_scan::{PeekScan, ScanOutcome};
 use crate::logging;
 use crate::logging::compute::{CollectionLogging, ComputeEvent, PeekEvent};
 use crate::logging::initialize::LoggingTraces;
 use crate::metrics::{CollectionMetrics, WorkerMetrics};
 use crate::render::{LinearJoinSpec, StartSignal};
 use crate::server::{ComputeInstanceContext, ResponseSender};
+use crate::yielding::{Budget, NestedBudget, YieldSpec};
 
 mod peek_result_iterator;
+mod peek_scan;
 mod peek_stash;
 
 /// Worker-local state that is maintained across dataflows.
@@ -122,6 +122,18 @@ pub struct ComputeState {
     max_result_size: u64,
     /// Specification for rendering linear joins.
     pub linear_join_spec: LinearJoinSpec,
+    /// How much scanning work one index peek may do before we move on to the
+    /// next pending peek.
+    pub peek_yielding: YieldSpec,
+    /// How much scanning work all index peeks together may do in one worker
+    /// activation.
+    pub peek_yielding_total: YieldSpec,
+    /// Peek to resume the round robin from in the next `process_peeks`.
+    ///
+    /// Once enough peeks are scanning at the same time, the activation budget
+    /// runs out before all of them have had a turn. Remembering where we
+    /// stopped is what keeps the peeks with low uuids from starving the rest.
+    peek_resume_at: Uuid,
     /// Metrics for this worker.
     pub metrics: WorkerMetrics,
     /// A process-global handle to tracing configuration.
@@ -202,6 +214,9 @@ impl ComputeState {
             command_history,
             max_result_size: u64::MAX,
             linear_join_spec: Default::default(),
+            peek_yielding: Default::default(),
+            peek_yielding_total: Default::default(),
+            peek_resume_at: Uuid::nil(),
             metrics,
             tracing_handle,
             context,
@@ -252,6 +267,9 @@ impl ComputeState {
 
         self.linear_join_spec = LinearJoinSpec::from_config(config);
 
+        self.peek_yielding = parse_yield_spec(&PEEK_YIELDING, config);
+        self.peek_yielding_total = parse_yield_spec(&PEEK_YIELDING_TOTAL, config);
+
         if ENABLE_LGALLOC.get(config) {
             if let Some(path) = &self.context.scratch_directory {
                 let clear_bytes = LGALLOC_SLOW_CLEAR_BYTES.get(config);
@@ -765,7 +783,13 @@ impl<'a> ActiveComputeState<'a> {
             logger.log(&pending.as_log_event(true));
         }
 
-        self.process_peek(&mut Antichain::new(), pending);
+        // We don't serve the peek here. `process_peeks` runs later in the same
+        // worker iteration, so the peek is served without extra latency, and
+        // going through there means a burst of peeks shares one work budget
+        // rather than each getting its own.
+        self.compute_state
+            .pending_peeks
+            .insert(pending.peek().uuid, pending);
     }
 
     fn handle_cancel_peek(&mut self, uuid: Uuid) {
@@ -1000,8 +1024,19 @@ impl<'a> ActiveComputeState<'a> {
         }
     }
 
-    /// Either complete the peek (and send the response) or put it in the pending set.
-    fn process_peek(&mut self, upper: &mut Antichain<Timestamp>, mut peek: PendingPeek) {
+    /// Either completes the peek (and sends the response) or puts it back in
+    /// the pending set.
+    ///
+    /// Scanning work is charged against `budget`, this peek's turn within the
+    /// worker's activation. Returns `true` if the peek has work left that only
+    /// this worker will pick up, meaning the worker must not park.
+    fn process_peek(
+        &mut self,
+        upper: &mut Antichain<Timestamp>,
+        budget: &mut NestedBudget<'_>,
+        mut peek: PendingPeek,
+    ) -> bool {
+        let mut work_pending = false;
         let response = match &mut peek {
             PendingPeek::Index(peek) => {
                 let start = Instant::now();
@@ -1042,7 +1077,10 @@ impl<'a> ActiveComputeState<'a> {
                         .compute_state
                         .metrics
                         .index_peek_row_iteration_seconds,
-                    result_sort_seconds: &self.compute_state.metrics.index_peek_result_sort_seconds,
+                    result_thinning_seconds: &self
+                        .compute_state
+                        .metrics
+                        .index_peek_result_thinning_seconds,
                     row_collection_seconds: &self
                         .compute_state
                         .metrics
@@ -1052,19 +1090,30 @@ impl<'a> ActiveComputeState<'a> {
                 let status = peek.seek_fulfillment(
                     upper,
                     self.compute_state.max_result_size,
-                    peek_stash_enabled && peek_stash_eligible,
-                    peek_stash_threshold_bytes,
+                    (peek_stash_enabled && peek_stash_eligible)
+                        .then_some(peek_stash_threshold_bytes),
+                    budget,
                     &metrics,
                 );
 
-                self.compute_state
-                    .metrics
-                    .index_peek_total_seconds
-                    .observe(start.elapsed().as_secs_f64());
+                // A peek can span many activations, so we only report the total
+                // once it is done. Otherwise a slow peek would show up as a
+                // string of fast ones.
+                peek.elapsed += start.elapsed();
+                if !matches!(status, PeekStatus::NotReady | PeekStatus::Yielded) {
+                    self.compute_state
+                        .metrics
+                        .index_peek_total_seconds
+                        .observe(peek.elapsed.as_secs_f64());
+                }
 
                 match status {
                     PeekStatus::Ready(result) => Some(result),
                     PeekStatus::NotReady => None,
+                    PeekStatus::Yielded => {
+                        work_pending = true;
+                        None
+                    }
                     PeekStatus::UsePeekStash => {
                         let _span =
                             span!(parent: &peek.span, Level::DEBUG, "process_stash_peek").entered();
@@ -1086,7 +1135,11 @@ impl<'a> ActiveComputeState<'a> {
                         self.compute_state
                             .pending_peeks
                             .insert(peek.peek.uuid, PendingPeek::Stash(stash_task));
-                        return;
+                        // The stash peek needs `pump_rows` to feed its upload
+                        // task, and we don't revisit it in this pass. Report
+                        // work pending so the worker doesn't park before the
+                        // first batch is pumped.
+                        return true;
                     }
                 }
             }
@@ -1118,20 +1171,72 @@ impl<'a> ActiveComputeState<'a> {
 
         if let Some(response) = response {
             let _span = span!(parent: peek.span(), Level::DEBUG, "process_peek_response").entered();
-            self.send_peek_response(peek, response)
+            self.send_peek_response(peek, response);
         } else {
             let uuid = peek.peek().uuid;
             self.compute_state.pending_peeks.insert(uuid, peek);
         }
+
+        work_pending
     }
 
-    /// Scan pending peeks and attempt to retire each.
-    pub fn process_peeks(&mut self) {
+    /// Scans pending peeks and attempts to retire each.
+    ///
+    /// Returns `true` if any peek yielded with work remaining, in which case
+    /// the worker must not park before calling this again.
+    ///
+    /// Peeks take turns: each gets a slice of the activation's budget, and we
+    /// stop once that budget is spent. Peeks that did not get a turn are
+    /// served first on the next activation.
+    pub fn process_peeks(&mut self) -> bool {
+        // Rotate the pending peeks so we resume where the last activation ran
+        // out of budget, rather than always starting from the lowest uuid.
+        let mut wrapped = std::mem::take(&mut self.compute_state.pending_peeks);
+        if wrapped.is_empty() {
+            return false;
+        }
+        let rest = wrapped.split_off(&self.compute_state.peek_resume_at);
+
+        let start = Instant::now();
+        let per_peek = self.compute_state.peek_yielding;
+        let mut activation = Budget::new(&self.compute_state.peek_yielding_total);
         let mut upper = Antichain::new();
-        let pending_peeks = std::mem::take(&mut self.compute_state.pending_peeks);
-        for (_uuid, peek) in pending_peeks {
-            self.process_peek(&mut upper, peek);
+        let mut work_pending = false;
+
+        let mut resume_at = None;
+        let mut last = None;
+        for (uuid, peek) in rest.into_iter().chain(wrapped) {
+            // Peeks reached after the budget is spent still get their frontiers
+            // checked, they just don't get to scan. Remember the first of them
+            // so the next activation starts there.
+            if resume_at.is_none() && activation.is_spent() {
+                resume_at = Some(uuid);
+            }
+            last = Some(uuid);
+
+            let mut budget = activation.nest(&per_peek);
+            work_pending |= self.process_peek(&mut upper, &mut budget, peek);
+        }
+
+        // If the budget ran out during the last peek's turn there was no later
+        // peek to record it, and resuming from the top would cut that same
+        // peek's turn short again on every activation.
+        if resume_at.is_none() && activation.is_spent() {
+            resume_at = last;
         }
+
+        self.compute_state.peek_resume_at = resume_at.unwrap_or(Uuid::nil());
+
+        // This is the number the yielding budgets exist to bound, so it is the
+        // one to look at when asking whether peeks are holding up the worker.
+        // Activations with no pending peeks are left out, they would otherwise
+        // bury the distribution in zeroes.
+        self.compute_state
+            .metrics
+            .peek_processing_seconds
+            .observe(start.elapsed().as_secs_f64());
+
+        work_pending
     }
 
     /// Sends a response for this peek's resolution to the coordinator.
@@ -1318,6 +1423,9 @@ impl PendingPeek {
             peek,
             trace_bundle,
             span: tracing::Span::current(),
+            scan: None,
+            elapsed: Duration::ZERO,
+            seek_fulfillment_time: Duration::ZERO,
         })
     }
 
@@ -1555,6 +1663,13 @@ pub struct IndexPeek {
     trace_bundle: TraceBundle,
     /// The `tracing::Span` tracking this peek's operation
     span: tracing::Span,
+    /// The result scan, created once the trace frontiers allow the read. It
+    /// persists across activations because the scan yields before it is done.
+    scan: Option<PeekScan>,
+    /// Worker time spent on this peek, summed over all activations.
+    elapsed: Duration,
+    /// Time spent inside `seek_fulfillment`, summed over all activations.
+    seek_fulfillment_time: Duration,
 }
 
 /// Histogram metrics for index peek phases.
@@ -1567,32 +1682,71 @@ pub(crate) struct IndexPeekMetrics<'a> {
     pub error_scan_seconds: &'a prometheus::Histogram,
     pub cursor_setup_seconds: &'a prometheus::Histogram,
     pub row_iteration_seconds: &'a prometheus::Histogram,
-    pub result_sort_seconds: &'a prometheus::Histogram,
+    pub result_thinning_seconds: &'a prometheus::Histogram,
     pub row_collection_seconds: &'a prometheus::Histogram,
 }
 
 impl IndexPeek {
-    /// Attempts to fulfill the peek and reports success.
-    ///
-    /// To produce output at `peek.timestamp`, we must be certain that
-    /// it is no longer changing. A trace guarantees that all future
-    /// changes will be greater than or equal to an element of `upper`.
+    /// Attempts to fulfill the peek, spending at most `budget` on scanning.
     ///
-    /// If an element of `upper` is less or equal to `peek.timestamp`,
-    /// then there can be further updates that would change the output.
-    /// If no element of `upper` is less or equal to `peek.timestamp`,
-    /// then for any time `t` less or equal to `peek.timestamp` it is
-    /// not the case that `upper` is less or equal to that timestamp,
-    /// and so the result cannot further evolve.
+    /// A [`PeekStatus::Yielded`] result obliges the caller to call this again.
+    /// Nothing else will wake the worker on the peek's behalf.
     fn seek_fulfillment(
         &mut self,
         upper: &mut Antichain<Timestamp>,
         max_result_size: u64,
-        peek_stash_eligible: bool,
-        peek_stash_threshold_bytes: usize,
+        peek_stash_threshold_bytes: Option<usize>,
+        budget: &mut NestedBudget<'_>,
         metrics: &IndexPeekMetrics<'_>,
     ) -> PeekStatus {
         let method_start = Instant::now();
+        let status = self.fulfill(
+            upper,
+            max_result_size,
+            peek_stash_threshold_bytes,
+            budget,
+            metrics,
+        );
+        self.seek_fulfillment_time += method_start.elapsed();
+
+        // The peek can span many activations, so the accumulated time is only
+        // meaningful once it has reached a terminal state.
+        if !matches!(status, PeekStatus::NotReady | PeekStatus::Yielded) {
+            metrics
+                .seek_fulfillment_seconds
+                .observe(self.seek_fulfillment_time.as_secs_f64());
+        }
+
+        status
+    }
+
+    /// Checks the peek's preconditions and advances its scan by one slice.
+    ///
+    /// To produce output at `peek.timestamp` we must be certain it is no
+    /// longer changing. A trace guarantees that all future changes are greater
+    /// than or equal to an element of `upper`, so an `upper` less or equal to
+    /// `peek.timestamp` means further updates could still change the output
+    /// and the peek is not ready.
+    ///
+    /// `max_result_size` and `peek_stash_threshold_bytes` only take effect on
+    /// the activation that creates the scan. The scan captures them, so a
+    /// config change mid-peek does not apply retroactively.
+    fn fulfill(
+        &mut self,
+        upper: &mut Antichain<Timestamp>,
+        max_result_size: u64,
+        peek_stash_threshold_bytes: Option<usize>,
+        budget: &mut NestedBudget<'_>,
+        metrics: &IndexPeekMetrics<'_>,
+    ) -> PeekStatus {
+        // Once the scan exists the frontier requirements have been checked and
+        // the cursor holds the batches it reads, so we go straight back to
+        // scanning.
+        if self.scan.is_some() {
+            return self.step_scan(budget, metrics);
+        }
+
+        let frontier_check_start = Instant::now();
 
         self.trace_bundle.oks_mut().read_upper(upper);
         if upper.less_equal(&self.peek.timestamp) {
@@ -1615,34 +1769,45 @@ impl IndexPeek {
 
         metrics
             .frontier_check_seconds
-            .observe(method_start.elapsed().as_secs_f64());
+            .observe(frontier_check_start.elapsed().as_secs_f64());
 
-        let result = self.collect_finished_data(
-            max_result_size,
-            peek_stash_eligible,
+        if let Some(status) = self.scan_error_trace(metrics) {
+            return status;
+        }
+
+        self.scan = Some(PeekScan::new(
+            &self.peek,
+            self.trace_bundle.oks_mut(),
+            usize::cast_from(max_result_size),
             peek_stash_threshold_bytes,
             metrics,
-        );
-
-        metrics
-            .seek_fulfillment_seconds
-            .observe(method_start.elapsed().as_secs_f64());
+        ));
 
-        result
+        self.step_scan(budget, metrics)
     }
 
-    /// Collects data for a known-complete peek from the ok stream.
-    fn collect_finished_data(
+    /// Advances the result scan by one budgeted slice.
+    fn step_scan(
         &mut self,
-        max_result_size: u64,
-        peek_stash_eligible: bool,
-        peek_stash_threshold_bytes: usize,
+        budget: &mut NestedBudget<'_>,
         metrics: &IndexPeekMetrics<'_>,
     ) -> PeekStatus {
-        let error_scan_start = Instant::now();
+        let scan = self.scan.as_mut().expect("scan exists");
+        match scan.step(budget, metrics) {
+            ScanOutcome::Yielded => PeekStatus::Yielded,
+            ScanOutcome::Complete(response) => PeekStatus::Ready(response),
+            ScanOutcome::UsePeekStash => PeekStatus::UsePeekStash,
+        }
+    }
+
+    /// Scans the error trace and returns a response if it holds any errors.
+    ///
+    /// NOTE: This scan is not budgeted. It walks every key of the errs trace,
+    /// which in practice holds at most a handful of rows, but a dataflow that
+    /// errors on a large fraction of its input could make it run long.
+    fn scan_error_trace(&mut self, metrics: &IndexPeekMetrics<'_>) -> Option<PeekStatus> {
+        let scan_start = Instant::now();
 
-        // Check if there exist any errors and, if so, return whatever one we
-        // find first.
         let (mut cursor, storage) = self.trace_bundle.errs_mut().cursor();
         while cursor.key_valid(&storage) {
             let mut copies = Diff::ZERO;
@@ -1657,187 +1822,43 @@ impl IndexPeek {
                     target = %self.peek.target.id(), diff = %copies, %error,
                     "index peek encountered negative multiplicities in error trace",
                 );
-                return PeekStatus::Ready(PeekResponse::Error(format!(
+                return Some(PeekStatus::Ready(PeekResponse::Error(format!(
                     "Invalid data in source errors, \
                     saw retractions ({}) for row that does not exist: {}",
                     -copies, error,
-                )));
+                ))));
             }
             if copies.is_positive() {
-                return PeekStatus::Ready(PeekResponse::Error(cursor.key(&storage).to_string()));
+                return Some(PeekStatus::Ready(PeekResponse::Error(
+                    cursor.key(&storage).to_string(),
+                )));
             }
             cursor.step_key(&storage);
         }
 
         metrics
             .error_scan_seconds
-            .observe(error_scan_start.elapsed().as_secs_f64());
+            .observe(scan_start.elapsed().as_secs_f64());
 
-        Self::collect_ok_finished_data(
-            &self.peek,
-            self.trace_bundle.oks_mut(),
-            max_result_size,
-            peek_stash_eligible,
-            peek_stash_threshold_bytes,
-            metrics,
-        )
+        None
     }
+}
 
-    /// Collects data for a known-complete peek from the ok stream.
-    fn collect_ok_finished_data<Tr>(
-        peek: &Peek,
-        oks_handle: &mut Tr,
-        max_result_size: u64,
-        peek_stash_eligible: bool,
-        peek_stash_threshold_bytes: usize,
-        metrics: &IndexPeekMetrics<'_>,
-    ) -> PeekStatus
-    where
-        Tr: TraceReader<Batch: Navigable>,
-        for<'a> BatchCursor<Tr>: Cursor<
-                Key<'a>: ExtendDatums + Eq,
-                KeyContainer: BatchContainer<Owned = Row>,
-                Val<'a>: ExtendDatums,
-                TimeGat<'a>: PartialOrder<Timestamp>,
-                DiffGat<'a> = &'a Diff,
-            >,
-    {
-        let max_result_size = usize::cast_from(max_result_size);
-        let count_byte_size = size_of::<NonZeroUsize>();
-
-        // Cursor setup timing
-        let cursor_setup_start = Instant::now();
-
-        // We clone `literal_constraints` here because we don't want to move the constraints
-        // out of the peek struct, and don't want to modify in-place.
-        let mut peek_iterator = peek_result_iterator::PeekResultIterator::new(
-            peek.target.id().clone(),
-            peek.map_filter_project.clone(),
-            peek.timestamp,
-            peek.literal_constraints.clone().as_deref_mut(),
-            oks_handle,
-        );
-
-        metrics
-            .cursor_setup_seconds
-            .observe(cursor_setup_start.elapsed().as_secs_f64());
-
-        // Accumulated `Vec<(row, count)>` results that we are likely to return.
-        let mut results = Vec::new();
-        let mut total_size: usize = 0;
-
-        // When set, a bound on the number of records we need to return.
-        // The requirements on the records are driven by the finishing's
-        // `order_by` field. Further limiting will happen when the results
-        // are collected, so we don't need to have exactly this many results,
-        // just at least those results that would have been returned.
-        let max_results = peek.finishing.num_rows_needed();
-
-        let comparator = RowComparator::new(peek.finishing.order_by.as_slice());
-
-        // Row iteration timing
-        let row_iteration_start = Instant::now();
-        let mut sort_time_accum = Duration::ZERO;
-
-        while let Some(row) = peek_iterator.next() {
-            let row: (Row, _) = match row {
-                Ok(row) => row,
-                Err(err) => return PeekStatus::Ready(PeekResponse::Error(err)),
-            };
-            let (row, copies) = row;
-            let copies: NonZeroUsize = NonZeroUsize::try_from(copies).expect("fits into usize");
-
-            total_size = total_size
-                .saturating_add(row.byte_len())
-                .saturating_add(count_byte_size);
-            if peek_stash_eligible && total_size > peek_stash_threshold_bytes {
-                return PeekStatus::UsePeekStash;
-            }
-            if total_size > max_result_size {
-                return PeekStatus::Ready(PeekResponse::Error(format!(
-                    "result exceeds max size of {}",
-                    ByteSize::b(u64::cast_from(max_result_size))
-                )));
-            }
-
-            results.push((row, copies));
-
-            // If we hold many more than `max_results` records, we can thin down
-            // `results` using `self.finishing.ordering`.
-            if let Some(max_results) = max_results {
-                // We use a threshold twice what we intend, to amortize the work
-                // across all of the insertions. We could tighten this, but it
-                // works for the moment.
-                //
-                // `max_results` is `limit + offset`, so a `LIMIT` near
-                // `i64::MAX` makes the doubling overflow. We then hold fewer
-                // rows than the threshold no matter what, and never thin. That
-                // is the right answer: such a peek cannot accumulate that many
-                // rows anyway, the result size limit stops it long before.
-                // Wrapping instead would make the threshold tiny, and we would
-                // thin while holding almost nothing, dropping rows past the end
-                // of the buffer.
-                if max_results
-                    .checked_mul(2)
-                    .is_some_and(|threshold| results.len() >= threshold)
-                {
-                    if peek.finishing.order_by.is_empty() {
-                        results.truncate(max_results);
-                        metrics
-                            .row_iteration_seconds
-                            .observe(row_iteration_start.elapsed().as_secs_f64());
-                        metrics
-                            .result_sort_seconds
-                            .observe(sort_time_accum.as_secs_f64());
-                        let row_collection_start = Instant::now();
-                        let collection = RowCollection::new(results, &peek.finishing.order_by);
-                        metrics
-                            .row_collection_seconds
-                            .observe(row_collection_start.elapsed().as_secs_f64());
-                        return PeekStatus::Ready(PeekResponse::Rows(vec![collection]));
-                    } else {
-                        // We can sort `results` and then truncate to `max_results`.
-                        // This has an effect similar to a priority queue, without
-                        // its interactive dequeueing properties.
-                        // TODO: Had we left these as `Vec<Datum>` we would avoid
-                        // the unpacking; we should consider doing that, although
-                        // it will require a re-pivot of the code to branch on this
-                        // inner test (as we prefer not to maintain `Vec<Datum>`
-                        // in the other case).
-                        let sort_start = Instant::now();
-                        results.sort_by(|left, right| {
-                            comparator.compare_rows(&left.0, &right.0, || left.0.cmp(&right.0))
-                        });
-                        sort_time_accum += sort_start.elapsed();
-                        let dropped = results.drain(max_results..);
-                        let dropped_size =
-                            dropped
-                                .into_iter()
-                                .fold(0, |acc: usize, (row, _count): (Row, _)| {
-                                    acc.saturating_add(
-                                        row.byte_len().saturating_add(count_byte_size),
-                                    )
-                                });
-                        total_size = total_size.saturating_sub(dropped_size);
-                    }
-                }
-            }
-        }
-
-        metrics
-            .row_iteration_seconds
-            .observe(row_iteration_start.elapsed().as_secs_f64());
-        metrics
-            .result_sort_seconds
-            .observe(sort_time_accum.as_secs_f64());
-
-        let row_collection_start = Instant::now();
-        let collection = RowCollection::new(results, &peek.finishing.order_by);
-        metrics
-            .row_collection_seconds
-            .observe(row_collection_start.elapsed().as_secs_f64());
-        PeekStatus::Ready(PeekResponse::Rows(vec![collection]))
+/// Reads a [`YieldSpec`] from its dyncfg, falling back to that config's own
+/// default if the value does not parse.
+///
+/// Falling back to `YieldSpec::default()` instead would be wrong: the two peek
+/// budgets have different defaults, and silently giving one of them the other's
+/// would undo the bound it exists to impose.
+fn parse_yield_spec(config: &Config<&'static str>, set: &ConfigSet) -> YieldSpec {
+    let raw = config.get(set);
+    if let Some(spec) = YieldSpec::try_from_str(&raw) {
+        return spec;
     }
+
+    let default = config.default();
+    error!(name = config.name(), %raw, "invalid yield spec, using {default}");
+    YieldSpec::try_from_str(default).expect("config default parses")
 }
 
 /// For keeping track of the state of pending or ready peeks, and managing
@@ -1846,6 +1867,9 @@ enum PeekStatus {
     /// The frontiers of objects are not yet advanced enough, peek is still
     /// pending.
     NotReady,
+    /// The peek is being served but ran out of budget for this activation. It
+    /// has to be stepped again, and nothing else will wake the worker for it.
+    Yielded,
     /// The result size is above the configured threshold and the peek is
     /// eligible for using the peek result stash.
     UsePeekStash,
diff --git a/src/compute/src/compute_state/peek_result_iterator.rs b/src/compute/src/compute_state/peek_result_iterator.rs
index fc38f024e0c79..19df543fd1a8b 100644
--- a/src/compute/src/compute_state/peek_result_iterator.rs
+++ b/src/compute/src/compute_state/peek_result_iterator.rs
@@ -176,19 +176,63 @@ where
     type Item = Result<(Row, NonZeroI64), String>;
 
     fn next(&mut self) -> Option<Self::Item> {
+        let mut fuel = usize::MAX;
+        match self.step(&mut fuel) {
+            Step::Row(row) => Some(row),
+            Step::Done => None,
+            Step::OutOfFuel => unreachable!("stepped with unbounded fuel"),
+        }
+    }
+}
+
+/// The outcome of a single fueled [`PeekResultIterator::step`].
+pub enum Step {
+    /// A result row, or the error that ended the scan.
+    Row(Result<(Row, NonZeroI64), String>),
+    /// The cursor is exhausted. Further steps also return `Done`.
+    Done,
+    /// The fuel ran out before a row was found. The cursor sits at the next
+    /// position to attempt, so stepping again resumes exactly there.
+    OutOfFuel,
+}
+
+impl<Tr> PeekResultIterator<Tr>
+where
+    Tr: TraceReader<Batch: Navigable>,
+    for<'a> BatchCursor<Tr>: Cursor<
+            Key<'a>: ExtendDatums + Eq,
+            KeyContainer: BatchContainer<Owned = Row>,
+            Val<'a>: ExtendDatums,
+            TimeGat<'a>: PartialOrder<mz_repr::Timestamp>,
+            DiffGat<'a> = &'a Diff,
+        >,
+{
+    /// Advances the cursor until it produces a row, the cursor is exhausted,
+    /// or `fuel` runs out, whichever comes first. Decrements `fuel` by the
+    /// number of cursor positions visited.
+    ///
+    /// Fuel is charged per cursor position, not per row returned, so a
+    /// selective `map_filter_project` cannot starve the caller of yield
+    /// points. Returning with fuel left over means the cursor is exhausted.
+    pub fn step(&mut self, fuel: &mut usize) -> Step {
         let result = loop {
+            if *fuel == 0 {
+                return Step::OutOfFuel;
+            }
+            *fuel -= 1;
+
             if self.literals_exhausted() {
-                return None;
+                return Step::Done;
             }
 
             if !self.cursor.key_valid(&self.storage) {
-                return None;
+                return Step::Done;
             }
 
             if !self.cursor.val_valid(&self.storage) {
                 let exhausted = self.step_key();
                 if exhausted {
-                    return None;
+                    return Step::Done;
                 }
             }
 
@@ -204,21 +248,9 @@ where
 
         self.cursor.step_val(&self.storage);
 
-        Some(result)
+        Step::Row(result)
     }
-}
 
-impl<Tr> PeekResultIterator<Tr>
-where
-    Tr: TraceReader<Batch: Navigable>,
-    for<'a> BatchCursor<Tr>: Cursor<
-            Key<'a>: ExtendDatums + Eq,
-            KeyContainer: BatchContainer<Owned = Row>,
-            Val<'a>: ExtendDatums,
-            TimeGat<'a>: PartialOrder<mz_repr::Timestamp>,
-            DiffGat<'a> = &'a Diff,
-        >,
-{
     /// Extracts and returns the row currently pointed at by our cursor. Returns
     /// `Ok(None)` if our MapFilterProject evaluates to `None`. Also returns any
     /// errors that arise from evaluating the MapFilterProject.
diff --git a/src/compute/src/compute_state/peek_scan.rs b/src/compute/src/compute_state/peek_scan.rs
new file mode 100644
index 0000000000000..0107902726ade
--- /dev/null
+++ b/src/compute/src/compute_state/peek_scan.rs
@@ -0,0 +1,310 @@
+// Copyright Materialize, Inc. and contributors. All rights reserved.
+//
+// Use of this software is governed by the Business Source License
+// included in the LICENSE file.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0.
+
+//! Cooperative accumulation of an index peek's result.
+//!
+//! Walking an arrangement can take arbitrarily long, and the compute worker is
+//! a single thread that also has to schedule dataflows and handle commands. A
+//! [`PeekScan`] therefore does its work in bounded slices: each
+//! [`step`](PeekScan::step) spends at most one [`NestedBudget`] and then hands
+//! the worker back, keeping enough state to resume where it left off.
+//!
+//! The scan owns its cursor, and the cursor owns the batches it reads rather
+//! than borrowing them from the trace. So a scan is self-contained and parking
+//! one between activations is safe. Nothing can compact the data out from under
+//! it either, since batches are immutable and a merge allocates new ones.
+//!
+//! The flip side is that a parked scan keeps its whole batch set alive. A peek
+//! already pinned those batches for its lifetime, but a small budget stretches
+//! that lifetime out in wall-clock terms, so the memory is held for longer.
+
+use std::num::{NonZeroI64, NonZeroUsize};
+use std::time::{Duration, Instant};
+
+use bytesize::ByteSize;
+use mz_compute_client::protocol::command::Peek;
+use mz_compute_client::protocol::response::PeekResponse;
+use mz_expr::row::RowCollection;
+use mz_expr::{ColumnOrder, RowComparator};
+use mz_ore::cast::CastFrom;
+use mz_repr::{Diff, Row, Timestamp};
+
+use crate::arrangement::manager::PaddedTrace;
+use crate::compute_state::IndexPeekMetrics;
+use crate::compute_state::peek_result_iterator::{PeekResultIterator, Step};
+use crate::typedefs::RowRowAgent;
+use crate::yielding::NestedBudget;
+
+/// Per-entry overhead we charge on top of the row itself, matching how the
+/// result is laid out in a [`RowCollection`].
+const COUNT_BYTE_SIZE: usize = size_of::<NonZeroUsize>();
+
+/// The result of one slice of scanning work.
+pub(super) enum ScanOutcome {
+    /// The budget ran out with work remaining. Call [`PeekScan::step`] again.
+    Yielded,
+    /// The scan is finished and this is the peek's response.
+    Complete(PeekResponse),
+    /// The result outgrew what we are willing to send inline. The caller
+    /// should abandon this scan and stash the response instead.
+    UsePeekStash,
+}
+
+/// Why the scan loop stopped.
+enum Stop {
+    /// No further rows are needed, or none are left. Either way the
+    /// accumulated results are the answer.
+    Complete,
+    /// The scan ended early with this response.
+    Response(PeekResponse),
+    /// The result outgrew the inline threshold.
+    UsePeekStash,
+}
+
+/// An index peek's result scan, resumable across worker activations.
+pub(super) struct PeekScan {
+    iter: PeekResultIterator<PaddedTrace<RowRowAgent<Timestamp, Diff>>>,
+    /// Rows accumulated so far, periodically thinned down to `max_results`.
+    results: Vec<(Row, NonZeroUsize)>,
+    /// Byte size of `results`, kept in sync as rows are added and thinned out.
+    total_size: usize,
+    /// A bound on the number of records the finishing can need, if it has one.
+    ///
+    /// Further limiting happens once the results are collected, so we don't
+    /// need exactly this many, just at least those that would be returned.
+    max_results: Option<usize>,
+    order_by: Vec<ColumnOrder>,
+    comparator: RowComparator,
+    max_result_size: usize,
+    /// When set, a result that grows past this many bytes goes to the peek
+    /// stash rather than being sent inline.
+    peek_stash_threshold_bytes: Option<usize>,
+    /// Wall time spent scanning, summed over all slices.
+    row_iteration_time: Duration,
+    /// Wall time spent sorting during thinning, summed over all slices.
+    thinning_time: Duration,
+}
+
+impl PeekScan {
+    /// Sets up a scan of `oks` at the peek's timestamp.
+    ///
+    /// The caller must have established that the trace's frontiers permit a
+    /// read at `peek.timestamp`. Taking the cursor fixes what the scan will
+    /// read, so that check has no meaning once the scan exists.
+    pub fn new(
+        peek: &Peek,
+        oks: &mut PaddedTrace<RowRowAgent<Timestamp, Diff>>,
+        max_result_size: usize,
+        peek_stash_threshold_bytes: Option<usize>,
+        metrics: &IndexPeekMetrics<'_>,
+    ) -> Self {
+        let cursor_setup_start = Instant::now();
+
+        // NOTE: Setting up the cursor is not budgeted. With literal constraints
+        // it sorts them and seeks to the first match, which for a large `IN`
+        // list is real work the peek cannot yield out of. Nor is a unit of fuel
+        // during the scan a bounded amount of work, see
+        // `PeekResultIterator::step`. So the budget bounds how often we get to
+        // yield, not the length of any one slice.
+        //
+        // We clone `literal_constraints` here because we don't want to move the
+        // constraints out of the peek struct, and don't want to modify in-place.
+        let iter = PeekResultIterator::new(
+            peek.target.id(),
+            peek.map_filter_project.clone(),
+            peek.timestamp,
+            peek.literal_constraints.clone().as_deref_mut(),
+            oks,
+        );
+
+        metrics
+            .cursor_setup_seconds
+            .observe(cursor_setup_start.elapsed().as_secs_f64());
+
+        Self {
+            iter,
+            results: Vec::new(),
+            total_size: 0,
+            max_results: peek.finishing.num_rows_needed(),
+            order_by: peek.finishing.order_by.clone(),
+            comparator: RowComparator::new(peek.finishing.order_by.clone()),
+            max_result_size,
+            peek_stash_threshold_bytes,
+            row_iteration_time: Duration::ZERO,
+            thinning_time: Duration::ZERO,
+        }
+    }
+
+    /// Performs one slice of scanning work, bounded by `budget`.
+    pub fn step(
+        &mut self,
+        budget: &mut NestedBudget<'_>,
+        metrics: &IndexPeekMetrics<'_>,
+    ) -> ScanOutcome {
+        let slice_start = Instant::now();
+        let stop = self.scan(budget);
+        self.row_iteration_time += slice_start.elapsed();
+
+        let Some(stop) = stop else {
+            return ScanOutcome::Yielded;
+        };
+
+        // The scan is over, so the accumulated timings are final.
+        metrics
+            .row_iteration_seconds
+            .observe(self.row_iteration_time.as_secs_f64());
+        metrics
+            .result_thinning_seconds
+            .observe(self.thinning_time.as_secs_f64());
+
+        match stop {
+            Stop::Complete => {
+                let collection_start = Instant::now();
+                let results = std::mem::take(&mut self.results);
+                let collection = RowCollection::new(results, &self.order_by);
+                metrics
+                    .row_collection_seconds
+                    .observe(collection_start.elapsed().as_secs_f64());
+                ScanOutcome::Complete(PeekResponse::Rows(vec![collection]))
+            }
+            Stop::Response(response) => ScanOutcome::Complete(response),
+            Stop::UsePeekStash => ScanOutcome::UsePeekStash,
+        }
+    }
+
+    /// Runs the cursor until the budget is spent, returning `None` in that
+    /// case, or until the scan reaches a terminal state.
+    ///
+    /// Always advances the cursor by at least one position, even on an
+    /// already-spent budget. A yielded peek keeps the worker from parking, so
+    /// a slice that does no work at all is a livelock rather than a slow peek.
+    /// That makes progress a property of this loop instead of something the
+    /// operator has to preserve when setting the budget.
+    fn scan(&mut self, budget: &mut NestedBudget<'_>) -> Option<Stop> {
+        loop {
+            // The iterator charges fuel per outer step, including steps its
+            // `map_filter_project` rejects, so a selective filter over a large
+            // arrangement still comes back here to have the budget checked.
+            let allowance = budget.allowance().max(1);
+            let mut fuel = allowance;
+            let step = self.iter.step(&mut fuel);
+            budget.charge(allowance - fuel);
+
+            match step {
+                Step::OutOfFuel => (),
+                Step::Done => return Some(Stop::Complete),
+                Step::Row(Err(err)) => return Some(Stop::Response(PeekResponse::Error(err))),
+                Step::Row(Ok((row, copies))) => {
+                    if let Some(stop) = self.absorb(row, copies) {
+                        return Some(stop);
+                    }
+                }
+            }
+
+            if budget.is_spent() {
+                return None;
+            }
+        }
+    }
+
+    /// Folds one result row into the accumulated results, thinning them down
+    /// if they have outgrown what the finishing can need.
+    fn absorb(&mut self, row: Row, copies: NonZeroI64) -> Option<Stop> {
+        let copies: NonZeroUsize = NonZeroUsize::try_from(copies).expect("fits into usize");
+
+        self.total_size = self
+            .total_size
+            .saturating_add(row.byte_len())
+            .saturating_add(COUNT_BYTE_SIZE);
+
+        if let Some(threshold) = self.peek_stash_threshold_bytes
+            && self.total_size > threshold
+        {
+            return Some(Stop::UsePeekStash);
+        }
+        if self.total_size > self.max_result_size {
+            return Some(Stop::Response(PeekResponse::Error(format!(
+                "result exceeds max size of {}",
+                ByteSize::b(u64::cast_from(self.max_result_size))
+            ))));
+        }
+
+        self.results.push((row, copies));
+
+        let Some(max_results) = self.max_results else {
+            return None;
+        };
+        // We use a threshold twice what we intend, to amortize the work across
+        // all of the insertions. We could tighten this, but it works for the
+        // moment.
+        //
+        // A `LIMIT` near `i64::MAX` makes that double overflow. Such a peek can
+        // never hold that many rows anyway, the result size limit stops it long
+        // before, so there is nothing to thin and we just keep accumulating.
+        // Wrapping instead would be a worker panic with an `ORDER BY` and a
+        // silently truncated answer without one.
+        let Some(thin_at) = max_results.checked_mul(2) else {
+            return None;
+        };
+        if self.results.len() < thin_at {
+            return None;
+        }
+
+        if self.order_by.is_empty() {
+            // Any `max_results` rows are as good as any others, so we're done.
+            self.results.truncate(max_results);
+            return Some(Stop::Complete);
+        }
+
+        // Partition rather than sort: we only need to know which rows fall
+        // outside the first `max_results`, not the order among those that stay.
+        // The final ordering is established once, when the results are
+        // collected. Sorting here would cost a log factor per row for an order
+        // we then throw away.
+        //
+        // Partitioning is not stable, so when rows tie across the cut it is
+        // unspecified which of them survives, and since entries carry a count
+        // the retained multiset genuinely differs between choices. What the
+        // client sees does not, for three reasons together:
+        //
+        //  - A tie under this comparator means the rows are byte-identical,
+        //    because the tiebreaker compares the whole encoded row.
+        //  - We keep exactly `max_results` entries, each with a count of at
+        //    least one, so the retained run expands to at least `max_results`
+        //    rows and its first `max_results` are the same either way.
+        //  - `max_results` is `limit + offset`, and the finishing reads exactly
+        //    `offset..offset + limit` of the merged result, whose prefix
+        //    depends only on the runs' prefixes.
+        //
+        // NOTE: The second and third points are why a peek result must not be
+        // consumed without applying the finishing's limit.
+        //
+        // TODO: Had we left these as `Vec<Datum>` we would avoid the unpacking.
+        // We should consider doing that, although it will require a re-pivot of
+        // the code to branch on this inner test (as we prefer not to maintain
+        // `Vec<Datum>` in the other case).
+        let sort_start = Instant::now();
+        let comparator = &self.comparator;
+        self.results
+            .select_nth_unstable_by(max_results, |left, right| {
+                comparator.compare_rows(&left.0, &right.0, || left.0.cmp(&right.0))
+            });
+        self.thinning_time += sort_start.elapsed();
+
+        let dropped_size = self
+            .results
+            .drain(max_results..)
+            .fold(0usize, |acc, (row, _count)| {
+                acc.saturating_add(row.byte_len().saturating_add(COUNT_BYTE_SIZE))
+            });
+        self.total_size = self.total_size.saturating_sub(dropped_size);
+
+        None
+    }
+}
diff --git a/src/compute/src/lib.rs b/src/compute/src/lib.rs
index 5e13276a34dc5..412aa7462be67 100644
--- a/src/compute/src/lib.rs
+++ b/src/compute/src/lib.rs
@@ -27,3 +27,4 @@ pub mod sink;
 #[cfg(not(feature = "bench"))]
 mod sink;
 mod typedefs;
+mod yielding;
diff --git a/src/compute/src/metrics.rs b/src/compute/src/metrics.rs
index 5246bf5ee5265..14fa1b97b44f1 100644
--- a/src/compute/src/metrics.rs
+++ b/src/compute/src/metrics.rs
@@ -59,13 +59,14 @@ pub struct ComputeMetrics {
     stashed_peek_seconds: HistogramVec,
     handle_command_duration_seconds: HistogramVec,
 
-    // Index peek timing phases (per-cluster, no worker label)
+    // Peek timing (per-cluster, no worker label)
+    peek_processing_seconds: Histogram,
     index_peek_total_seconds: Histogram,
     index_peek_seek_fulfillment_seconds: Histogram,
     index_peek_error_scan_seconds: Histogram,
     index_peek_cursor_setup_seconds: Histogram,
     index_peek_row_iteration_seconds: Histogram,
-    index_peek_result_sort_seconds: Histogram,
+    index_peek_result_thinning_seconds: Histogram,
     index_peek_frontier_check_seconds: Histogram,
     index_peek_row_collection_seconds: Histogram,
 
@@ -178,14 +179,19 @@ impl ComputeMetrics {
                 var_labels: ["worker_id", "command_type"],
                 buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0),
             )),
+            peek_processing_seconds: registry.register(metric!(
+                name: "mz_peek_processing_seconds",
+                help: "Time a worker spent serving pending peeks in one activation, before returning to scheduling dataflows and handling commands. This is the latency the peek yielding budgets exist to bound. Only recorded for activations that had at least one pending peek.",
+                buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0),
+            )),
             index_peek_total_seconds: registry.register(metric!(
                 name: "mz_index_peek_total_seconds",
-                help: "Total time processing index peeks, from process_peek entry to response. Excluding peeks that use the peek response stash.",
+                help: "Worker time spent serving an index peek, summed over the activations it took and reported once it is done. Peeks that are cancelled or dropped before finishing report nothing.",
                 buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0),
             )),
             index_peek_seek_fulfillment_seconds: registry.register(metric!(
                 name: "mz_index_peek_seek_fulfillment_seconds",
-                help: "Time in seek_fulfillment method including frontier checks and data collection.",
+                help: "Worker time spent in seek_fulfillment for an index peek, including every not-yet-ready frontier check, summed over activations and reported once the peek is done.",
                 buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0),
             )),
             index_peek_error_scan_seconds: registry.register(metric!(
@@ -200,12 +206,12 @@ impl ComputeMetrics {
             )),
             index_peek_row_iteration_seconds: registry.register(metric!(
                 name: "mz_index_peek_row_iteration_seconds",
-                help: "Time iterating rows and evaluating MFP.",
+                help: "Time iterating rows and evaluating MFP, summed over the activations an index peek took and reported once it is done. Peeks that are cancelled or dropped before finishing report nothing.",
                 buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0),
             )),
-            index_peek_result_sort_seconds: registry.register(metric!(
-                name: "mz_index_peek_result_sort_seconds",
-                help: "Time sorting intermediate results during peek collection.",
+            index_peek_result_thinning_seconds: registry.register(metric!(
+                name: "mz_index_peek_result_thinning_seconds",
+                help: "Time thinning intermediate peek results down to the rows the finishing can need, summed over the activations an index peek took and reported once it is done. Peeks that are cancelled or dropped before finishing report nothing.",
                 buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0),
             )),
             index_peek_frontier_check_seconds: registry.register(metric!(
@@ -263,12 +269,13 @@ impl ComputeMetrics {
             self.handle_command_duration_seconds
                 .with_label_values(&[worker.as_ref(), typ])
         });
+        let peek_processing_seconds = self.peek_processing_seconds.clone();
         let index_peek_total_seconds = self.index_peek_total_seconds.clone();
         let index_peek_seek_fulfillment_seconds = self.index_peek_seek_fulfillment_seconds.clone();
         let index_peek_error_scan_seconds = self.index_peek_error_scan_seconds.clone();
         let index_peek_cursor_setup_seconds = self.index_peek_cursor_setup_seconds.clone();
         let index_peek_row_iteration_seconds = self.index_peek_row_iteration_seconds.clone();
-        let index_peek_result_sort_seconds = self.index_peek_result_sort_seconds.clone();
+        let index_peek_result_thinning_seconds = self.index_peek_result_thinning_seconds.clone();
         let index_peek_frontier_check_seconds = self.index_peek_frontier_check_seconds.clone();
         let index_peek_row_collection_seconds = self.index_peek_row_collection_seconds.clone();
         let replica_expiration_timestamp_seconds = self
@@ -290,12 +297,13 @@ impl ComputeMetrics {
             persist_peek_seconds,
             stashed_peek_seconds,
             handle_command_duration_seconds,
+            peek_processing_seconds,
             index_peek_total_seconds,
             index_peek_seek_fulfillment_seconds,
             index_peek_error_scan_seconds,
             index_peek_cursor_setup_seconds,
             index_peek_row_iteration_seconds,
-            index_peek_result_sort_seconds,
+            index_peek_result_thinning_seconds,
             index_peek_frontier_check_seconds,
             index_peek_row_collection_seconds,
             replica_expiration_timestamp_seconds,
@@ -327,6 +335,8 @@ pub struct WorkerMetrics {
     pub(crate) stashed_peek_seconds: Histogram,
     /// Histogram of command handling durations.
     pub(crate) handle_command_duration_seconds: CommandMetrics<Histogram>,
+    /// Histogram of how long one pass over the pending peeks took.
+    pub(crate) peek_processing_seconds: Histogram,
     /// Histogram of total index peek durations.
     pub(crate) index_peek_total_seconds: Histogram,
     /// Histogram of index peek seek_fulfillment durations.
@@ -337,8 +347,8 @@ pub struct WorkerMetrics {
     pub(crate) index_peek_cursor_setup_seconds: Histogram,
     /// Histogram of index peek row iteration durations.
     pub(crate) index_peek_row_iteration_seconds: Histogram,
-    /// Histogram of index peek result sort durations.
-    pub(crate) index_peek_result_sort_seconds: Histogram,
+    /// Histogram of index peek result thinning durations.
+    pub(crate) index_peek_result_thinning_seconds: Histogram,
     /// Histogram of index peek frontier check durations.
     pub(crate) index_peek_frontier_check_seconds: Histogram,
     /// Histogram of index peek row collection construction durations.
diff --git a/src/compute/src/render/join/linear_join.rs b/src/compute/src/render/join/linear_join.rs
index 13d6d910d0d91..216dcc76194f3 100644
--- a/src/compute/src/render/join/linear_join.rs
+++ b/src/compute/src/render/join/linear_join.rs
@@ -11,7 +11,7 @@
 //!
 //! Consult [LinearJoinPlan] documentation for details.
 
-use std::time::{Duration, Instant};
+use std::time::Instant;
 
 use differential_dataflow::consolidation::ConsolidatingContainerBuilder;
 use differential_dataflow::lattice::Lattice;
@@ -42,6 +42,7 @@ use crate::render::context::{ArrangementFlavor, CollectionBundle, Context};
 use crate::render::errors::DataflowErrorSer;
 use crate::render::join::mz_join_core::mz_join_core;
 use crate::typedefs::{RowRowAgent, RowRowEnter};
+use crate::yielding::YieldSpec;
 use mz_row_spine::{RowRowBuilder, RowRowColPagedBuilder, RowRowSpine};
 
 /// Available linear join implementations.
@@ -140,53 +141,6 @@ impl LinearJoinSpec {
     }
 }
 
-/// Specification of a dataflow operator's yielding behavior.
-#[derive(Clone, Copy)]
-struct YieldSpec {
-    /// Yield after the given amount of work was performed.
-    after_work: Option<usize>,
-    /// Yield after the given amount of time has elapsed.
-    after_time: Option<Duration>,
-}
-
-impl Default for YieldSpec {
-    fn default() -> Self {
-        Self {
-            after_work: Some(1_000_000),
-            after_time: Some(Duration::from_millis(100)),
-        }
-    }
-}
-
-impl YieldSpec {
-    fn try_from_str(s: &str) -> Option<Self> {
-        let mut after_work = None;
-        let mut after_time = None;
-
-        let options = s.split(',').map(|o| o.trim());
-        for option in options {
-            let mut iter = option.split(':').map(|p| p.trim());
-            match std::array::from_fn(|_| iter.next()) {
-                [Some("work"), Some(amount), None] => {
-                    let amount = amount.parse().ok()?;
-                    after_work = Some(amount);
-                }
-                [Some("time"), Some(millis), None] => {
-                    let millis = millis.parse().ok()?;
-                    let duration = Duration::from_millis(millis);
-                    after_time = Some(duration);
-                }
-                _ => return None,
-            }
-        }
-
-        Some(Self {
-            after_work,
-            after_time,
-        })
-    }
-}
-
 /// Different forms the streamed data might take.
 enum JoinedFlavor<'scope, T: RenderTimestamp> {
     /// Streamed data as a collection.
diff --git a/src/compute/src/server.rs b/src/compute/src/server.rs
index 3588044852835..315943f185ad0 100644
--- a/src/compute/src/server.rs
+++ b/src/compute/src/server.rs
@@ -376,6 +376,9 @@ impl<'w> Worker<'w> {
         // The last time we did periodic maintenance.
         let mut last_maintenance = Instant::now();
 
+        // Whether a peek yielded with scanning work left to do.
+        let mut peek_work_pending = false;
+
         // Commence normal operation.
         loop {
             // Get the maintenance interval, default to zero if we don't have a compute state.
@@ -407,6 +410,16 @@ impl<'w> Worker<'w> {
                 sleep_duration = Some(next_maintenance.saturating_duration_since(now))
             };
 
+            // A yielded peek is work we owe ourselves, and nothing outside this
+            // loop will activate the worker on its behalf. Parking would stall
+            // it until some unrelated event happens to wake us, so we step
+            // without parking instead.
+            let sleep_duration = if peek_work_pending {
+                Some(Duration::ZERO)
+            } else {
+                sleep_duration
+            };
+
             // Step the timely worker, recording the time taken.
             let timer = self.metrics.timely_step_duration_seconds.start_timer();
             self.timely_worker.step_or_park(sleep_duration);
@@ -414,8 +427,9 @@ impl<'w> Worker<'w> {
 
             self.handle_pending_commands()?;
 
+            peek_work_pending = false;
             if let Some(mut compute_state) = self.activate_compute() {
-                compute_state.process_peeks();
+                peek_work_pending = compute_state.process_peeks();
                 compute_state.process_subscribes();
                 compute_state.process_copy_tos();
             }
diff --git a/src/compute/src/yielding.rs b/src/compute/src/yielding.rs
new file mode 100644
index 0000000000000..8a3f485e5ec0d
--- /dev/null
+++ b/src/compute/src/yielding.rs
@@ -0,0 +1,409 @@
+// Copyright Materialize, Inc. and contributors. All rights reserved.
+//
+// Use of this software is governed by the Business Source License
+// included in the LICENSE file.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0.
+
+//! Cooperative scheduling of long-running work on a compute worker.
+//!
+//! A compute worker is a single thread that multiplexes dataflow scheduling,
+//! command handling, and peek processing. Any one of those refusing to return
+//! for seconds at a time costs the whole worker its interactivity. Work that
+//! can run long is therefore expected to run in bounded slices: perform a
+//! slice, return, and get called again on the next activation.
+//!
+//! [`YieldSpec`] says how large a slice may be, and [`Budget`] tracks the
+//! remaining allowance within one slice. When several items share an
+//! activation, [`Budget::nest`] gives each one its own allowance inside the
+//! shared one, so that no single item can hog the activation or keep the
+//! others from being reached.
+
+use std::time::{Duration, Instant};
+
+/// Specification of how large a slice of cooperative work may be.
+///
+/// Both bounds are optional and independent. Omitting one disables that
+/// dimension entirely rather than falling back to a default, and omitting both
+/// disables yielding.
+#[derive(Clone, Copy, Debug)]
+pub(crate) struct YieldSpec {
+    /// Yield after the given amount of work was performed.
+    pub after_work: Option<usize>,
+    /// Yield after the given amount of time has elapsed.
+    pub after_time: Option<Duration>,
+}
+
+impl Default for YieldSpec {
+    fn default() -> Self {
+        Self {
+            after_work: Some(1_000_000),
+            after_time: Some(Duration::from_millis(100)),
+        }
+    }
+}
+
+impl YieldSpec {
+    /// Parses the dyncfg representation: `work:<amount>`, `time:<millis>`, or
+    /// both separated by a comma. Returns `None` if the string does not parse.
+    pub fn try_from_str(s: &str) -> Option<Self> {
+        let mut after_work = None;
+        let mut after_time = None;
+
+        let options = s.split(',').map(|o| o.trim());
+        for option in options {
+            let mut iter = option.split(':').map(|p| p.trim());
+            match std::array::from_fn(|_| iter.next()) {
+                [Some("work"), Some(amount), None] => {
+                    let amount = amount.parse().ok()?;
+                    after_work = Some(amount);
+                }
+                [Some("time"), Some(millis), None] => {
+                    let millis = millis.parse().ok()?;
+                    let duration = Duration::from_millis(millis);
+                    after_time = Some(duration);
+                }
+                _ => return None,
+            }
+        }
+
+        Some(Self {
+            after_work,
+            after_time,
+        })
+    }
+}
+
+/// The remaining allowance for one slice of cooperative work.
+///
+/// Work is measured in caller-defined units. The caller asks for an
+/// [`allowance`](Budget::allowance), spends up to that much, reports what it
+/// spent with [`charge`](Budget::charge), and yields once the budget
+/// [`is_spent`](Budget::is_spent).
+pub(crate) struct Budget {
+    /// Work units left, `None` when the spec imposes no work bound.
+    work: Option<usize>,
+    deadline: Option<Instant>,
+    /// Work units left until we read the clock again. Reading the clock is
+    /// cheap but not free, and a slice can be tens of millions of units, so we
+    /// only consult the deadline every `CLOCK_INTERVAL` units.
+    until_clock_check: usize,
+    /// Latched once the deadline has passed.
+    expired: bool,
+}
+
+impl Budget {
+    /// Work units charged between clock reads.
+    ///
+    /// Reading the clock per unit is measurable once a slice runs to tens of
+    /// millions of units, so we amortize it. The cost is resolution: a time
+    /// bound cannot be observed sooner than `CLOCK_INTERVAL` units, so at a
+    /// few microseconds per unit the deadline can overshoot by milliseconds.
+    /// Set a work bound as well if that matters.
+    const CLOCK_INTERVAL: usize = 1024;
+
+    pub fn new(spec: &YieldSpec) -> Self {
+        Self {
+            work: spec.after_work,
+            deadline: spec.after_time.map(|d| Instant::now() + d),
+            until_clock_check: Self::CLOCK_INTERVAL,
+            expired: false,
+        }
+    }
+
+    /// How much work the caller may perform before it must call
+    /// [`charge`](Budget::charge) again.
+    ///
+    /// Non-zero while the budget is not spent, and zero once it is. A caller
+    /// that must make progress regardless has to impose its own floor: a spec
+    /// of `work:0` yields a budget that is spent before any work happens.
+    pub fn allowance(&self) -> usize {
+        // The time bound has to be checked here and not only in `is_spent`,
+        // otherwise an expired budget keeps handing out full allowances. That
+        // matters most for a shared budget: a nested allowance is the minimum
+        // of the two, so items reached after the shared deadline passed would
+        // each still get a full turn, and the time bound would not cap the
+        // round at all.
+        if self.expired {
+            return 0;
+        }
+        match self.work {
+            Some(work) => work.min(self.until_clock_check),
+            None => self.until_clock_check,
+        }
+    }
+
+    /// Charges `units` of performed work against the budget.
+    pub fn charge(&mut self, units: usize) {
+        if let Some(work) = &mut self.work {
+            *work = work.saturating_sub(units);
+        }
+        self.until_clock_check = self.until_clock_check.saturating_sub(units);
+
+        if self.until_clock_check == 0 {
+            self.until_clock_check = Self::CLOCK_INTERVAL;
+            if let Some(deadline) = self.deadline {
+                self.expired |= Instant::now() >= deadline;
+            }
+        }
+    }
+
+    /// Whether the allowance is used up and the caller should yield.
+    pub fn is_spent(&self) -> bool {
+        self.work == Some(0) || self.expired
+    }
+
+    /// Carves a per-item allowance out of this budget.
+    pub fn nest(&mut self, spec: &YieldSpec) -> NestedBudget<'_> {
+        NestedBudget {
+            own: Budget::new(spec),
+            shared: self,
+        }
+    }
+}
+
+/// A per-item allowance nested inside one shared by all items.
+///
+/// Work charged here is charged against both, and the pair is spent as soon as
+/// either bound is reached. So an item yields once it has had its turn, and
+/// the shared bound still caps what all items together may spend.
+pub(crate) struct NestedBudget<'a> {
+    own: Budget,
+    shared: &'a mut Budget,
+}
+
+impl NestedBudget<'_> {
+    /// See [`Budget::allowance`].
+    pub fn allowance(&self) -> usize {
+        self.own.allowance().min(self.shared.allowance())
+    }
+
+    /// See [`Budget::charge`].
+    pub fn charge(&mut self, units: usize) {
+        self.own.charge(units);
+        self.shared.charge(units);
+    }
+
+    /// See [`Budget::is_spent`].
+    pub fn is_spent(&self) -> bool {
+        self.own.is_spent() || self.shared.is_spent()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[mz_ore::test]
+    fn parses_yield_spec() {
+        let spec = YieldSpec::try_from_str("work:100,time:5").unwrap();
+        assert_eq!(spec.after_work, Some(100));
+        assert_eq!(spec.after_time, Some(Duration::from_millis(5)));
+
+        let spec = YieldSpec::try_from_str("work:100").unwrap();
+        assert_eq!(spec.after_work, Some(100));
+        assert_eq!(spec.after_time, None);
+
+        let spec = YieldSpec::try_from_str("time:5").unwrap();
+        assert_eq!(spec.after_work, None);
+        assert_eq!(spec.after_time, Some(Duration::from_millis(5)));
+
+        assert!(YieldSpec::try_from_str("work:banana").is_none());
+        assert!(YieldSpec::try_from_str("fuel:100").is_none());
+        assert!(YieldSpec::try_from_str("").is_none());
+        assert!(YieldSpec::try_from_str("work:1,").is_none());
+
+        // Repeating a key is accepted and the last one wins. Nothing depends on
+        // this, it just isn't worth rejecting.
+        let spec = YieldSpec::try_from_str("work:1,work:2").unwrap();
+        assert_eq!(spec.after_work, Some(2));
+    }
+
+    /// A zero work bound produces a budget that is spent before anything
+    /// happens. Callers that must make progress have to floor the allowance
+    /// themselves, so pin the behavior they are flooring against.
+    #[mz_ore::test]
+    fn zero_work_budget_is_spent_immediately() {
+        let spec = YieldSpec {
+            after_work: Some(0),
+            after_time: None,
+        };
+        let mut budget = Budget::new(&spec);
+        assert!(budget.is_spent());
+        assert_eq!(budget.allowance(), 0);
+        assert!(budget.nest(&YieldSpec::default()).is_spent());
+
+        // And the other way around: a spent per-item budget inside a healthy
+        // shared one.
+        let mut shared = Budget::new(&YieldSpec::default());
+        assert!(shared.nest(&spec).is_spent());
+    }
+
+    /// Charging more than the allowance is legal and saturates.
+    #[mz_ore::test]
+    fn overcharging_saturates() {
+        let spec = YieldSpec {
+            after_work: Some(10),
+            after_time: None,
+        };
+        let mut budget = Budget::new(&spec);
+        budget.charge(usize::MAX);
+        assert!(budget.is_spent());
+        assert_eq!(budget.allowance(), 0);
+    }
+
+    /// A per-item bound wider than the shared one is dead: the shared bound
+    /// decides. This is what a misconfigured `peek_yielding` looks like.
+    #[mz_ore::test]
+    fn nested_budget_wider_than_shared_is_dead() {
+        let shared_spec = YieldSpec {
+            after_work: Some(10),
+            after_time: None,
+        };
+        let own = YieldSpec {
+            after_work: Some(usize::MAX),
+            after_time: None,
+        };
+        let mut shared = Budget::new(&shared_spec);
+        let mut nested = shared.nest(&own);
+
+        assert_eq!(nested.allowance(), 10);
+        nested.charge(10);
+        assert!(nested.is_spent());
+        assert!(shared.is_spent());
+    }
+
+    /// A work bound that is not a multiple of the clock interval still hands
+    /// out exactly that much.
+    #[mz_ore::test]
+    fn work_bound_off_clock_interval() {
+        let spec = YieldSpec {
+            after_work: Some(Budget::CLOCK_INTERVAL + 7),
+            after_time: None,
+        };
+        let mut budget = Budget::new(&spec);
+
+        let mut spent = 0;
+        while !budget.is_spent() {
+            let allowance = budget.allowance();
+            assert!(allowance > 0);
+            budget.charge(allowance);
+            spent += allowance;
+        }
+        assert_eq!(spent, Budget::CLOCK_INTERVAL + 7);
+    }
+
+    #[mz_ore::test]
+    fn budget_allowance_is_never_zero_before_spent() {
+        let spec = YieldSpec {
+            after_work: Some(3 * Budget::CLOCK_INTERVAL),
+            after_time: None,
+        };
+        let mut budget = Budget::new(&spec);
+
+        let mut spent = 0;
+        while !budget.is_spent() {
+            let allowance = budget.allowance();
+            assert!(allowance > 0);
+            budget.charge(allowance);
+            spent += allowance;
+        }
+        assert_eq!(spent, 3 * Budget::CLOCK_INTERVAL);
+    }
+
+    #[mz_ore::test]
+    fn nested_budget_is_bounded_by_both() {
+        let own = YieldSpec {
+            after_work: Some(Budget::CLOCK_INTERVAL),
+            after_time: None,
+        };
+        let shared_spec = YieldSpec {
+            after_work: Some(3 * Budget::CLOCK_INTERVAL),
+            after_time: None,
+        };
+
+        // The shared budget covers three turns of the per-item budget.
+        let mut shared = Budget::new(&shared_spec);
+        for _ in 0..3 {
+            let mut nested = shared.nest(&own);
+            let mut spent = 0;
+            while !nested.is_spent() {
+                let allowance = nested.allowance();
+                assert!(allowance > 0);
+                nested.charge(allowance);
+                spent += allowance;
+            }
+            assert_eq!(spent, Budget::CLOCK_INTERVAL);
+        }
+
+        assert!(shared.is_spent());
+        assert!(shared.nest(&own).is_spent());
+    }
+
+    #[mz_ore::test]
+    fn budget_expires_on_deadline() {
+        let spec = YieldSpec {
+            after_work: None,
+            after_time: Some(Duration::ZERO),
+        };
+        let mut budget = Budget::new(&spec);
+
+        // The deadline is only consulted once a full clock interval is charged.
+        assert!(!budget.is_spent());
+        budget.charge(Budget::CLOCK_INTERVAL);
+        assert!(budget.is_spent());
+    }
+
+    #[mz_ore::test]
+    fn expired_budget_hands_out_no_allowance() {
+        let spec = YieldSpec {
+            after_work: None,
+            after_time: Some(Duration::ZERO),
+        };
+        let mut budget = Budget::new(&spec);
+
+        assert!(budget.allowance() > 0);
+        budget.charge(Budget::CLOCK_INTERVAL);
+
+        // These two have to agree, so that a caller can drive off either one.
+        assert!(budget.is_spent());
+        assert_eq!(budget.allowance(), 0);
+    }
+
+    #[mz_ore::test]
+    fn expired_shared_budget_leaves_nothing_to_nest() {
+        let shared_spec = YieldSpec {
+            after_work: None,
+            after_time: Some(Duration::ZERO),
+        };
+        let own_spec = YieldSpec {
+            after_work: Some(100),
+            after_time: None,
+        };
+
+        let mut shared = Budget::new(&shared_spec);
+        shared.charge(Budget::CLOCK_INTERVAL);
+        assert!(shared.is_spent());
+
+        // A nested allowance is the minimum of the two. If the shared one kept
+        // reporting an allowance after its deadline passed, every item reached
+        // afterwards would still get a full turn and the shared time bound
+        // would not cap the round.
+        let nested = shared.nest(&own_spec);
+        assert!(nested.is_spent());
+        assert_eq!(nested.allowance(), 0);
+    }
+
+    #[mz_ore::test]
+    fn unbounded_budget_never_spends() {
+        let spec = YieldSpec {
+            after_work: None,
+            after_time: None,
+        };
+        let mut budget = Budget::new(&spec);
+        budget.charge(usize::MAX);
+        assert!(!budget.is_spent());
+    }
+}
diff --git a/test/clusterd-test-driver/mzcompose.py b/test/clusterd-test-driver/mzcompose.py
index 3db70d857c7c2..93ddf52ce247f 100644
--- a/test/clusterd-test-driver/mzcompose.py
+++ b/test/clusterd-test-driver/mzcompose.py
@@ -96,6 +96,7 @@ def __init__(self, name: str = "headless-driver") -> None:
     "join.spec",
     "index_and_mv.spec",
     "create_time_config.spec",
+    "peek_yielding.spec",
 ]
 
 
diff --git a/test/clusterd-test-driver/scripts/peek_yielding.spec b/test/clusterd-test-driver/scripts/peek_yielding.spec
new file mode 100644
index 0000000000000..a2d23ea0dae32
--- /dev/null
+++ b/test/clusterd-test-driver/scripts/peek_yielding.spec
@@ -0,0 +1,112 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+# peek_yielding scenario: serve an index peek under a work budget so small that
+# the arrangement scan cannot finish in one worker activation, and check that
+# resuming it returns the same collection a single-slice scan would.
+#
+# `work:1` lets the peek advance one cursor position per turn, so the peeks
+# below resume on the order of 2000 times each.
+#
+# `peek-count` rather than `count`: `count` tallies through an ephemeral reduce
+# dataflow and then peeks that dataflow's single-row output, so the peek itself
+# would only ever walk one cursor position and none of this would be exercised.
+# `peek-count` peeks the index directly.
+#
+# What this pins: no row is lost or duplicated across a yield, `Step::Done`
+# stays sticky once the cursor is exhausted, a finished scan leaves nothing
+# behind that corrupts the next peek, a peek spanning many activations still
+# produces exactly one response, and the same holds for the literal-constraint
+# path, whose cursor seeks rather than steps.
+#
+# What it does not pin. Row contents, since the golden is a count — that
+# coverage comes from running sqllogictest with the small CI budgets, where the
+# goldens compare rows positionally. And `peek_yielding_total`, because the
+# driver awaits each peek before sending the next, so two peeks are never
+# pending at once and the activation budget never binds.
+create-instance
+----
+ok
+
+update-configuration
+peek_yielding string work:1,time:60000
+----
+ok
+
+initialization-complete
+----
+ok
+
+write-single-ts shard=data ts=0 count=2000
+----
+wrote 2000
+
+define-index source=1000 index=1001 shard=data key=[0] as-of=0 upper=1
+----
+ok
+
+schedule id=1001
+----
+ok
+
+await-frontier id=1001 ts=1
+----
+ok
+
+# The whole collection survives a scan that yielded after every cursor position.
+peek-count id=1001 ts=0
+----
+2000
+
+# Again, to check a completed scan left nothing behind for the next peek.
+peek-count id=1001 ts=0
+----
+2000
+
+# And the reduce path still works with peeks yielding underneath it.
+count id=1001 ts=0
+----
+2000
+
+# Literal constraints take a different path through the scan. The cursor seeks
+# from one literal to the next instead of stepping, and the position in the
+# literal list has to survive a yield just as the cursor position does. Keys run
+# 0..1999.
+
+# Unsorted literals, all matching. The replica sorts them itself.
+peek-count id=1001 ts=0 literals=[1999,5,500]
+----
+3
+
+# Literals matching no key are skipped. Each skip is a seek, and a run of them
+# happens within one unit of fuel.
+peek-count id=1001 ts=0 literals=[9999,5,2500,500,3000]
+----
+2
+
+# A run of non-matching literals after the last match, so the scan finishes by
+# exhausting the literal list rather than the cursor.
+peek-count id=1001 ts=0 literals=[1999,2000,2001,2002,2003]
+----
+1
+
+# Nothing matches at all.
+peek-count id=1001 ts=0 literals=[3000,4000]
+----
+0
+
+# An empty literal list selects nothing.
+peek-count id=1001 ts=0 literals=[]
+----
+0
+
+# Enough matching literals to yield repeatedly between them.
+peek-count id=1001 ts=0 literals=[0,100,200,300,400,500,600,700,800,900,1000,1100,1200,1300,1400,1500,1600,1700,1800,1900]
+----
+20