perf(runner): save provider state once per poll instead of once per frame (#20217)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - Native OpenCode, Codex and Cursor Cloud chat turns run through
runnerd. runnerd reads provider frames in order and keeps the run's
provider state in a durable file
> - For every frame that it reads, runnerd rewrote the whole provider
state file and called `fsync` on the file and on its directory. The
state holds every tool result kept in the turn, so each save got more
expensive as the turn went on
> - A tool call request is just the next frame. It waited behind one
full save for every frame before it. #20209 measured tool-call
turnaround that grew from about 1.4 s to about 15 s in one turn
> - #20209 stopped the OpenCode proxy from sending frames that runnerd
discards. Every frame that runnerd still reads still paid a full save,
and other proxies (for example Cursor Cloud) still send such frames
> - This pull request makes runnerd save the frames of one poll once,
before anything derived from them leaves runnerd. A save that would
write the bytes already on disk is skipped
> - The benefit is that the number of saves before a tool call no longer
grows with the number of streamed frames. In a local measurement, a tool
call after 400 frames with 830 KB of kept state took 0.57 s instead of
more than 30 s

## Linked Issues or Issue Description

Refs #20209
Related: #17514 (the same per-event full-state rewrite, on the server
control-plane side)

**What happened**

During a native OpenCode chat turn, tool-call turnaround grew from about
1.4 s to about 15 s as the turn went on. runnerd saved its complete
durable provider state, with `fsync`, after every provider frame. That
state includes the tool results kept so far in the turn. A tool call
request waited behind one full save for each earlier frame.

**Expected behavior**

The time to deliver a tool call must not grow with the number of
assistant-text, reasoning or item frames that came before it. A crash or
restart must still never lose state that a resumed run needs.

**Steps to reproduce**

1. Start a native OpenCode (or Codex app-server or Cursor Cloud) turn
that keeps a few tool results and then streams many text or reasoning
frames before its next tool call.
2. Measure the time from the tool call frame to the
`semantic_tool.input` event.
3. See that the time grows with the number of earlier frames and with
the size of the kept state. Locally, `cargo test -p
paperclip-runner-core --lib measure_streamed_frame -- --ignored
--nocapture` on `master` reports 102 saves and 9.3 s for 100 frames with
830 KB of kept state, and more than 30 s for 400 frames.

## What Changed

-
`packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs`
and `codex_provider.rs`:
- The provider frame loop moves into `poll_provider_frames`. An ordinary
notification frame and a descendant notification frame no longer save
the state. They mark it as unsaved.
- After the loop, `poll_current_provider_with_control` saves the unsaved
state once. It also saves when a frame fails, before it returns the
error.
- Frames that already saved at once still do so: turn terminals, Cursor
Cloud terminal deliveries, `reconciliation_required`, protocol failures,
MCP permission and runtime requests, process exit, and new tool calls.
- A tool call that runnerd answers at once (an exact completed replay or
the receipt-limit stop) saves unsaved state before it writes to the
provider. A rejected tool call already saved first.
- `retained_events`, `execute` and `host_park_ready` save unsaved state
first, so no event, command result or park decision can expose state
that is not on disk.
- Courtesy rejections that `CodexProvider::poll` writes by itself
(unsupported requests, descendant requests, delayed requests after a
terminal, full runtime-request capacity) are held while runnerd reads a
batch (`hold_courtesy_responses`). runnerd sends them only after it
saves the state that covers every earlier frame
(`save_before_provider_writes`). A terminal frame also saves and
releases them before runnerd asks the provider for goal state or drains
the accounting tail. Released courtesy replies are best-effort, as
courtesy replies were before: a provider that already closed stdin
cannot turn a received terminal into a polling failure. The
receipt-limit stop sends held cancellations before its `turn/interrupt`
request, so their order to the provider does not change.
- The trace records for runnerd's normalization
(`rust_durable_normalization`), which name emitted event ids, are kept
until the state that holds those ids is saved. The trace therefore only
names saved events. Saved mappings are written before an exited or
failed provider is dropped, so they are not lost. Raw frame records
still go to the trace when a frame is read; they name no event ids.
- `save_unsaved_frame_state` does not write when the serialized state is
byte-for-byte the same as the saved file. All other saves write as
before.
- Serialization moves to `serialize_provider_state`, and the file write
moves to `write_state_bytes`. The write path (private temporary file,
`fsync`, rename, directory `fsync`, `0600`) does not change.
- New tests in
`packages/paperclip-runner/runner/crates/runner-core/src/tests/provider_frame_save_batching.rs`:
- `streamed_frames_do_not_multiply_durable_saves_before_a_tool_call`: 8
and 48 streamed frame pairs before a tool call need the same number of
saves (1). On `master` they need 18 and 98.
-
`batched_frame_state_is_saved_before_events_leave_and_survives_a_restart`:
at every poll, the saved file is equal to the in-memory state and holds
exactly the delivered events. A new executor loads the file after the
old one stops and gets the same state, the same unacknowledged events
and the pending tool call.
- `batched_frame_state_is_saved_when_a_later_frame_fails`: a frame that
fails after 20 batched frames still leaves those 20 frames on disk.
- `previous_release_state_is_rewritten_by_the_first_batched_save`: a
state file in an older shape (compact JSON without newer default fields)
loads, and the first batched save rewrites it. Unchanged state is not
rewritten. Changed state is.
-
`courtesy_rejections_reach_the_provider_only_after_earlier_frames_are_saved`:
when the courtesy rejection reaches the provider, the saved file (what a
restarted runnerd would load) already holds the 20 earlier frames and
equals the in-memory state. It fails if the hold is removed.
- `trace_interpretations_name_only_saved_events`: every event id that a
trace record names is in saved state at the moment the record is
written. It fails if records are written before the save.
-
`a_crash_inside_an_unsaved_batch_leaves_saved_state_and_provider_output_unchanged`:
after 122 frames are read and not yet saved, the saved file is
byte-for-byte unchanged and runnerd has written nothing to the provider.
A reload gets the pre-batch event sequence and no pending events.
-
`a_held_cancellation_that_cannot_reach_the_provider_keeps_the_completed_turn`:
the provider closes stdin and then completes the turn with a tool call
still pending. The held cancellation cannot be written, and the turn
still completes and is saved.
- `trace_interpretations_survive_a_provider_exit_in_the_same_poll`: the
trace keeps the mappings for frames that arrived in the same poll as the
provider exit.
- `measure_streamed_frame_cost_before_a_tool_call` (ignored): prints
saves and tool-call latency for different frame counts and kept-state
sizes.

## Verification

