perf(runner): poll the provider right after an ACK releases it (#20374)

## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The Runner (`paperclip-runnerd`) sends provider events to the controller over one authenticated socket > - Its main loop reads one control frame, then polls the provider only after a read finds no frame > - After the controller ACKs the last sent event, the next read waits the full 250 ms `RUNTIME_READ_TIMEOUT` before the Runner polls the provider again > - A Codex turn start needs one or two such rounds before the Runner reads `turn/started`, so every Codex turn waits 250 ms or 500 ms for no reason > - This pull request reads only the frames that are already queued after a consumed control frame, and then polls at once, when this connection produced the unacknowledged output > - The benefit is 250 ms less time to `turn/started` for each ACK round, and faster event delivery after each ACK ## Linked Issues or Issue Description This is an enhancement. The fields follow `.github/ISSUE_TEMPLATE/enhancement.yml`. **What existing behavior does this improve?** Codex turns wait for the Runner read timeout before the Runner sends `turn/started`. **Subsystem affected** The Runner (`paperclip-runnerd`) durable loop in `packages/paperclip-runner/runner/crates/runner-core/src/durable/`, for the local Codex provider. **Current behavior** On a hosted test deployment (48 h, 114 Codex turns), `runner.transport.started_event` for Codex has two modes: | Codex turn | n | `started_event` p50 | | --- | --- | --- | | no provider cold start in the trace | 57 | 468 ms (all < 700 ms) | | provider cold start in the trace | 57 | 918 ms (51 of 57 ≥ 700 ms) | For Claude (ACPX) turns, the p50 is 90 ms. ACPX returns `turn.started` with the command result. Codex reads `turn/started` from the provider in a later poll. **Proposed behavior** The Runner polls the provider as soon as the controller ACK releases the poll gate. The time to `turn/started` then follows the controller round trip, not the read timeout. **Reason and benefit** - Real Codex (`codex app-server` 0.157, local) sends `turn/started` 2–16 ms after its `turn/start` response. The provider is not the cause. - `runner.transport.started_event` starts when the `turn.start` command result arrives. It ends when the controller commits `turn.started`. In this time the Runner sends `turn.accepted`, waits for its ACK, waits 250 ms, polls, and sends `turn.started`. - On the first turn after a provider launch, the first poll returns only the launch diagnostic that the executor retained. A second ACK round and a second 250 ms wait come before the poll that reads `turn/started`. This is the difference between the two modes above (about 450 ms: 250 ms wait and one event round). - So 250 ms (warm turn) or 500 ms (first turn) of each Codex `started_event` is the read timeout. The rest is one or two controller event rounds (about 200 ms each on that deployment), less than 20 ms of Codex time, and the 10 ms pump interval of the TypeScript transport. - The expected benefit is 250 ms less per warm Codex turn start and 500 ms less for the first turn after a provider launch, plus faster Codex event delivery after each ACK. **Breaking changes** None. No protocol, schema, or controller change. ## What Changed - `AuthenticatedTransport::receive_queued_json` reads with a 1 ms timeout and then restores `RUNTIME_READ_TIMEOUT`. - New `CommandExecutor::provider_poll_is_live` hook (default `false`). The local Codex executor returns `true` only when it owns a running provider with no open startup attempt, so a poll cannot restore or launch a provider. The native executor forwards it for the local Codex facade only. - In the steady-state loop, after a consumed control frame, the Runner uses `receive_queued_json` when it has sent output that this connection produced (a command or a provider poll) and the executor reports a live provider. If no frame is queued, it polls the provider at once. - When a provider poll returns no output, or when no frame was consumed, the Runner uses the 250 ms read as before. A 1 ms read follows only a consumed frame, so an idle loop does not spin. - The replay boundary is the highest source sequence before the welcome's pending commands run. A fresh command delivered in the welcome counts as this connection's output. Replayed commands only resend their stored results. - Replayed output from an earlier connection or process, ACPX, and the other providers keep the 250 ms read. The controller can still send its next command before a poll restores a provider. (Without the live-provider gate, six ACPX source-recovery tests in `runnerd-acpx-source-reconciliation.test.ts` failed: the fast poll restored the ACPX sidecar before the next recovery command.) - Control frames that are already queued behind the ACK are still read and executed before the provider poll. The ACK gate in `poll_executor_events_after_controller_ack` is unchanged. ## Verification - `cargo test -p paperclip-runner-core --lib durable::` (121 passed). Five new tests in `src/durable/tests/ack_release_poll.rs`, included by the transport test module like `hosted_retention.rs`: - `acknowledged_output_polls_the_provider_without_waiting_out_the_read_timeout`: the provider poll follows the ACK read in less than `RUNTIME_READ_TIMEOUT`. Without the change it fails (the poll waited 280–356 ms). - `control_queued_behind_an_ack_runs_before_the_next_provider_poll`: a command sent in the same write as the ACK executes before the provider poll. - `welcome_command_output_polls_the_provider_without_waiting_out_the_read_timeout`: the output of a fresh command in the welcome also releases the poll at once. With the replay boundary taken after the welcome commands, it fails. - `replayed_output_ack_keeps_the_read_timeout_before_the_provider_poll`: after a reconnect, a command sent 50 ms after the ACK of replayed output executes before the provider poll. Without the replay condition it fails. - `restorable_provider_ack_keeps_the_read_timeout_before_the_provider_poll`: when the executor does not report a live provider, a command sent 50 ms after the ACK executes before the provider poll. - `npx vitest run src/live/runnerd-` in `packages/paperclip-runner` with the new debug `paperclip-runnerd`: run with `--no-file-parallelism`. Before the welcome-boundary fix: 576 passed, 3 skipped. After it: 575 passed, 3 skipped, and 1 failure once (`stops a lost runnerd's provider and escaped tool job from its lineage journal (journal=true)`, a SIGKILL timing test) while `cargo test` ran at the same time. `runnerd-codex-transport.test.ts` then passed 293 of 293 three times with this build, and three times with the base build. - Local before and after measurement with the real Runner, the TypeScript transport, and the fake Codex app-server (8 turns, 2 runs each, debug build): | span | before p50 | after p50 | | --- | --- | --- | | `runner.transport.started_event` | 377 ms (first turn 682–695 ms) | 119–122 ms (first turn 117–119 ms) | | `runner.transport.turn_start` | 462–471 ms (first turn 775–780 ms) | 204–212 ms (first turn 247–251 ms) | - The saving on the hosted deployment is not measured. From the analysis above, the expected saving is about 250 ms for a warm Codex turn and about 500 ms for the first turn after a provider launch. ## Risks - Only local Codex gets the fast path. ACPX streaming keeps the old 250 ms wait after each ACK. A later change can opt ACPX in when its poll can prove that it does not restore the sidecar. - More provider polls during Codex streaming: a poll can now follow each ACK at once instead of after 250 ms. Batches can become smaller, so the provider state save for each batch can happen more often. Each event already has its own durable save. - A command that the controller sends more than 1 ms after its ACK can now arrive after a provider poll. This was already possible after 250 ms. The change applies only when this connection produced the output, so the provider is already active. - If the Runner cannot restore the 250 ms read timeout, it treats the socket as failed and reconnects. It does not continue with a 1 ms timeout. - No protocol, schema, or controller change. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`) in Claude Code, with tool use (shell, code search, Honeycomb queries) and extended reasoning. ## 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 - [ ] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] 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>
2026-11-11 05:11:49 +01:00 · 2026-11-09 15:38:58 -08:00
1 parent c985541a57
commit 5858cef8f9
@@ -387,6 +387,10 @@ pub trait CommandExecutor {
        Ok(())
    }

+    fn provider_poll_is_live(&self) -> bool {
+        false
+    }
+
    /// Removes the prefix returned by `poll_events` after every event in that
    /// prefix is durably committed to the PRP outbox. Implementations that retain
    /// provider events must not remove them before this acknowledgement.
@@ -899,6 +903,7 @@ pub fn run_durable_runner_with_host_control<E: CommandExecutor>(
            notifications.poll(&config, &mut state, &store);
        }
        let mut sent_source_seq = state.acked_source_seq;
+        let replayed_through_source_seq = state.highest_source_seq();
        let mut ack_persistence = CumulativeAckPersistence::default();

        let mut lifecycle_after_reply = CommandLifecycle::Continue;
@@ -1055,6 +1060,8 @@ pub fn run_durable_runner_with_host_control<E: CommandExecutor>(
                    .min(reconnect_deadline.saturating_duration_since(Instant::now()));
            continue;
        }
+        let mut control_frame_consumed = false;
+        let mut quiet_through_source_seq = replayed_through_source_seq;
        loop {
            if !config.max_runtime.is_zero() && started.elapsed() >= config.max_runtime {
                break;
@@ -1153,7 +1160,19 @@ pub fn run_durable_runner_with_host_control<E: CommandExecutor>(
                    }
                }
            }
-            let control_message = transport.receive_json();
+            let control_message = if control_frame_consumed
+                && state.highest_source_seq() > quiet_through_source_seq
+                && executor.provider_poll_is_live()
+            {
+                transport.receive_queued_json()
+            } else {
+                transport.receive_json()
+            };
+            control_frame_consumed = matches!(control_message, Ok(Some(_)));
+            let source_seq_before_poll = state.highest_source_seq();
+            let polls_provider = matches!(control_message, Ok(None))
+                && state.pending_provider_cleanup.is_none()
+                && state.acked_source_seq >= sent_source_seq;
            poll_executor_events_when_control_idle(
                &mut state,
                &store,
@@ -1162,6 +1181,9 @@ pub fn run_durable_runner_with_host_control<E: CommandExecutor>(
                sent_source_seq,
                &control_message,
            )?;
+            if polls_provider {
+                quiet_through_source_seq = source_seq_before_poll;
+            }
            let message = match control_message {
                Ok(Some(message)) => message,
                Ok(None) => continue,
@@ -0,0 +1,322 @@
+#[test]
+fn acknowledged_output_polls_the_provider_without_waiting_out_the_read_timeout() {
+    acknowledged_prefix_releases_provider_output(AckRelease::Idle);
+}
+
+#[test]
+fn control_queued_behind_an_ack_runs_before_the_next_provider_poll() {
+    acknowledged_prefix_releases_provider_output(AckRelease::QueuedCommand);
+}
+
+#[test]
+fn replayed_output_ack_keeps_the_read_timeout_before_the_provider_poll() {
+    acknowledged_prefix_releases_provider_output(AckRelease::Replayed);
+}
+
+#[test]
+fn welcome_command_output_polls_the_provider_without_waiting_out_the_read_timeout() {
+    acknowledged_prefix_releases_provider_output(AckRelease::WelcomeCommand);
+}
+
+#[test]
+fn restorable_provider_ack_keeps_the_read_timeout_before_the_provider_poll() {
+    acknowledged_prefix_releases_provider_output(AckRelease::NotLive);
+}
+
+#[derive(Clone, Copy, Debug, PartialEq)]
+enum AckRelease {
+    Idle,
+    QueuedCommand,
+    Replayed,
+    NotLive,
+    WelcomeCommand,
+}
+
+fn acknowledged_prefix_releases_provider_output(mode: AckRelease) {
+    struct PollingExecutor {
+        live: bool,
+        opened: bool,
+        emitted: bool,
+        control_frame_read_at: Option<Instant>,
+        poll_delay: Arc<std::sync::Mutex<Option<Duration>>>,
+        log: Arc<std::sync::Mutex<Vec<String>>>,
+    }
+
+    impl super::super::CommandExecutor for PollingExecutor {
+        fn execute(
+            &mut self,
+            command: &Command,
+        ) -> Result<super::super::CommandExecution, DurableRunnerError> {
+            self.log.lock().unwrap().push(command.command_type.clone());
+            if command.command_type != "session.open" {
+                return Ok(super::super::CommandExecution::result(
+                    json!({"status": "completed"}),
+                ));
+            }
+            self.opened = true;
+            Ok(super::super::CommandExecution {
+                result: json!({"status": "completed"}),
+                events: vec![(
+                    "provider.notice.recorded".to_owned(),
+                    super::super::EventPriority::P1,
+                    json!({"message": "accepted"}),
+                )],
+            })
+        }
+
+        fn poll_events(&mut self) -> Result<Vec<super::super::PolledEvent>, DurableRunnerError> {
+            if !self.opened || self.emitted {
+                return Ok(Vec::new());
+            }
+            self.emitted = true;
+            *self.poll_delay.lock().unwrap() =
+                self.control_frame_read_at.map(|read_at| read_at.elapsed());
+            self.log.lock().unwrap().push("poll".to_owned());
+            Ok(vec![super::super::PolledEvent {
+                executor_event_id: "provider_event_1".to_owned(),
+                event_type: "provider.notice.recorded".to_owned(),
+                priority: super::super::EventPriority::P1,
+                payload: json!({"message": "started"}),
+            }])
+        }
+
+        fn maintain_backpressured_provider(&mut self) -> Result<(), DurableRunnerError> {
+            self.control_frame_read_at = Some(Instant::now());
+            Ok(())
+        }
+
+        fn provider_poll_is_live(&self) -> bool {
+            self.live && self.opened
+        }
+    }
+
+    let command = |seq: u64, command_type: &str| {
+        json!({
+            "schema": "paperclip.prp.command.v1",
+            "commandId": format!("command_{seq}"),
+            "controllerSeq": seq,
+            "type": command_type,
+            "issuedAt": "2026-08-24T00:00:00.000Z",
+            "payload": {},
+        })
+    };
+    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+    let port = listener.local_addr().unwrap().port();
+    let mut config = config(port);
+    config.max_runtime = Duration::from_secs(5);
+    let directory = std::env::temp_dir().join(format!(
+        "paperclip-runner-ack-poll-{}-{mode:?}",
+        std::process::id()
+    ));
+    let _ = std::fs::remove_dir_all(&directory);
+    config.state_dir = directory.clone();
+    let state = test_state(&config);
+    let expires = current_unix_ms().unwrap() + 60_000;
+    let server_config = config.clone();
+    let server_state = state.clone();
+    let open_command = command(1, "session.open");
+    let snapshot_command = command(2, "session.snapshot");
+    let fast_poll = matches!(mode, AckRelease::Idle | AckRelease::WelcomeCommand);
+    let shutdown_seq = if fast_poll { 2 } else { 3 };
+    let shutdown_command = command(shutdown_seq, "runner.shutdown");
+    let server = thread::spawn(move || {
+        let (stream, _) = listener.accept().unwrap();
+        let mut socket = accept(stream).unwrap();
+        let mut secure = server_authenticate(
+            &mut socket,
+            &server_config,
+            &server_state,
+            ServerCredential {
+                token: "bootstrap-secret",
+                kind: "bootstrap",
+                lease_id: None,
+                expires_at_unix_ms: expires,
+                revocation_epoch: 0,
+            },
+        );
+        send_secure(
+            &mut socket,
+            &mut secure,
+            &server_config,
+            &welcome(
+                &server_state,
+                "connection_1",
+                Some("lease-secret"),
+                expires,
+                0,
+                if mode == AckRelease::WelcomeCommand {
+                    vec![open_command.clone()]
+                } else {
+                    Vec::new()
+                },
+            ),
+        );
+        if mode != AckRelease::WelcomeCommand {
+            send_secure(
+                &mut socket,
+                &mut secure,
+                &server_config,
+                &control(
+                    &server_state,
+                    "connection_1",
+                    "command",
+                    open_command.clone(),
+                ),
+            );
+        }
+        let open_result = receive_secure(&mut socket, &mut secure, &server_config);
+        assert_eq!(open_result["kind"], "command_result");
+        let open_event = receive_secure(&mut socket, &mut secure, &server_config);
+        assert_eq!(open_event["kind"], "event");
+        let mut connection = "connection_1";
+        if mode == AckRelease::Replayed {
+            drop(socket);
+            let (stream, _) = listener.accept().unwrap();
+            socket = accept(stream).unwrap();
+            secure = server_authenticate(
+                &mut socket,
+                &server_config,
+                &server_state,
+                ServerCredential {
+                    token: "lease-secret",
+                    kind: "lease",
+                    lease_id: Some("lease_1"),
+                    expires_at_unix_ms: expires,
+                    revocation_epoch: 1,
+                },
+            );
+            connection = "connection_2";
+            send_secure(
+                &mut socket,
+                &mut secure,
+                &server_config,
+                &welcome(
+                    &server_state,
+                    connection,
+                    None,
+                    expires,
+                    0,
+                    vec![open_command],
+                ),
+            );
+            assert_eq!(
+                receive_secure(&mut socket, &mut secure, &server_config),
+                open_result
+            );
+            assert_eq!(
+                receive_secure(&mut socket, &mut secure, &server_config),
+                open_event
+            );
+        }
+        let ack = control(
+            &server_state,
+            connection,
+            "ack",
+            json!({"ackedSourceSeq": 1}),
+        );
+        let snapshot = control(&server_state, connection, "command", snapshot_command);
+        match mode {
+            AckRelease::Idle | AckRelease::WelcomeCommand => {
+                send_secure(&mut socket, &mut secure, &server_config, &ack)
+            }
+            AckRelease::QueuedCommand => {
+                for value in [ack, snapshot] {
+                    let encrypted = secure
+                        .encrypt(&serde_json::to_vec(&value).unwrap(), true)
+                        .unwrap();
+                    socket
+                        .write(
+                            encode_plain_message(&encrypted, server_config.max_frame_bytes)
+                                .unwrap(),
+                        )
+                        .unwrap();
+                }
+                socket.flush().unwrap();
+            }
+            AckRelease::Replayed | AckRelease::NotLive => {
+                send_secure(&mut socket, &mut secure, &server_config, &ack);
+                thread::sleep(Duration::from_millis(50));
+                send_secure(&mut socket, &mut secure, &server_config, &snapshot);
+            }
+        }
+        if !fast_poll {
+            let result = receive_secure(&mut socket, &mut secure, &server_config);
+            assert_eq!(result["kind"], "command_result");
+            assert_eq!(result["payload"]["commandType"], "session.snapshot");
+        }
+        let polled = receive_secure(&mut socket, &mut secure, &server_config);
+        assert_eq!(polled["kind"], "event");
+        assert_eq!(polled["payload"]["payload"]["message"], "started");
+        send_secure(
+            &mut socket,
+            &mut secure,
+            &server_config,
+            &control(
+                &server_state,
+                connection,
+                "ack",
+                json!({"ackedSourceSeq": 2}),
+            ),
+        );
+        send_secure(
+            &mut socket,
+            &mut secure,
+            &server_config,
+            &control(&server_state, connection, "command", shutdown_command),
+        );
+        let shutdown_result = receive_secure(&mut socket, &mut secure, &server_config);
+        assert_eq!(shutdown_result["kind"], "command_result");
+        send_secure(
+            &mut socket,
+            &mut secure,
+            &server_config,
+            &control(
+                &server_state,
+                connection,
+                "command_result_ack",
+                json!({
+                    "commandId": format!("command_{shutdown_seq}"),
+                    "commandType": "runner.shutdown",
+                    "controllerSeq": shutdown_seq,
+                    "status": "completed",
+                }),
+            ),
+        );
+    });
+
+    let log = Arc::new(std::sync::Mutex::new(Vec::new()));
+    let poll_delay = Arc::new(std::sync::Mutex::new(None));
+    super::super::run_durable_runner(
+        config,
+        BootstrapTicket::new("bootstrap-secret".to_owned()).unwrap(),
+        PollingExecutor {
+            live: mode != AckRelease::NotLive,
+            opened: false,
+            emitted: false,
+            control_frame_read_at: None,
+            poll_delay: poll_delay.clone(),
+            log: log.clone(),
+        },
+    )
+    .unwrap();
+    server.join().unwrap();
+    let expected: &[&str] = if fast_poll {
+        &["session.open", "poll", "runner.shutdown"]
+    } else {
+        &[
+            "session.open",
+            "session.snapshot",
+            "poll",
+            "runner.shutdown",
+        ]
+    };
+    assert_eq!(*log.lock().unwrap(), expected);
+    if fast_poll {
+        let delay = poll_delay.lock().unwrap().unwrap();
+        assert!(
+            delay < RUNTIME_READ_TIMEOUT,
+            "the provider poll waited {delay:?} after its ACK was read"
+        );
+    }
+    std::fs::remove_dir_all(directory).unwrap();
+}
@@ -28,6 +28,7 @@ const SECURE_FRAME_SCHEMA: &str = "paperclip.runner.secure-frame.v1";
 const AUTH_TIMEOUT: Duration = Duration::from_secs(2);
 const WELCOME_TIMEOUT: Duration = CONNECT_TOTAL_TIMEOUT;
 const RUNTIME_READ_TIMEOUT: Duration = Duration::from_millis(250);
+const QUEUED_CONTROL_READ_TIMEOUT: Duration = Duration::from_millis(1);
 const CONNECT_ATTEMPT_TIMEOUT: Duration = Duration::from_secs(3);
 const CONNECT_TOTAL_TIMEOUT: Duration = Duration::from_secs(10);
 const MAX_CA_BUNDLE_BYTES: u64 = 4 * 1024 * 1024;
@@ -1163,6 +1164,19 @@ impl AuthenticatedTransport {
        self.receive_json_until(None)
    }

+    pub(crate) fn receive_queued_json(&mut self) -> Result<Option<Value>, DurableRunnerError> {
+        self.set_runtime_read_timeout(QUEUED_CONTROL_READ_TIMEOUT)?;
+        let message = self.receive_json_until(None);
+        self.set_runtime_read_timeout(RUNTIME_READ_TIMEOUT)?;
+        message
+    }
+
+    fn set_runtime_read_timeout(&mut self, timeout: Duration) -> Result<(), DurableRunnerError> {
+        self.socket
+            .set_read_timeout(Some(timeout))
+            .map_err(|error| DurableRunnerError::invalid(error.to_string()))
+    }
+
    fn receive_json_until(
        &mut self,
        deadline: Option<Instant>,
@@ -1749,6 +1763,7 @@ fn hex_nibble(byte: u8) -> Result<u8, DurableRunnerError> {
 mod tests {
    #[cfg(unix)]
    include!("tests/hosted_retention.rs");
+    include!("tests/ack_release_poll.rs");

    use std::net::{IpAddr, Ipv4Addr, TcpListener};
    use std::path::PathBuf;
@@ -95,6 +95,13 @@ impl CommandExecutor for SelectedExecutor {
        }
    }

+    fn provider_poll_is_live(&self) -> bool {
+        match self {
+            Self::LocalFacade(executor) => executor.provider_poll_is_live(),
+            _ => false,
+        }
+    }
+
    fn acknowledge_events(&mut self, count: usize) -> Result<(), DurableRunnerError> {
        match self {
            Self::LocalFacade(executor) => executor.acknowledge_events(count),
@@ -454,6 +461,12 @@ impl CommandExecutor for NativeProviderCommandExecutor {
            .map_or(Ok(None), CommandExecutor::maintain_disconnected_provider)
    }

+    fn provider_poll_is_live(&self) -> bool {
+        self.selected
+            .as_ref()
+            .is_some_and(CommandExecutor::provider_poll_is_live)
+    }
+
    fn maintain_backpressured_provider(&mut self) -> Result<(), DurableRunnerError> {
        // A provider that has not yet been selected/restored cannot have
        // in-process cleanup to advance. Never launch one merely for ACK debt.
@@ -5958,6 +5958,13 @@ impl CommandExecutor for CodexCommandExecutor {
        Ok(drained.next_drain_deadline)
    }

+    fn provider_poll_is_live(&self) -> bool {
+        self.provider.is_some()
+            && self.state.as_ref().is_some_and(|state| {
+                state.startup_attempt.is_none() && state.lifecycle != "reconciliation_required"
+            })
+    }
+
    fn maintain_backpressured_provider(&mut self) -> Result<(), DurableRunnerError> {
        if self.state.as_ref().is_some_and(|state| {
            state.receipt_limit_interrupt_pending && state.active_provider_turn_id.is_some()