Retain passive Codex output while Core reconnects (#19029)

## Thinking Path > - Paperclip manages work that AI agents execute through a Runner. > - Core controls the work while the Runner owns the provider process. > - Core can restart while an admitted provider turn is active. > - Reconnect waits used to leave provider output in the process pipe. > - Ordinary provider polling can also wait on a provider RPC before Core reconnects. > - This pull request drains passive local Codex output into bounded durable storage between reconnect attempts. > - Core receives the original events after recovery and keeps authority over tools and approvals. ## Linked Issues or Issue Description Refs #17594. Related work: merged #19353 supplies the Codex receipt-limit fence and deadline-aware reconnect scheduler. This PR now contains only shared passive intake and the already-live native Codex adapter. A controller outage must preserve admitted work and event identity without restoring a provider, sending a provider RPC, or executing a tool during reconnect. ## What Changed - Add a default no-op disconnected-poll hook. Managed providers and ACPX retain that default. - Drain passive frames from the existing local Codex process between reconnect attempts. Requests, terminal reconciliation and lineage checks wait for the connected path. - Persist each accepted event before acknowledging the provider queue. Soft-limit pressure retains the rejected suffix until Core ACKs free space. Hard failures stop the owned provider and preserve the durable prefix. - Preserve credential observations and the existing lease, runtime, grace and provider deadline bounds. Document the intake contract and test replay through the real PRP controller. ## Verification The slice was replayed onto master `bf10bc695a`. `git range-diff f453509e6f..20626230b9 bf10bc695a..HEAD` reports an unchanged patch. The diff has 11 files and contains no ACPX adapter or sidecar changes. Targeted local checks on this head: - `cargo test --manifest-path packages/paperclip-runner/runner/Cargo.toml --locked -p paperclip-runner-core --lib disconnected -- --test-threads=1`: 12 passed. These cover passive deferral, no restoration, the real managed HTTP worker, durable backpressure, oversized output and existing receipt-limit cleanup. - The 10 `credential_observer::tests::` checks passed. The Runner daemon and recovery example build, and workspace Rust formatting passes. - `vitest run src/control-plane/runner-recovery-fixture.test.ts --maxWorkers=1`: four passed. Both controller replacement modes retain the same process and replay event identities. The deadline and silent-handshake cases also pass. - `vitest run src/live/runnerd-codex-transport.test.ts -t late-remote-release-error --maxWorkers=1`: one passed. Whitespace and added-line secret checks pass. Earlier review found that disconnected ordinary polling could perform synchronous managed history reads. The real-worker regression retained here proves no history or stream-reopen request occurs until connected polling. Earlier slice preparation hit a sandbox loopback denial and a missing local daemon; the permitted loopback test and the test after building the daemon both passed. The current-base checks above run with those prerequisites in place. Historical CI on the old combined head failed an unordered AI-connections migration comparison. The faithful full-file local check passed 158 tests, and the failure remains recorded [here](https://github.com/paperclipai/paperclip/actions/runs/42112385671/job/127543554397). Earlier receipt-limit delivery failures are addressed by merged #19353. Previous review scores belong to the old combined diff. On current head `b92d4b604ee6992eeefbc02c8d41ec4f2a608a4d`, [CI run 42125840239](https://github.com/paperclipai/paperclip/actions/runs/42125840239) passed. All 59 checks completed with 55 successes and four intentional skips. Standard review 13 and Apex review 14 each scored 5/5 on this exact head, with zero comments and no unresolved threads. The single Apex request is [recorded here](https://github.com/paperclipai/paperclip/pull/19029#issuecomment-6671993072). ## Risks During an outage, native Codex intake uses 25 ms ticks between reconnect attempts. A provider request or terminal waits for Core. Saturation can pause intake or produce a failure. Managed remote streams retain their existing limits. Tests use controlled local providers; live provider qualification and deployment are separate. ## Model Used OpenAI GPT-6.1 Sol (`gpt-6.1-sol`), high reasoning effort, through Codex. Used source inspection, local native and PRP tests, and code review. ## 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 - [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 Co-authored-by: Paperclip <noreply@paperclip.ing>
2026-11-09 16:02:36 +01:00 · 2026-11-05 15:29:38 -08:00
1 parent 5ac05871b5
commit 08f3e7ebec
@@ -41,6 +41,12 @@ source sequence the runner has produced; acknowledged prefixes are removed
 atomically from durable state. After disconnect, every remaining event is sent
 again with the same identity and source sequence.

+After disconnect, runnerd tries Core when reconnect is due. Between attempts,
+live Codex processes can drain local output into the outbox. Work that
+needs a provider request waits for Core, including managed-provider history
+reads and unanswered tools or approvals. Runnerd suspends
+intake during terminal delivery or an authority transition.
+
 If Codex hits its tool-receipt limit during reconnect, runnerd stops its local
 process group at the original deadline. It reads the remaining stdout in short
 intervals, then replays the frames in order across as many batches as needed.
@@ -76,6 +82,11 @@ the P0 reserve. When the soft limit is reached, the runner enters backpressure
 and rejects the new non-P0 event rather than silently losing it. Exhausting the
 P0 reserve is an explicit unrecoverable condition.

+While disconnected, soft-limit backpressure retains the rejected event and its
+suffix at the provider owner until replay ACKs free outbox space. Oversized
+events or exhausted hard limits stop the owned provider and preserve the
+accepted durable prefix.
+
 ## Recovering Codex between turns

 Suppose run A finishes a turn, its Runner exits, and run B wants to continue
@@ -187,6 +187,10 @@ impl CommandExecutor for ScriptedExecutor {
        Ok(self.stop_at)
    }

+    fn poll_disconnected_events(&mut self) -> Result<Vec<PolledEvent>, DurableRunnerError> {
+        self.retained_events()
+    }
+
    fn retained_events(&mut self) -> Result<Vec<PolledEvent>, DurableRunnerError> {
        Ok(self
            .shared
@@ -2491,7 +2491,7 @@ fn extract_event_list(value: Value) -> Vec<Value> {
 }

 #[cfg(test)]
-mod tests {
+pub(crate) mod tests {
    use super::*;
    use std::io::{ErrorKind, Read, Write};
    use std::net::{TcpListener, TcpStream};
@@ -3000,6 +3000,33 @@ mod tests {
        .map_err(|error| error.cause)
    }

+    pub(crate) fn disconnected_executor_fixture(
+        config: ClaudeManagedProviderConfig,
+    ) -> (Box<dyn Provider>, impl Fn() -> Vec<String>) {
+        let service = FakeAnthropicService::start(Vec::new(), Vec::new());
+        let origin = service.origin.clone();
+        let provider = ClaudeManagedProvider::start_with_worker(
+            &config,
+            Vec::new(),
+            "test-scope",
+            None,
+            None,
+            0,
+            Vec::new(),
+            SensitiveApiKey("anthropic-test-secret".to_owned()),
+            move |key, beta| NetworkWorker::start_at(key, beta, &origin, false),
+        )
+        .map_err(|error| error.cause)
+        .unwrap();
+        (Box::new(provider), move || {
+            service
+                .captured()
+                .iter()
+                .map(|request| format!("{} {}", request.method, request.path))
+                .collect()
+        })
+    }
+
    #[test]
    fn tool_names_are_anthropic_safe_stable_and_collision_resistant() {
        let first = remote_tool_name("issues.comment:create");
@@ -2571,6 +2571,59 @@ impl CodexProvider {
    }

    pub fn poll(&mut self) -> Result<Option<CodexProviderEvent>, LocalRunnerError> {
+        self.poll_with_control(true)
+    }
+
+    pub(crate) fn poll_disconnected(
+        &mut self,
+    ) -> Result<Option<CodexProviderEvent>, LocalRunnerError> {
+        self.poll_with_control(false)
+    }
+
+    fn disconnected_message_is_passive(&self, message: &Value) -> bool {
+        let Some(method) = message.get("method").and_then(Value::as_str) else {
+            return false;
+        };
+        let params = message.get("params").unwrap_or(&Value::Null);
+        // Preserve frames that need a request, authority reconciliation or a
+        // terminal tail drain for the ordinary connected path.
+        if method == "turn/started"
+            && !notification_turn_id(params)
+                .is_some_and(|id| Some(id) == self.active_provider_turn_id.as_deref())
+        {
+            return false;
+        }
+        message.get("id").is_none()
+            && normalized_codex_terminal_event_type(method, params).is_none()
+            && params.pointer("/item/type").and_then(Value::as_str) != Some("collabAgentToolCall")
+            && matches!(
+                classify_notification_thread(
+                    method,
+                    &self.thread_id,
+                    &self.descendant_thread_ids,
+                    params
+                ),
+                Ok(NotificationThread::Root | NotificationThread::UnrelatedInformation)
+            )
+            && validate_notification_binding(
+                &self.thread_id,
+                self.active_provider_turn_id.as_deref(),
+                params,
+            )
+            .is_ok()
+    }
+
+    fn poll_with_control(
+        &mut self,
+        allow_control: bool,
+    ) -> Result<Option<CodexProviderEvent>, LocalRunnerError> {
+        if !allow_control
+            && (self.quarantined
+                || self.ambiguous_turn_start_pending
+                || self.process.try_wait()?.is_some())
+        {
+            return Ok(None);
+        }
        if self.process.stdout_failed() {
            return Err(LocalRunnerError::invalid(
                "Codex stdout closed without a valid reader EOF",
@@ -2676,6 +2729,22 @@ impl CodexProvider {
            })?;
            (message, trace_frame_id)
        };
+        if !allow_control && !self.disconnected_message_is_passive(&message) {
+            let bytes = json_size(&message, "buffered Codex provider message")?;
+            self.pending_message_bytes =
+                retain_buffered_message_bytes(self.pending_message_bytes, bytes).ok_or_else(
+                    || {
+                        LocalRunnerError::invalid(
+                            "Codex buffered messages exceed the 16 MiB aggregate limit",
+                        )
+                    },
+                )?;
+            self.pending_messages.push_front(BufferedProviderMessage {
+                value: message,
+                trace_frame_id,
+            });
+            return Ok(None);
+        }
        self.last_trace_frame_id = trace_frame_id;
        if let (Some(trace), Some(frame_id)) = (self.trace.as_mut(), trace_frame_id) {
            let method = message.get("method").and_then(Value::as_str);
@@ -4960,6 +5029,86 @@ done
        assert!(provider.completed_tool_call_ids.contains("denied-call"));
    }

+    #[cfg(unix)]
+    #[test]
+    fn disconnected_poll_retains_terminal_frame_and_exact_connected_replay() {
+        let mut provider = completion_tail_provider();
+        provider.active_provider_turn_id = Some("offline-turn".to_owned());
+        provider.completed_turn_authority = None;
+        let accepted_start = json!({"method":"turn/started","params":{
+            "threadId":"reader-tail-thread","turn":{"id":"offline-turn"}}});
+        let progress = json!({"method":"item/agentMessage/delta", "params":{
+            "threadId":"reader-tail-thread", "turnId":"offline-turn", "delta":"offline output"}});
+        let terminal = json!({"method":"turn/completed", "params":{
+            "threadId":"reader-tail-thread", "turn":{"id":"offline-turn","status":"completed"}}});
+        for (value, trace_frame_id) in [
+            (accepted_start, Some(9)),
+            (progress, Some(10)),
+            (terminal.clone(), Some(11)),
+        ] {
+            provider.pending_message_bytes += json_size(&value, "fixture").unwrap();
+            provider
+                .pending_messages
+                .push_back(BufferedProviderMessage {
+                    value,
+                    trace_frame_id,
+                });
+        }
+        assert!(matches!(provider.poll_disconnected().unwrap(),
+            Some(CodexProviderEvent::Notification { method, .. }) if method == "turn/started"));
+        assert_eq!(provider.take_provider_trace_frame_id(), Some(9));
+        assert!(
+            matches!(provider.poll_disconnected().unwrap(), Some(CodexProviderEvent::Notification { method, .. }) if method == "item/agentMessage/delta")
+        );
+        assert_eq!(provider.take_provider_trace_frame_id(), Some(10));
+        let bytes = provider.pending_message_bytes;
+        let started = std::time::Instant::now();
+        for _ in 0..5 {
+            assert!(provider.poll_disconnected().unwrap().is_none());
+            assert_eq!(provider.pending_messages.front().unwrap().value, terminal);
+            assert_eq!(provider.pending_message_bytes, bytes);
+            assert_eq!(provider.active_provider_turn_id(), Some("offline-turn"));
+        }
+        assert!(started.elapsed() < Duration::from_secs(1));
+        assert!(
+            matches!(provider.poll().unwrap(), Some(CodexProviderEvent::Notification { method, params }) if method == "turn/completed" && params == terminal["params"])
+        );
+        assert_eq!(provider.take_provider_trace_frame_id(), Some(11));
+        assert!(provider.pending_messages.is_empty());
+        assert_eq!(provider.pending_message_bytes, 0);
+        assert_eq!(provider.active_provider_turn_id(), None);
+        provider.shutdown().unwrap();
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn disconnected_poll_defers_requests_lineage_and_turn_reconciliation() {
+        let mut provider = completion_tail_provider();
+        provider.active_provider_turn_id = Some("offline-turn".to_owned());
+        for message in [
+            json!({"id":91,"method":"item/tool/call","params":{}}),
+            json!({"method":"item/agentMessage/delta","params":{"threadId":"unverified-child","delta":"child"}}),
+            json!({"method":"turn/started","params":{"threadId":"reader-tail-thread"}}),
+            json!({"method":"turn/started","params":{"threadId":"reader-tail-thread","turn":{"id":"next-turn"}}}),
+        ] {
+            provider.pending_message_bytes = json_size(&message, "fixture").unwrap();
+            provider
+                .pending_messages
+                .push_back(BufferedProviderMessage {
+                    value: message.clone(),
+                    trace_frame_id: Some(12),
+                });
+            let started = std::time::Instant::now();
+            assert!(provider.poll_disconnected().unwrap().is_none());
+            assert!(started.elapsed() < Duration::from_secs(1));
+            let retained = provider.pending_messages.pop_front().unwrap();
+            assert_eq!(retained.value, message);
+            assert_eq!(retained.trace_frame_id, Some(12));
+            assert_eq!(provider.active_provider_turn_id(), Some("offline-turn"));
+        }
+        provider.shutdown().unwrap();
+    }
+
    #[cfg(unix)]
    #[test]
    fn fenced_receipt_output_keeps_prefix_when_stdout_never_closes() {
@@ -329,6 +329,12 @@ pub trait CommandExecutor {
        Ok(Vec::new())
    }

+    /// Drain live local output between controller reconnect attempts. Providers
+    /// whose ordinary poll can start recovery or wait for RPCs must defer that work.
+    fn poll_disconnected_events(&mut self) -> Result<Vec<PolledEvent>, DurableRunnerError> {
+        Ok(Vec::new())
+    }
+
    /// Advance already-owned local cleanup without provider RPCs or ordinary intake.
    /// A pending deadline bounds both controller backoff and authentication.
    fn maintain_disconnected_provider(&mut self) -> Result<Option<Instant>, DurableRunnerError> {
@@ -559,6 +565,9 @@ pub fn run_durable_runner<E: CommandExecutor>(
            None
        };
        if Instant::now() < next_connection_attempt {
+            if authenticated_once {
+                poll_disconnected_executor_events(&mut state, &store, &config, &mut executor)?;
+            }
            let wake_at = next_connection_attempt.min(connection_attempt_deadline(
                &config,
                started,
@@ -570,7 +579,7 @@ pub fn run_durable_runner<E: CommandExecutor>(
            let wake_at = provider_deadline.map_or(wake_at, |due| due.min(wake_at));
            let remaining = wake_at.saturating_duration_since(Instant::now());
            if !remaining.is_zero() {
-                thread::sleep(remaining);
+                thread::sleep(remaining.min(Duration::from_millis(25)));
            }
            continue;
        }
@@ -1778,6 +1787,40 @@ fn stop_after_terminal_result_delivery_failure<E: CommandExecutor>(
    Err(error)
 }

+fn poll_disconnected_executor_events<E: CommandExecutor>(
+    state: &mut DurableState,
+    store: &DurableStateStore,
+    config: &DurableRunnerConfig,
+    executor: &mut E,
+) -> Result<(), DurableRunnerError> {
+    // Maintenance has already run. Leave the rejected suffix at the provider until
+    // a reconnect ACK frees space.
+    if state.warm_transition.is_some()
+        || state.pending_terminal_delivery.is_some()
+        || state.pending_provider_cleanup.is_some()
+        || state.backpressure
+    {
+        return Ok(());
+    }
+    let events = executor.poll_disconnected_events();
+    let polled = persist_polled_events(state, store, config, executor, events);
+    match polled {
+        Err(error)
+            if error.classification() == Some("outbox_backpressure")
+                && !state.outbox.is_empty() =>
+        {
+            store.save(state)
+        }
+        Err(error) => {
+            let _ = shutdown_preserving_cleanup(state, executor);
+            state.record_diagnostic(format!("disconnected provider ingestion failed: {error}"));
+            store.save(state)?;
+            Err(error)
+        }
+        Ok(()) => Ok(()),
+    }
+}
+
 fn poll_executor_events_after_controller_ack<E: CommandExecutor>(
    state: &mut DurableState,
    store: &DurableStateStore,
@@ -1838,7 +1881,18 @@ fn poll_executor_events<E: CommandExecutor>(
    if state.pending_provider_cleanup.is_some() {
        return Ok(());
    }
-    let events = match executor.poll_events() {
+    let events = executor.poll_events();
+    persist_polled_events(state, store, config, executor, events)
+}
+
+fn persist_polled_events<E: CommandExecutor>(
+    state: &mut DurableState,
+    store: &DurableStateStore,
+    config: &DurableRunnerConfig,
+    executor: &mut E,
+    events: Result<Vec<PolledEvent>, DurableRunnerError>,
+) -> Result<(), DurableRunnerError> {
+    let events = match events {
        Ok(events) => events,
        Err(error) => {
            drain_retained_events(state, store, config, executor).map_err(|secondary| {
@@ -2608,6 +2662,10 @@ mod tests {
            Ok(self.events.iter().cloned().collect())
        }

+        fn poll_disconnected_events(&mut self) -> Result<Vec<PolledEvent>, DurableRunnerError> {
+            self.poll_events()
+        }
+
        fn acknowledge_events(&mut self, count: usize) -> Result<(), DurableRunnerError> {
            self.acknowledgements.push(count);
            if self.fail_acknowledgement {
@@ -4139,6 +4197,91 @@ mod tests {
        fs::remove_dir_all(directory).unwrap();
    }

+    #[test]
+    fn disconnected_ingestion_keeps_durable_prefix_and_resumes_after_backpressure_ack() {
+        let directory = std::env::temp_dir().join(format!(
+            "paperclip-runner-offline-backpressure-{}",
+            std::process::id()
+        ));
+        let _ = fs::remove_dir_all(&directory);
+        let mut config = config(directory.clone());
+        config.max_outbox_bytes = 16 * 1024;
+        let store = DurableStateStore::new(&directory).unwrap();
+        let (mut state, _) = store.load_or_create(&config).unwrap();
+        let mut executor = RetainingEventExecutor {
+            events: (1..=128)
+                .map(|index| PolledEvent {
+                    executor_event_id: format!("offline-{index}"),
+                    event_type: "item.delta".to_owned(),
+                    priority: EventPriority::P1,
+                    payload: json!({"index":index,"text":"x".repeat(512)}),
+                })
+                .collect(),
+            fail_acknowledgement: false,
+            acknowledgements: Vec::new(),
+        };
+        poll_disconnected_executor_events(&mut state, &store, &config, &mut executor).unwrap();
+        assert!(state.backpressure);
+        assert!(!state.outbox.is_empty());
+        assert!(!executor.events.is_empty());
+        assert_eq!(state.outbox.len() + executor.events.len(), 128);
+        assert_eq!(state.last_controller_command_seq, 0);
+        let (reloaded, _) = store.load_or_create(&config).unwrap();
+        assert_eq!(reloaded.outbox, state.outbox);
+        assert!(reloaded.backpressure);
+        let prefix = state.highest_source_seq();
+        poll_disconnected_executor_events(&mut state, &store, &config, &mut executor).unwrap();
+        assert_eq!(state.highest_source_seq(), prefix);
+        while !executor.events.is_empty() {
+            state.apply_ack(state.highest_source_seq(), 2).unwrap();
+            poll_disconnected_executor_events(&mut state, &store, &config, &mut executor).unwrap();
+        }
+        assert_eq!(state.highest_source_seq(), 128);
+        assert_eq!(executor.acknowledgements.iter().sum::<usize>(), 128);
+        assert!(state.peak_outbox_bytes <= config.max_outbox_bytes);
+        fs::remove_dir_all(directory).unwrap();
+    }
+
+    #[test]
+    fn disconnected_ingestion_retires_provider_on_unrecoverable_output() {
+        struct OversizedExecutor {
+            shutdowns: usize,
+        }
+        impl CommandExecutor for OversizedExecutor {
+            fn execute(&mut self, _: &Command) -> Result<CommandExecution, DurableRunnerError> {
+                panic!("offline ingestion cannot execute commands");
+            }
+            fn poll_disconnected_events(&mut self) -> Result<Vec<PolledEvent>, DurableRunnerError> {
+                Ok(vec![PolledEvent {
+                    executor_event_id: "oversized".to_owned(),
+                    event_type: "item.delta".to_owned(),
+                    priority: EventPriority::P1,
+                    payload: json!({"text":"x".repeat(2048)}),
+                }])
+            }
+            fn shutdown(&mut self) -> Result<(), DurableRunnerError> {
+                self.shutdowns += 1;
+                Ok(())
+            }
+        }
+        let directory = std::env::temp_dir().join(format!(
+            "paperclip-runner-offline-output-failure-{}",
+            std::process::id()
+        ));
+        let _ = fs::remove_dir_all(&directory);
+        let mut config = config(directory.clone());
+        config.max_frame_bytes = 1024;
+        let store = DurableStateStore::new(&directory).unwrap();
+        let (mut state, _) = store.load_or_create(&config).unwrap();
+        let mut executor = OversizedExecutor { shutdowns: 0 };
+        let error = poll_disconnected_executor_events(&mut state, &store, &config, &mut executor)
+            .unwrap_err();
+        assert!(error.to_string().contains("transport frame limit"));
+        assert_eq!(executor.shutdowns, 1);
+        assert_eq!(state.highest_source_seq(), 0);
+        fs::remove_dir_all(directory).unwrap();
+    }
+
    #[test]
    fn provider_poll_yields_until_the_sent_controller_ack_prefix_is_consumed() {
        let directory = std::env::temp_dir().join(format!(
@@ -662,7 +662,8 @@ impl DurableState {
            self.record_diagnostic("outbox soft limit reached; non-P0 event rejected");
            return Err(DurableRunnerError::invalid(
                "outbox soft limit reached; reserved storage is available only to P0 events",
-            ));
+            )
+            .with_classification("outbox_backpressure"));
        }
        if projected > config.max_outbox_bytes {
            self.lifecycle = "unrecoverable".to_owned();
@@ -2698,6 +2698,70 @@ mod tests {
        })
    }

+    #[test]
+    fn disconnected_managed_executor_does_not_restore_or_reconcile_remote_streams() {
+        let directory =
+            std::env::temp_dir().join(format!("paperclip-managed-disconnected-{}", Uuid::new_v4()));
+        let config = test_config(&directory);
+        let mut executor = ManagedProviderCommandExecutor::with_runner_config(&directory, &config);
+        let mut payload = claude_prepare_payload();
+        payload["provider"]["runtimeContext"] = Value::Null;
+        executor
+            .execute(&test_command(1, "run.prepare", payload))
+            .unwrap();
+        let ManagedProviderDescriptor::ClaudeManaged(descriptor) =
+            executor.state.as_ref().unwrap().descriptor.clone()
+        else {
+            panic!("expected Claude managed descriptor");
+        };
+        let (provider, requests) =
+            crate::claude_managed_provider::tests::disconnected_executor_fixture(descriptor);
+        executor.provider = Some(provider);
+        executor.state.as_mut().unwrap().lifecycle = "session_open".to_owned();
+        executor.refresh_provider_checkpoint();
+        executor.save_state().unwrap();
+        let retained = fs::read(executor.state_path()).unwrap();
+        let before = requests();
+        assert_eq!(
+            before
+                .iter()
+                .filter(|path| path.contains("/events/stream"))
+                .count(),
+            1
+        );
+
+        for _ in 0..64 {
+            assert!(executor.poll_disconnected_events().unwrap().is_empty());
+        }
+        assert_eq!(requests(), before);
+        assert_eq!(fs::read(executor.state_path()).unwrap(), retained);
+
+        let mut replacement =
+            ManagedProviderCommandExecutor::with_runner_config(&directory, &config);
+        assert!(replacement.poll_disconnected_events().unwrap().is_empty());
+        assert!(!replacement.restore_checked);
+        assert_eq!(requests(), before);
+
+        // The real worker's closed stream needs history and a reopen on ordinary
+        // polling. This control proves disconnected polling left it untouched.
+        let deadline = std::time::Instant::now() + Duration::from_secs(2);
+        while requests().len() == before.len() && std::time::Instant::now() < deadline {
+            let _ = executor.poll_events();
+            std::thread::sleep(Duration::from_millis(1));
+        }
+        let after = requests();
+        assert!(after.iter().any(|path| path.contains("/events?")));
+        assert_eq!(
+            after
+                .iter()
+                .filter(|path| path.contains("/events/stream"))
+                .count(),
+            2
+        );
+        executor.shutdown().unwrap();
+        fs::remove_dir_all(directory).unwrap();
+    }
+
    #[test]
    fn attach_refreshes_tools_in_the_same_session_and_preserves_profile_fences() {
        let directory =
@@ -61,6 +61,15 @@ impl CommandExecutor for SelectedExecutor {
        }
    }

+    fn poll_disconnected_events(&mut self) -> Result<Vec<PolledEvent>, DurableRunnerError> {
+        match self {
+            Self::LocalFacade(executor) => executor.poll_disconnected_events(),
+            Self::Acpx(executor) => executor.poll_disconnected_events(),
+            Self::Managed(executor) => executor.poll_disconnected_events(),
+            Self::Dot(executor) => executor.poll_disconnected_events(),
+        }
+    }
+
    fn maintain_disconnected_provider(&mut self) -> Result<Option<Instant>, DurableRunnerError> {
        match self {
            Self::LocalFacade(executor) => executor.maintain_disconnected_provider(),
@@ -215,6 +224,35 @@ impl NativeProviderCommandExecutor {
        Ok(())
    }

+    fn poll_with_credentials(
+        &mut self,
+        poll: fn(&mut SelectedExecutor) -> Result<Vec<PolledEvent>, DurableRunnerError>,
+    ) -> Result<Vec<PolledEvent>, DurableRunnerError> {
+        self.credentials.poll(false)?;
+        if let Some(event) = self.credentials.pending() {
+            return Ok(vec![event]);
+        }
+        let events = self
+            .selected
+            .as_mut()
+            .map_or_else(|| Ok(Vec::new()), poll)?;
+        let terminal = events.iter().any(|event| {
+            matches!(
+                event.event_type.as_str(),
+                "turn.completed"
+                    | "turn.failed"
+                    | "turn.interrupted"
+                    | "turn.cancelled"
+                    | "harness.exited"
+            )
+        });
+        self.credentials.poll(terminal)?;
+        Ok(self
+            .credentials
+            .pending()
+            .map_or(events, |event| vec![event]))
+    }
+
    fn select_from_payload(&mut self, payload: &Value) -> Result<(), DurableRunnerError> {
        let kind = payload
            .pointer("/provider/kind")
@@ -375,29 +413,11 @@ impl CommandExecutor for NativeProviderCommandExecutor {

    fn poll_events(&mut self) -> Result<Vec<PolledEvent>, DurableRunnerError> {
        self.select_recovery()?;
-        self.credentials.poll(false)?;
-        if let Some(event) = self.credentials.pending() {
-            return Ok(vec![event]);
-        }
-        let events = self
-            .selected
-            .as_mut()
-            .map_or_else(|| Ok(Vec::new()), CommandExecutor::poll_events)?;
-        let terminal = events.iter().any(|event| {
-            matches!(
-                event.event_type.as_str(),
-                "turn.completed"
-                    | "turn.failed"
-                    | "turn.interrupted"
-                    | "turn.cancelled"
-                    | "harness.exited"
-            )
-        });
-        self.credentials.poll(terminal)?;
-        Ok(self
-            .credentials
-            .pending()
-            .map_or(events, |event| vec![event]))
+        self.poll_with_credentials(CommandExecutor::poll_events)
+    }
+
+    fn poll_disconnected_events(&mut self) -> Result<Vec<PolledEvent>, DurableRunnerError> {
+        self.poll_with_credentials(CommandExecutor::poll_disconnected_events)
    }

    fn maintain_disconnected_provider(&mut self) -> Result<Option<Instant>, DurableRunnerError> {
@@ -4806,6 +4806,13 @@ impl CodexCommandExecutor {
    }

    fn poll_current_provider(&mut self) -> Result<(), DurableRunnerError> {
+        self.poll_current_provider_with_control(true)
+    }
+
+    fn poll_current_provider_with_control(
+        &mut self,
+        allow_control: bool,
+    ) -> Result<(), DurableRunnerError> {
        // Receipt-limit interruption is autonomous recovery. It must advance
        // even while older durable events await acknowledgement, otherwise a
        // slow or disconnected controller can keep an exhausted provider turn
@@ -4845,14 +4852,18 @@ impl CodexCommandExecutor {
            return self.settle_receipt_limit_interrupt_if_deadline_elapsed();
        }
        for _ in 0..MAX_EVENTS_PER_POLL {
-            let event = self
+            let provider = self
                .provider
                .as_mut()
-                .expect("provider remains present while polling")
-                .poll()
-                .map_err(|error| {
-                    DurableRunnerError::invalid(format!("Codex provider failed: {error}"))
-                })?;
+                .expect("provider remains present while polling");
+            let event = (if allow_control {
+                provider.poll()
+            } else {
+                provider.poll_disconnected()
+            })
+            .map_err(|error| {
+                DurableRunnerError::invalid(format!("Codex provider failed: {error}"))
+            })?;
            let Some(event) = event else { break };
            let trace_frame_id = self
                .provider
@@ -5513,6 +5524,21 @@ impl CommandExecutor for CodexCommandExecutor {
        self.retained_events()
    }

+    fn poll_disconnected_events(&mut self) -> Result<Vec<PolledEvent>, DurableRunnerError> {
+        // Only an already-admitted local Codex process can supply passive
+        // output. Recovery, facades and cleanup retain their connected path.
+        if self.provider.is_some()
+            && self.state.as_ref().is_some_and(|state| {
+                state.config.provider == "codex"
+                    && state.startup_attempt.is_none()
+                    && !state.receipt_limit_interrupt_pending
+            })
+        {
+            self.poll_current_provider_with_control(false)?;
+        }
+        self.retained_events()
+    }
+
    fn retained_events(&mut self) -> Result<Vec<PolledEvent>, DurableRunnerError> {
        if let Some(error) = &self.startup_evidence_error {
            return Err(error.clone());
@@ -6008,6 +6034,123 @@ mod tests {
        );
    }

+    #[cfg(unix)]
+    #[test]
+    fn disconnected_poll_never_restores_an_unopened_facade() {
+        let directory = std::env::temp_dir().join(format!(
+            "paperclip-offline-no-restore-{}",
+            uuid::Uuid::new_v4()
+        ));
+        fs::create_dir_all(&directory).unwrap();
+        fs::write(
+            directory.join(CODEX_PROVIDER_STATE_FILE),
+            b"not a provider state",
+        )
+        .unwrap();
+        let mut executor = CodexCommandExecutor::new(&directory);
+        assert!(executor.poll_disconnected_events().unwrap().is_empty());
+        assert!(executor.maintain_disconnected_provider().unwrap().is_none());
+        assert!(!executor.restore_checked);
+        executor.state = Some(opencode_result_state());
+        assert!(executor.poll_disconnected_events().unwrap().is_empty());
+        assert!(executor.maintain_disconnected_provider().unwrap().is_none());
+        assert!(!executor.restore_checked);
+        fs::remove_dir_all(directory).unwrap();
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn disconnected_codex_output_is_durable_and_terminal_control_waits_for_reconnect() {
+        let directory =
+            std::env::temp_dir().join(format!("paperclip-offline-codex-{}", uuid::Uuid::new_v4()));
+        fs::create_dir_all(&directory).unwrap();
+        let goal_called = directory.join("goal-called");
+        let script = format!(
+            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"}}}}}}'
+      printf '%s\n' '{{"method":"item/agentMessage/delta","params":{{"threadId":"thread-1","turnId":"turn-1","delta":"retained offline output"}}}}'
+      printf '%s\n' '{{"method":"turn/completed","params":{{"threadId":"thread-1","turn":{{"id":"turn-1","status":"completed"}}}}}}' ;;
+    *'"method":"thread/goal/get"'*)
+      touch '{}'
+      printf '%s\n' '{{"id":4,"result":{{"goal":null}}}}' ;;
+  esac
+done
+"#,
+            goal_called.display()
+        );
+        let mut state = opencode_result_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.goal_capability = Some(SessionGoalCapability::codex_available());
+        let mut provider = CodexProvider::start(&state.config, None).unwrap();
+        provider.start_turn("Test", &state.config.cwd).unwrap();
+        state.provider_process_generation = provider.process_generation();
+        let mut executor = CodexCommandExecutor::new(&directory);
+        executor.state = Some(state);
+        executor.provider = Some(provider);
+        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
+        let mut observed = Vec::new();
+        loop {
+            assert!(std::time::Instant::now() < deadline);
+            let events = executor.poll_disconnected_events().unwrap();
+            if events.is_empty() {
+                continue;
+            }
+            let saved: CodexProviderState =
+                serde_json::from_slice(&fs::read(executor.state_path()).unwrap()).unwrap();
+            assert_eq!(saved.pending_events, events);
+            let count = events.len();
+            observed.extend(events);
+            executor.acknowledge_events(count).unwrap();
+            if observed
+                .iter()
+                .any(|event| event.payload["text"] == "retained offline output")
+            {
+                break;
+            }
+        }
+        let start = observed
+            .iter()
+            .position(|event| event.event_type == "turn.started")
+            .unwrap();
+        let text = observed
+            .iter()
+            .position(|event| event.payload["text"] == "retained offline output")
+            .unwrap();
+        assert!(start < text);
+        let started = std::time::Instant::now();
+        for _ in 0..5 {
+            assert!(executor.poll_disconnected_events().unwrap().is_empty());
+        }
+        assert!(started.elapsed() < std::time::Duration::from_secs(1));
+        assert!(!goal_called.exists());
+        assert_eq!(
+            executor
+                .state
+                .as_ref()
+                .unwrap()
+                .active_provider_turn_id
+                .as_deref(),
+            Some("turn-1")
+        );
+        let events = executor.poll_events().unwrap();
+        assert!(goal_called.exists());
+        assert!(events
+            .iter()
+            .any(|event| event.event_type == "turn.completed"));
+        executor.provider.as_mut().unwrap().shutdown().unwrap();
+        fs::remove_dir_all(directory).unwrap();
+    }
+
    fn opencode_result_state() -> CodexProviderState {
        let mut state = CodexProviderState::new(
            CodexProviderConfig {
@@ -99,12 +99,17 @@ it.each([false, true])("preserves useful in-process work through repeated replac
    expect(run.receipt().completedSteps).toBe(0);
    await run.stopController();
    run.release();
-    await vi.waitFor(() => expect(run.receipt()).toMatchObject({ phase: "backpressured", completedSteps: 8, retainedEvents: 8 }), { timeout: 5000 });
+    await vi.waitFor(() => expect(run.receipt()).toMatchObject({ phase: "awaiting_controller", completedSteps: 32, retainedEvents: 0, acknowledgedEvents: 32 }), { timeout: 10_000 });
    expect(run.receipt().pid).toBe(run.child.pid);
    expect(run.child.exitCode).toBeNull();
    expect(run.delivered).toHaveLength(0);
+    const offline = JSON.parse(readFileSync(resolve(run.root, "runner/runner-state.json"), "utf8"));
+    expect(offline.ackedSourceSeq).toBe(0);
+    expect(offline.outbox).toHaveLength(32);
+    expect(offline.peakOutboxBytes).toBeLessThanOrEqual(offline.maxOutboxBytes);
    await run.replaceController();
    await vi.waitFor(() => expect(run.delivered).toHaveLength(32), { timeout: 10_000 });
+    expect(run.delivered.map(event => event.sourceEventId)).toEqual(offline.outbox.map((event: any) => event.envelope.payload.sourceEventId));
    expect(run.receipt()).toMatchObject({ phase: "awaiting_controller", completedSteps: 32 });
    await run.stopController();
    expect(run.receipt().completedSteps).toBe(32);
@@ -125,7 +130,7 @@ it.each([false, true])("preserves useful in-process work through repeated replac
    await run.command("runner.shutdown", "shutdown");
    expect((await run.exit)[0]).toBe(0);
  } finally { await run.cleanup(); }
-}, 35_000);
+}, 60_000);

 it("stops after the original reconnect deadline while retaining recovery state", async () => {
  const run = await fixture(true, 1500, undefined, 5000);
@@ -3900,7 +3900,7 @@ it.each(["admission-only", "admission-only-mismatched-model", "complete", "api-k
    } else {
      await expect(probe).rejects.toThrow(mode === "missing-runtime" ? /ENOENT|unavailable|cleanup/i : mode === "admission-only-mismatched-model" || mode === "mismatched-model" || mode === "remote-mismatched-model" || mode === "model-refresh-error" || mode === "model-cleanup-error" || mode === "missing-model" ? /observed model|mismatched provider model/i : /timed out|cleanup|suspension/i);
      if (mode === "late-remote-release-error") {
-        expect(registrationReleased).toBe(true); expect(cleanupConfirmed).toBe(false);
+        await vi.waitFor(() => expect(registrationReleased).toBe(true)); expect(cleanupConfirmed).toBe(false);
        expect((await stat(runtimeDirectory)).isDirectory()).toBe(true);
      }
      if (mode === "model-refresh-error") {