Runner: retain the provider across hosted park and reclaim (#19140)

packages/paperclip-runner/runner/DURABLE_TRANSPORT.md

@@ -162,3 +162,32 @@ after that drain, so a late active-goal notification cannot be hidden by the
 older executor journal. This hook does not detach a controller, stop a process,
 or advertise hosted parking support. The host-control consumer must separately
 validate the exact Runner identity and acknowledged sequence cut.
+
+### Hosted park and reclaim loop
+
+A trusted Launcher may pass a connected Unix stream at fd 4 and start durable
+listener mode with `--host-control-fd 4`. The descriptor is close-on-exec before
+provider startup. The private channel accepts the `park`, `claim`, and `retire`
+requests defined in the host-control codec; it cannot execute PRP commands.
+
+`park` requires the exact six-field authority, acknowledged command/event cut,
+no unfinished transition or cleanup, and the provider eligibility proof above.
+The Runner drops its controller connection and credentials, retains its provider
+and journals, and blocks on host control without provider polling. Ordinary
+reconnect grace does not expire a parked Runner. The configured process lifetime
+still applies. Mount ownership stays with Launcher.
+
+`claim` names the accepted park request and supplies a fresh bootstrap ticket.
+It resumes authenticated PRP under the retained authority; ordinary `run.attach`
+must still establish the next authority before new work. Pending reconnect and
+live commands both reject execution before that attachment; snapshots, drain,
+and cleanup remain available. Replaying the same claim returns its acknowledgement without consuming another ticket. A parked
+retirement names the park request. Cancelling a claim additionally requires its
+claim ID, the unchanged authority and sequence cut, and provider quiescence.
+Retirement, parked host-channel loss, or process-lifetime expiry uses the existing
+suspended lifecycle and provider shutdown path. There is no new checkpoint state.
+
+This native loop does not advertise hosted parking support: Launcher retention,
+Core orchestration, provider-specific qualification, and live VM sleep/wake
+verification are required before rollout. Providers without an eligibility
+implementation remain ineligible.

packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs

@@ -9,8 +9,8 @@ use std::os::unix::fs::{MetadataExt, OpenOptionsExt, PermissionsExt};
 
 use paperclip_runner_core::acpx_candidate_admission::AcpxCandidateAdmission;
 use paperclip_runner_core::durable::{
-    capture_bootstrap_ticket, redact_diagnostic_text, run_durable_runner, AcpxLaunchProfile,
-    CursorCloudLaunchProfile, DurableRunnerConfig, OpenCodeLaunchProfile, QualifiedLaunchArtifact,
+    capture_bootstrap_ticket, redact_diagnostic_text, AcpxLaunchProfile, CursorCloudLaunchProfile,
+    DurableRunnerConfig, OpenCodeLaunchProfile, QualifiedLaunchArtifact,
 };
 use paperclip_runner_core::local_runner::{run_local_runner, LocalRunnerError, RunnerConfig};
 use paperclip_runner_core::native_provider_backend::NativeProviderCommandExecutor;
@@ -417,6 +417,38 @@ fn durable_connect_url(args: &[String], run_id: &str) -> Result<String, LocalRun
 }
 
 fn run_durable(args: &[String]) -> Result<(), LocalRunnerError> {
+    #[cfg(unix)]
+    let host_control = if let Some(fd) = optional_u64(args, "--host-control-fd")? {
+        use std::os::fd::{FromRawFd, OwnedFd};
+        use std::os::unix::fs::FileTypeExt;
+        if fd != 4
+            || !fs::metadata("/dev/fd/4")
+                .map(|metadata| metadata.file_type().is_socket())
+                .unwrap_or(false)
+        {
+            return Err(LocalRunnerError::invalid(
+                "host control requires an inherited socket at fd 4",
+            ));
+        }
+        // SAFETY: fd 4 was inherited from the trusted host and verified open.
+        // No other Runner owner takes or closes it; HostControl sets CLOEXEC
+        // before any provider can spawn and validates connected AF_UNIX STREAM.
+        let socket = std::os::unix::net::UnixStream::from(unsafe { OwnedFd::from_raw_fd(4) });
+        Some(
+            paperclip_runner_core::host_control::HostControl::new(socket)
+                .map_err(|error| LocalRunnerError::invalid(error.to_string()))?,
+        )
+    } else {
+        None
+    };
+    #[cfg(not(unix))]
+    let host_control = if args.iter().any(|arg| arg == "--host-control-fd") {
+        return Err(LocalRunnerError::invalid(
+            "host control is unavailable on this platform",
+        ));
+    } else {
+        None
+    };
     let ticket = capture_bootstrap_ticket()
         .map_err(|error| LocalRunnerError::invalid(error.to_string()))?
         .ok_or_else(|| {
@@ -462,6 +494,11 @@ fn run_durable(args: &[String]) -> Result<(), LocalRunnerError> {
         reconnect_grace: optional_u64(args, "--reconnect-grace-ms")?.map(Duration::from_millis),
         max_runtime: duration("--max-runtime-ms", 0)?,
     };
+    if host_control.is_some() && !config.connect_url.starts_with("listen://") {
+        return Err(LocalRunnerError::invalid(
+            "host control requires trusted Runner listener mode",
+        ));
+    }
     #[cfg(unix)]
     std::env::set_var(
         paperclip_runner_core::runtime_notifications::CONTEXT_ENV,
@@ -479,8 +516,13 @@ fn run_durable(args: &[String]) -> Result<(), LocalRunnerError> {
         }
     }
     let executor = NativeProviderCommandExecutor::with_runner_config(state_dir, &config);
-    run_durable_runner(config, ticket, executor)
-        .map_err(|error| LocalRunnerError::invalid(error.to_string()))
+    paperclip_runner_core::durable::run_durable_runner_with_host_control(
+        config,
+        ticket,
+        executor,
+        host_control,
+    )
+    .map_err(|error| LocalRunnerError::invalid(error.to_string()))
 }
 
 fn apply_oom_score_adj(args: &[String]) -> Result<(), LocalRunnerError> {
@@ -545,6 +587,11 @@ fn run(args: &[String]) -> Result<(), LocalRunnerError> {
             "ACPX candidate admission requires trusted durable runner startup",
         ));
     }
+    if args.iter().any(|arg| arg == "--host-control-fd") {
+        return Err(LocalRunnerError::invalid(
+            "host control requires trusted durable runner startup",
+        ));
+    }
     run_local_runner(RunnerConfig {
         run_id: value(args, "--run-id")?,
         normalized_session_id: value(args, "--session-id")?,
@@ -678,6 +725,7 @@ mod tests {
         assert!(metadata.get("codexSourceRecoveryVersion").is_none());
         assert_eq!(metadata["acpxRecoveryIntegrityVersion"], 1);
         assert!(metadata.get("acpxSourceRecoveryVersion").is_none());
+        assert!(metadata.get("hostedParkingVersion").is_none());
         assert_eq!(metadata["schema"], RUNNERD_BUILD_METADATA_SCHEMA);
         assert_eq!(metadata["binaryContractVersion"], 2);
         assert_eq!(
@@ -694,6 +742,20 @@ mod tests {
         );
     }
 
+    #[test]
+    #[cfg(unix)]
+    fn host_control_rejects_other_fds_before_bootstrap_or_provider_startup() {
+        let args = vec!["--host-control-fd".to_owned(), "3".to_owned()];
+        assert_eq!(
+            run_durable(&args).unwrap_err().to_string(),
+            "host control requires an inherited socket at fd 4"
+        );
+        assert_eq!(
+            run(&args).unwrap_err().to_string(),
+            "host control requires trusted durable runner startup"
+        );
+    }
+
     #[test]
     #[cfg(all(target_os = "macos", target_arch = "aarch64"))]
     fn candidate_startup_requires_a_complete_trusted_launch_profile() {

packages/paperclip-runner/runner/crates/runner-core/src/durable/mod.rs

@@ -12,8 +12,8 @@ use sha2::{Digest, Sha256};
 use crate::stable_identity::{is_stable_id, DURABLE_STABLE_ID_CHARS, SHORT_STABLE_ID_CHARS};
 
 pub use runner::{
-    run_durable_runner, CommandExecution, CommandExecutor, PolledEvent,
-    TerminalDeliveryReconciliation,
+    run_durable_runner, run_durable_runner_with_host_control, CommandExecution, CommandExecutor,
+    PolledEvent, TerminalDeliveryReconciliation,
 };
 pub(crate) use state::{
     create_private_temporary_file, open_private_regular_file, redact_sensitive_text_values,

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

@@ -447,10 +447,44 @@ impl CumulativeAckPersistence {
 }
 
 pub fn run_durable_runner<E: CommandExecutor>(
+    config: DurableRunnerConfig,
+    bootstrap_ticket: BootstrapTicket,
+    executor: E,
+) -> Result<(), DurableRunnerError> {
+    run_durable_runner_with_host_control(config, bootstrap_ticket, executor, None)
+}
+
+#[cfg(not(unix))]
+pub struct HostControl;
+#[cfg(unix)]
+use crate::host_control::{HostControl, HostIdentity, HostRequest};
+
+#[cfg(unix)]
+struct HostedRunnerControl {
+    channel: HostControl,
+    park_id: Option<String>,
+    claiming: Option<(String, String)>,
+}
+
+#[cfg(unix)]
+impl HostedRunnerControl {
+    fn new(channel: HostControl) -> Self {
+        Self {
+            channel,
+            park_id: None,
+            claiming: None,
+        }
+    }
+}
+
+pub fn run_durable_runner_with_host_control<E: CommandExecutor>(
     mut config: DurableRunnerConfig,
     bootstrap_ticket: BootstrapTicket,
     mut executor: E,
+    #[allow(unused_variables)] host_control: Option<HostControl>,
 ) -> Result<(), DurableRunnerError> {
+    #[cfg(unix)]
+    let mut host_control = host_control.map(HostedRunnerControl::new);
     let started = Instant::now();
     config.validate()?;
     if let Some(admission) = config
@@ -508,7 +542,50 @@ pub fn run_durable_runner<E: CommandExecutor>(
     let mut reconnect_attempt = 0_u32;
     let mut next_connection_attempt = Instant::now();
 
-    loop {
+    'session: loop {
+        #[cfg(unix)]
+        if let Some(control) = host_control.as_mut() {
+            if control.park_id.is_some() {
+                let remaining = (!config.max_runtime.is_zero())
+                    .then(|| config.max_runtime.saturating_sub(started.elapsed()));
+                if remaining.is_some_and(|duration| duration.is_zero()) {
+                    persist_lifecycle_before_shutdown(
+                        &mut state,
+                        &store,
+                        &mut executor,
+                        "suspended",
+                    )?;
+                    return Ok(());
+                }
+                match process_host_control(
+                    control,
+                    &mut state,
+                    &store,
+                    &mut executor,
+                    true,
+                    remaining,
+                ) {
+                    Ok(HostAction::Claim(ticket)) => {
+                        bootstrap_ticket = Some(ticket);
+                        lease = None;
+                        authenticated_once = false;
+                        disconnected_since = None;
+                        reconnect_attempt = 0;
+                    }
+                    Ok(HostAction::Retire) => return Ok(()),
+                    Ok(_) => continue,
+                    Err(error) => {
+                        let _ = persist_lifecycle_before_shutdown(
+                            &mut state,
+                            &store,
+                            &mut executor,
+                            "suspended",
+                        );
+                        return Err(error);
+                    }
+                }
+            }
+        }
         if authenticated_once {
             let disconnected_at = disconnected_since.get_or_insert_with(Instant::now);
             if config
@@ -584,6 +661,22 @@ pub fn run_durable_runner<E: CommandExecutor>(
             continue;
         }
         let using_bootstrap = lease.is_none();
+        #[cfg(unix)]
+        if let Some(control) = host_control.as_mut() {
+            match process_host_control(control, &mut state, &store, &mut executor, false, None) {
+                Ok(HostAction::Park) => {
+                    lease = None;
+                    bootstrap_ticket = None;
+                    continue 'session;
+                }
+                Ok(HostAction::Retire) => return Ok(()),
+                Err(error) => {
+                    let _ = shutdown_preserving_cleanup(&state, &mut executor);
+                    return Err(error);
+                }
+                _ => {}
+            }
+        }
         let connect_deadline = connection_attempt_deadline(
             &config,
             started,
@@ -743,6 +836,10 @@ pub fn run_durable_runner<E: CommandExecutor>(
                             &mut endpoint,
                             next,
                         )?;
+                        #[cfg(unix)]
+                        if let Some(control) = host_control.as_mut() {
+                            control.claiming = None;
+                        }
                     }
                     Err(error) => {
                         state.record_diagnostic(format!(
@@ -786,6 +883,8 @@ pub fn run_durable_runner<E: CommandExecutor>(
                 welcome.warm_transition_version,
                 connection.protocol_version,
             )?;
+            #[cfg(unix)]
+            admit_host_claim(host_control.as_ref(), &command, next_authority.as_ref())?;
             let (result, lifecycle) =
                 process_command(&mut state, &store, &config, &mut executor, &command)?;
             let next_authority = next_authority.filter(|_| completed_attachment(&result));
@@ -861,6 +960,10 @@ pub fn run_durable_runner<E: CommandExecutor>(
         }
         if let Some(next) = authority_rotation {
             apply_authority_rotation(&mut state, &store, &mut config, &mut endpoint, next)?;
+            #[cfg(unix)]
+            if let Some(control) = host_control.as_mut() {
+                control.claiming = None;
+            }
             disconnected_since = Some(Instant::now());
             continue;
         }
@@ -994,6 +1097,34 @@ pub fn run_durable_runner<E: CommandExecutor>(
             // Read control before starting another fsynced provider batch.
             // Consuming the last cumulative ACK must not let a new output
             // suffix overtake the stop/suspend already queued behind it.
+            #[cfg(unix)]
+            {
+                if let Some(control) = host_control.as_mut() {
+                    match process_host_control(
+                        control,
+                        &mut state,
+                        &store,
+                        &mut executor,
+                        false,
+                        None,
+                    ) {
+                        Ok(HostAction::Park) => {
+                            // The private owner now holds the exact idle cut. No
+                            // old PRP frame, lease renewal, or provider poll may follow.
+                            lease = None;
+                            bootstrap_ticket = None;
+                            drop(transport);
+                            continue 'session;
+                        }
+                        Ok(HostAction::Retire) => return Ok(()),
+                        Err(error) => {
+                            let _ = shutdown_preserving_cleanup(&state, &mut executor);
+                            return Err(error);
+                        }
+                        _ => {}
+                    }
+                }
+            }
             let control_message = transport.receive_json();
             poll_executor_events_when_control_idle(
                 &mut state,
@@ -1058,6 +1189,8 @@ pub fn run_durable_runner<E: CommandExecutor>(
                         welcome.warm_transition_version,
                         connection.protocol_version,
                     )?;
+                    #[cfg(unix)]
+                    admit_host_claim(host_control.as_ref(), &command, next_authority.as_ref())?;
                     let (result, lifecycle) =
                         process_command(&mut state, &store, &config, &mut executor, &command)?;
                     let next_authority = next_authority.filter(|_| completed_attachment(&result));
@@ -1135,6 +1268,10 @@ pub fn run_durable_runner<E: CommandExecutor>(
                             &mut endpoint,
                             next,
                         )?;
+                        #[cfg(unix)]
+                        if let Some(control) = host_control.as_mut() {
+                            control.claiming = None;
+                        }
                         disconnected_since = Some(Instant::now());
                         break;
                     }
@@ -1238,6 +1375,148 @@ pub fn run_durable_runner<E: CommandExecutor>(
     }
 }
 
+#[cfg(unix)]
+enum HostAction {
+    None,
+    Park,
+    Claim(BootstrapTicket),
+    Retire,
+}
+
+#[cfg(unix)]
+fn admit_host_claim(
+    control: Option<&HostedRunnerControl>,
+    command: &Command,
+    next_authority: Option<&DurableRunnerConfig>,
+) -> Result<(), DurableRunnerError> {
+    if control.is_some_and(|control| control.claiming.is_some())
+        && !(matches!(
+            command.command_type.as_str(),
+            "session.snapshot"
+                | "runner.drain"
+                | "runner.suspend"
+                | "runner.shutdown"
+                | "turn.stop"
+        ) || (command.command_type == "run.attach" && next_authority.is_some()))
+    {
+        return Err(DurableRunnerError::invalid(
+            "claimed Runner requires authority attachment before execution",
+        ));
+    }
+    Ok(())
+}
+
+#[cfg(unix)]
+fn process_host_control<E: CommandExecutor>(
+    control: &mut HostedRunnerControl,
+    state: &mut DurableState,
+    store: &DurableStateStore,
+    executor: &mut E,
+    parked: bool,
+    remaining: Option<Duration>,
+) -> Result<HostAction, DurableRunnerError> {
+    let Some(request) = control.channel.receive(parked, remaining)? else {
+        return Ok(HostAction::None);
+    };
+    if control.channel.replay(&request)? {
+        return Ok(HostAction::None);
+    }
+    match &request {
+        HostRequest::Park {
+            id,
+            identity,
+            last_controller_command_seq,
+            acked_source_seq,
+        } if !parked
+            && control.park_id.is_none()
+            && control.claiming.is_none()
+            && host_park_cut_matches(
+                state,
+                identity,
+                *last_controller_command_seq,
+                *acked_source_seq,
+            )
+            && executor.host_park_ready()? =>
+        {
+            control.park_id = Some(id.clone());
+            control.channel.reply(&request, "parked")?;
+            Ok(HostAction::Park)
+        }
+        HostRequest::Claim {
+            id,
+            park_id,
+            bootstrap_ticket,
+            ..
+        } if parked && control.park_id.as_ref() == Some(park_id) => {
+            let ticket = BootstrapTicket::new(bootstrap_ticket.clone())?;
+            control.claiming = Some((park_id.clone(), id.clone()));
+            control.park_id = None;
+            control.channel.reply(&request, "claimed")?;
+            Ok(HostAction::Claim(ticket))
+        }
+        HostRequest::Retire {
+            park_id,
+            claim_id,
+            identity,
+            last_controller_command_seq,
+            acked_source_seq,
+            ..
+        } if match (
+            claim_id,
+            identity,
+            last_controller_command_seq,
+            acked_source_seq,
+        ) {
+            (None, None, None, None) => parked && control.park_id.as_ref() == Some(park_id),
+            (Some(claim_id), Some(identity), Some(command_seq), Some(acked_seq)) => {
+                !parked
+                    && control
+                        .claiming
+                        .as_ref()
+                        .is_some_and(|(park, claim)| park == park_id && claim == claim_id)
+                    && host_park_cut_matches(state, identity, *command_seq, *acked_seq)
+                    && executor.host_park_ready()?
+            }
+            _ => false,
+        } =>
+        {
+            persist_lifecycle_before_shutdown(state, store, executor, "suspended")?;
+            control.channel.reply(&request, "retired")?;
+            Ok(HostAction::Retire)
+        }
+        _ => {
+            control.channel.reply(&request, "conflict")?;
+            Ok(HostAction::None)
+        }
+    }
+}
+
+#[cfg(unix)]
+fn host_park_cut_matches(
+    state: &DurableState,
+    identity: &HostIdentity,
+    command_seq: u64,
+    acked_seq: u64,
+) -> bool {
+    identity.runner_instance_id == state.runner_instance_id
+        && identity.environment_lease_id == state.environment_lease_id
+        && identity.run_id == state.run_id
+        && identity.normalized_session_id == state.normalized_session_id
+        && identity.turn_id == state.turn_id
+        && identity.item_id == state.item_id
+        && command_seq == state.last_controller_command_seq
+        && acked_seq == state.acked_source_seq
+        && state.outbox.is_empty()
+        && state.acked_source_seq == state.highest_source_seq()
+        && state.warm_transition.is_none()
+        && state.pending_terminal_delivery.is_none()
+        && state.pending_provider_cleanup.is_none()
+        && state
+            .processed_commands
+            .values()
+            .all(|command| !matches!(command.status.as_str(), "pending" | "indeterminate"))
+}
+
 fn lease_renewal_deadline(now: u64, expires_at: u64) -> u64 {
     now.saturating_add(expires_at.saturating_sub(now) / 2)
 }
@@ -2721,6 +3000,271 @@ mod tests {
         }
     }
 
+    #[test]
+    #[cfg(unix)]
+    fn hosted_parking_requires_exact_cut_and_replays_claim_without_reusing_ticket() {
+        use std::io::{BufRead, BufReader, Write};
+        use std::os::unix::net::UnixStream;
+        struct IdleExecutor {
+            proofs: usize,
+            shutdowns: usize,
+            ready: bool,
+        }
+        impl CommandExecutor for IdleExecutor {
+            fn execute(&mut self, _: &Command) -> Result<CommandExecution, DurableRunnerError> {
+                panic!("private control cannot execute PRP commands");
+            }
+            fn host_park_ready(&mut self) -> Result<bool, DurableRunnerError> {
+                self.proofs += 1;
+                Ok(self.ready)
+            }
+            fn shutdown(&mut self) -> Result<(), DurableRunnerError> {
+                self.shutdowns += 1;
+                Ok(())
+            }
+        }
+        let directory = std::env::temp_dir().join(format!(
+            "paperclip-host-retirement-{}",
+            uuid::Uuid::new_v4()
+        ));
+        let config = config(directory.clone());
+        let store = DurableStateStore::new(&config.state_dir).unwrap();
+        let mut state = DurableState::new(&config);
+        let (socket, mut host) = UnixStream::pair().unwrap();
+        let mut reader = BufReader::new(host.try_clone().unwrap());
+        let mut control = HostedRunnerControl::new(HostControl::new(socket).unwrap());
+        let mut executor = IdleExecutor {
+            proofs: 0,
+            shutdowns: 0,
+            ready: true,
+        };
+        let park = json!({"op":"park","id":"park-1","identity":{
+            "runnerInstanceId":state.runner_instance_id,"environmentLeaseId":state.environment_lease_id,
+            "runId":state.run_id,"normalizedSessionId":state.normalized_session_id,"turnId":state.turn_id,"itemId":state.item_id,
+        },"lastControllerCommandSeq":0,"ackedSourceSeq":0});
+        let mut exchange = |request: &Value,
+                            state: &mut DurableState,
+                            control: &mut HostedRunnerControl,
+                            executor: &mut IdleExecutor,
+                            parked: bool| {
+            writeln!(host, "{request}").unwrap();
+            let action = process_host_control(
+                control,
+                state,
+                &store,
+                executor,
+                parked,
+                Some(Duration::from_secs(1)),
+            )
+            .unwrap();
+            let mut line = String::new();
+            reader.read_line(&mut line).unwrap();
+            (action, serde_json::from_str::<Value>(&line).unwrap())
+        };
+        state.last_controller_command_seq = 1;
+        assert_eq!(
+            exchange(&park, &mut state, &mut control, &mut executor, false).1["status"],
+            "conflict"
+        );
+        state.last_controller_command_seq = 0;
+        assert!(matches!(
+            exchange(&park, &mut state, &mut control, &mut executor, false).0,
+            HostAction::Park
+        ));
+        assert!(matches!(
+            exchange(&park, &mut state, &mut control, &mut executor, true).0,
+            HostAction::None
+        ));
+        assert_eq!(executor.proofs, 1);
+        let claim = json!({"op":"claim","id":"claim-1","parkId":"park-1","bootstrapTicket":"fresh-private-ticket"});
+        assert!(matches!(
+            exchange(&claim, &mut state, &mut control, &mut executor, true).0,
+            HostAction::Claim(_)
+        ));
+        assert!(matches!(
+            exchange(&claim, &mut state, &mut control, &mut executor, false).0,
+            HostAction::None
+        ));
+        assert!(admit_host_claim(Some(&control), &command("turn.start"), None).is_err());
+        assert!(admit_host_claim(Some(&control), &command("session.snapshot"), None).is_ok());
+        assert!(admit_host_claim(Some(&control), &command("run.attach"), None).is_err());
+        let mut next_authority = config.clone();
+        next_authority.run_id = "next-run".to_owned();
+        assert!(admit_host_claim(
+            Some(&control),
+            &command("run.attach"),
+            Some(&next_authority)
+        )
+        .is_ok());
+        let mut changed = claim;
+        changed["bootstrapTicket"] = json!("different-private-ticket");
+        assert_eq!(
+            exchange(&changed, &mut state, &mut control, &mut executor, false).1["status"],
+            "conflict"
+        );
+        assert_eq!(executor.shutdowns, 0);
+        let mut premature_park = park.clone();
+        premature_park["id"] = json!("park-before-attach");
+        assert_eq!(
+            exchange(
+                &premature_park,
+                &mut state,
+                &mut control,
+                &mut executor,
+                false
+            )
+            .1["status"],
+            "conflict"
+        );
+        let retire = json!({"op":"retire","id":"retire-park","parkId":"park-1"});
+        assert_eq!(
+            exchange(&retire, &mut state, &mut control, &mut executor, false).1["status"],
+            "conflict"
+        );
+        let conditional_retire = json!({"op":"retire","id":"retire-claim","parkId":"park-1",
+            "claimId":"claim-1","identity":park["identity"],
+            "lastControllerCommandSeq":0,"ackedSourceSeq":0});
+        for (field, value) in [
+            ("parkId", json!("wrong-park")),
+            ("claimId", json!("wrong-claim")),
+            ("lastControllerCommandSeq", json!(1)),
+            ("ackedSourceSeq", json!(1)),
+        ] {
+            let mut stale = conditional_retire.clone();
+            stale["id"] = json!(format!("stale-{field}"));
+            stale[field] = value;
+            assert_eq!(
+                exchange(&stale, &mut state, &mut control, &mut executor, false).1["status"],
+                "conflict"
+            );
+        }
+        let mut wrong_identity = conditional_retire.clone();
+        wrong_identity["id"] = json!("stale-identity");
+        wrong_identity["identity"]["runId"] = json!("another-run");
+        assert_eq!(
+            exchange(
+                &wrong_identity,
+                &mut state,
+                &mut control,
+                &mut executor,
+                false
+            )
+            .1["status"],
+            "conflict"
+        );
+        executor.ready = false;
+        assert_eq!(
+            exchange(
+                &conditional_retire,
+                &mut state,
+                &mut control,
+                &mut executor,
+                false
+            )
+            .1["status"],
+            "conflict"
+        );
+        assert_eq!(executor.shutdowns, 0);
+        executor.ready = true;
+        // Simulate completed run.attach before a later exact parked cut.
+        control.claiming = None;
+        assert!(admit_host_claim(Some(&control), &command("turn.start"), None).is_ok());
+        let mut next_park = park.clone();
+        next_park["id"] = json!("park-2");
+        assert!(matches!(
+            exchange(&next_park, &mut state, &mut control, &mut executor, false).0,
+            HostAction::Park
+        ));
+        let retire = json!({"op":"retire","id":"retire-park","parkId":"park-2"});
+        assert!(matches!(
+            exchange(&retire, &mut state, &mut control, &mut executor, true).0,
+            HostAction::Retire
+        ));
+        assert_eq!(executor.shutdowns, 1);
+        fs::remove_dir_all(directory).unwrap();
+    }
+
+    #[test]
+    #[cfg(unix)]
+    fn hosted_park_blocks_without_provider_polls_and_retires_on_eof_or_lifetime() {
+        use std::io::{BufRead, BufReader, Write};
+        use std::os::unix::net::UnixStream;
+        use std::sync::{
+            atomic::{AtomicUsize, Ordering},
+            Arc,
+        };
+        struct ParkedExecutor(Arc<AtomicUsize>);
+        impl CommandExecutor for ParkedExecutor {
+            fn execute(&mut self, _: &Command) -> Result<CommandExecution, DurableRunnerError> {
+                panic!("parked provider cannot execute");
+            }
+            fn host_park_ready(&mut self) -> Result<bool, DurableRunnerError> {
+                Ok(true)
+            }
+            fn poll_events(&mut self) -> Result<Vec<PolledEvent>, DurableRunnerError> {
+                panic!("parked provider cannot poll");
+            }
+            fn shutdown(&mut self) -> Result<(), DurableRunnerError> {
+                self.0.fetch_add(1, Ordering::SeqCst);
+                Ok(())
+            }
+        }
+        for eof in [false, true] {
+            let directory = std::env::temp_dir().join(format!(
+                "paperclip-hosted-park-stop-{}-{eof}",
+                std::process::id()
+            ));
+            let _ = fs::remove_dir_all(&directory);
+            let mut config = config(directory.clone());
+            config.max_runtime = if eof {
+                Duration::ZERO
+            } else {
+                // Include durable startup under parallel-suite filesystem load.
+                Duration::from_secs(2)
+            };
+            let (socket, mut host) = UnixStream::pair().unwrap();
+            let control = HostControl::new(socket).unwrap();
+            writeln!(host,"{}",json!({"op":"park","id":"park-stop","identity":{
+                "runnerInstanceId":config.runner_instance_id,"environmentLeaseId":config.environment_lease_id,
+                "runId":config.run_id,"normalizedSessionId":config.normalized_session_id,"turnId":config.turn_id,"itemId":config.item_id,
+            },"lastControllerCommandSeq":0,"ackedSourceSeq":0})).unwrap();
+            let shutdowns = Arc::new(AtomicUsize::new(0));
+            let executor = ParkedExecutor(shutdowns.clone());
+            let saved_config = config.clone();
+            let task = std::thread::spawn(move || {
+                run_durable_runner_with_host_control(
+                    config,
+                    BootstrapTicket::new("unused-bootstrap".to_owned()).unwrap(),
+                    executor,
+                    Some(control),
+                )
+            });
+            let mut reply = String::new();
+            BufReader::new(&host).read_line(&mut reply).unwrap();
+            assert_eq!(
+                serde_json::from_str::<Value>(&reply).unwrap()["status"],
+                "parked"
+            );
+            if eof {
+                drop(host);
+                assert!(task.join().unwrap().is_err());
+            } else {
+                assert!(task.join().unwrap().is_ok());
+            }
+            assert_eq!(shutdowns.load(Ordering::SeqCst), 1);
+            assert_eq!(
+                DurableStateStore::new(&directory)
+                    .unwrap()
+                    .load_or_create(&saved_config)
+                    .unwrap()
+                    .0
+                    .lifecycle,
+                "suspended"
+            );
+            fs::remove_dir_all(directory).unwrap();
+        }
+    }
+
     #[test]
     fn indeterminate_recovery_result_is_a_failed_wire_result() {
         let state = DurableState::new(&config(PathBuf::from("unused")));

packages/paperclip-runner/runner/crates/runner-core/src/durable/tests/hosted_retention.rs

@@ -0,0 +1,258 @@
+// Shares this module's authenticated PRP server helpers.
+use crate::durable::{
+    run_durable_runner_with_host_control, CommandExecution, CommandExecutor, DurableStateStore,
+    PolledEvent,
+};
+use crate::host_control::HostControl;
+use std::io::BufRead;
+use std::os::unix::net::UnixStream;
+
+struct RetentionExecutor {
+    polls: Arc<AtomicUsize>,
+    shutdowns: Arc<AtomicUsize>,
+}
+impl CommandExecutor for RetentionExecutor {
+    fn execute(&mut self, _: &Command) -> Result<CommandExecution, DurableRunnerError> {
+        panic!("host control must not execute business commands");
+    }
+    fn host_park_ready(&mut self) -> Result<bool, DurableRunnerError> {
+        Ok(true)
+    }
+    fn poll_events(&mut self) -> Result<Vec<PolledEvent>, DurableRunnerError> {
+        self.polls.fetch_add(1, Ordering::SeqCst);
+        Ok(vec![])
+    }
+    fn shutdown(&mut self) -> Result<(), DurableRunnerError> {
+        self.shutdowns.fetch_add(1, Ordering::SeqCst);
+        Ok(())
+    }
+}
+
+fn host_exchange(host: &mut UnixStream, request: Value) -> Value {
+    writeln!(host, "{request}").unwrap();
+    let mut line = String::new();
+    BufReader::new(host).read_line(&mut line).unwrap();
+    serde_json::from_str(&line).unwrap()
+}
+
+#[test]
+fn hosted_loop_parks_and_reclaims_with_fresh_bootstrap_during_busy_prp() {
+    exercise_hosted_loop(None);
+}
+
+#[test]
+fn hosted_claim_blocks_pending_and_live_work_before_attachment() {
+    for pending in [true, false] {
+        exercise_hosted_loop(Some(pending));
+    }
+}
+
+fn exercise_hosted_loop(blocked_work: Option<bool>) {
+    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+    let mut config = config(listener.local_addr().unwrap().port());
+    let directory =
+        std::env::temp_dir().join(format!("paperclip-host-loop-{}", uuid::Uuid::new_v4()));
+    config.state_dir = directory.clone();
+    config.max_runtime = Duration::from_secs(15);
+    config.reconnect_grace = Some(Duration::from_millis(20));
+    let state = test_state(&config);
+    let server_config = config.clone();
+    let server_state = state.clone();
+    let (busy_tx, busy_rx) = mpsc::channel();
+    let (claimed_tx, claimed_rx) = mpsc::channel();
+    let (finish_tx, finish_rx) = mpsc::channel();
+    let server = thread::spawn(move || {
+        let expires = current_unix_ms().unwrap() + 60_000;
+        let (stream, _) = listener.accept().unwrap();
+        stream
+            .set_read_timeout(Some(Duration::from_secs(5)))
+            .unwrap();
+        stream
+            .set_write_timeout(Some(Duration::from_secs(5)))
+            .unwrap();
+        let mut socket = accept(stream).unwrap();
+        let mut secure = server_authenticate(
+            &mut socket,
+            &server_config,
+            &server_state,
+            ServerCredential {
+                token: "initial-private-bootstrap",
+                kind: "bootstrap",
+                lease_id: None,
+                expires_at_unix_ms: expires,
+                revocation_epoch: 0,
+            },
+        );
+        send_secure(
+            &mut socket,
+            &mut secure,
+            &server_config,
+            &welcome(
+                &server_state,
+                "connection-1",
+                Some("old-private-lease"),
+                expires,
+                0,
+                vec![],
+            ),
+        );
+        let mut announced = false;
+        // Keep PRP readable. Host control must not depend on an idle socket read.
+        loop {
+            let ping = control(&server_state, "connection-1", "ping", json!({}));
+            let encrypted = secure
+                .encrypt(&serde_json::to_vec(&ping).unwrap(), true)
+                .unwrap();
+            if send_plain(&mut socket, &encrypted, server_config.max_frame_bytes).is_err() {
+                break;
+            }
+            let Ok(frame) = receive_plain(&mut socket, server_config.max_frame_bytes) else {
+                break;
+            };
+            assert_eq!(secure.decrypt(&frame, true).unwrap()["kind"], "pong");
+            if !announced {
+                busy_tx.send(()).unwrap();
+                announced = true;
+            }
+        }
+        drop(socket);
+        // A parked Runner must accept the new bootstrap, never the old lease.
+        let (stream, _) = listener.accept().unwrap();
+        stream
+            .set_read_timeout(Some(Duration::from_secs(5)))
+            .unwrap();
+        let mut socket = accept(stream).unwrap();
+        let mut secure = server_authenticate(
+            &mut socket,
+            &server_config,
+            &server_state,
+            ServerCredential {
+                token: "fresh-private-bootstrap",
+                kind: "bootstrap",
+                lease_id: None,
+                expires_at_unix_ms: expires,
+                revocation_epoch: 0,
+            },
+        );
+        let forbidden = json!({
+            "schema":"paperclip.prp.command.v1", "commandId":"premature-turn",
+            "controllerSeq":1, "type":"turn.start", "issuedAt":"2026-01-01T00:00:00Z", "payload":{}
+        });
+        send_secure(
+            &mut socket,
+            &mut secure,
+            &server_config,
+            &welcome(
+                &server_state,
+                "connection-2",
+                Some("new-private-lease"),
+                expires,
+                0,
+                if blocked_work == Some(true) { vec![forbidden.clone()] } else { vec![] },
+            ),
+        );
+        if let Some(pending) = blocked_work {
+            if !pending {
+                send_secure(&mut socket, &mut secure, &server_config,
+                    &control(&server_state, "connection-2", "command", forbidden));
+            }
+            finish_rx.recv_timeout(Duration::from_secs(5)).unwrap();
+            return;
+        }
+        send_secure(
+            &mut socket,
+            &mut secure,
+            &server_config,
+            &control(&server_state, "connection-2", "ping", json!({})),
+        );
+        assert_eq!(
+            receive_secure(&mut socket, &mut secure, &server_config)["kind"],
+            "pong"
+        );
+        claimed_tx.send(()).unwrap();
+        finish_rx.recv_timeout(Duration::from_secs(5)).unwrap();
+    });
+    let polls = Arc::new(AtomicUsize::new(0));
+    let shutdowns = Arc::new(AtomicUsize::new(0));
+    let executor = RetentionExecutor {
+        polls: polls.clone(),
+        shutdowns: shutdowns.clone(),
+    };
+    let (socket, mut host) = UnixStream::pair().unwrap();
+    host.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
+    let runner_config = config.clone();
+    let runner = thread::spawn(move || {
+        run_durable_runner_with_host_control(
+            runner_config,
+            BootstrapTicket::new("initial-private-bootstrap".to_owned()).unwrap(),
+            executor,
+            Some(HostControl::new(socket).unwrap()),
+        )
+    });
+    busy_rx.recv_timeout(Duration::from_secs(5)).unwrap();
+    let identity = json!({"runnerInstanceId":state.runner_instance_id,"environmentLeaseId":state.environment_lease_id,
+        "runId":state.run_id,"normalizedSessionId":state.normalized_session_id,"turnId":state.turn_id,"itemId":state.item_id});
+    assert_eq!(
+        host_exchange(
+            &mut host,
+            json!({"op":"park","id":"park-1","identity":identity,
+        "lastControllerCommandSeq":0,"ackedSourceSeq":0})
+        )["status"],
+        "parked"
+    );
+    let parked_polls = polls.load(Ordering::SeqCst);
+    // Outlast the ordinary reconnect grace without any polling or shutdown.
+    thread::sleep(Duration::from_millis(80));
+    assert_eq!(polls.load(Ordering::SeqCst), parked_polls);
+    assert_eq!(shutdowns.load(Ordering::SeqCst), 0);
+    assert_eq!(
+        host_exchange(
+            &mut host,
+            json!({"op":"claim","id":"claim-1","parkId":"park-1",
+        "bootstrapTicket":"fresh-private-bootstrap"})
+        )["status"],
+        "claimed"
+    );
+    if blocked_work.is_some() {
+        let error = runner.join().unwrap().unwrap_err();
+        assert_eq!(error.to_string(), "claimed Runner requires authority attachment before execution");
+        finish_tx.send(()).unwrap();
+        server.join().unwrap();
+        let retained = DurableStateStore::new(&directory).unwrap().load_or_create(&config).unwrap().0;
+        assert_eq!(retained.last_controller_command_seq, 0);
+        assert!(retained.processed_commands.is_empty());
+        std::fs::remove_dir_all(directory).unwrap();
+        return;
+    }
+    claimed_rx.recv_timeout(Duration::from_secs(5)).unwrap();
+    // A stale parked-only retirement cannot kill a newly claimed process.
+    assert_eq!(
+        host_exchange(
+            &mut host,
+            json!({"op":"retire","id":"stale-retire","parkId":"park-1"})
+        )["status"],
+        "conflict"
+    );
+    assert_eq!(
+        host_exchange(
+            &mut host,
+            json!({"op":"retire","id":"retire-claim","parkId":"park-1",
+        "claimId":"claim-1","identity":identity,"lastControllerCommandSeq":0,"ackedSourceSeq":0})
+        )["status"],
+        "retired"
+    );
+    assert!(runner.join().unwrap().is_ok());
+    finish_tx.send(()).unwrap();
+    server.join().unwrap();
+    assert_eq!(shutdowns.load(Ordering::SeqCst), 1);
+    assert_eq!(
+        DurableStateStore::new(&directory)
+            .unwrap()
+            .load_or_create(&config)
+            .unwrap()
+            .0
+            .lifecycle,
+        "suspended"
+    );
+    std::fs::remove_dir_all(directory).unwrap();
+}

packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs

@@ -1747,6 +1747,9 @@ fn hex_nibble(byte: u8) -> Result<u8, DurableRunnerError> {
 
 #[cfg(test)]
 mod tests {
+    #[cfg(unix)]
+    include!("tests/hosted_retention.rs");
+
     use std::net::{IpAddr, Ipv4Addr, TcpListener};
     use std::path::PathBuf;
     use std::sync::atomic::{AtomicUsize, Ordering};