diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index d14ce0d70a3..e774ef88914 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1226,6 +1226,12 @@ Blocked readiness probes back off within the reconnect deadline so they do not fill the durable command journal while waiting. The fast ready path keeps its short second barrier. If readiness never arrives, attachment fails closed with the last observed blocker; a full journal is not a substitute for that diagnosis. +Both native providers publish this readiness contract. ACPX reports its durable +session identity, active turn, pending audit events, closed state, and unproven +provider exit as blockers. Explicit readiness probes let the durable runner +commit and acknowledge retained events under the old run authority; snapshotting +alone never discards them. ACPX checkpoints its process during the subsequent +attachment before resuming the same provider session under the new run. Safe native replacement may clear a Blocked status only with a durable receipt that the same failed run projected that exact status version. Explicitly 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 6805294ec00..a093f564c43 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 @@ -1260,6 +1260,27 @@ impl AcpxCommandExecutor { .state .as_ref() .ok_or_else(|| DurableRunnerError::invalid("ACPX provider has not been prepared"))?; + // The shared transport waits for an explicit readiness contract before + // rotating run authority. ACPX checkpoints its process during attach, + // but must first settle the same durable guards enforced by attach_run. + // Do not consume events here: the durable runner commits and ACKs the + // old authority's retained prefix before a later probe can report ready. + let mut warm_attach_blockers = Vec::new(); + if state.lifecycle == "closed" { + warm_attach_blockers.push("durable_closed"); + } + if state.provider_exit_unconfirmed { + warm_attach_blockers.push("provider_exit_unconfirmed"); + } + if state.identity.is_none() { + warm_attach_blockers.push("provider_identity_unavailable"); + } + if state.active_turn_id.is_some() { + warm_attach_blockers.push("durable_active_turn"); + } + if !state.pending_events.is_empty() { + warm_attach_blockers.push("durable_pending_events"); + } Ok(CommandExecution::result(json!({ "status": state.lifecycle, "provider": "acpx", @@ -1270,6 +1291,8 @@ impl AcpxCommandExecutor { "providerAccountSessionId": state.identity.as_ref().map(|value| value.agent_session_id.as_str()), "providerIdentity": state.identity, "activeProviderTurnId": state.active_turn_id, + "warmAttachReady": warm_attach_blockers.is_empty(), + "warmAttachBlockers": warm_attach_blockers, }))) } @@ -2099,6 +2122,59 @@ mod tests { let original_config = test_config(&directory, Some(launch_profile.clone())); let mut original = AcpxCommandExecutor::with_runner_config(&directory, &original_config); original.state = Some(state); + let settled = original.state.clone().unwrap(); + let ready = original.snapshot().unwrap().result; + assert_eq!(ready["warmAttachReady"], true); + assert_eq!(ready["warmAttachBlockers"], json!([])); + let blocked_states: [(&str, fn(&mut AcpxDurableState)); 4] = [ + ("durable_closed", |state| state.lifecycle = "closed".to_owned()), + ("provider_exit_unconfirmed", |state| { + state.lifecycle = "prepared".to_owned(); + state.provider_exit_unconfirmed = true; + }), + ("provider_identity_unavailable", |state| { + state.lifecycle = "prepared".to_owned(); + state.identity = None; + }), + ("durable_active_turn", |state| { + state.lifecycle = "turn_active".to_owned(); + state.active_turn_id = Some("turn-1".to_owned()); + }), + ]; + for (blocker, mutate) in blocked_states { + original.state = Some(settled.clone()); + mutate(original.state.as_mut().unwrap()); + let snapshot = original.snapshot().unwrap().result; + assert_eq!(snapshot["warmAttachReady"], false, "{blocker}"); + assert_eq!(snapshot["warmAttachBlockers"], json!([blocker])); + } + // A readiness probe must retain the old authority's audit events until + // the durable runner commits and acknowledges them, including recovery + // notices that run.attach itself is allowed to consume. + for event_type in ["session.resumed", "harness.diagnostic"] { + original.state = Some(settled.clone()); + original + .state + .as_mut() + .unwrap() + .push(NormalizedProviderEvent { + event_type: event_type.to_owned(), + priority: EventPriority::P0, + payload: json!({}), + }) + .unwrap(); + let snapshot = original.snapshot().unwrap().result; + assert_eq!(snapshot["warmAttachReady"], false); + assert_eq!( + snapshot["warmAttachBlockers"], + json!(["durable_pending_events"]) + ); + assert_eq!(original.retained_events().unwrap().len(), 1); + original.acknowledge_events(1).unwrap(); + assert_eq!(original.snapshot().unwrap().result["warmAttachReady"], true); + } + original.state = Some(settled); + assert!(!marker.exists(), "readiness must not start a provider"); original.save_state().unwrap(); let mut wrong_session_config = original_config.clone();