- `cargo test --locked -p paperclip-runner-core` (debug, macOS arm64):
library 595 passed, 1 failed; all 35 integration test binaries passed
(`--no-fail-fast`). The one library failure is
`task_memory::tests::monitor_bounds_notices`. It also fails when run
alone, takes about 9 minutes, and waits on file-watch timing in
`task_memory.rs`, which this PR does not touch. In one earlier run, two
ACPX park tests in `acpx_park_preparation` failed. The same binary also
failed on unchanged `master` (with a different test), and it passed in
the final run.
- `cargo test -p paperclip-runner-core --lib measure_streamed_frame --
--ignored --nocapture`, debug build, macOS APFS (`sync_all` is
`F_FULLFSYNC`). Each row is one turn: frames streamed, then one tool
call.

| Kept state | Frames | Saves (master) | Tool-call latency (master) |
Saves (this PR) | Tool-call latency (this PR) |
| --- | --- | --- | --- | --- | --- |
| 4 KB | 100 | 102 | 946 ms | 1 | 18 ms |
| 4 KB | 400 | 405 | 3,791 ms | 7 | 70 ms |
| 211 KB | 100 | 102 | 3,207 ms | 1 | 37 ms |
| 211 KB | 400 | 405 | 12,803 ms | 7 | 193 ms |
| 833 KB | 100 | 102 | 9,251 ms | 1 | 93 ms |
| 833 KB | 400 | about 405 | more than 30 s (test deadline) | 7 | 567 ms
|

With 400 frames, the remaining saves are one per poll of up to 128
frames and one per controller acknowledgement.
- With the debug `paperclip-runnerd` and `fake-codex-app-server`
binaries from this branch: `npx vitest run` on
`src/live/runnerd-opencode-streaming-tools`,
`runnerd-opencode-assigned-tools`, `runnerd-codex-transport`,
`runnerd-codex-transport-settlement`,
`runnerd-codex-transport-cold-restore`, `runnerd-restart-continuation`,
`runnerd-saved-continuation`, `runnerd-detach-durability`, and
`src/cli/opencode-proxy-output`, `opencode-proxy-events`,
`opencode-app-server-proxy`: 11 files, 353 tests passed.

## Risks

- Durability. The rule is: nothing derived from a frame leaves runnerd
before the state that holds that frame is on disk. That covers delivered
events, command results, park decisions, tool results and rejections,
courtesy rejections, Cursor Cloud terminal acknowledgements, and trace
records that name event ids. The tests above check each of these.
- The unsaved window is at most one poll (128 frames). Every frame in it
is undelivered: no event from it reached the controller, and runnerd
wrote nothing to the provider while it was unsaved.
`a_crash_inside_an_unsaved_batch_leaves_saved_state_and_provider_output_unchanged`
proves this. A crash in the window therefore leaves the saved state and
everything the controller and provider have seen exactly as a crash just
before those frames were read. The resumed run starts from the same
event sequence, so it cannot deliver an event id twice. Its normal
recovery (thread resume and `thread/read` reconciliation) handles those
frames in the same way as frames still in the provider pipe. On
`master`, a crash could already lose the one frame being processed.
- One provider write can still happen in the window: `thread/read`,
which `CodexProvider::poll` sends to check the lineage of an unknown
helper thread. It is a read-only query that waits for its answer, and it
changes nothing on the provider side.
- Courtesy rejections now reach the provider at the end of the poll
instead of during it. The delay is one batch. The provider gets the same
replies.
- Previous-release story: the saved file shape does not change, and no
new field is saved. A file from the previous release loads as before.
The skip check compares exact bytes, so an older file is never treated
as already saved. The first save after an upgrade rewrites it in the
current shape. A deploy that lands mid-turn sees a file that the
previous release saved after each frame. That file is valid input for
this code.
- The ACPX backend has its own frame loop and is not changed here.

## Model Used

- Anthropic Claude Opus 5.5 (`claude-opus-5-5`), used through Claude
Code with tool use (shell, file editing) and extended thinking.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

packages/paperclip-runner/runner/CODEX_PROVIDER.md

@@ -29,7 +29,10 @@ execution and finalization path.
 
 The provider descriptor, Codex thread ID, account session ID, active Codex turn
 ID, and unacknowledged normalized-event prefix are written to a private,
-bounded, atomically replaced sidecar in the runner state directory. On restart,
+bounded, atomically replaced sidecar in the runner state directory. runnerd
+writes the sidecar once for the provider frames read in one poll, not once per
+frame. The write always comes before an event, a command result, or a reply to
+the provider that depends on those frames leaves runnerd. On restart,
 runnerd resumes that exact thread and reconciles the active turn from
 `thread/read`. An unexpected provider exit retains the last active turn until
 that reconciliation proves whether it is still running, so cancellation and a

packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs

