Reconcile qualified ACPX predecessors through ordinary recovery (#19153)

## Thinking Path > - Paperclip runs agent work through a controller and a Runner. > - A completed provider session can outlive an abrupt Runner exit. > - The next owner must reconcile the retained source before it attaches a new run. > - The ordinary recovery foundation in #19132 first supports Codex App Server. > - Qualified ACPX providers need their own saved and live consistency checks. > - This pull request adds those checks and a separate provider attestation. > - The benefit is session continuity through existing recovery and attachment. ## Linked Issues or Issue Description Builds on #19132, which has merged. This PR targets `master` and contains only the ACPX changes. **What existing behavior does this improve?** Recovery of a completed ACPX session after the Runner and provider stop. **Current behavior** Ordinary ACPX recovery and attachment preserve a qualified session. A completed predecessor also needs trusted admission by its new owner before restoration. Codex checks cannot attest ACPX state. **Proposed behavior** Reuse ordinary exact-source recovery. Compare live restored state with saved dormant state when no uncertain effect explains new work. Require the selected driver's separate ACPX prelaunch attestation. Settle retained events before authenticated target attachment. ## What Changed - Preserve baseline prepared cleanup and ordinary continuation. New predecessor admission excludes prepared provider state; snapshots and cleanup do not restart it. - Observe existing recovery command facts. Keep ordinary active same-run behavior and existing profile, workspace, model, permission, session and catalog checks. - Verify the restored qualified session's identity, turn, goal and outstanding requests when saved state was dormant. Failed verification reaps the owned process group. - Validate scoped goal updates arriving during the final status snapshot after the authoritative goal reply. Active or unknown updates fail admission; earlier and benign queued events remain available to ordinary polling. - Recognize the existing inactive `limited`, `usage_limited` and `budget_limited` goal statuses, so an unexpected active goal after restore cannot bypass live verification. Treat malformed nested working-state metadata as unproven while keeping missing-field compatibility. - Keep existing ordinary recovery of qualified session-open and suspended ACPX sessions. The new integrity check does not expand prepared restoration. - Advertise `acpxRecoveryIntegrityVersion: 1`, separate from Codex integrity. No native source mode or CLI remains. - Require `acpxSourceRecoveryVersion: 1` from the selected trusted execution driver for a stopped predecessor. Codex capability alone cannot grant it. - Reuse dynamic authenticated owner admission, retained source accounting, and existing ordinary ACPX attachment. ## Verification Current head `b294d8ae88d0` is one commit on `master` at `e80d745060fd`, the merge of #19132. The ten-file diff is byte-for-byte identical to the reviewed ACPX diff against its former base. The guide shows where `paperclip-runnerd --build-metadata` emits `acpxRecoveryIntegrityVersion: 1` and how host admission uses a separate capability. [ACPX recovery guide](https://github.com/paperclipai/paperclip/blob/b294d8ae88d019ef02a942e240bc5c14287239e0/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md#recovering-acpx-between-turns) The rebased head passed all 14 ACPX process cases, three native snapshot checks, Runner typecheck and the binary metadata check locally. GitHub CI, Docker and Windows passed on their first attempts. Standard Greptile review 12 scored the rebased commit 5/5 with zero unresolved threads. [CI run](https://github.com/paperclipai/paperclip/actions/runs/42063188178) · [Standard review](https://github.com/paperclipai/paperclip/pull/19153#issuecomment-6646809075) The preceding code commit `08df48de20d613efcd861c278a7f9929c21d6670` includes the reviewed parent `c09ebb4a2a6c140521698cf9fd681de67cfb9745`. Paid review 9 scores this exact head 5/5 with no unresolved findings (summary updated 2026-11-05 09:13:11 UTC after the 09:07:15 UTC request). Standard review 8 also scored this head 5/5. [PR workflow 41995103666](https://github.com/paperclipai/paperclip/actions/runs/41995103666), [Runner Windows 41995103419](https://github.com/paperclipai/paperclip/actions/runs/41995103419), and [Docker Runner 41995103366](https://github.com/paperclipai/paperclip/actions/runs/41995103366) passed on this exact head. The initial main attempt lost self-hosted runners: the Runner package log records a shutdown signal, and the other affected jobs report lost communication or cancellation. Only unsuccessful jobs were retried; their successful retry and aggregate gate are green. No source, assertion or timeout change was made for this interruption. - Combined `runnerd-acpx-source-reconciliation.test.ts` and `runnerd-source-reconciliation.test.ts`: 25 real native/fake-provider process cases passed. ACPX contributes 14 cases, including the final-snapshot goal race, process-group cleanup, same-PID adoption, exact provider identities, retained accounting, limited/paused goals and ordinary active same-run recovery. - The maintained fake sidecar emits a real `runtime.goal` frame between the dormant goal reply and final snapshot response. Before the fix, recovery succeeded. Afterward it refuses admission, reaps the owned sidecar/provider group, and records only the original provider turn. - Three native wire tests passed. They cover active updates with low/equal/missing revisions, foreign run/session identity, and unchanged ordered delivery of earlier history plus benign new events. The queue is inspected without consuming frames. - Both parent Codex final-read tests passed in the combined checkout: active-goal refusal/reap and safe paused-goal delivery. - The real native backend completes A, settles its journals, kills Runner and provider, recovers the same ACPX session, adopts the replacement Runner after controller restart, attaches B and completes a second turn, then uses ordinary warm attachment for C and completes a third successful turn. Record, backend and agent session identities remain exact; ordinary ACPX attachment may restart its sidecar. - Provider compatibility checks also passed while preparing this stack: 30 permission ordering/forwarding cases, 19 factory cases, the native inactive-goal classification unit, and six native warm/suspend integration cases. - Combined offline Runner/fake-provider build, Runner `tsc --noEmit`, `cargo fmt --all -- --check`, parent-relative `git diff --check`, and added-line secret/personal-data scan passed. - Only focused local checks ran. Full suites and platform checks run in CI. ## Risks - Fake providers prove protocol behavior, not vendor restore inertness or all vendor background work. Real provider qualification remains required. - The trusted process owner must reject unsafe completed predecessors before launch. Ordinary same-run active-goal recovery remains supported. - Probes use the existing bounded sidecar budget. A timeout can terminate the sidecar through existing transport cleanup. Null goal projections remain supported. - Ordinary ACPX attachment can restart the sidecar. Record, backend and agent session identities remain exact; the Runner PID remains the same. - This draft adds no deployment, release pin change or parking hook. ## Model Used OpenAI GPT-6.1 Sol (`gpt-6.1-sol`), high reasoning, through Codex with local code and command tools. ## Checklist - [x] I have included a thinking path and specified the model - [x] I have checked ROADMAP.md and confirmed this does not duplicate planned work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have not included internal or instance-local references - [x] My branch name describes the change - [x] Focused local tests pass - [x] Tests and documentation cover the changes - [x] Risks are documented - [x] All Paperclip CI gates are green at the revised head - [x] Greptile is 5/5 with no unresolved comments at the revised head - [x] I will address review findings before requesting merge Co-authored-by: Paperclip <noreply@paperclip.ing>
2026-11-08 12:33:05 +01:00 · 2026-11-05 08:37:08 -08:00
1 parent b0fa2ef9bf
commit 3d0d178ff7
@@ -107,3 +107,16 @@ Cancellation cleanup only settles retained work.

 Live qualification still needs to verify that restoring a provider starts no
 work, deployed storage honors fsync, and the old process has exited.
+
+## Recovering ACPX between turns
+
+`paperclip-runnerd --build-metadata` prints `acpxRecoveryIntegrityVersion: 1`
+when the Runner implements these ACPX checks. The hosted ROM build copies
+that field into its manifest. The host driver advertises a separate
+`acpxSourceRecoveryVersion: 1` and checks whether this run qualifies.
+
+The Runner compares the restored session identity, pending requests, turn
+and goal with A’s saved idle state. After reading the goal, it checks updates
+that arrive during the final status read and keeps them queued for delivery.
+As with Codex, a prepared session can continue or be cleaned up. This handoff
+requires a completed turn. Suspended sessions use their usual recovery path.
@@ -17,8 +17,9 @@ use sha2::{Digest, Sha256};

 use crate::acpx_provider_capabilities::{run_attachment_policy, RunAttachmentPolicy};
 use crate::acpx_provider_session::{
-    AcpxPermissionMode, AcpxProviderRuntimePolicy, AcpxProviderSession, AcpxProviderSessionConfig,
-    AcpxProviderSessionIdentity, AcpxTurnControlCapabilities, PiThinkingLevel,
+    goal_is_dormant, AcpxPermissionMode, AcpxProviderRuntimePolicy, AcpxProviderSession,
+    AcpxProviderSessionConfig, AcpxProviderSessionIdentity, AcpxTurnControlCapabilities,
+    PiThinkingLevel,
 };
 use crate::acpx_sidecar_transport::AcpxSidecarTransportConfig;
 #[cfg(test)]
@@ -885,6 +886,7 @@ pub struct AcpxCommandExecutor {
    #[cfg(test)]
    fail_after_state_rename: bool,
    launch_profile: Option<AcpxLaunchProfile>,
+    recovery_pending_effects: Option<usize>,
 }

 // The durable command's turn and the live provider callback have separate IDs.
@@ -923,6 +925,7 @@ impl AcpxCommandExecutor {
            state: None,
            session: None,
            restore_checked: false,
+            recovery_pending_effects: None,
            restore_error: None,
            persistence_error: None,
            #[cfg(test)]
@@ -935,6 +938,10 @@ impl AcpxCommandExecutor {
        self.state_dir.join(ACPX_PROVIDER_STATE_FILE)
    }

+    pub(crate) fn observe_recovery_state(&mut self, state: &crate::durable::DurableState) {
+        self.recovery_pending_effects = Some(state.recovery_pending_effects());
+    }
+
    fn candidate_admission(
        &self,
    ) -> Option<&crate::acpx_candidate_admission::AcpxCandidateAdmission> {
@@ -1032,6 +1039,11 @@ impl AcpxCommandExecutor {
    }

    fn restore_once(&mut self) -> Result<(), DurableRunnerError> {
+        self.load_state_without_session()?;
+        self.restore_session_if_needed()
+    }
+
+    fn load_state_without_session(&mut self) -> Result<(), DurableRunnerError> {
        let path = self.state_path();
        let mut file = match open_private_regular_file(&path) {
            Ok(file) => file,
@@ -1078,7 +1090,7 @@ impl AcpxCommandExecutor {
        }
        state.validate(&self.context, &launch_profile_digest)?;
        self.state = Some(state);
-        self.restore_session_if_needed()
+        Ok(())
    }

    fn restore_session_if_needed(&mut self) -> Result<(), DurableRunnerError> {
@@ -1154,6 +1166,7 @@ impl AcpxCommandExecutor {
            priority: EventPriority::P0,
            payload: session_event_payload(&state.descriptor, &identity, process_id, turn_controls),
        })?;
+        self.recovery_pending_effects = None;
        self.session = Some(session);
        self.save_state()
    }
@@ -1179,6 +1192,21 @@ impl AcpxCommandExecutor {
                "ACPX provider identity did not attest the qualified command digest",
            ));
        }
+        let expected_dormant = recovering
+            && self.recovery_pending_effects == Some(0)
+            && self.candidate_admission().is_none()
+            && acpx_release_profile(&state.descriptor.agent)
+                .is_some_and(|profile| !profile.qualification_pending)
+            && matches!(state.lifecycle.as_str(), "session_open" | "suspended")
+            && state.active_turn_id.is_none()
+            && !state.provider_exit_unconfirmed
+            && state.pending_runtime_requests.is_empty()
+            && (state.goal_projection.is_null() || goal_is_dormant(&state.goal_projection));
+        if expected_dormant {
+            session
+                .verify_dormant_session()
+                .map_err(|error| DurableRunnerError::invalid(error.to_string()))?;
+        }
        Ok(session)
    }

