diff --git a/packages/paperclip-runner/docs/durable-recovery.md b/packages/paperclip-runner/docs/durable-recovery.md index 63e25852e53..617398eec88 100644 --- a/packages/paperclip-runner/docs/durable-recovery.md +++ b/packages/paperclip-runner/docs/durable-recovery.md @@ -92,6 +92,12 @@ returns as soon as the drain is proved; it adds no fixed delay to a ready Runner ## ACPX checkpoints after a rejected turn start +Recovered ACPX shutdown uses the same bounded five-second lifetime-fence proof +as suspension. It waits for a transient listener holder to release the original +quorum, then returns immediately when cleanup is proved. A timeout leaves +`providerExitUnconfirmed` set and refuses reuse. Live-session shutdown and normal +task dispatch keep their existing paths. + A matching `turn.start` response with `SESSION_RESUME_REQUIRED` and `retryable: true` can retain the exact saved conversation. The session must have no active turn, pending request or buffered event. The saved identity must still diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs index da20602953d..6797bdcb6d0 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs @@ -115,16 +115,6 @@ fn try_acquire_provider_lifetime_fence( Ok(None) } -fn acquire_provider_lifetime_fence( - candidates: [u16; 3], -) -> Result, DurableRunnerError> { - try_acquire_provider_lifetime_fence(candidates)?.ok_or_else(|| { - DurableRunnerError::invalid( - "ACPX original provider lifetime remains active; cleanup is not yet proven", - ) - }) -} - fn turn_start_request(payload: &Value) -> Result<(&str, String), DurableRunnerError> { let text = payload .get("text") @@ -153,7 +143,7 @@ fn await_provider_lifetime_fence( } if Instant::now() >= deadline { return Err(DurableRunnerError::invalid( - "ACPX original provider lifetime remains active after suspension; provider exit is not confirmed", + "ACPX original provider lifetime remains active; provider exit is not confirmed", )); } std::thread::sleep(PROVIDER_LIFETIME_CONFIRMATION_RETRY); @@ -3166,7 +3156,7 @@ impl CommandExecutor for AcpxCommandExecutor { .and_then(|state| state.identity.as_ref()) .expect("provider cleanup state has a validated identity") .provider_lifetime_fence_candidates; - Some(acquire_provider_lifetime_fence(candidates)?) + Some(await_provider_lifetime_fence(candidates)?) } else { None }; @@ -5611,8 +5601,28 @@ mod tests { assert!(attach_error .to_string() .contains("requires the same settled ACPX provider profile and session")); + let (started_tx, started_rx) = std::sync::mpsc::channel(); + let (shutdown_tx, shutdown_rx) = std::sync::mpsc::channel(); + let cleanup = std::thread::spawn(move || { + started_tx.send(()).unwrap(); + let result = executor.shutdown(); + shutdown_tx.send((executor, result)).unwrap(); + }); + started_rx.recv_timeout(Duration::from_secs(1)).unwrap(); + assert!( + matches!( + shutdown_rx.recv_timeout(Duration::from_millis(100)), + Err(std::sync::mpsc::RecvTimeoutError::Timeout) + ), + "shutdown must wait while the original lifetime still holds a quorum" + ); drop(original_lifetime_fence); - executor.shutdown().unwrap(); + let (mut executor, result) = shutdown_rx + .recv_timeout(PROVIDER_LIFETIME_CONFIRMATION_TIMEOUT + Duration::from_secs(1)) + .unwrap(); + cleanup.join().unwrap(); + result.unwrap(); + assert!(executor.session.is_none()); let recovered = executor.state.as_ref().unwrap(); assert_eq!(recovered.lifecycle, "prepared"); assert!(!recovered.provider_exit_unconfirmed); @@ -5640,7 +5650,7 @@ mod tests { "another fixture must not impersonate a retired provider lifetime" ); assert_eq!( - acquire_provider_lifetime_fence(original_candidates) + await_provider_lifetime_fence(original_candidates) .expect("unrelated live fixture must not block the original cleanup proof") .len(), 2 diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/native_provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/native_provider_backend.rs index d6ccb98ee10..a8922c17568 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/native_provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/native_provider_backend.rs @@ -605,42 +605,7 @@ fn assert_no_replayed_turn_or_request(events: &[PolledEvent]) { } fn shutdown_recovered_acpx_fixture(executor: &mut NativeProviderCommandExecutor, directory: &Path) { - // The fake sidecar advertises the same three lifetime-fence ports for all - // fixtures. Parallel recovered shutdowns hold a quorum through state fsync, - // temporarily preventing another fixture from proving cleanup. Retry only - // that rejection, using the real production quorum proof on every attempt. - // This is fixture synchronization, not a production cleanup workaround. - let deadline = std::time::Instant::now() + Duration::from_secs(5); - let mut blocked_attempts = 0; - loop { - match executor.shutdown() { - Ok(()) => break, - Err(error) => { - assert_eq!( - error.to_string(), - "ACPX original provider lifetime remains active; cleanup is not yet proven", - "unexpected recovered fixture shutdown failure" - ); - let persisted: Value = serde_json::from_slice( - &fs::read(directory.join("acpx-provider-state.json")).unwrap(), - ) - .unwrap(); - assert_eq!(persisted["providerExitUnconfirmed"], true); - blocked_attempts += 1; - if blocked_attempts == 1 { - eprintln!("recovered fixture cleanup waiting for the shared lifetime quorum"); - } - assert!( - std::time::Instant::now() < deadline, - "recovered fixture cleanup never acquired the lifetime quorum" - ); - std::thread::sleep(Duration::from_millis(10)); - } - } - } - if blocked_attempts > 0 { - eprintln!("recovered fixture cleanup proved after {blocked_attempts} quorum rejections"); - } + executor.shutdown().unwrap(); let persisted: Value = serde_json::from_slice(&fs::read(directory.join("acpx-provider-state.json")).unwrap()) .unwrap();