fix: publish ACPX warm attachment readiness

Report durable attachment blockers so Claude follow-ups can reuse a settled runner without exhausting the command journal. Preserve the retained event acknowledgement barrier and cover missing, blocked, and drained snapshots.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

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

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();