@@ -869,6 +869,11 @@ pub struct CodexProvider {
     quarantined: bool,
     trace: Option<ProviderTraceSink>,
     last_trace_frame_id: Option<u64>,
+    held_courtesy_responses: Option<Vec<Value>>,
+    #[cfg(test)]
+    pub(crate) sent_frame_observer: Option<Box<dyn FnMut(&Value) + Send>>,
+    #[cfg(test)]
+    pub(crate) interpretation_observer: Option<Box<dyn FnMut(&[String]) + Send>>,
     opencode_launch_profile: Option<OpenCodeLaunchProfile>,
     cursor_cloud_launch_profile: Option<CursorCloudLaunchProfile>,
     provider_termination_receipt: Option<Value>,
@@ -1300,6 +1305,11 @@ impl CodexProvider {
             quarantined: false,
             trace: ProviderTraceSink::from_environment(),
             last_trace_frame_id: None,
+            held_courtesy_responses: None,
+            #[cfg(test)]
+            sent_frame_observer: None,
+            #[cfg(test)]
+            interpretation_observer: None,
             opencode_launch_profile: opencode_launch_profile.cloned(),
             cursor_cloud_launch_profile: cursor_cloud_launch_profile.cloned(),
             provider_termination_receipt: None,
@@ -1467,6 +1477,11 @@ impl CodexProvider {
         self.last_trace_frame_id.take()
     }
 
+    #[cfg(test)]
+    pub(crate) fn enable_trace_for_test(&mut self, trace_path: PathBuf) {
+        self.trace = Some(ProviderTraceSink::at_path(trace_path, 64 * 1024 * 1024));
+    }
+
     pub(crate) fn record_provider_trace_interpretation(
         &mut self,
         frame_id: u64,
@@ -1475,6 +1490,10 @@ impl CodexProvider {
         emitted_event_ids: Vec<String>,
         reason: &str,
     ) {
+        #[cfg(test)]
+        if let Some(observer) = self.interpretation_observer.as_mut() {
+            observer(&emitted_event_ids);
+        }
         if let Some(trace) = self.trace.as_mut() {
             trace.interpretation(
                 frame_id,
@@ -3831,10 +3850,36 @@ impl CodexProvider {
         {
             return Ok(());
         }
+        if let Some(held) = self.held_courtesy_responses.as_mut() {
+            held.push(value.clone());
+            return Ok(());
+        }
         self.send_frame(value)
     }
 
+    pub(crate) fn hold_courtesy_responses(&mut self) {
+        self.held_courtesy_responses.get_or_insert_with(Vec::new);
+    }
+
+    pub(crate) fn release_courtesy_responses(&mut self, keep_holding: bool) {
+        let held = if keep_holding {
+            self.held_courtesy_responses
+                .as_mut()
+                .map(std::mem::take)
+                .unwrap_or_default()
+        } else {
+            self.held_courtesy_responses.take().unwrap_or_default()
+        };
+        for value in held {
+            let _ = self.send_frame(&value);
+        }
+    }
+
     fn send_frame(&mut self, value: &Value) -> Result<(), LocalRunnerError> {
+        #[cfg(test)]
+        if let Some(observer) = self.sent_frame_observer.as_mut() {
+            observer(value);
+        }
         if let Some(trace) = self.trace.as_mut() {
             let raw = serde_json::to_vec(value).unwrap_or_default();
             if let Some(frame_id) = trace.frame("client_to_provider", &raw) {

packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs

@@ -79,6 +79,18 @@ fn invalid_tool_call_result(
     }
 }
 
+fn serialize_provider_state(state: &CodexProviderState) -> Result<Vec<u8>, DurableRunnerError> {
+    let bytes = serde_json::to_vec_pretty(state).map_err(|error| {
+        DurableRunnerError::invalid(format!("failed to serialize Codex provider state: {error}"))
+    })?;
+    if bytes.len() as u64 > MAX_PROVIDER_STATE_BYTES {
+        return Err(DurableRunnerError::invalid(
+            "Codex provider state exceeds the 16 MiB limit",
+        ));
+    }
+    Ok(bytes)
+}
+
 fn receipt_limit_deadline_after(timeout_ms: u64) -> Result<u64, DurableRunnerError> {
     current_unix_ms()?.checked_add(timeout_ms).ok_or_else(|| {
         DurableRunnerError::invalid("Codex receipt-limit interruption deadline overflowed")
@@ -1973,6 +1985,13 @@ impl CodexProviderState {
     }
 }
 
+struct NormalizationTrace {
+    frame_id: u64,
+    rule_id: String,
+    mapped: bool,
+    emitted_event_ids: Vec<String>,
+}
+
 pub struct CodexCommandExecutor {
     state_dir: PathBuf,
     state: Option<CodexProviderState>,
@@ -1985,6 +2004,10 @@ pub struct CodexCommandExecutor {
     startup_command: Option<ProviderStartupCommand>,
     startup_evidence_error: Option<DurableRunnerError>,
     recovery_pending_effects: Option<usize>,
+    unsaved_frame_state: std::cell::Cell<bool>,
+    unsaved_trace_interpretations: Vec<NormalizationTrace>,
+    #[cfg(test)]
+    persist_count: std::cell::Cell<usize>,
 }
 
 impl CodexCommandExecutor {
@@ -2001,6 +2024,10 @@ impl CodexCommandExecutor {
             startup_command: None,
             startup_evidence_error: None,
             recovery_pending_effects: None,
+            unsaved_frame_state: std::cell::Cell::new(false),
+            unsaved_trace_interpretations: Vec::new(),
+            #[cfg(test)]
+            persist_count: std::cell::Cell::new(0),
         }
     }
 
@@ -2960,11 +2987,69 @@ impl CodexCommandExecutor {
             .state
             .as_ref()
             .ok_or_else(|| DurableRunnerError::invalid("Codex provider state is unavailable"))?;
-        self.persist_state(state)
+        self.persist_state(state)?;
+        self.unsaved_frame_state.set(false);
+        Ok(())
+    }
+
+    fn save_unsaved_frame_state(&self) -> Result<(), DurableRunnerError> {
+        if !self.unsaved_frame_state.get() {
+            return Ok(());
+        }
+        let state = self
+            .state
+            .as_ref()
+            .ok_or_else(|| DurableRunnerError::invalid("Codex provider state is unavailable"))?;
+        let bytes = serialize_provider_state(state)?;
+        if fs::read(self.state_path()).is_ok_and(|saved| saved == bytes) {
+            self.unsaved_frame_state.set(false);
+            return Ok(());
+        }
+        state.validate()?;
+        self.write_state_bytes(&bytes)?;
+        self.unsaved_frame_state.set(false);
+        Ok(())
+    }
+
+    fn record_saved_trace_interpretations(&mut self) {
+        let traces = std::mem::take(&mut self.unsaved_trace_interpretations);
+        let Some(provider) = self.provider.as_mut() else {
+            return;
+        };
+        for trace in traces {
+            provider.record_provider_trace_interpretation(
+                trace.frame_id,
+                &trace.rule_id,
+                if trace.mapped { "mapped" } else { "ignored" },
+                trace.emitted_event_ids,
+                if trace.mapped {
+                    "Provider notification normalized into durable PRP events"
+                } else {
+                    "Provider notification did not produce a durable PRP event"
+                },
+            );
+        }
+    }
+
+    fn save_before_provider_writes(
+        &mut self,
+        still_polling: bool,
+    ) -> Result<(), DurableRunnerError> {
+        self.save_unsaved_frame_state()?;
+        self.record_saved_trace_interpretations();
+        if let Some(provider) = self.provider.as_mut() {
+            provider.release_courtesy_responses(still_polling);
+        }
+        Ok(())
     }
 
     fn persist_state(&self, state: &CodexProviderState) -> Result<(), DurableRunnerError> {
         state.validate()?;
+        let bytes = serialize_provider_state(state)?;
+        self.write_state_bytes(&bytes)
+    }
+
+    fn write_state_bytes(&self, bytes: &[u8]) -> Result<(), DurableRunnerError> {
         fs::create_dir_all(&self.state_dir).map_err(|error| {
             DurableRunnerError::invalid(format!(
                 "failed to create provider state directory: {error}"
@@ -2980,19 +3065,9 @@ impl CodexCommandExecutor {
         )?;
         verify_private_directory(&self.state_dir)?;
         let path = self.state_path();
-        let bytes = serde_json::to_vec_pretty(state).map_err(|error| {
-            DurableRunnerError::invalid(format!(
-                "failed to serialize Codex provider state: {error}"
-            ))
-        })?;
-        if bytes.len() as u64 > MAX_PROVIDER_STATE_BYTES {
-            return Err(DurableRunnerError::invalid(
-                "Codex provider state exceeds the 16 MiB limit",
-            ));
-        }
         let (temporary, mut file) = create_private_temporary_file(&path)?;
         let result = (|| -> std::io::Result<()> {
-            file.write_all(&bytes)?;
+            file.write_all(bytes)?;
             file.sync_all()?;
             drop(file);
             fs::rename(&temporary, &path)?;
@@ -3010,6 +3085,8 @@ impl CodexCommandExecutor {
         fs::set_permissions(&path, fs::Permissions::from_mode(0o600)).map_err(|error| {
             DurableRunnerError::invalid(format!("failed to protect provider state: {error}"))
         })?;
+        #[cfg(test)]
+        self.persist_count.set(self.persist_count.get() + 1);
         Ok(())
     }
 
@@ -4515,6 +4592,7 @@ impl CodexCommandExecutor {
         call_id: String,
         operation_id: String,
     ) -> Result<(), DurableRunnerError> {
+        self.save_before_provider_writes(true)?;
         let deadline_unix_ms =
             receipt_limit_deadline_after(RECEIPT_LIMIT_INTERRUPT_TERMINAL_DEADLINE_MS)?;
         let state = self
@@ -4562,7 +4640,12 @@ impl CodexCommandExecutor {
                 .expect("Codex state remains available before receipt-limit interruption")
                 .record_receipt_limit_interrupt_attempt()?;
             self.save_state()?;
-            match self.interrupt_turn("semantic_tool_turn_receipt_limit") {
+            self.save_before_provider_writes(false)?;
+            let interrupted = self.interrupt_turn("semantic_tool_turn_receipt_limit");
+            if let Some(provider) = self.provider.as_mut() {
+                provider.hold_courtesy_responses();
+            }
+            match interrupted {
                 Ok(_) => {
                     let accepted_deadline_unix_ms =
                         receipt_limit_deadline_after(RECEIPT_LIMIT_ACCEPTED_TERMINAL_DEADLINE_MS)?;
@@ -4782,6 +4865,7 @@ impl CodexCommandExecutor {
             .admit_tool_call(&call_id, &operation_id, &input);
         match admission {
             Ok(ToolCallAdmission::CompletedReplay(result)) => {
+                self.save_before_provider_writes(true)?;
                 self.provider
                     .as_mut()
                     .expect("provider remains present while handling its tool call")
@@ -5113,6 +5197,25 @@ impl CodexCommandExecutor {
         if self.provider.is_none() {
             return self.settle_receipt_limit_interrupt_if_deadline_elapsed();
         }
+        if let Some(provider) = self.provider.as_mut() {
+            provider.hold_courtesy_responses();
+        }
+        let polled = self.poll_provider_frames(allow_control, receipt_limit_terminal_poll);
+        let saved = self.save_before_provider_writes(false);
+        polled?;
+        saved?;
+        // Check the durable deadline only after a complete provider poll. A
+        // terminal that arrived after interruption acceptance but before this
+        // observation remains authoritative even when several fast controller
+        // polls have already exhausted the interruption-attempt budget.
+        self.settle_receipt_limit_interrupt_if_deadline_elapsed()
+    }
+
+    fn poll_provider_frames(
+        &mut self,
+        allow_control: bool,
+        receipt_limit_terminal_poll: bool,
+    ) -> Result<(), DurableRunnerError> {
         for _ in 0..MAX_EVENTS_PER_POLL {
             let provider = self
                 .provider
@@ -5175,6 +5278,7 @@ impl CodexCommandExecutor {
                     ))?;
                     // Commit the authoritative failure before best-effort provider cleanup.
                     self.save_state()?;
+                    self.record_saved_trace_interpretations();
                     if let Some(mut provider) = self.provider.take() {
                         if let Some(frame_id) = trace_frame_id {
                             provider.record_provider_trace_interpretation(
@@ -5225,7 +5329,7 @@ impl CodexCommandExecutor {
                             "method": method.chars().take(128).collect::<String>(), "receivedThreadId": child, "expectedThreadId": root_thread,
                             "receivedTurnId": child_turn, "expectedTurnId": root_turn }),
                     }])?;
-                    self.save_state()?;
+                    self.unsaved_frame_state.set(true);
                     if let (Some(frame_id), Some(provider)) =
                         (trace_frame_id, self.provider.as_mut())
                     {
@@ -5245,6 +5349,9 @@ impl CodexCommandExecutor {
                     };
                     let normalized_terminal_type =
                         normalized_codex_terminal_event_type(&method, &params);
+                    if normalized_terminal_type.is_some() {
+                        self.save_before_provider_writes(true)?;
+                    }
                     let cursor_cloud_delivery = if self
                         .state
                         .as_ref()
@@ -5486,12 +5593,16 @@ impl CodexCommandExecutor {
                                 .collect::<Vec<_>>()
                         })
                         .unwrap_or_default();
-                    self.save_state()?;
-                    if self
-                        .state
-                        .as_ref()
-                        .is_some_and(|state| state.lifecycle == "reconciliation_required")
+                    let reconciliation_required = state.lifecycle == "reconciliation_required";
+                    if terminal_event_type.is_some()
+                        || cursor_cloud_delivery.is_some()
+                        || reconciliation_required
                     {
+                        self.save_state()?;
+                    } else {
+                        self.unsaved_frame_state.set(true);
+                    }
+                    if reconciliation_required {
                         break;
                     }
                     if let (Some(delivery), Some(provider)) =
@@ -5499,24 +5610,16 @@ impl CodexCommandExecutor {
                     {
                         let _ = provider.acknowledge_cursor_cloud_terminal(delivery);
                     }
-                    if let (Some(frame_id), Some(provider)) =
-                        (trace_frame_id, self.provider.as_mut())
-                    {
-                        provider.record_provider_trace_interpretation(
+                    if let Some(frame_id) = trace_frame_id {
+                        self.unsaved_trace_interpretations.push(NormalizationTrace {
                             frame_id,
-                            &format!("codex.normalize.{}", method.replace('/', ".")),
-                            if normalized_event_count > 0 {
-                                "mapped"
-                            } else {
-                                "ignored"
-                            },
-                            trace_emitted_event_ids,
-                            if normalized_event_count > 0 {
-                                "Provider notification normalized into durable PRP events"
-                            } else {
-                                "Provider notification did not produce a durable PRP event"
-                            },
-                        );
+                            rule_id: format!("codex.normalize.{}", method.replace('/', ".")),
+                            mapped: normalized_event_count > 0,
+                            emitted_event_ids: trace_emitted_event_ids,
+                        });
+                    }
+                    if !self.unsaved_frame_state.get() {
+                        self.record_saved_trace_interpretations();
                     }
                 }
                 CodexProviderEvent::McpPermissionRequest { request_id, params } => {
@@ -5582,6 +5685,8 @@ impl CodexCommandExecutor {
                     process_generation,
                     completed_turn_process_generation,
                 } => {
+                    self.save_unsaved_frame_state()?;
+                    self.record_saved_trace_interpretations();
                     self.provider = None;
                     if !success {
                         let state = self
@@ -5624,16 +5729,13 @@ impl CodexCommandExecutor {
                 }
             }
         }
-        // Check the durable deadline only after a complete provider poll. A
-        // terminal that arrived after interruption acceptance but before this
-        // observation remains authoritative even when several fast controller
-        // polls have already exhausted the interruption-attempt budget.
-        self.settle_receipt_limit_interrupt_if_deadline_elapsed()
+        Ok(())
     }
 }
 
 impl CommandExecutor for CodexCommandExecutor {
     fn host_park_ready(&mut self) -> Result<bool, DurableRunnerError> {
+        self.save_before_provider_writes(false)?;
         let Some(state) = self.state.as_ref() else {
             return Ok(false);
         };
@@ -5701,6 +5803,7 @@ impl CommandExecutor for CodexCommandExecutor {
             command_type: command.command_type.clone(),
         });
         let outcome = (|| {
+            self.save_before_provider_writes(false)?;
             self.restore()?;
             if self
                 .state
@@ -5805,6 +5908,7 @@ impl CommandExecutor for CodexCommandExecutor {
         if let Some(error) = &self.startup_evidence_error {
             return Err(error.clone());
         }
+        self.save_before_provider_writes(false)?;
         Ok(self
             .state
             .as_ref()
@@ -9312,6 +9416,10 @@ done
 #[path = "tests/provider_api_validation.rs"]
 mod api_validation_tests;
 
+#[cfg(test)]
+#[path = "tests/provider_frame_save_batching.rs"]
+mod frame_save_batching_tests;
+
 #[cfg(test)]
 #[path = "tests/provider_assigned_tool_validation.rs"]
 mod assigned_tool_validation_tests;

packages/paperclip-runner/runner/crates/runner-core/src/tests/provider_frame_save_batching.rs

@@ -0,0 +1,579 @@
+use super::*;
+
+fn frame_test_state() -> CodexProviderState {
+    let mut state = CodexProviderState::new(
+        CodexProviderConfig {
+            provider: "opencode".to_owned(),
+            driver: "opencode_server".to_owned(),
+            provider_version: "1.18.34".to_owned(),
+            command: PathBuf::from("node"),
+            args: Vec::new(),
+            cwd: std::env::current_dir()
+                .unwrap()
+                .to_string_lossy()
+                .into_owned(),
+            model: Some("openrouter/model".to_owned()),
+            provider_session_id: None,
+            instructions: String::new(),
+            approval_policy: "never".to_owned(),
+            externally_sandboxed: false,
+            read_only: false,
+            include_skill_instructions: None,
+            conversation_mode: None,
+        },
+        Some(CompletionContractBinding {
+            revision: "revision-1".to_owned(),
+            criterion_ids: vec!["criterion-1".to_owned()],
+        }),
+        ProviderToolBridge::default(),
+    );
+    state.thread_id = Some("thread-1".to_owned());
+    state.active_provider_turn_id = Some("turn-1".to_owned());
+    state.lifecycle = "turn_active".to_owned();
+    state
+}
+
+fn streamed_task_context_operation() -> crate::provider_bridge::AuthorizedTool {
+    crate::provider_bridge::AuthorizedTool {
+        operation_id: "get_task_context".to_owned(),
+        version: 1,
+        description: "Read the active task context.".to_owned(),
+        input_schema: json!({"type": "object"}),
+        response_schema: json!({"type": "object"}),
+    }
+}
+
+#[cfg(unix)]
+fn streamed_turn_before_tool_call(
+    frames: usize,
+    retained_results: usize,
+    tool: &str,
+    wait_for_stream: bool,
+) -> (PathBuf, CodexCommandExecutor) {
+    let tail = format!(
+        r#"{{"id":"tool-1","method":"item/tool/call","params":{{"threadId":"thread-1","turnId":"turn-1","callId":"call-streamed","tool":"{tool}","arguments":{{}}}}}}"#
+    );
+    streamed_turn(frames, retained_results, &tail, wait_for_stream, |_| {})
+}
+
+#[cfg(unix)]
+fn streamed_turn(
+    frames: usize,
+    retained_results: usize,
+    tail: &str,
+    wait_for_stream: bool,
+    prepare_provider: impl FnOnce(&mut CodexProvider),
+) -> (PathBuf, CodexCommandExecutor) {
+    streamed_turn_then(
+        frames,
+        retained_results,
+        tail,
+        ":",
+        wait_for_stream,
+        prepare_provider,
+    )
+}
+
+#[cfg(unix)]
+fn streamed_turn_then(
+    frames: usize,
+    retained_results: usize,
+    tail: &str,
+    after: &str,
+    wait_for_stream: bool,
+    prepare_provider: impl FnOnce(&mut CodexProvider),
+) -> (PathBuf, CodexCommandExecutor) {
+    let directory = std::env::temp_dir().join(format!(
+        "paperclip-streamed-frames-{}",
+        uuid::Uuid::new_v4()
+    ));
+    fs::create_dir_all(&directory).unwrap();
+    let streamed = directory.join("streamed");
+    let script = r#"
+while IFS= read -r line; do
+  case "$line" in
+    *'"method":"initialize"'*) printf '%s\n' '{"id":1,"result":{}}' ;;
+    *'"method":"thread/start"'*) printf '%s\n' '{"id":2,"result":{"thread":{"id":"thread-1"}}}' ;;
+    *'"method":"turn/start"'*)
+      printf '%s\n' '{"id":3,"result":{"turn":{"id":"turn-1"}}}'
+      printf '%s\n' '{"method":"turn/started","params":{"threadId":"thread-1","turn":{"id":"turn-1","status":"inProgress"}}}'
+      i=0
+      while [ "$i" -lt FRAMES ]; do
+        printf '%s\n' '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","itemId":"message-1","delta":"x"}}'
+        printf '%s\n' '{"method":"item/updated","params":{"threadId":"thread-1","turnId":"turn-1","item":{"id":"reasoning-1","type":"reasoning","summary":[],"content":[]}}}'
+        i=$((i+1))
+      done
+      printf '%s\n' 'TAIL'
+      touch 'STREAMED'
+      AFTER ;;
+  esac
+done
+"#
+    .replace("FRAMES", &frames.to_string())
+    .replace("TAIL", tail)
+    .replace("AFTER", after)
+    .replace("STREAMED", &streamed.display().to_string());
+    let operation = streamed_task_context_operation();
+    let mut bridge = ProviderToolBridge::default();
+    bridge
+        .prepare(AuthorizedToolSet {
+            schema: TOOL_SET_SCHEMA.to_owned(),
+            schema_version: 1,
+            catalog_digest: authorized_tool_catalog_digest(std::slice::from_ref(&operation))
+                .unwrap(),
+            operations: vec![operation.clone()],
+        })
+        .unwrap();
+    for index in 0..retained_results {
+        let call_id = format!("retained-{index}");
+        bridge
+            .begin_call(call_id.clone(), "get_task_context".to_owned(), json!({}))
+            .unwrap();
+        bridge
+            .apply_result(ToolResult {
+                call_id,
+                operation_id: "get_task_context".to_owned(),
+                result: json!({"body": "x".repeat(10_000)}),
+                is_error: false,
+            })
+            .unwrap();
+    }
+    let mut state = frame_test_state();
+    state.config.provider = "codex".to_owned();
+    state.config.driver = "codex_app_server".to_owned();
+    state.config.command = PathBuf::from("/bin/sh");
+    state.config.args = vec!["-c".to_owned(), script];
+    state.config.model = None;
+    state.tool_bridge = bridge;
+    let mut provider = CodexProvider::start_with_tools(&state.config, [operation], None).unwrap();
+    prepare_provider(&mut provider);
+    provider.start_turn("Test", &state.config.cwd).unwrap();
+    state.provider_process_generation = provider.process_generation();
+    let mut executor = CodexCommandExecutor::new(&directory);
+    executor.event_identity = Some(ProviderEventIdentity {
+        runner_instance_id: "runner-1".to_owned(),
+        run_id: "run-1".to_owned(),
+        normalized_session_id: "session-1".to_owned(),
+        turn_id: "turn-1".to_owned(),
+        item_id: "item-1".to_owned(),
+    });
+    executor.state = Some(state);
+    executor.provider = Some(provider);
+    executor.restore_checked = true;
+    executor.save_state().unwrap();
+    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
+    while wait_for_stream && !streamed.exists() {
+        assert!(std::time::Instant::now() < deadline);
+        std::thread::sleep(std::time::Duration::from_millis(5));
+    }
+    (directory, executor)
+}
+
+fn poll_until_semantic_tool_call(executor: &mut CodexCommandExecutor) -> Vec<PolledEvent> {
+    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
+    let mut observed = Vec::new();
+    loop {
+        assert!(std::time::Instant::now() < deadline);
+        let events = executor.poll_events().unwrap();
+        let count = events.len();
+        let delivered = events
+            .iter()
+            .any(|event| event.event_type == "semantic_tool.input");
+        observed.extend(events);
+        if delivered {
+            return observed;
+        }
+        executor.acknowledge_events(count).unwrap();
+    }
+}
+
+#[cfg(unix)]
+#[test]
+fn streamed_frames_do_not_multiply_durable_saves_before_a_tool_call() {
+    let mut saves = Vec::new();
+    for frames in [8, 48] {
+        let (directory, mut executor) =
+            streamed_turn_before_tool_call(frames, 4, "get_task_context", true);
+        let before = executor.persist_count.get();
+        let observed = poll_until_semantic_tool_call(&mut executor);
+        saves.push(executor.persist_count.get() - before);
+        assert_eq!(
+            observed
+                .iter()
+                .filter(|event| event.event_type == "item.delta")
+                .count(),
+            frames
+        );
+        executor.provider.as_mut().unwrap().shutdown().unwrap();
+        fs::remove_dir_all(directory).unwrap();
+    }
+    eprintln!("durable saves before the tool call: {saves:?}");
+    assert_eq!(
+        saves[0], saves[1],
+        "saves grew with streamed frames: {saves:?}"
+    );
+}
+
+fn saved_provider_state(executor: &CodexCommandExecutor) -> CodexProviderState {
+    serde_json::from_slice(&fs::read(executor.state_path()).unwrap()).unwrap()
+}
+
+#[cfg(unix)]
+#[test]
+fn batched_frame_state_is_saved_before_events_leave_and_survives_a_restart() {
+    let (directory, mut executor) =
+        streamed_turn_before_tool_call(300, 4, "get_task_context", false);
+    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
+    loop {
+        assert!(std::time::Instant::now() < deadline);
+        let events = executor.poll_events().unwrap();
+        assert!(!executor.unsaved_frame_state.get());
+        let saved = saved_provider_state(&executor);
+        assert_eq!(
+            serde_json::to_value(&saved).unwrap(),
+            serde_json::to_value(executor.state.as_ref().unwrap()).unwrap()
+        );
+        assert_eq!(saved.pending_events, events);
+        if events
+            .iter()
+            .any(|event| event.event_type == "semantic_tool.input")
+        {
+            break;
+        }
+        let count = events.len();
+        executor.acknowledge_events(count).unwrap();
+    }
+    let live = executor.state.clone().unwrap();
+    executor.provider.as_mut().unwrap().shutdown().unwrap();
+    drop(executor);
+
+    let mut resumed = CodexCommandExecutor::new(&directory);
+    resumed.load_state_without_provider().unwrap();
+    let resumed_state = resumed.state.as_ref().unwrap();
+    assert_eq!(
+        serde_json::to_value(resumed_state).unwrap(),
+        serde_json::to_value(&live).unwrap()
+    );
+    assert!(resumed_state
+        .tool_bridge
+        .pending_calls()
+        .any(|call| call.call_id == "call-streamed"));
+    assert_eq!(
+        resumed.retained_events().unwrap(),
+        live.pending_events.iter().cloned().collect::<Vec<_>>()
+    );
+    fs::remove_dir_all(directory).unwrap();
+}
+
+#[cfg(unix)]
+#[test]
+fn batched_frame_state_is_saved_when_a_later_frame_fails() {
+    let (directory, mut executor) =
+        streamed_turn_before_tool_call(20, 0, "unauthorized_tool", true);
+    assert!(executor.poll_events().is_err());
+    assert!(!executor.unsaved_frame_state.get());
+    let saved = saved_provider_state(&executor);
+    assert_eq!(
+        serde_json::to_value(&saved).unwrap(),
+        serde_json::to_value(executor.state.as_ref().unwrap()).unwrap()
+    );
+    assert_eq!(
+        saved
+            .pending_events
+            .iter()
+            .filter(|event| event.event_type == "item.delta")
+            .count(),
+        20
+    );
+    let _ = executor.provider.as_mut().unwrap().shutdown();
+    fs::remove_dir_all(directory).unwrap();
+}
+
+fn previous_release_state() -> CodexProviderState {
+    let mut state = frame_test_state();
+    state.config.provider = "codex".to_owned();
+    state.config.driver = "codex_app_server".to_owned();
+    state.config.model = None;
+    state
+}
+
+#[test]
+fn previous_release_state_is_rewritten_by_the_first_batched_save() {
+    let directory = std::env::temp_dir().join(format!(
+        "paperclip-batched-save-upgrade-{}",
+        uuid::Uuid::new_v4()
+    ));
+    fs::create_dir_all(&directory).unwrap();
+    #[cfg(unix)]
+    fs::set_permissions(&directory, fs::Permissions::from_mode(0o700)).unwrap();
+    let mut previous = serde_json::to_value(previous_release_state()).unwrap();
+    let object = previous.as_object_mut().unwrap();
+    assert!(object.remove("activeTerminalToolRejected").is_some());
+    assert!(object.remove("descendantThreadIds").is_some());
+    let previous_bytes = serde_json::to_vec(&previous).unwrap();
+    let path = directory.join(CODEX_PROVIDER_STATE_FILE);
+    fs::write(&path, &previous_bytes).unwrap();
+    #[cfg(unix)]
+    fs::set_permissions(&path, fs::Permissions::from_mode(0o600)).unwrap();
+
+    let mut resumed = CodexCommandExecutor::new(&directory);
+    resumed.load_state_without_provider().unwrap();
+    assert_eq!(resumed.state.as_ref().unwrap(), &previous_release_state());
+    resumed.unsaved_frame_state.set(true);
+    resumed.save_unsaved_frame_state().unwrap();
+    assert_eq!(resumed.persist_count.get(), 1);
+    assert_ne!(fs::read(&path).unwrap(), previous_bytes);
+    assert_eq!(
+        serde_json::to_value(saved_provider_state(&resumed)).unwrap(),
+        serde_json::to_value(resumed.state.as_ref().unwrap()).unwrap()
+    );
+
+    resumed.unsaved_frame_state.set(true);
+    resumed.save_unsaved_frame_state().unwrap();
+    assert_eq!(resumed.persist_count.get(), 1);
+
+    resumed.state.as_mut().unwrap().last_agent_message = Some("streamed".to_owned());
+    resumed.unsaved_frame_state.set(true);
+    resumed.save_unsaved_frame_state().unwrap();
+    assert_eq!(resumed.persist_count.get(), 2);
+    assert_eq!(
+        saved_provider_state(&resumed).last_agent_message.as_deref(),
+        Some("streamed")
+    );
+    fs::remove_dir_all(directory).unwrap();
+}
+
+#[cfg(unix)]
+#[test]
+#[ignore = "timing measurement; run with --ignored --nocapture"]
+fn measure_streamed_frame_cost_before_a_tool_call() {
+    for retained in [0, 20, 80] {
+        for frames in [50, 200] {
+            let (directory, mut executor) =
+                streamed_turn_before_tool_call(frames, retained, "get_task_context", false);
+            let state_bytes = fs::metadata(executor.state_path()).unwrap().len();
+            let before = executor.persist_count.get();
+            let started = std::time::Instant::now();
+            poll_until_semantic_tool_call(&mut executor);
+            let elapsed = started.elapsed();
+            eprintln!(
+                "retained={retained} stateBytes={state_bytes} frames={} saves={} toolCallLatencyMs={:.1}",
+                frames * 2,
+                executor.persist_count.get() - before,
+                elapsed.as_secs_f64() * 1000.0
+            );
+            executor.provider.as_mut().unwrap().shutdown().unwrap();
+            fs::remove_dir_all(directory).unwrap();
+        }
+    }
+}
+
+#[cfg(unix)]
+fn event_identity_for_tests() -> ProviderEventIdentity {
+    ProviderEventIdentity {
+        runner_instance_id: "runner-1".to_owned(),
+        run_id: "run-1".to_owned(),
+        normalized_session_id: "session-1".to_owned(),
+        turn_id: "turn-1".to_owned(),
+        item_id: "item-1".to_owned(),
+    }
+}
+
+#[cfg(unix)]
+fn read_saved_state(path: &std::path::Path) -> CodexProviderState {
+    serde_json::from_slice(&fs::read(path).unwrap()).unwrap()
+}
+
+#[cfg(unix)]
+#[test]
+fn courtesy_rejections_reach_the_provider_only_after_earlier_frames_are_saved() {
+    let directory_slot = std::sync::Arc::new(std::sync::Mutex::new(None::<PathBuf>));
+    let saved_at_rejection = std::sync::Arc::new(std::sync::Mutex::new(None::<CodexProviderState>));
+    let observer_directory = std::sync::Arc::clone(&directory_slot);
+    let observer_saved = std::sync::Arc::clone(&saved_at_rejection);
+    let tail = r#"{"id":"courtesy-1","method":"paperclip/test/unsupported","params":{"threadId":"thread-1","turnId":"turn-1"}}"#;
+    let (directory, mut executor) = streamed_turn(20, 0, tail, true, move |provider| {
+        provider.sent_frame_observer = Some(Box::new(move |frame: &Value| {
+            if frame["id"] != "courtesy-1" {
+                return;
+            }
+            let directory = observer_directory.lock().unwrap().clone().unwrap();
+            *observer_saved.lock().unwrap() =
+                Some(read_saved_state(&directory.join(CODEX_PROVIDER_STATE_FILE)));
+        }));
+    });
+    *directory_slot.lock().unwrap() = Some(directory.clone());
+    let events = executor.poll_events().unwrap();
+    assert!(events.iter().any(|event| event.event_type == "item.delta"));
+
+    let resumed = saved_at_rejection
+        .lock()
+        .unwrap()
+        .take()
+        .expect("runnerd sent the courtesy rejection");
+    assert_eq!(
+        resumed
+            .pending_events
+            .iter()
+            .filter(|event| event.event_type == "item.delta")
+            .count(),
+        20
+    );
+    assert_eq!(
+        serde_json::to_value(&resumed).unwrap(),
+        serde_json::to_value(executor.state.as_ref().unwrap()).unwrap()
+    );
+    executor.provider.as_mut().unwrap().shutdown().unwrap();
+    fs::remove_dir_all(directory).unwrap();
+}
+
+#[cfg(unix)]
+#[test]
+fn trace_interpretations_name_only_saved_events() {
+    let directory_slot = std::sync::Arc::new(std::sync::Mutex::new(None::<PathBuf>));
+    let checked = std::sync::Arc::new(std::sync::Mutex::new((0usize, Vec::<String>::new())));
+    let observer_directory = std::sync::Arc::clone(&directory_slot);
+    let observer_checked = std::sync::Arc::clone(&checked);
+    let trace_path = std::env::temp_dir().join(format!(
+        "paperclip-frame-trace-{}.jsonl",
+        uuid::Uuid::new_v4()
+    ));
+    let provider_trace_path = trace_path.clone();
+    let (directory, mut executor) = streamed_turn_before_tool_call_with(40, move |provider| {
+        provider.enable_trace_for_test(provider_trace_path);
+        provider.interpretation_observer = Some(Box::new(move |ids: &[String]| {
+            if ids.is_empty() {
+                return;
+            }
+            let directory = observer_directory.lock().unwrap().clone().unwrap();
+            let saved = read_saved_state(&directory.join(CODEX_PROVIDER_STATE_FILE));
+            let identity = event_identity_for_tests();
+            let saved_ids = (0..saved.next_provider_event_seq)
+                .map(|sequence| identity.source_event_id(&provider_event_id(sequence)))
+                .collect::<HashSet<_>>();
+            let mut checked = observer_checked.lock().unwrap();
+            checked.0 += ids.len();
+            checked
+                .1
+                .extend(ids.iter().filter(|id| !saved_ids.contains(*id)).cloned());
+        }));
+    });
+    *directory_slot.lock().unwrap() = Some(directory.clone());
+    poll_until_semantic_tool_call(&mut executor);
+    let (count, unsaved) = checked.lock().unwrap().clone();
+    assert!(count >= 40, "trace named {count} events");
+    assert!(
+        unsaved.is_empty(),
+        "trace named unsaved events: {unsaved:?}"
+    );
+    executor.provider.as_mut().unwrap().shutdown().unwrap();
+    fs::remove_dir_all(directory).unwrap();
+    let _ = fs::remove_file(trace_path);
+}
+
+#[cfg(unix)]
+#[test]
+fn a_crash_inside_an_unsaved_batch_leaves_saved_state_and_provider_output_unchanged() {
+    let sent = std::sync::Arc::new(std::sync::Mutex::new(Vec::<Value>::new()));
+    let observer_sent = std::sync::Arc::clone(&sent);
+    let tail = r#"{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","itemId":"message-1","delta":"last"}}"#;
+    let (directory, mut executor) = streamed_turn(60, 0, tail, true, move |provider| {
+        provider.sent_frame_observer = Some(Box::new(move |frame: &Value| {
+            observer_sent.lock().unwrap().push(frame.clone());
+        }));
+    });
+    let path = executor.state_path();
+    let saved_before = fs::read(&path).unwrap();
+    let sent_before = sent.lock().unwrap().len();
+    executor
+        .provider
+        .as_mut()
+        .unwrap()
+        .hold_courtesy_responses();
+    executor.poll_provider_frames(true, false).unwrap();
+    assert!(executor.unsaved_frame_state.get());
+    let live_next = executor.state.as_ref().unwrap().next_provider_event_seq;
+    let saved_next = read_saved_state(&path).next_provider_event_seq;
+    assert!(live_next >= saved_next + 121);
+    assert_eq!(fs::read(&path).unwrap(), saved_before);
+    assert_eq!(sent.lock().unwrap().len(), sent_before);
+    executor.provider.as_mut().unwrap().shutdown().unwrap();
+    drop(executor);
+
+    let mut resumed = CodexCommandExecutor::new(&directory);
+    resumed.load_state_without_provider().unwrap();
+    let resumed_state = resumed.state.as_ref().unwrap();
+    assert_eq!(resumed_state.next_provider_event_seq, saved_next);
+    assert!(resumed_state.pending_events.is_empty());
+    fs::remove_dir_all(directory).unwrap();
+}
+
+#[cfg(unix)]
+fn streamed_turn_before_tool_call_with(
+    frames: usize,
+    prepare_provider: impl FnOnce(&mut CodexProvider),
+) -> (PathBuf, CodexCommandExecutor) {
+    let tail = r#"{"id":"tool-1","method":"item/tool/call","params":{"threadId":"thread-1","turnId":"turn-1","callId":"call-streamed","tool":"get_task_context","arguments":{}}}"#;
+    streamed_turn(frames, 0, tail, true, prepare_provider)
+}
+
+#[cfg(unix)]
+#[test]
+fn a_held_cancellation_that_cannot_reach_the_provider_keeps_the_completed_turn() {
+    let tail = r#"{"id":"tool-1","method":"item/tool/call","params":{"threadId":"thread-1","turnId":"turn-1","callId":"call-streamed","tool":"get_task_context","arguments":{}}}"#;
+    let after = r#"exec 0<&-; printf '%s\n' '{"method":"turn/completed","params":{"threadId":"thread-1","turn":{"id":"turn-1","status":"completed"}}}'; sleep 2; exit 0"#;
+    let (directory, mut executor) = streamed_turn_then(10, 0, tail, after, true, |_| {});
+    std::thread::sleep(std::time::Duration::from_millis(200));
+    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
+    let mut observed = Vec::new();
+    while !observed
+        .iter()
+        .any(|event: &PolledEvent| event.event_type == "turn.completed")
+    {
+        assert!(std::time::Instant::now() < deadline);
+        let events = executor.poll_events().unwrap();
+        let count = events.len();
+        observed.extend(events);
+        executor.acknowledge_events(count).unwrap();
+    }
+    assert!(observed
+        .iter()
+        .any(|event| event.event_type == "semantic_tool.input"));
+    let saved = read_saved_state(&executor.state_path());
+    assert!(saved.completed_turn_authoritative);
+    if let Some(provider) = executor.provider.as_mut() {
+        let _ = provider.shutdown();
+    }
+    fs::remove_dir_all(directory).unwrap();
+}
+
+#[cfg(unix)]
+#[test]
+fn trace_interpretations_survive_a_provider_exit_in_the_same_poll() {
+    let named = std::sync::Arc::new(std::sync::Mutex::new(0usize));
+    let observer_named = std::sync::Arc::clone(&named);
+    let tail = r#"{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","itemId":"message-1","delta":"last"}}"#;
+    let trace_path = std::env::temp_dir().join(format!(
+        "paperclip-frame-trace-exit-{}.jsonl",
+        uuid::Uuid::new_v4()
+    ));
+    let provider_trace_path = trace_path.clone();
+    let (directory, mut executor) =
+        streamed_turn_then(20, 0, tail, "exit 0", true, move |provider| {
+            provider.enable_trace_for_test(provider_trace_path);
+            provider.interpretation_observer = Some(Box::new(move |ids: &[String]| {
+                *observer_named.lock().unwrap() += ids.len();
+            }));
+        });
+    std::thread::sleep(std::time::Duration::from_millis(200));
+    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
+    while executor.provider.is_some() {
+        assert!(std::time::Instant::now() < deadline);
+        let events = executor.poll_events().unwrap();
+        let count = events.len();
+        executor.acknowledge_events(count).unwrap();
+    }
+    assert!(*named.lock().unwrap() >= 41);
+    fs::remove_dir_all(directory).unwrap();
+    let _ = fs::remove_file(trace_path);
+}