@@ -2513,6 +2541,7 @@ impl CommandExecutor for AcpxCommandExecutor {
    }

    fn rotate_authority(&mut self, config: &DurableRunnerConfig) {
+        self.recovery_pending_effects = None;
        self.context.run_id = config.run_id.clone();
        self.context.normalized_session_id = config.normalized_session_id.clone();
        self.context.turn_id = config.turn_id.clone();
@@ -4959,6 +4988,9 @@ mod tests {
        let config = test_config(&directory, Some(launch_profile));
        let mut executor = AcpxCommandExecutor::with_runner_config(&directory, &config);
        executor.state = Some(state);
+        executor.recovery_pending_effects = Some(0);
+        executor.restore_session_if_needed().unwrap();
+        assert!(executor.session.is_none());
        let open_error = executor.open_session().unwrap_err();
        assert!(open_error
            .to_string()
@@ -353,6 +353,26 @@ pub struct AcpxProviderSession {
    runtime_retired: bool,
 }

+pub(crate) fn goal_is_dormant(projection: &Value) -> bool {
+    projection["schema"] == "paperclip.session_goal.snapshot.v1"
+        && projection["workingNow"].as_bool() == Some(false)
+        && projection
+            .pointer("/goal/workingNow")
+            .is_none_or(|working| working.as_bool() == Some(false))
+        && (projection.get("goal") == Some(&Value::Null)
+            || matches!(
+                projection.pointer("/goal/status").and_then(Value::as_str),
+                Some(
+                    "paused"
+                        | "blocked"
+                        | "limited"
+                        | "usage_limited"
+                        | "budget_limited"
+                        | "complete"
+                )
+            ))
+}
+
 impl AcpxProviderSession {
    pub fn start(config: &AcpxProviderSessionConfig) -> Result<Self, LocalRunnerError> {
        config.validate()?;
@@ -410,6 +430,56 @@ impl AcpxProviderSession {
        &self.state
    }

+    pub(crate) fn verify_dormant_session(&mut self) -> Result<(), LocalRunnerError> {
+        if self.state.active_turn_id().is_some() || self.state.has_pending_requests() {
+            return Err(LocalRunnerError::invalid(
+                "saved dormant ACPX resumed pending work",
+            ));
+        }
+        let goal = self.goal_control(GeneratedAcpxSidecarCommand::SessionGoalGet, json!({}))?;
+        let prior_events = self.transport.buffered_events().len();
+        if !goal_is_dormant(&goal) {
+            return Err(LocalRunnerError::invalid(
+                "saved dormant ACPX resumed an active or unknown goal",
+            ));
+        }
+        let snapshot = self
+            .transport
+            .request(GeneratedAcpxSidecarCommand::SessionSnapshot, json!({}))?;
+        let identity: AcpxProviderSessionIdentity =
+            serde_json::from_value(snapshot["identity"].clone()).map_err(|_| {
+                LocalRunnerError::invalid("saved dormant ACPX snapshot identity is invalid")
+            })?;
+        if identity != self.identity
+            || snapshot["runId"].as_str() != Some(self.config.run_id.as_str())
+            || snapshot.get("turnId") != Some(&Value::Null)
+            || snapshot["pendingToolCount"].as_u64() != Some(0)
+            || snapshot["pendingInputCount"].as_u64() != Some(0)
+            || !snapshot["pendingRuntimeRequests"]
+                .as_array()
+                .is_some_and(Vec::is_empty)
+        {
+            return Err(LocalRunnerError::invalid(
+                "saved dormant ACPX snapshot has unexpected work or identity",
+            ));
+        }
+        // The status barrier can queue a newer goal while the snapshot is in flight.
+        // Inspect without consuming facts that ordinary polling still owns.
+        let mut checked_state = self.state.clone();
+        for event in self.transport.buffered_events().skip(prior_events) {
+            if event.event_type != GeneratedAcpxSidecarEventType::RuntimeGoal {
+                continue;
+            }
+            checked_state.accept_event(event)?;
+            if !goal_is_dormant(&event.payload) {
+                return Err(LocalRunnerError::invalid(
+                    "saved dormant ACPX received an active or unknown goal during its snapshot",
+                ));
+            }
+        }
+        Ok(())
+    }
+
    /// Read the same live sidecar before exposing its in-memory pending ledger.
    /// A durable file alone cannot authorize a request after process loss.
    pub fn verify_live_request_snapshot(
@@ -1718,6 +1788,54 @@ fn with_cleanup_error(

 #[cfg(test)]
 mod tests {
+    #[test]
+    fn dormant_goals_accept_known_inactive_statuses_and_reject_active_or_unknown() {
+        use super::*;
+        for (status, expected) in [
+            ("paused", true),
+            ("blocked", true),
+            ("limited", true),
+            ("usage_limited", true),
+            ("budget_limited", true),
+            ("complete", true),
+            ("active", false),
+            ("unknown", false),
+        ] {
+            let mut projection = json!({
+                "schema":"paperclip.session_goal.snapshot.v1", "workingNow":false,
+                "goal":{"status":status,"workingNow":false}
+            });
+            assert_eq!(goal_is_dormant(&projection), expected, "{status}");
+            projection["goal"]
+                .as_object_mut()
+                .unwrap()
+                .remove("workingNow");
+            assert_eq!(
+                goal_is_dormant(&projection),
+                expected,
+                "missing workingNow: {status}"
+            );
+            for working in [
+                json!(true),
+                Value::Null,
+                json!("false"),
+                json!("true"),
+                json!(0),
+                json!({}),
+                json!([]),
+            ] {
+                projection["goal"]["workingNow"] = working;
+                assert!(
+                    !goal_is_dormant(&projection),
+                    "{status} has invalid workingNow"
+                );
+            }
+            projection["goal"]["workingNow"] = json!(false);
+            projection["workingNow"] = json!(true);
+            assert!(!goal_is_dormant(&projection), "{status} session is working");
+        }
+    }
+
    #[test]
    fn startup_cleanup_failure_keeps_the_validated_provider_cause() {
        use super::*;
@@ -1939,6 +2057,180 @@ mod mode_tests {
            provider_lifetime_fence_candidates: [60_001, 60_002, 60_003],
        }
    }
+
+    #[cfg(unix)]
+    fn snapshot_wire_session(
+        prior_events: &[crate::acpx_sidecar_transport::AcpxSidecarEvent],
+        new_events: &[crate::acpx_sidecar_transport::AcpxSidecarEvent],
+        foreign_session: bool,
+    ) -> AcpxProviderSession {
+        let mut cfg = config();
+        let identity = identity();
+        let mut snapshot_identity = identity.clone();
+        if foreign_session {
+            snapshot_identity.acpx_record_id = "foreign-record".to_owned();
+        }
+        let reply = |id, result| {
+            json!({"protocolVersion":GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION,
+            "id":id,"ok":true,"result":result})
+        };
+        let print = |value: Value| {
+            format!(
+                "printf '%s\\n' '{}'\n",
+                value.to_string().replace('\'', "'\"'\"'")
+            )
+        };
+        let mut script = String::new();
+        for (id, events, result) in [
+            (
+                1,
+                prior_events,
+                json!({"schema":"paperclip.session_goal.snapshot.v1","providerRevision":2,"goal":null,"workingNow":false}),
+            ),
+            (
+                2,
+                new_events,
+                json!({"identity":snapshot_identity,"runId":"run-1","turnId":null,
+                "pendingToolCount":0,"pendingInputCount":0,"pendingRuntimeRequests":[]}),
+            ),
+            (3, &[][..], json!({"closed":true})),
+        ] {
+            script.push_str("IFS= read -r request_line\n");
+            for event in events {
+                let event_type = match event.event_type {
+                    GeneratedAcpxSidecarEventType::RuntimeGoal => "runtime.goal",
+                    GeneratedAcpxSidecarEventType::RuntimeDiagnostic => "runtime.diagnostic",
+                    _ => panic!("unsupported fixture event"),
+                };
+                script.push_str(&print(
+                    json!({"protocolVersion":GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION,
+                    "sequence":event.sequence,"eventType":event_type,"runId":event.run_id,
+                    "turnId":event.turn_id,"payload":event.payload}),
+                ));
+            }
+            script.push_str(&print(reply(id, result)));
+        }
+        cfg.transport.command = PathBuf::from("/bin/sh");
+        cfg.transport.args = vec!["-c".to_owned(), script];
+        let transport = AcpxSidecarTransport::start_for_agent(&cfg.transport, &cfg.agent).unwrap();
+        AcpxProviderSession {
+            transport,
+            state: AcpxProviderState::new(&cfg.run_id).unwrap(),
+            identity,
+            tool_bridge: ProviderToolBridge::default(),
+            reserved_tool_bridge: ProviderToolBridge::default(),
+            turn_controls: AcpxTurnControlCapabilities::default(),
+            catalog_revision: cfg.catalog_revision,
+            working_directory: cfg.working_directory.clone(),
+            config: cfg,
+            closed: false,
+            failure: None,
+            transport_terminated: false,
+            runtime_retired: false,
+        }
+    }
+
+    #[cfg(unix)]
+    fn wire_goal(
+        sequence: u64,
+        revision: Option<u64>,
+        status: &str,
+    ) -> crate::acpx_sidecar_transport::AcpxSidecarEvent {
+        crate::acpx_sidecar_transport::AcpxSidecarEvent {
+            sequence,
+            event_type: GeneratedAcpxSidecarEventType::RuntimeGoal,
+            run_id: Some("run-1".to_owned()),
+            turn_id: None,
+            payload: json!({"schema":"paperclip.session_goal.snapshot.v1","providerRevision":revision,
+                "goal":{"status":status,"workingNow":false},"workingNow":false}),
+        }
+    }
+
+    #[test]
+    #[cfg(unix)]
+    fn dormant_snapshot_rejects_new_active_goals_regardless_of_revision() {
+        for (revision, status) in [
+            (Some(3), "active"),
+            (Some(2), "active"),
+            (Some(1), "active"),
+            (None, "active"),
+            (Some(3), "unknown"),
+        ] {
+            let events = vec![wire_goal(1, revision, status)];
+            let mut session = snapshot_wire_session(&[], &events, false);
+            let error = session.verify_dormant_session().unwrap_err();
+            assert!(error.to_string().contains("during its snapshot"), "{error}");
+            assert_eq!(
+                session
+                    .transport
+                    .buffered_events()
+                    .cloned()
+                    .collect::<Vec<_>>(),
+                events
+            );
+            session.shutdown("test complete").unwrap();
+        }
+    }
+
+    #[test]
+    #[cfg(unix)]
+    fn dormant_snapshot_preserves_prior_and_benign_new_events_for_polling() {
+        let events = vec![
+            crate::acpx_sidecar_transport::AcpxSidecarEvent {
+                sequence: 1,
+                event_type: GeneratedAcpxSidecarEventType::RuntimeDiagnostic,
+                run_id: Some("run-1".to_owned()),
+                turn_id: None,
+                payload: json!({"code":"probe-observation","message":"Retained observation"}),
+            },
+            wire_goal(2, Some(1), "active"),
+            wire_goal(3, Some(3), "paused"),
+        ];
+        let mut session = snapshot_wire_session(&events[..2], &events[2..], false);
+        session.verify_dormant_session().unwrap();
+        assert_eq!(
+            session
+                .transport
+                .buffered_events()
+                .cloned()
+                .collect::<Vec<_>>(),
+            events
+        );
+        for expected in &events {
+            let delivered = session.poll_event(Duration::ZERO).unwrap().unwrap();
+            if expected.event_type == GeneratedAcpxSidecarEventType::RuntimeGoal {
+                assert!(
+                    matches!(&delivered[..], [AcpxProviderStateEvent::Goal(goal)] if goal == &expected.payload)
+                );
+            } else {
+                assert_eq!(delivered.len(), 1);
+            }
+        }
+        assert!(session.poll_event(Duration::ZERO).unwrap().is_none());
+        session.shutdown("test complete").unwrap();
+    }
+
+    #[test]
+    #[cfg(unix)]
+    fn dormant_snapshot_requires_the_same_run_and_session() {
+        for foreign_session in [false, true] {
+            let mut event = wire_goal(1, Some(3), "paused");
+            if !foreign_session {
+                event.run_id = Some("foreign-run".to_owned());
+            }
+            let mut session = snapshot_wire_session(&[], &[event], foreign_session);
+            let error = session.verify_dormant_session().unwrap_err();
+            assert!(
+                error.to_string().contains(if foreign_session {
+                    "identity"
+                } else {
+                    "stale run"
+                }),
+                "{error}"
+            );
+            session.shutdown("test complete").unwrap();
+        }
+    }
    #[test]
    fn pending_profile_cannot_launch_even_with_explicit_policy() {
        let mut config = config();
@@ -402,6 +402,10 @@ impl AcpxSidecarTransport {
        self.process.id()
    }

+    pub(crate) fn buffered_events(&self) -> impl ExactSizeIterator<Item = &AcpxSidecarEvent> {
+        self.buffered_events.iter()
+    }
+
    pub fn request(
        &mut self,
        command: GeneratedAcpxSidecarCommand,
@@ -66,6 +66,29 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
    let mut goal = Value::Null;
    let mut provider_lost = false;
    let mut session_identity = Value::Null;
+    let mut source_run = String::new();
+    let mut source_pending_completion: Option<String> = None;
+    let mut source_goal_revision = 0_u64;
+    let mut source_journal = if mode == "source-reconcile" {
+        let path = args
+            .windows(2)
+            .find(|pair| pair[0] == "--journal")
+            .ok_or("source journal is missing")?;
+        let mut file = std::fs::OpenOptions::new()
+            .create(true)
+            .append(true)
+            .open(&path[1])?;
+        write_json(&mut file, &json!({"processStart": std::process::id()}))?;
+        if args.iter().any(|arg| arg == "--spawn-owned-child") {
+            let child = std::process::Command::new("/bin/sleep")
+                .arg("120")
+                .spawn()?;
+            write_json(&mut file, &json!({"childPid": child.id()}))?;
+        }
+        Some(file)
+    } else {
+        None
+    };
    for line in stdin.lock().lines() {
        let request: Value = serde_json::from_str(&line?)?;
        let id = request
@@ -76,6 +99,115 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
            .get("command")
            .and_then(Value::as_str)
            .ok_or("request command is missing")?;
+        if let Some(journal) = source_journal.as_mut() {
+            write_json(journal, &request)?;
+            if command == "run.attach" {
+                source_run = request["params"]["runId"].as_str().unwrap().to_owned();
+            }
+            let fault = args
+                .windows(2)
+                .find(|pair| pair[0] == "--source-fault")
+                .and_then(|pair| std::fs::read_to_string(&pair[1]).ok())
+                .unwrap_or_default();
+            let mut response = bootstrap_success(
+                id,
+                command,
+                &request,
+                mode,
+                profile_digest,
+                lifetime_fence_candidates,
+                &mut session_identity,
+            );
+            if command == "session.snapshot" {
+                response["result"] = json!({"identity":session_identity, "runId":source_run,
+                    "turnId":if fault == "active" {json!("unexpected-turn")} else {json!(source_pending_completion)},
+                    "pendingToolCount":if fault == "tool" || source_pending_completion.is_some() {1} else {0}, "pendingInputCount":0,
+                    "pendingRuntimeRequests":[]});
+            }
+            if command == "session.goal.get" {
+                source_goal_revision += 1;
+                response["result"]["providerRevision"] = json!(source_goal_revision);
+            }
+            if command == "session.snapshot" && fault == "goal-during-snapshot" {
+                source_goal_revision += 1;
+                let frame = json!({
+                    "protocolVersion":GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION,
+                    "sequence":next_sequence,"eventType":"runtime.goal","runId":source_run,"turnId":null,
+                    "payload":{"schema":"paperclip.session_goal.snapshot.v1","providerRevision":source_goal_revision,
+                        "goal":{"objective":"Changed during final status","status":"active","workingNow":false},"workingNow":false}
+                });
+                write_json(journal, &json!({"emitted":frame}))?;
+                write_json(&mut stdout, &frame)?;
+                next_sequence += 1;
+            }
+            if command == "session.goal.get" && matches!(fault.as_str(), "goal" | "paused") {
+                response["result"]["goal"] = json!({"objective":"Continue independently", "status":if fault == "goal" {"active"} else {"paused"},
+                    "tokenBudget":null,"tokensUsed":null,"elapsedSeconds":null,"iterations":null,
+                    "lastReason":null,"createdAt":null,"updatedAt":null,"completedAt":null,"workingNow":false});
+                response["result"]["sessionGoals"] = json!({"availability":"available", "actions":["set","pause","resume","clear"],
+                    "autonomousUpdates":true,"persistentAcrossResume":true,"maxObjectiveChars":4000,"tokenBudgetControl":false,"usageReporting":false});
+            }
+            write_json(&mut stdout, &response)?;
+            if command == "turn.start" {
+                let turn = request["params"]["turnId"].as_str().unwrap();
+                if args.iter().any(|arg| arg == "--wait-completion-feedback") {
+                    let report = json!({"schema":"paperclip.run_result.v1", "reportedWorkDisposition":"done", "summary":"Completed fake ACPX work",
+                    "completionClaim":{"contractRevision":"1","objectiveSatisfied":true,
+                        "criteria":[{"criterionId":"objective","status":"satisfied","evidenceRefs":[]}],"remainingWork":[]},
+                    "evidence":[],"verification":[],"attentionRequests":[],"artifacts":[]});
+                    let call_id = format!("finish-{turn}");
+                    write_turn_event(
+                        &mut stdout,
+                        next_sequence,
+                        "runtime.tool_called",
+                        &source_run,
+                        turn,
+                        json!({"callId":call_id,"operationId":"paperclip_finish","input":report}),
+                    )?;
+                    next_sequence += 1;
+                    source_pending_completion = Some(turn.to_owned());
+                    continue;
+                }
+                write_turn_event(
+                    &mut stdout,
+                    next_sequence,
+                    "runtime.event",
+                    &source_run,
+                    turn,
+                    json!({"type":"text_delta", "text":"Completed fake ACPX work."}),
+                )?;
+                next_sequence += 1;
+                write_turn_event(
+                    &mut stdout,
+                    next_sequence,
+                    "runtime.turn_terminal",
+                    &source_run,
+                    turn,
+                    json!({"status":"completed"}),
+                )?;
+                next_sequence += 1;
+            }
+            if command == "tool.resolve" {
+                let turn = source_pending_completion
+                    .take()
+                    .ok_or("missing source completion")?;
+                if request["params"]["turnId"] != turn
+                    || request["params"]["callId"] != format!("finish-{turn}")
+                {
+                    return Err("source completion identity mismatch".into());
+                }
+                write_turn_event(
+                    &mut stdout,
+                    next_sequence,
+                    "runtime.turn_terminal",
+                    &source_run,
+                    &turn,
+                    json!({"status":if request["params"]["error"].is_null() {"completed"} else {"failed"}}),
+                )?;
+                next_sequence += 1;
+            }
+            continue;
+        }
        if mode == "turns-provider-death" && provider_lost && command == "session.goal.get" {
            write_json(
                &mut stdout,
@@ -122,6 +122,7 @@ fn build_metadata() -> serde_json::Value {
        "packageVersion": env!("CARGO_PKG_VERSION"),
        "binaryContractVersion": 2,
        "recoveryIntegrityVersion": 1,
+        "acpxRecoveryIntegrityVersion": 1,
        "durableSessionCapabilities": ["unlimited_runtime", "connection_lease_renewal"],
        "externalProviderCapabilities": ["openai_dot_mcp", "grok_bot_mcp", "muse_external_v1"],
        "nativeExecutionVersion": 1,
@@ -674,6 +675,8 @@ mod tests {
        let metadata = build_metadata();
        assert_eq!(metadata["recoveryIntegrityVersion"], 1);
        assert!(metadata.get("sourceRecoveryVersion").is_none());
+        assert_eq!(metadata["acpxRecoveryIntegrityVersion"], 1);
+        assert!(metadata.get("acpxSourceRecoveryVersion").is_none());
        assert_eq!(metadata["schema"], RUNNERD_BUILD_METADATA_SCHEMA);
        assert_eq!(metadata["binaryContractVersion"], 2);
        assert_eq!(
@@ -286,8 +286,10 @@ impl CommandExecutor for NativeProviderCommandExecutor {
            return Ok(());
        }
        self.select_recovery()?;
-        if let Some(SelectedExecutor::LocalFacade(executor)) = self.selected.as_mut() {
-            executor.observe_recovery_state(state);
+        match self.selected.as_mut() {
+            Some(SelectedExecutor::LocalFacade(executor)) => executor.observe_recovery_state(state),
+            Some(SelectedExecutor::Acpx(executor)) => executor.observe_recovery_state(state),
+            _ => {}
        }
        Ok(())
    }
@@ -11,6 +11,8 @@ import type { DurableRecoveryIdentity } from "../contracts/durable-recovery.js";
 export interface RunnerExecutionDriver {
  /** Attests stopped ownership and dormant prelaunch admission for the selected provider. */
  sourceRecoveryVersion?: 1;
+  /** Same trusted prelaunch attestation for a qualified ACPX provider. */
+  acpxSourceRecoveryVersion?: 1;
  /** Current authenticated lease/checkpoint-writer owner; may change during warm reuse. */
  authorityRunId?: string;
  /** Own provider configuration and file preparation in the execution environment. */
@@ -0,0 +1,272 @@
+import { createHash } from "node:crypto";
+import { execFileSync, spawn } from "node:child_process";
+import { mkdir, mkdtemp, readFile, realpath, rm, writeFile } from "node:fs/promises";
+import { appendFileSync, closeSync, openSync, readFileSync } from "node:fs";
+import { createServer } from "node:http";
+import { tmpdir } from "node:os";
+import { join, resolve } from "node:path";
+import { expect, it, vi } from "vitest";
+import type { DurablePrpControlPlane } from "../control-plane/durable-prp-control-plane.js";
+import { QUALIFIED_ACPX_PROFILES, resolveQualifiedAcpxProfile } from "../drivers/acpx/qualified-profiles.js";
+import { createCapabilityRunnerdCodexTransport, readRunnerdArtifactBinding, type CapabilityRunnerdCodexTransportOptions } from "./runnerd-codex-transport.js";
+import { createRunnerProviderLaunch, runnerLaunchRequirementsDigest, type RunnerProviderLaunchRequirements } from "./runner-provider-launch.js";
+import { defaultCapabilityRunnerdBinary } from "../../test/support/runnerd-transport.js";
+import type { PrpEvent } from "../protocol/replay-contract.js";
+import { createNativeSessionBackend } from "../backends/native-backend-factory.js";
+import type { NativeExecutionInput } from "../contracts/native-execution.js";
+import type { PersistedNativeSession } from "../contracts/native-session-backend.js";
+
+type Bundle = ReturnType<typeof createCapabilityRunnerdCodexTransport>;
+const read = async (path: string) => JSON.parse(await readFile(path, "utf8"));
+const sha = (value: string | Buffer) => `sha256:${createHash("sha256").update(value).digest("hex")}`;
+const quote = (value: string) => `'${value.replaceAll("'", "'\"'\"'")}'`;
+const kill = (pid: number) => { try { process.kill(-pid, "SIGKILL"); } catch (error) { if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error; } };
+const absent = (pid: number) => { try { process.kill(pid, 0); return false; } catch (error) { if ((error as NodeJS.ErrnoException).code === "ESRCH") return true; throw error; } };
+const stopped = (pid: number) => {
+  try { process.kill(pid, 0); } catch (error) { if ((error as NodeJS.ErrnoException).code === "ESRCH") return true; throw error; }
+  try { return execFileSync("ps", ["-p", String(pid), "-o", "stat="], { encoding: "utf8" }).trim().startsWith("Z"); }
+  catch (error) { if ((error as { status?: number }).status === 1) return true; throw error; }
+};
+
+async function fixture(agent: "codex" | "claude" = "codex", nativeSource = false) {
+  const root = await realpath(await mkdtemp(join(tmpdir(), "runner-acpx-source-")));
+  const journal = join(root, "calls"), fault = join(root, "fault"), wrapper = join(root, "sidecar"), script = join(root, "sidecar-entry");
+  const binary = defaultCapabilityRunnerdBinary(), sidecar = resolve(import.meta.dirname, "../../runner/target/debug/fake-acpx-sidecar");
+  await mkdir(join(root, "acpx"), { mode: 0o700 }); await mkdir(join(root, "diagnostics"), { mode: 0o700 });
+  await writeFile(script, "qualified fixture entry");
+  await writeFile(wrapper, `#!/bin/sh\nexec ${quote(sidecar)} --mode source-reconcile --profile-digest ${quote(QUALIFIED_ACPX_PROFILES[agent].commandDigest)} --journal ${quote(journal)} --source-fault ${quote(fault)} --spawn-owned-child${nativeSource ? " --wait-completion-feedback" : ""}\n`, { mode: 0o700 });
+  let authority: DurablePrpControlPlane | null = null;
+  let registrationGeneration = 0;
+  const server = createServer();
+  server.on("upgrade", (request, socket, head) => authority ? authority.handleUpgrade(request, socket, "/runner", head) : socket.destroy());
+  await new Promise<void>(resolve => server.listen(0, "127.0.0.1", resolve));
+  const address = server.address(); if (!address || typeof address === "string") throw new Error("missing fixture listener");
+  const identity = { runnerInstanceId: "runner-source", environmentLeaseId: "lease-source", runId: "run-source", normalizedSessionId: "session-source", turnId: "turn-source", itemId: "item-source" };
+  const nativeIdentity = { companyId: "company", agentId: "agent", issueId: "issue", runId: identity.runId, sessionId: identity.normalizedSessionId };
+  const execution: NativeExecutionInput = { schema: "paperclip.native-execution-input.v1",
+    binding: { companyId: nativeIdentity.companyId, agentId: nativeIdentity.agentId, issueId: nativeIdentity.issueId, runId: identity.runId, executionWorkspaceId: "workspace" },
+    task: { identifier: "TASK-1", title: "Complete fake work", description: null, prompt: "Complete fake work", workMode: "standard" },
+    workspace: { cwd: root, repoUrl: null, repoRef: null, branchName: null },
+    session: { normalizedSessionId: identity.normalizedSessionId, driverKind: "acpx_runtime", protocolVersion: 1 },
+    provider: { kind: "acpx", agent, model: "explicit-source-model", permissionPolicy: "interactive", profile: resolveQualifiedAcpxProfile(agent, "explicit-source-model") },
+    completionContract: { id: "contract", sha256: sha("contract"), schemaVersion: "paperclip.completion-contract.v1",
+      contract: { revision: "1", objective: "Complete fake work", criteria: [{ id: "objective", requirement: "Complete fake work" }] } }, interactionResponses: [], credentialBindings: [] };
+  const bundles: Bundle[] = [], runnerPids = new Set<number>();
+  let requirements: RunnerProviderLaunchRequirements | undefined;
+  let recovering = false;
+  const options: CapabilityRunnerdCodexTransportOptions = {
+    provider: "acpx", acpxAgent: agent, runnerBinary: binary, stateDirectory: root, prpIdentity: identity,
+    runnerArtifact: readRunnerdArtifactBinding(binary), sourceRecoveryVersion: 1, acpxSourceRecoveryVersion: 1,
+    lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 }, runnerProcessReconnectGraceMs: null, runnerReconnectGraceMs: 3_000,
+    controlPlaneRegistration: async next => {
+      const generation = ++registrationGeneration; authority = next;
+      return { connectUrl: `ws://127.0.0.1:${address.port}/runner`, release: () => { if (generation === registrationGeneration) authority = null; } };
+    },
+    prepareLaunch: async input => {
+      requirements ??= structuredClone(input);
+      return { configurationId: `${recovering ? "source" : "initial"}-${input.identity.runId}`, requirementsDigest: runnerLaunchRequirementsDigest(input),
+        provider: createRunnerProviderLaunch(input, { options: {}, root, cwd: root, codexHome: root, codexCommand: "", providerNodeCommand: wrapper, opencodeProxyPath: "", acpxSidecarPath: script }) };
+    },
+    runnerProcessLauncher: spec => {
+      const stderr = openSync(join(root, "native.stderr"), "a", 0o600);
+      const child = spawn(binary, [...spec.args, "--acpx-launch-authority-digest", sha("fixture authority"),
+        "--acpx-sidecar-command", wrapper, "--acpx-sidecar-command-sha256", sha(execFileSync("cat", [wrapper])),
+        "--acpx-sidecar-script", script, "--acpx-sidecar-script-sha256", sha("qualified fixture entry")],
+        { env: { PATH: process.env.PATH, HOME: root, ...spec.environment }, stdio: ["ignore", "ignore", stderr], detached: true });
+      closeSync(stderr);
+      if (child.pid) runnerPids.add(child.pid);
+      return { child, completion: new Promise((resolve, reject) => { child.once("error", reject); child.once("exit", (code, signal) => {
+        try { appendFileSync(join(root, "native.stderr"), readFileSync(join(root, "diagnostics", "runnerd.stderr.log"))); } catch {}
+        resolve({ code, signal, stdout: "", stderr: "" });
+      }); }) };
+    },
+  };
+  const runnerFile = join(root, "runner", "runner-state.json"), providerFile = join(root, "runner", "acpx-provider-state.json"), coreFile = join(root, "control-plane", "control-plane-state.json");
+  const calls = async () => (await readFile(journal, "utf8")).trim().split("\n").map(line => JSON.parse(line));
+  const count = async (command: string) => (await calls()).filter(entry => command === "processStart" ? entry.processStart : entry.command === command).length;
+  const settle = async () => vi.waitFor(async () => {
+    const runner = await read(runnerFile), provider = await read(providerFile), core = await read(coreFile);
+    expect(runner.outbox).toEqual([]); expect(runner.ackedSourceSeq).toBe(runner.nextSourceSeq - 1);
+    expect(provider.activeTurnId).toBeNull(); expect(provider.pendingEvents).toEqual([]); expect(provider.pendingRuntimeRequests).toEqual({});
+    expect(core.commands.some((command: { status: string }) => command.status === "pending")).toBe(false);
+    expect(authority?.semanticToolResultsSettled()).toBe(true);
+  }, { timeout: 5_000 });
+  const complete = async (bundle: Bundle) => {
+    await bundle.transport.request("turn/start", { input: [{ type: "text", text: "Complete the fake ACPX turn." }] });
+    for await (const event of bundle.transport.notifications()) if (event.method === "turn/completed") return;
+    throw new Error("missing completed turn");
+  };
+  const cleanup = async () => {
+    for (const bundle of bundles.reverse()) await bundle.detachControllerForRestart().catch(() => {});
+    for (const pid of runnerPids) kill(pid);
+    for (const entry of await calls().catch(() => [])) if (entry.processStart) kill(entry.processStart);
+    server.closeAllConnections(); await new Promise<void>(resolve => server.close(() => resolve()));
+    await rm(root, { recursive: true, force: true });
+  };
+  try {
+    const first = createCapabilityRunnerdCodexTransport(options); bundles.push(first);
+    let nativeSnapshot: PersistedNativeSession | undefined;
+    let thread: { id: string; sessionId: string };
+    if (nativeSource) {
+      const session = await createNativeSessionBackend(execution, { codexTransportFactory: () => first.transport,
+        environment: { PAPERCLIP_WORKSPACE_CWD: root }, dynamicTools: [] }).openSession({ identity: nativeIdentity, workingDirectory: root });
+      await session.startTurn({ message: { role: "user", text: "Complete before abrupt loss." } });
+      for await (const event of session.events()) if (event.eventType === "turn.completed") break;
+      nativeSnapshot = await session.snapshot();
+      expect(nativeSnapshot.terminal?.runTerminalState).toBe("succeeded");
+      expect(nativeSnapshot.semanticResult?.reportedWorkDisposition).toBe("done");
+      thread = { id: nativeSnapshot.sessionId, sessionId: nativeSnapshot.providerSessionId! };
+    } else {
+      const opened = await first.transport.request("thread/start", { cwd: root, model: "explicit-source-model", dynamicTools: [] });
+      thread = { id: (opened.thread as { id: string }).id, sessionId: (opened.thread as { sessionId: string }).sessionId };
+      await complete(first);
+    }
+    await settle();
+    expect((await read(coreFile)).committedEvents.some((event: { eventType: string; envelope: { payload: { payload: { runTerminalState?: string } } } }) => event.eventType === "run.terminal" && event.envelope.payload.payload.runTerminalState === "succeeded")).toBe(true);
+    const providerIdentity = (await read(providerFile)).identity;
+    const pid = first.evidence().runnerPid!;
+    await first.detachControllerForRestart(); kill(pid);
+    for (const entry of await calls()) if (entry.processStart) kill(entry.processStart);
+    await vi.waitFor(async () => { expect(stopped(pid)).toBe(true); for (const entry of await calls()) if (entry.processStart || entry.childPid) expect(stopped(entry.processStart ?? entry.childPid)).toBe(true); });
+    const recover = (overrides: Partial<CapabilityRunnerdCodexTransportOptions> = {}) => {
+      recovering = true;
+      const bundle = createCapabilityRunnerdCodexTransport({ ...options, recoveryLaunch: { requirements: requirements! },
+        resumeProviderSession: { driverSessionId: thread.id, providerSessionId: thread.sessionId, providerIdentity }, onRecoveryAccountingEvent: async () => {}, ...overrides, get authorityRunId() { return overrides.authorityRunId ?? "run-target"; } });
+      bundles.push(bundle); return bundle;
+    };
+    return { root, fault, runnerFile, providerFile, coreFile, identity, providerIdentity, thread, calls, count, settle, complete, recover, cleanup, nativeIdentity, nativeSnapshot, execution };
+  } catch (error) { await cleanup(); throw error; }
+}
+
+it.skipIf(process.platform === "win32")("recovers and adopts ACPX through the real native backend before attachment", async () => {
+  const f = await fixture("claude", true);
+  try {
+    let owner = "run-target";
+    let source = f.recover({ get authorityRunId() { return owner; } });
+    const backend = () => createNativeSessionBackend(f.execution, { codexTransportFactory: () => source.transport,
+      environment: { PAPERCLIP_WORKSPACE_CWD: f.root }, dynamicTools: [] });
+    let recovered = await backend().recoverSession!(f.nativeSnapshot!, { signal: new AbortController().signal });
+    expect(recovered.recovered).toBe(true);
+    const pid = source.evidence().runnerPid, startedAt = source.transport.processInfo().startedAt;
+    expect(await f.count("turn.start")).toBe(1); expect(await f.count("processStart")).toBe(2);
+    await recovered.session!.detachControllerForRestart!();
+    const launch = vi.fn(() => { throw new Error("duplicate source launch"); });
+    source = f.recover({ get authorityRunId() { return owner; }, runnerProcessLauncher: launch, authorizeRunnerAdoption: async () => { throw new Error("old source configuration"); },
+      adoptExistingRunner: { pid, processGroupId: pid, startedAt, isAlive: () => !stopped(pid!), signal: () => false } });
+    recovered = await backend().recoverSession!(f.nativeSnapshot!, { signal: new AbortController().signal });
+    expect(recovered.recovered).toBe(true); expect(launch).not.toHaveBeenCalled(); expect(source.evidence().runnerPid).toBe(pid);
+    await expect(source.transport.request("turn/start", {})).rejects.toThrow("source_attachment_pending");
+    for (const runId of ["run-target", "run-warm-third"]) {
+      owner = runId;
+      await recovered.session!.attachRun!({ identity: { ...f.nativeIdentity, runId } });
+      expect(source.evidence().runnerPid).toBe(pid);
+      await recovered.session!.startTurn({ message: { role: "user", text: "Continue after authenticated attachment." } });
+      let completed = false;
+      for await (const event of recovered.session!.events()) {
+        if (event.eventType === "turn.completed" && event.runId === runId) { completed = true; break; }
+      }
+      expect(completed).toBe(true); await f.settle();
+      const snapshot = await recovered.session!.snapshot();
+      expect(snapshot.sessionId).toBe(f.thread.id); expect(snapshot.providerIdentity).toEqual(f.providerIdentity);
+      expect(snapshot.semanticResult?.reportedWorkDisposition).toBe("done"); expect(snapshot.terminal).toMatchObject({ runTerminalState: "succeeded" });
+    }
+    expect(await f.count("turn.start")).toBe(3); expect(await f.count("processStart")).toBe(4);
+    expect(await f.count("tool.resolve")).toBe(3);
+    await recovered.session!.close({ reason: "test complete" });
+  } catch (error) {
+    const diagnostics = await readFile(join(f.root, "native.stderr"), "utf8").catch(() => "");
+    throw new Error(`${String(error)}\n${diagnostics}`, { cause: error });
+  } finally { await f.cleanup(); }
+}, 30_000);
+
+it.skipIf(process.platform === "win32").each(["codex", "claude", "adopt", "paused", "retained"] as const)("recovers qualified ACPX %s under source authority before the second turn", async mode => {
+  const f = await fixture(mode === "claude" ? "claude" : "codex");
+  try {
+    await expect(f.recover({ acpxSourceRecoveryVersion: undefined }).transport.request("thread/read", {})).rejects.toThrow("recovery_integrity_unsupported");
+    if (mode === "paused") await writeFile(f.fault, "paused");
+    if (mode === "retained") {
+      const state = await read(f.providerFile);
+      for (const [eventType, payload] of [["usage.reported", { cumulative: { inputTokens: 3 } }], ["harness.diagnostic", { code: "retained-source-maintenance", message: "Retained maintenance fact" }]]) {
+        state.pendingEvents.push({ executorEventId: `acpx_provider_${String(state.nextEventSequence++).padStart(16, "0")}`, eventType, priority: "p0", payload });
+      }
+      await writeFile(f.providerFile, JSON.stringify(state));
+    }
+    const accounting = vi.fn(async (_event: PrpEvent) => {});
+    let source = f.recover({ onRecoveryAccountingEvent: accounting });
+    expect((await source.transport.request("thread/read", {})).thread).toMatchObject(f.thread); await f.settle();
+    await expect(source.transport.request("turn/start", {})).rejects.toThrow("source_attachment_pending");
+    const pid = source.evidence().runnerPid;
+    if (mode === "adopt") {
+      const startedAt = source.transport.processInfo().startedAt;
+      await source.detachControllerForRestart();
+      const launch = vi.fn(() => { throw new Error("duplicate source launch"); });
+      source = f.recover({ runnerProcessLauncher: launch, authorizeRunnerAdoption: async () => { throw new Error("old source configuration"); },
+        adoptExistingRunner: { pid, processGroupId: pid, startedAt, isAlive: () => !stopped(pid!), signal: () => false } });
+      expect((await source.transport.request("thread/read", {})).thread).toMatchObject(f.thread);
+      expect(source.evidence().runnerPid).toBe(pid); expect(launch).not.toHaveBeenCalled();
+    }
+    await source.transport.attachRun!({ runId: "run-target", turnId: "turn-target", itemId: "item-target" });
+    expect(source.evidence().runnerPid).toBe(pid);
+    await f.complete(source); await f.settle();
+    expect((await read(f.providerFile)).identity).toEqual(f.providerIdentity);
+    expect(await f.count("turn.start")).toBe(2); expect(await f.count("processStart")).toBe(3);
+    if (mode === "retained") {
+      for (const type of ["usage.reported", "harness.diagnostic"]) {
+        const delivered = accounting.mock.calls.filter(call => call[0].eventType === type);
+        expect(delivered).toHaveLength(1); expect(delivered[0]![0]).toHaveProperty("runId", f.identity.runId);
+      }
+    }
+    await source.transport.close(); expect((await read(f.runnerFile)).lifecycle).toBe("suspended");
+    expect(await f.count("processStart")).toBe(3);
+  } catch (error) {
+    const diagnostics = (await readFile(join(f.root, "native.stderr"), "utf8").catch(() => ""))
+      + (await readFile(join(f.root, "diagnostics", "runnerd.stderr.log"), "utf8").catch(() => ""));
+    throw new Error(`${String(error)}\n${diagnostics}`, { cause: error });
+  } finally { await f.cleanup(); }
+}, 30_000);
+
+it.skipIf(process.platform === "win32").each(["goal", "active", "tool", "goal-during-snapshot"] as const)("stops the owned ACPX process group after live %s admission fails", async fault => {
+  const f = await fixture();
+  try {
+    await writeFile(f.fault, fault);
+    await expect(f.recover().transport.request("thread/read", {})).rejects.toThrow("saved dormant ACPX");
+    await vi.waitFor(async () => {
+      const calls = await f.calls();
+      expect(absent(calls.filter(entry => entry.processStart).at(-1)!.processStart)).toBe(true);
+      for (const entry of calls) if (entry.processStart || entry.childPid) expect(stopped(entry.processStart ?? entry.childPid)).toBe(true);
+    }, { timeout: 5_000 });
+    expect(await f.count("processStart")).toBe(2); expect(await f.count("turn.start")).toBe(1);
+    if (fault === "goal-during-snapshot") {
+      expect((await f.calls()).find(entry => entry.emitted?.eventType === "runtime.goal")?.emitted)
+        .toMatchObject({ runId: f.identity.runId, turnId: null, payload: { providerRevision: 2, goal: { status: "active" } } });
+    }
+  } finally { await f.cleanup(); }
+}, 30_000);
+
+it.skipIf(process.platform === "win32").each(["limited", "usage_limited", "budget_limited"] as const)("rejects a saved %s ACPX goal that resumes active and reaps the sidecar", async status => {
+  const f = await fixture("claude");
+  try {
+    const state = await read(f.providerFile);
+    state.goalProjection = { schema: "paperclip.session_goal.snapshot.v1", goal: { objective: "Settled limited goal", status, workingNow: false }, workingNow: false };
+    await writeFile(f.providerFile, JSON.stringify(state)); await writeFile(f.fault, "goal");
+    await expect(f.recover().transport.request("thread/read", {})).rejects.toThrow("saved dormant ACPX resumed an active or unknown goal");
+    await vi.waitFor(async () => {
+      const calls = await f.calls();
+      expect(absent(calls.filter(entry => entry.processStart).at(-1)!.processStart)).toBe(true);
+      for (const entry of calls) if (entry.processStart || entry.childPid) expect(stopped(entry.processStart ?? entry.childPid)).toBe(true);
+    }, { timeout: 5_000 });
+    expect(await f.count("processStart")).toBe(2); expect(await f.count("turn.start")).toBe(1);
+  } finally { await f.cleanup(); }
+}, 30_000);
+
+it.skipIf(process.platform === "win32")("preserves ordinary same-run active goal restoration", async () => {
+  const f = await fixture();
+  try {
+    const state = await read(f.providerFile);
+    state.goalProjection = { schema: "paperclip.session_goal.snapshot.v1", goal: { objective: "Authorized same-run goal", status: "active" }, workingNow: false };
+    await writeFile(f.providerFile, JSON.stringify(state)); await writeFile(f.fault, "goal");
+    const sameRun = f.recover({ authorityRunId: f.identity.runId });
+    expect((await sameRun.transport.request("thread/read", {})).thread).toMatchObject(f.thread);
+    expect(await f.count("processStart")).toBe(2); expect(await f.count("turn.start")).toBe(1);
+  } finally { await f.cleanup(); }
+}, 30_000);
@@ -5605,8 +5605,9 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
      })))) {
      throw new Error("native_runner_source_reconciliation_requires_exact_authority");
    }
-    if (lostPredecessor && (this.options.provider ?? "codex") !== "codex") throw new Error("native_runner_recovery_integrity_unsupported");
-    if (lostPredecessor && this.options.sourceRecoveryVersion !== 1) {
+    if (lostPredecessor && (!["codex", "acpx"].includes(this.options.provider ?? "codex") || this.options.acpxCandidateAdmission)) throw new Error("native_runner_recovery_integrity_unsupported");
+    const recoveryCapability = this.options.provider === "acpx" ? this.options.acpxSourceRecoveryVersion : this.options.sourceRecoveryVersion;
+    if (lostPredecessor && recoveryCapability !== 1) {
      throw new Error("native_runner_recovery_integrity_unsupported");
    }
    const hasControlPlaneWarmBoundary =