diff --git a/docs/DAEMON_MEASUREMENTS.md b/docs/DAEMON_MEASUREMENTS.md
index 0c4a2d849..7bc6167aa 100644
--- a/docs/DAEMON_MEASUREMENTS.md
+++ b/docs/DAEMON_MEASUREMENTS.md
@@ -5644,3 +5644,28 @@ Hot-path screen, in the same alternating run: the previous pacer change against
 Every range overlaps the other build's. The medians for completion p99 and
 lifecycle CPU are a little higher; neither bench waits on a learned limit,
 so the new path costs one look at an empty list of bookings per call.
+
+## Calls in a row run together (#291)
+
+The Perf workflow's matched screen, base main 93e57b8 against the PR head
+78c5a4d, on the GitHub runner (2026-10-28):
+
+| Bench | Measure | Base | Head | Change |
+| --- | --- | ---: | ---: | ---: |
+| admission | lone submit CPU, ms | 38.914 | 38.556 | -0.9% |
+| admission | burst submit p99, ms | 6.025 | 5.86 | -2.7% |
+| completion | release to terminal p99, ms | 13.442 | 13.297 | -1.1% |
+| footprint | idle RSS, MiB | 17.418 | 17.51 | +0.5% |
+| footprint | per idle bot, KiB | 1.578 | 1.61 | +2.0% |
+| footprint | RSS after a turn each, MiB | 26.357 | 26.639 | +1.1% |
+| footprint | CPU per turn, ms | 1.284 | 1.268 | -1.2% |
+| lifecycle | CPU, s | 1.225 | 1.165 | -4.9% |
+| lifecycle | p95 turn, ms | 583.266 | 581.296 | -0.3% |
+| lifecycle | peak RSS, MiB | 23.912 | 24.0 | +0.4% |
+
+Store jobs per turn did not change. Every change is within the gate's 15%.
+These benches make one call per round, so they show the lone path unchanged.
+The batch path's gain is wall time: in `tests/test_wait.py`, two commands
+that each sleep 0.6 s finish in under 1.1 s. Results of a batch are queued
+to the store in call order without waiting on each other, so calls that
+finish together share a commit.
diff --git a/src/server/turn.rs b/src/server/turn.rs
index ec652a6c9..01fcf084f 100644
--- a/src/server/turn.rs
+++ b/src/server/turn.rs
@@ -247,10 +247,13 @@ impl Lift {
     }
 }
 const LIFT_GRACE: Duration = Duration::from_millis(250);
+/// The most of one round's calls a bot runs at once. Each command also
+/// waits for the daemon's process budget, so this is a bot's share of it.
+const PARALLEL_CALLS: usize = 4;
 
 /// A tool's result, or the command a steer moved to the background.
