fix(runner): retain exact ACPX checkpoints after rejected turn starts (#19762)

Preserve the exact idle ACPX conversation when a matching turn.start
response rejects work with retryable SESSION_RESUME_REQUIRED. Require
verified identity, no pending work, confirmed process cleanup and the
provider lifetime fence before normal drain and suspension can publish a
reusable checkpoint. Unknown acceptance and arbitrary closed states remain
closed.

The real Runner regression completes the next turn on attempts 2 and 3,
requires successful terminal status for the exact submitted turn, and
checks expectedIdentity on both resume requests. Controlled wire faults
reject failed, cancelled and missing-identity recovery.

Validation: 193 affected Rust tests on unchanged Rust source/fixtures;
134 recovery integration tests, 14 environment-gated skips; 21 Claude effort
tests; all server typecheck layers. Exact-head CI: 59 successful checks and
statuses, 6 intentional skips. Greptile Apex 5/5, no unresolved findings.
Hosted deployment and task acceptance remain separate.

Fixes: #19751
Refs: #19366
Refs: #19745

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Devin FoleyandPaperclip authored and GitHub committed 2026-11-06 23:45:35 -08:00
1 parent 7475577cdb
commit 2d4f80504e
7 files changed
+341 -14

No files matched your search

@@ -83,6 +83,22 @@ a scripted harness, removes the controller's launch files, then recovers the
same process and provider session and completes another turn. This exercises the
real transport; it does not qualify a hosted OpenCode or ACPX deployment.
## ACPX checkpoints after a rejected turn start
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
match, with no retained work event or active goal. A pending `session.resumed`
notification is retained for the normal drain; it does not start provider work. The Runner first reaps
the sidecar, then confirms the provider lifetime fence. It stores this boundary
as `prepared`, with exit confirmation, before the normal drain and suspension
make it reusable. The scheduled recovery keeps its existing attempt limit.
Unknown acceptance, malformed replies, other rejection codes, pending work and
unconfirmed cleanup do not qualify. Error messages cannot authorize recovery.
An arbitrary `closed` provider state stays closed. This path does not establish
recovery for a Runner killed during an accepted turn.
## Settled ACPX sessions across a Runner upgrade
Saved ACPX provider state records the digest of the launch profile that wrote
@@ -1900,15 +1900,54 @@ impl AcpxCommandExecutor {
.expect("ACPX session exists before turn start")
.start_turn_timed(&provider_turn_id, text, &working_directory, measure);
if let Err(error) = &start_result {
let rejected_checkpoint = self
.session
.as_ref()
.and_then(AcpxProviderSession::rejected_start_checkpoint)
.filter(|identity| {
self.state.as_ref().is_some_and(|state| {
state.identity.as_ref() == Some(*identity)
&& !state.provider_exit_unconfirmed
&& state.pending_runtime_requests.is_empty()
&& state
.pending_events
.iter()
.all(|event| event.event_type == "session.resumed")
&& (state.goal_projection.is_null()
|| goal_is_dormant(&state.goal_projection))
})
})
.cloned();
let state = self
.state
.as_mut()
.expect("ACPX state remains available after failed turn start");
state.lifecycle = "closed".to_owned();
state.lifecycle = if rejected_checkpoint.is_some() {
"prepared"
} else {
"closed"
}
.to_owned();
state.provider_exit_unconfirmed = rejected_checkpoint.is_some();
state.active_turn_id = None;
self.context.provider_turn_id = None;
self.session = None;
self.save_state()?;
if let Some(identity) = rejected_checkpoint {
let _provider_lifetime_fence =
await_provider_lifetime_fence(identity.provider_lifetime_fence_candidates)?;
self.state
.as_mut()
.expect("ACPX rejected checkpoint remains available")
.provider_exit_unconfirmed = false;
if let Err(error) = self.save_state() {
self.state
.as_mut()
.expect("ACPX rejected checkpoint remains available")
.provider_exit_unconfirmed = true;
return Err(error);
}
}
return Err(DurableRunnerError::invalid(format!(
"ACPX turn start failed closed: {error}"
)));
@@ -350,6 +350,7 @@ pub struct AcpxProviderSession {
closed: bool,
failure: Option<LocalRunnerError>,
transport_terminated: bool,
rejected_start_resumable: bool,
runtime_retired: bool,
completed_turn_observed: bool,
}
@@ -407,6 +408,7 @@ impl AcpxProviderSession {
closed: false,
failure: None,
transport_terminated: false,
rejected_start_resumable: false,
runtime_retired: false,
completed_turn_observed: false,
})
@@ -424,6 +426,10 @@ impl AcpxProviderSession {
&self.identity
}
pub(crate) fn rejected_start_checkpoint(&self) -> Option<&AcpxProviderSessionIdentity> {
self.rejected_start_resumable.then_some(&self.identity)
}
pub fn turn_control_capabilities(&self) -> AcpxTurnControlCapabilities {
self.turn_controls
}
@@ -598,6 +604,7 @@ impl AcpxProviderSession {
working_directory: &Path,
performance: bool,
) -> Result<Value, LocalRunnerError> {
self.rejected_start_resumable = false;
self.ensure_open()?;
if self.runtime_retired {
return Err(LocalRunnerError::invalid(
@@ -686,12 +693,19 @@ impl AcpxProviderSession {
"ACPX reserved tool receipt rotation failed: {error}"
))));
}
let response = match self.transport.request(
GeneratedAcpxSidecarCommand::TurnStart,
let response = match self.transport.request_turn_start(
json!({"turnId":turn_id,"message":message,"performance":performance}),
) {
Ok(response) => response,
Err(error) => return Err(self.fail_closed(error)),
Err(failure) => {
let resumable = failure.resume_required
&& self.state.active_turn_id().is_none()
&& !self.state.has_pending_requests()
&& self.transport.buffered_events().len() == 0;
let error = self.fail_closed(failure.error);
self.rejected_start_resumable = resumable && self.transport_terminated;
return Err(error);
}
};
if response.get("turnId").and_then(Value::as_str) != Some(turn_id) {
return Err(self.fail_closed(LocalRunnerError::invalid(
@@ -2227,6 +2241,7 @@ mod mode_tests {
closed: false,
failure: None,
transport_terminated: false,
rejected_start_resumable: false,
runtime_retired: false,
completed_turn_observed: false,
}
@@ -76,6 +76,11 @@ pub struct AcpxSidecarEvent {
pub payload: Value,
}
pub(crate) struct AcpxTurnStartFailure {
pub error: LocalRunnerError,
pub resume_required: bool,
}
pub struct AcpxSidecarTransport {
process: SupervisedProcess,
request_timeout: Duration,
@@ -415,6 +420,37 @@ impl AcpxSidecarTransport {
command: GeneratedAcpxSidecarCommand,
params: Value,
) -> Result<Value, LocalRunnerError> {
match self.request_outcome(command, params)? {
CommandOutcome::Success(value) => Ok(value),
CommandOutcome::Rejected { error, .. } => Err(error),
}
}
pub(crate) fn request_turn_start(
&mut self,
params: Value,
) -> Result<Value, AcpxTurnStartFailure> {
match self.request_outcome(GeneratedAcpxSidecarCommand::TurnStart, params) {
Ok(CommandOutcome::Success(value)) => Ok(value),
Ok(CommandOutcome::Rejected {
error,
resume_required,
}) => Err(AcpxTurnStartFailure {
error,
resume_required,
}),
Err(error) => Err(AcpxTurnStartFailure {
error,
resume_required: false,
}),
}
}
fn request_outcome(
&mut self,
command: GeneratedAcpxSidecarCommand,
params: Value,
) -> Result<CommandOutcome, LocalRunnerError> {
if self.poisoned {
return Err(LocalRunnerError::invalid(
"ACPX sidecar transport is unavailable after a protocol failure",
@@ -422,8 +458,7 @@ impl AcpxSidecarTransport {
}
let result = self.request_inner(command, params);
match result {
Ok(CommandOutcome::Success(value)) => Ok(value),
Ok(CommandOutcome::Rejected(error)) => Err(error),
Ok(outcome) => Ok(outcome),
Err(error) => {
self.poison();
Err(error)
@@ -517,7 +552,7 @@ impl AcpxSidecarTransport {
});
match result {
Ok(Some(CommandOutcome::Success(value))) => Ok(Some(value)),
Ok(Some(CommandOutcome::Rejected(_))) => Ok(None),
Ok(Some(CommandOutcome::Rejected { .. })) => Ok(None),
Ok(None) => Ok(None),
Err(error) => {
self.poison();
@@ -623,14 +658,17 @@ impl AcpxSidecarTransport {
}
let error = response.error.expect("failed response has validated error");
let provider_failure = error.provider_failure.clone();
return Ok(Some(CommandOutcome::Rejected(LocalRunnerError::invalid(
let resume_required = command == GeneratedAcpxSidecarCommand::TurnStart
&& error.retryable
&& error.code == "SESSION_RESUME_REQUIRED";
return Ok(Some(CommandOutcome::Rejected { error: LocalRunnerError::invalid(
format!(
"ACPX sidecar command {} was rejected (retryable={}, classification={})",
command.as_str(),
error.retryable,
response_error_classification(&error),
),
).with_provider_failure(provider_failure))));
).with_provider_failure(provider_failure), resume_required }));
}
}
}
@@ -823,7 +861,10 @@ impl AcpxSidecarTransport {
enum CommandOutcome {
Success(Value),
Rejected(LocalRunnerError),
Rejected {
error: LocalRunnerError,
resume_required: bool,
},
}
enum ParsedFrame {
@@ -1154,6 +1195,30 @@ fn response_error_classification(error: &ResponseError) -> &'static str {
#[cfg(test)]
mod tests {
#[test]
#[cfg(unix)]
fn a_timed_out_turn_start_never_establishes_a_resumable_rejection() {
let mut transport = AcpxSidecarTransport::start(&AcpxSidecarTransportConfig {
command: "/bin/sh".into(),
args: vec!["-c".to_owned(), "read request; sleep 10".to_owned()],
verified_launch: None,
request_timeout: Duration::from_millis(30),
shutdown_grace: Duration::from_millis(20),
})
.unwrap();
let failure = transport
.request_turn_start(json!({"turnId":"turn-1"}))
.unwrap_err();
assert!(!failure.resume_required);
assert!(failure.error.to_string().contains("timed out"));
let reuse = transport
.request_turn_start(json!({"turnId":"turn-2"}))
.unwrap_err();
assert!(!reuse.resume_required);
assert!(reuse.error.to_string().contains("unavailable"));
transport.shutdown().unwrap();
}
#[test]
fn pi_credentials_require_a_bounded_binding_without_process_control_variables() {
assert!(pi_credential_environment_keys(None).unwrap().is_empty());
@@ -99,6 +99,39 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
.get("command")
.and_then(Value::as_str)
.ok_or("request command is missing")?;
if command == "turn.start" {
if let Some(fault) = args
.windows(2)
.find(|pair| pair[0] == "--reject-turn-start")
{
let fault = fault[1].as_str();
if fault == "event" {
write_turn_event(
&mut stdout,
next_sequence,
"runtime.event",
"run-1",
request["params"]["turnId"].as_str().unwrap(),
json!({"type":"text_delta","text":"Work may have started."}),
)?;
next_sequence += 1;
}
let code = match fault {
"unknown" | "message-only" => "UNKNOWN",
"lowercase" => "session_resume_required",
_ => "SESSION_RESUME_REQUIRED",
};
let mut error = json!({"protocolVersion":GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION,
"id":if fault == "wrong-id" {id + 1} else {id},"ok":false,
"error":{"code":code,"message":"SESSION_RESUME_REQUIRED retryable=true classification=session_resume_required",
"retryable":fault != "permanent"}});
if fault == "malformed" {
error["error"]["retryable"] = json!("true");
}
write_json(&mut stdout, &error)?;
continue;
}
}
if let Some(journal) = source_journal.as_mut() {
write_json(journal, &request)?;
if command == "run.attach" {
@@ -3099,6 +3099,141 @@ fn idle_pi_stop_cannot_publish_a_reusable_identity_while_its_lifetime_fence_is_h
fs::remove_dir_all(directory).unwrap();
}
fn rejected_turn_start_fixture(directory: &Path, fault: &str) -> (DurableRunnerConfig, Value) {
let mut config = acpx_config(directory, "bootstrap");
config
.acpx_launch_profile
.as_mut()
.unwrap()
.args
.extend(["--reject-turn-start".to_owned(), fault.to_owned()]);
let mut payload = prepare_payload_with_mode(directory, "codex", "bootstrap");
payload["provider"]["sidecarArgs"] = json!(config.acpx_launch_profile.as_ref().unwrap().args);
(config, payload)
}
#[test]
#[cfg_attr(windows, ignore = "verified launch is not supported on Windows yet")]
fn rejected_turn_start_retains_only_the_definite_retryable_checkpoint() {
for fault in [
"resume",
"permanent",
"unknown",
"lowercase",
"message-only",
"event",
"wrong-id",
"malformed",
] {
let directory = temporary_directory(&format!("rejected-start-{fault}"));
let (config, payload) = rejected_turn_start_fixture(&directory, fault);
let mut executor = NativeProviderCommandExecutor::with_runner_config(&directory, &config);
executor
.execute(&command(1, "run.prepare", payload))
.unwrap();
executor
.execute(&command(2, "session.open", json!({})))
.unwrap();
let path = directory.join("acpx-provider-state.json");
let opened: Value = serde_json::from_slice(&fs::read(&path).unwrap()).unwrap();
executor
.execute(&command(
3,
"turn.start",
json!({"turnId":"rejected-turn", "text":"Continue saved work."}),
))
.unwrap_err();
let rejected: Value = serde_json::from_slice(&fs::read(&path).unwrap()).unwrap();
assert_eq!(
rejected["lifecycle"],
if fault == "resume" {
"prepared"
} else {
"closed"
},
"{fault}"
);
assert_eq!(rejected["providerExitUnconfirmed"], false, "{fault}");
assert_eq!(rejected["identity"], opened["identity"], "{fault}");
assert_eq!(rejected["activeTurnId"], Value::Null, "{fault}");
executor
.execute(&command(4, "runner.drain", json!({})))
.unwrap();
executor
.execute(&command(5, "runner.suspend", json!({})))
.unwrap();
let suspended: Value = serde_json::from_slice(&fs::read(&path).unwrap()).unwrap();
assert_eq!(
suspended["lifecycle"],
if fault == "resume" {
"suspended"
} else {
"closed"
},
"{fault}"
);
executor.shutdown().unwrap();
fs::remove_dir_all(directory).unwrap();
}
}
#[test]
#[cfg_attr(windows, ignore = "verified launch is not supported on Windows yet")]
fn rejected_turn_start_cannot_publish_a_checkpoint_while_its_provider_lifetime_is_held() {
let directory = temporary_directory("rejected-start-held-lifetime");
let fence = (61_000..=u16::MAX)
.filter_map(|port| std::net::TcpListener::bind(("127.0.0.1", port)).ok())
.take(3)
.collect::<Vec<_>>();
assert_eq!(fence.len(), 3);
let ports = fence
.iter()
.map(|listener| listener.local_addr().unwrap().port().to_string())
.collect::<Vec<_>>()
.join(",");
let (mut config, mut payload) = rejected_turn_start_fixture(&directory, "resume");
config
.acpx_launch_profile
.as_mut()
.unwrap()
.args
.extend(["--lifetime-fence-ports".to_owned(), ports]);
payload["provider"]["sidecarArgs"] = json!(config.acpx_launch_profile.as_ref().unwrap().args);
let mut executor = NativeProviderCommandExecutor::with_runner_config(&directory, &config);
executor
.execute(&command(1, "run.prepare", payload))
.unwrap();
executor
.execute(&command(2, "session.open", json!({})))
.unwrap();
let error = executor
.execute(&command(
3,
"turn.start",
json!({"turnId":"rejected-turn", "text":"Continue saved work."}),
))
.unwrap_err();
assert!(
error
.to_string()
.contains("original provider lifetime remains active"),
"{error}"
);
let path = directory.join("acpx-provider-state.json");
let state: Value = serde_json::from_slice(&fs::read(&path).unwrap()).unwrap();
assert_eq!(state["lifecycle"], "prepared");
assert_eq!(state["providerExitUnconfirmed"], true);
assert!(executor
.execute(&command(4, "runner.suspend", json!({})))
.is_err());
assert!(executor.poll_events().unwrap().is_empty());
let after_poll: Value = serde_json::from_slice(&fs::read(&path).unwrap()).unwrap();
assert_eq!(after_poll, state);
drop(fence);
executor.shutdown().unwrap();
fs::remove_dir_all(directory).unwrap();
}
#[test]
#[cfg_attr(windows, ignore = "verified launch is not supported on Windows yet")]
fn publishes_only_changed_pi_controls_before_the_new_turn() {
@@ -191,12 +191,21 @@ process.on("SIGTERM", () => child.kill("SIGTERM"));
bundles.push(bundle);
return bundle;
};
const expectCompleted = async (bundle: ReturnType<typeof create>) => {
const expectCompleted = async (bundle: ReturnType<typeof create>, expectedTurnId: string) => {
expect(expectedTurnId).toEqual(expect.any(String));
expect(expectedTurnId).not.toBe("");
let timer: ReturnType<typeof setTimeout> | undefined;
try {
await Promise.race([
(async () => {
for await (const event of bundle.transport.notifications()) if (event.method === "turn/completed") return;
for await (const event of bundle.transport.notifications()) {
if (event.method !== "turn/completed") continue;
const turn = event.params?.turn as { id?: string; status?: string; error?: unknown } | undefined;
if (turn?.id !== expectedTurnId) continue;
expect(turn).toMatchObject({ id: expectedTurnId, status: "completed" });
expect(turn?.error ?? null).toBeNull();
return;
}
throw new Error("Fixture turn ended without completion");
})(),
new Promise<never>((_, reject) => { timer = setTimeout(() => reject(new Error("Fixture turn completion timed out")), 5_000); }),
@@ -208,8 +217,8 @@ process.on("SIGTERM", () => child.kill("SIGTERM"));
const seed = create(seedDirectory);
const opened = await seed.transport.request("thread/start", { cwd: root, model: "fixture-model", dynamicTools: [] });
const thread = opened.thread as { id: string; sessionId: string };
await seed.transport.request("turn/start", { input: [{ type: "text", text: "Save the provider checkpoint." }] });
await expectCompleted(seed);
const seedStarted = await seed.transport.request("turn/start", { input: [{ type: "text", text: "Save the provider checkpoint." }] });
await expectCompleted(seed, (seedStarted.turn as { id: string }).id);
await seed.transport.close();
const provider = JSON.parse(await readFile(join(seedDirectory, "runner", "acpx-provider-state.json"), "utf8"));
const resume = { driverSessionId: thread.id, providerSessionId: thread.sessionId, providerIdentity: provider.identity };
@@ -230,6 +239,8 @@ process.on("SIGTERM", () => child.kill("SIGTERM"));
`fixture_nonce\n${pid}\n${new Date().toISOString()}\n${identity.runnerInstanceId}\n${fingerprint}\n`);
await expect(fresh.transport.request("turn/start", { input: [{ type: "text", text: "Resume saved work." }] }))
.rejects.toThrow("classification=session_resume_required");
const rejectedCheckpoint = JSON.parse(await readFile(join(stateDirectory, "runner", "acpx-provider-state.json"), "utf8"));
expect(rejectedCheckpoint).toMatchObject({ lifecycle: "prepared", providerExitUnconfirmed: false, identity: provider.identity });
await fresh.transport.close();
await vi.waitFor(() => expect(child.exitCode).not.toBeNull(), { timeout: 5_000 });
const resumedControl = JSON.parse(await readFile(join(stateDirectory, "control-plane", "control-plane-state.json"), "utf8"));
@@ -239,6 +250,9 @@ process.on("SIGTERM", () => child.kill("SIGTERM"));
expect(saved.pendingTerminalDelivery ?? null).toBeNull();
expect(saved.pendingProviderCleanup ?? null).toBeNull();
expect(saved.warmTransition ?? null).toBeNull();
const suspendedCheckpoint = JSON.parse(await readFile(join(stateDirectory, "runner", "acpx-provider-state.json"), "utf8"));
expect(suspendedCheckpoint).toMatchObject({ lifecycle: "suspended", providerExitUnconfirmed: false,
activeTurnId: null, pendingEvents: [], pendingRuntimeRequests: {}, identity: provider.identity });
const runner = { execute: async (request: { command: string; args: string[]; timeoutMs?: number }) =>
new Promise<{ stdout: string; stderr: string; exitCode: number; timedOut: boolean }>(resolveResult => {
execFile(request.command, request.args, { timeout: request.timeoutMs }, (error, stdout, stderr) => resolveResult({
@@ -256,6 +270,16 @@ process.on("SIGTERM", () => child.kill("SIGTERM"));
const retried = create(stateDirectory, resume, adopted);
await retried.transport.request("thread/read", { threadId: thread.id, includeTurns: false });
await retried.transport.request("thread/resume", { threadId: thread.id, cwd: root, model: "fixture-model", dynamicTools: [] });
const retryStarted = await retried.transport.request("turn/start", { input: [{ type: "text", text: "Continue after the bounded retry." }] });
const retryTurnId = (retryStarted.turn as { id: string }).id;
await expectCompleted(retried, retryTurnId);
const calls = (await readFile(journal, "utf8")).trim().split("\n").map(line => JSON.parse(line));
const turnStarts = calls.filter(call => call.command === "turn.start");
expect(turnStarts).toHaveLength(3);
expect(new Set(turnStarts.map(call => call.params.turnId)).size).toBe(3);
expect(turnStarts.at(-1).params.turnId).toBe(retryTurnId);
expect(calls.filter(call => call.command === "session.open").map(call => call.params.expectedIdentity))
.toEqual([null, provider.identity, provider.identity]);
expect(launches.mock.calls).toEqual([[undefined], [undefined], [undefined]]);
expect(retried.evidence().runnerPid).toBeGreaterThan(0);
expect(retried.evidence().runnerPid).not.toBe(pid);