-enum Ran<F> {
-    Done(Result<Outcome>),
+enum Ran<F, T = Outcome> {
+    Done(Result<T>),
     Lifted(std::pin::Pin<Box<F>>),
 }
 
@@ -3151,14 +3154,14 @@ impl Turn {
     /// the run to the background instead, handed back unfinished, once the
     /// paired signal says it is running: a command still waiting for a
     /// process slot stays in line in the foreground.
-    async fn run_tool<F: Future<Output = Result<Outcome>>>(
+    async fn run_tool<T, F: Future<Output = Result<T>>>(
         &self,
         run: F,
         lift: Option<(&Lift, oneshot::Receiver<()>)>,
         warm: Option<&mut Warm<'_>>,
-    ) -> Result<Ran<F>> {
-        enum Next {
-            Ran(Result<Outcome>),
+    ) -> Result<Ran<F, T>> {
+        enum Next<T> {
+            Ran(Result<T>),
             Lifted,
             Refreshed(Result<(agent_runtime::provider::Usage, tokio::time::Instant)>),
         }
@@ -3365,6 +3368,26 @@ impl Turn {
         let (turn, allowed) = (self.turn, record.callable());
         let mut calls = calls.into_iter();
         while let Some(call) = calls.next() {
+            // Calls in a row that only read or run commands, none of them
+            // gated, run together, as Codex runs them.
+            if let Some(first) = self.parallel(record, &call, workspace)
+                && calls
+                    .as_slice()
+                    .first()
+                    .is_some_and(|next| self.parallel(record, next, workspace).is_some())
+            {
+                let mut batch = vec![(call, first)];
+                while let Some(prepared) = calls
+                    .as_slice()
+                    .first()
+                    .and_then(|next| self.parallel(record, next, workspace))
+                {
+                    batch.push((calls.next().expect("a next call"), prepared));
+                }
+                self.execute_parallel(batch, record, workspace, environment, warm.as_deref_mut())
+                    .await?;
+                continue;
+            }
             // An escape from a sandbox runs unconfined only once a gate
             // allowed it.
             let approved = record.gated(&call, workspace);
@@ -3419,46 +3442,10 @@ impl Turn {
                          run it without sandbox_permissions",
                     ))
                 }
-                Ok(prepared) if confined => {
-                    // Reading the roots touches the filesystem: off the
-                    // runtime thread.
-                    let (settings, daemon, bot) = (
-                        record.settings.clone(),
-                        self.registry.daemon().clone(),
-                        record.bot_id,
-                    );
-                    let root = record
-                        .workspace
-                        .as_ref()
-                        .map_or_else(|| workspace.to_owned(), PathBuf::from);
-                    tokio::task::spawn_blocking(move || {
-                        // The probe for a sandbox runs once, here off the
-                        // runtime thread, also for a bot from before a restart.
-                        agent_runtime::sandbox::available()?;
-                        // The bot's own socket, so its commands may use the
-                        // daemon as far as a sandboxed bot may; opened first,
-                        // so its folder is there to be kept read-only.
-                        let capability = daemon.capability(bot)?;
-                        let policy = Policy::new(
-                            settings.sandbox,
-                            Some(&root),
-                            &settings.writable,
-                            settings.network,
-                            &daemon,
-                        )?;
-                        Ok(policy.map(|policy| Policy {
-                            capability,
-                            ..policy
-                        }))
-                    })
-                    .await
-                    .map_err(|_| Error::new("sandbox_unavailable"))
-                    .flatten()
-                    .map(|policy| {
-                        sandbox = policy.map(Arc::new);
-                        prepared
-                    })
-                }
+                Ok(prepared) if confined => self.sandbox(record, workspace).await.map(|policy| {
+                    sandbox = policy;
+                    prepared
+                }),
                 other => other,
             };
             let answered = record
@@ -3641,6 +3628,228 @@ impl Turn {
         Ok(None)
     }
 
+    /// The call prepared, when it may run beside others: an ungated
+    /// command in the foreground or a file read, which the bot may call. Codex
+    /// runs these together too; a change to a file runs alone, in order.
+    /// A sandboxed bot's command runs beside others only inside its
+    /// sandbox; one that asks to leave it goes the lone way, to be refused.
+    fn parallel(
+        &self,
+        record: &Bot,
+        call: &ToolCall,
+        workspace: &std::path::Path,
+    ) -> Option<Prepared> {
+        if !matches!(call.name.as_str(), "shell" | "read")
+            || record.gated(call, workspace)
+            || record.declared(&call.name).is_some()
+            || !record.callable().iter().any(|name| name == &call.name)
+            || (call.name == "shell" && record.sandboxed() && !record.confines(call, workspace))
+        {
+            return None;
+        }
+        match self.registry.prepare(&call.name, &call.arguments) {
+            // A stored stream or result is paged to fit the context left,
+            // which a read beside it would spend too: it runs alone.
+            Ok(
+                prepared @ (Prepared::Shell {
+                    background: false, ..
+                }
+                | Prepared::Read {
+                    source: ReadSource::Path(_),
+                    ..
+                }),
+            ) => Some(prepared),
+            _ => None,
+        }
+    }
+
+    /// Run a batch of calls [`Self::parallel`] admits together, up to
+    /// [`PARALLEL_CALLS`] at a time and no more than half the process
+    /// budget: all started in one commit, each result recorded in call
+    /// order as soon as it and those before it are done. A steer moves every command still running to the
+    /// background, as it does a lone one.
+    async fn execute_parallel(
+        &self,
+        batch: Vec<(ToolCall, Prepared)>,
+        record: &Bot,
+        workspace: &std::path::Path,
+        environment: &[(String, String)],
+        warm: Option<&mut Warm<'_>>,
+    ) -> Result<()> {
+        let turn = self.turn;
+        let starting: Vec<ToolCall> = batch.iter().map(|(call, _)| call.clone()).collect();
+        self.store
+            .op("tool_start", move |db| {
+                starting
+                    .iter()
+                    .try_for_each(|call| db.tool_start(turn, call).map(drop))
+            })
+            .await?;
+        // A sandboxed bot's commands all run in its one sandbox; when it
+        // can't be had, each command fails with why.
+        let shells = batch
+            .iter()
+            .any(|(_, prepared)| matches!(prepared, Prepared::Shell { .. }));
+        let sandbox = match shells && record.sandboxed() {
+            true => self.sandbox(record, workspace).await,
+            false => Ok(None),
+        };
+        let runs = batch.into_iter().map(|(call, prepared)| {
+            let sandbox = match (&prepared, &sandbox) {
+                (Prepared::Shell { .. }, Err(error)) => Err(error.clone()),
+                (_, sandbox) => Ok(sandbox.clone().unwrap_or_default()),
+            };
+            self.run_parallel(call.call_id, prepared, workspace, environment, sandbox)
+        });
+        let record = async {
+            // Results go to the store in call order as they come, each
+            // queued without waiting on the one before, so results done
+            // together share a commit. Waiting for room on the store's
+            // queue happens beside the calls still running, which go on
+            // being polled; meanwhile the channel holds at most one result
+            // per call in the batch.
+            let (done, mut recorded) = tokio::sync::mpsc::unbounded_channel();
+            let run = async move {
+                let mut results = stream::iter(runs).buffered(self.registry.wide(PARALLEL_CALLS));
+                while let Some((id, outcome)) = results.next().await {
+                    let _ = done.send((id, outcome?));
+                }
+                Ok::<_, Error>(())
+            };
+            // Each write's acknowledgement carries its result: dropped as
+            // it comes, so committed results aren't held for the batch.
+            let write = async {
+                let mut written = stream::FuturesOrdered::new();
+                loop {
+                    tokio::select! {
+                        Some(answer) = written.next(), if !written.is_empty() => drop(answer?),
+                        next = recorded.recv() => match next {
+                            Some((id, outcome)) => written.push_back(
+                                self.store.slot().await?.op("tool_finish", move |db| {
+                                    db.tool_finish(turn, &id, &outcome)
+                                }),
+                            ),
+                            None => break,
+                        },
+                    }
+                }
+                while let Some(answer) = written.next().await {
+                    answer?;
+                }
+                Ok::<_, Error>(())
+            };
+            let (ran, wrote) = tokio::join!(run, write);
+            ran.and(wrote)
+        };
+        match self.run_tool(record, None, warm).await? {
+            Ran::Done(result) => result,
+            Ran::Lifted(_) => unreachable!("a batch is not lifted whole"),
+        }
+    }
+
+    /// One call of a batch, run to its outcome: an error only when the
+    /// scheduler closed.
+    async fn run_parallel(
+        &self,
+        call_id: String,
+        prepared: Prepared,
+        workspace: &std::path::Path,
+        environment: &[(String, String)],
+        sandbox: Result<Option<Arc<Policy>>>,
+    ) -> (String, Result<Outcome>) {
+        let sandbox = match sandbox {
+            Ok(sandbox) => sandbox,
+            Err(error) => return (call_id, Ok(failure(error))),
+        };
+        let ran = match prepared {
+            Prepared::Shell {
+                command,
+                timeout_ms,
+                background: false,
+            } => {
+                let (started, running) = oneshot::channel();
+                let mut run = Box::pin(self.registry.foreground(
+                    command,
+                    workspace.to_path_buf(),
+                    timeout_ms,
+                    environment.to_vec(),
+                    sandbox,
+                    started,
+                ));
+                let lifted = async {
+                    // A run that ends before it starts wins the race below.
+                    match running.await {
+                        Ok(()) => self.lift.lifted().await,
+                        Err(_) => std::future::pending().await,
+                    }
+                };
+                let finished = tokio::select! {
+                    biased;
+                    result = &mut run => Some(result),
+                    () = lifted => None,
+                };
+                match finished {
+                    Some(result) => result,
+                    None => return (call_id.clone(), self.lift_shell(&call_id, run).await),
+                }
+            }
+            prepared => {
+                self.registry
+                    .execute(prepared, workspace, environment, sandbox)
+                    .await
+            }
+        };
+        let outcome = match ran {
+            Ok(outcome) => Ok(annotate(outcome, self.turn, &call_id)),
+            Err(error) if error.code == "tool_scheduler_closed" => Err(error),
+            Err(error) => Ok(failure(error)),
+        };
+        (call_id, outcome)
+    }
+
+    /// The bot's sandbox, built from its settings. Reading its roots
+    /// touches the filesystem, so it runs off the runtime thread; it fails
+    /// rather than leave a call that must be confined unconfined.
+    async fn sandbox(
+        &self,
+        record: &Bot,
+        workspace: &std::path::Path,
+    ) -> Result<Option<Arc<Policy>>> {
+        let (settings, daemon, bot) = (
+            record.settings.clone(),
+            self.registry.daemon().clone(),
+            record.bot_id,
+        );
+        let root = record
+            .workspace
+            .as_ref()
+            .map_or_else(|| workspace.to_owned(), PathBuf::from);
+        tokio::task::spawn_blocking(move || {
+            // The probe for a sandbox runs once, here off the runtime
+            // thread, also for a bot from before a restart.
+            agent_runtime::sandbox::available()?;
+            // The bot's own socket, so its commands may use the daemon as
+            // far as a sandboxed bot may; opened first, so its folder is
+            // there to be kept read-only.
+            let capability = daemon.capability(bot)?;
+            let policy = Policy::new(
+                settings.sandbox,
+                Some(&root),
+                &settings.writable,
+                settings.network,
+                &daemon,
+            )?;
+            Ok(policy.map(|policy| Policy {
+                capability,
+                ..policy
+            }))
+        })
+        .await
+        .map_err(|_| Error::new("sandbox_unavailable"))
+        .flatten()
+        .map(|policy| policy.map(Arc::new))
+    }
+
     /// Wait for a gated call's verdict: live for the hold, then parked like
     /// `wait`, holding no task and no slot. Answers wake the task through
     /// the worker, so each check is one job and no commit.
diff --git a/src/tools.rs b/src/tools.rs
index ea2e751e2..e42290734 100644
--- a/src/tools.rs
+++ b/src/tools.rs
@@ -853,6 +853,15 @@ impl Registry {
         }));
         self
     }
+    /// How many of one bot's calls may run together: `cap` at most, and
+    /// no more than half the process budget, so one bot's wide round
+    /// leaves room for the others.
+    pub fn wide(&self, cap: usize) -> usize {
+        match self.budget {
+            0 => cap,
+            budget => cap.min(budget / 2).max(1),
+        }
+    }
     /// Take a place in line for a background command, or refuse when as many
     /// are already waiting as may run: a counter and a comparison, so a fleet
     /// cannot accumulate an invisible backlog behind the process bound.
@@ -1957,6 +1966,19 @@ async fn shell(
 mod tests {
     use super::*;
     #[test]
+    fn one_bot_runs_at_most_half_the_process_budget_at_once() {
+        let wide = |budget| {
+            Registry::new("shell")
+                .unwrap()
+                .with_process_budget(budget)
+                .wide(4)
+        };
+        assert_eq!(
+            [wide(0), wide(1), wide(2), wide(4), wide(64)],
+            [4, 1, 1, 2, 4]
+        );
+    }
+    #[test]
     fn binary_files_are_named_and_text_passes() {
         assert_eq!(binary(b"\x89PNG\r\n\x1a\n\0\0"), Some("a PNG image"));
         assert_eq!(binary(b"RIFF\0\0\0\0WEBPVP8 "), Some("a WebP image"));
diff --git a/tests/test_approvals.py b/tests/test_approvals.py
index 37f168fb5..fb02af9b1 100644
--- a/tests/test_approvals.py
+++ b/tests/test_approvals.py
@@ -250,6 +250,28 @@ def test_a_sandboxed_bot_uses_the_daemon_as_itself_and_no_further(self):
         self.assertIn('result', daemon.request('delete', bot='Carl'))
         self.assertFalse(own.exists())
 
+    @unittest.skipUnless(shutil.which('bwrap') or sys.platform == 'darwin', 'sandboxes need bubblewrap')
+    def test_a_sandboxed_bots_commands_run_together_in_its_sandbox(self):
+        (ROOT / '.local').mkdir(exist_ok=True)
+        outside = Path(tempfile.mkdtemp(dir=ROOT / '.local'))
+        self.addCleanup(shutil.rmtree, outside)
+        client = self.client('shell', settings={'sandbox': 'workspace-write'})
+        client.request('create', bot='Bob', workspace=str(self.path))
+        calls = [['shell', {'command': 'sleep .6; touch a'}],
+                 ['shell', {'command': f'sleep .6; touch b; touch {outside}/x'}],
+                 ['shell', {'command': f'touch {outside}/escaped', 'sandbox_permissions': 'require_escalated'}]]
+        started = time.monotonic()
+        turn = client.request('submit', bot='Bob', request_id='t',
+                              prompt='calls:' + json.dumps(calls))['result']['turn']
+        self.assertEqual(client.finished(turn)['data']['status'], 'completed')
+        # The two confined commands ran at once, each in the sandbox.
+        self.assertLess(time.monotonic() - started, 1.1)
+        self.assertTrue((self.path / 'a').exists() and (self.path / 'b').exists())
+        self.assertNotEqual(self.tool_output(client, 'shell-2')[1]['exit_code'], 0)
+        # The one that asks to leave it still runs alone, and is refused.
+        self.assertEqual(self.tool_output(client, 'shell-3')[1]['error'], 'escalation_refused')
+        self.assertEqual(list(outside.iterdir()), [])
+
     def test_a_denial_is_the_result_and_the_turn_goes_on(self):
         client = self.gated()
         turn = client.request('submit', bot='Bob', request_id='t', prompt='shell:touch ran')['result']['turn']
diff --git a/tests/test_runtime.py b/tests/test_runtime.py
index 8be9ee7b1..72a7ed75c 100644
--- a/tests/test_runtime.py
+++ b/tests/test_runtime.py
@@ -319,6 +319,21 @@ def do_POST(self):
                     text = 'echo:' + json.dumps(outputs[-1])
                     output = [{'type': 'message', 'role': 'assistant',
                                'content': [{'type': 'output_text', 'text': text}]}]
+            elif user.startswith('para:'):
+                # `para:CMD|CMD|...`: every command in one response, then
+                # the results in the order they were recorded.
+                start = max(n for n, i in enumerate(request['input']) if i.get('role') == 'user'
+                            and not i['content'][0]['text'].startswith('steer:'))
+                outputs = [i for i in request['input'][start:] if i.get('type') == 'function_call_output']
+                text = ''
+                if not outputs:
+                    output = [{'type': 'function_call', 'name': 'shell', 'call_id': f'para-{n}',
+                               'arguments': json.dumps({'command': command, 'timeout_ms': 5000})}
+                              for n, command in enumerate(user[5:].split('|'), 1)]
+                else:
+                    text = 'echo:' + json.dumps([i['call_id'] for i in outputs])
+                    output = [{'type': 'message', 'role': 'assistant',
+                               'content': [{'type': 'output_text', 'text': text}]}]
             elif user.startswith('sayslow:'):
                 # An answer and a slow command in one response, then a word
                 # for whatever steered it.
diff --git a/tests/test_wait.py b/tests/test_wait.py
index 8d7795f17..ebeb5ad78 100644
--- a/tests/test_wait.py
+++ b/tests/test_wait.py
@@ -403,6 +403,46 @@ def test_a_steer_moves_a_running_command_to_the_background(self):
         result = self.tool_output(client, 'Bob', 'wait-2')['results'][lifted['handle']]
         self.assertEqual((result['stdout'], result['exit_code']), ('done', 0))
 
+    def test_commands_in_one_response_run_together_and_record_in_order(self):
+        client = self.client('echo,shell,wait')
+        client.request('create', bot='Bob', workspace=str(self.path))
+        # The first ends last: results still go in call order.
+        commands = ['sleep .9; printf 1'] + [f'sleep .6; printf {n}' for n in range(2, 7)]
+        started = time.monotonic()
+        submitted = client.request('submit', bot='Bob', request_id='p', prompt='para:' + '|'.join(commands))['result']
+        turn = submitted['turn']
+        finished = client.finished(turn)['data']
+        elapsed = time.monotonic() - started
+        self.assertEqual(finished['status'], 'completed')
+        # Four at a time: two waves, not six commands in a row.
+        self.assertGreater(elapsed, 1.2)
+        self.assertLess(elapsed, 3.0)
+        ids = [f'para-{n}' for n in range(1, 7)]
+        result = client.request('wait', handles=[submitted['handle']])['result']['results'][submitted['handle']]
+        self.assertEqual(result['text'], 'echo:' + json.dumps(ids))
+        events = client.request('events', bot='Bob', after=0, limit=256)['result']['events']
+        self.assertEqual([e['data']['call_id'] for e in events if e['event'] == 'tool_completed'], ids)
+        for n, call in enumerate(ids, 1):
+            self.assertEqual(self.tool_output(client, 'Bob', call)['stdout'], str(n))
+
+    def test_a_steer_moves_every_running_command_to_the_background(self):
+        client = self.client('echo,shell,wait')
+        client.request('create', bot='Bob', workspace=str(self.path))
+        turn = client.request('submit', bot='Bob', request_id='p',
+                              prompt='para:sleep 1.5; printf a|sleep 1.5; printf b|printf c')['result']['turn']
+        client.receive(lambda m: m.get('event') == 'tool_started' and m.get('turn') == turn
+                       and m['data']['call_id'] == 'para-3')
+        time.sleep(.2)
+        sent = time.monotonic()
+        steer = client.request('submit', bot='Bob', request_id='s', prompt='steer:look', delivery='steer')['result']
+        done = client.finished(steer['turn'])['data']
+        self.assertEqual((done['status'], done['into']), ('steered', turn))
+        self.assertLess(time.monotonic() - sent, 1.0)
+        for call in ('para-1', 'para-2'):
+            lifted = self.tool_output(client, 'Bob', call)
+            self.assertEqual((lifted['background'], lifted['steered']), (True, True))
+        self.assertEqual(self.tool_output(client, 'Bob', 'para-3')['stdout'], 'c')
+
     def test_a_command_that_ends_soon_after_a_steer_returns_as_usual(self):
         client = self.client('echo,shell,wait')
         client.request('create', bot='Bob', workspace=str(self.path))