diff --git a/packages/paperclip-runner/docs/phase-03-durable-transport.md b/packages/paperclip-runner/docs/phase-03-durable-transport.md index 1635dfd8f81..84ce2851caf 100644 --- a/packages/paperclip-runner/docs/phase-03-durable-transport.md +++ b/packages/paperclip-runner/docs/phase-03-durable-transport.md @@ -204,7 +204,12 @@ acknowledged. A `revoke` envelope persists the `revoked` lifecycle. The runner flushes any existing durable events and exits. It does not accept new work or delete -unacknowledged facts. +unacknowledged facts. If the revoked connection fails before the flush is +acknowledged, the runner exits unsuccessfully and records +`revocation_flush_requires_bootstrap`. A replacement process may use a new +one-time bootstrap only to replay the preserved outbox. It skips pending +commands and exits after the cumulative ACK empties the outbox. A restarted +revoked runner with no pending events stays network-silent. ## Diagnostics diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/phase3.rs b/packages/paperclip-runner/runner/crates/runner-core/src/phase3.rs index 7d4c3881f2d..baae80236b9 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/phase3.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/phase3.rs @@ -2080,14 +2080,14 @@ fn fail_revocation_flush( store: &DurableStateStore, reason: impl AsRef, ) -> Result<(), Phase3Error> { - state.unrecoverable_outcome = Some("revocation_flush_incomplete".to_owned()); + state.recoverable_failure = Some("revocation_flush_requires_bootstrap".to_owned()); state.record_diagnostic(format!( - "revocation flush incomplete; unacked durable events require operator recovery: {}", + "revocation flush interrupted; unacked durable events require a fresh bootstrap: {}", reason.as_ref() )); store.save(state)?; Err(Phase3Error::invalid( - "revocation flush incomplete; unacked durable events require operator recovery", + "revocation flush interrupted; unacked durable events require a fresh bootstrap", )) } @@ -2220,17 +2220,22 @@ pub fn run_durable_runner( } let store = DurableStateStore::new(&config.state_dir)?; let (mut state, recovered) = store.load_or_create(&config)?; - if recovered && state.lifecycle == "revoked" { - // Revocation is durable and terminal. Dropping the fresh bootstrap capability before - // destination resolution guarantees a restarted daemon cannot authenticate, reconnect, - // advance cursors, or rewrite the preserved operator-recovery state. + if recovered && state.lifecycle == "revoked" && state.outbox.is_empty() { + // A fully flushed revocation is durable and terminal. A revoked state with pending + // events may use a newly issued bootstrap only to finish delivery below. drop(bootstrap_ticket); return Ok(()); } // Resolve once before authentication. Reconnects use only this validated, // concrete address set, so DNS cannot redirect a retry. let target = ResolvedWsTarget::resolve(&config.connect_url)?; - if recovered { + if recovered && state.lifecycle == "revoked" { + state.reconnect_count += 1; + state.record_diagnostic( + "revoked runner restored with a fresh bootstrap for durable outbox flush only", + ); + store.save(&state)?; + } else if recovered { state.reconnect_count += 1; state.record_diagnostic("runner process restored the same durable identity"); enqueue_event( @@ -2375,15 +2380,17 @@ pub fn run_durable_runner( store.save(&state)?; let initial_delivery = (|| -> Result<(), Phase3Error> { - if let Some(commands) = payload.get("pendingCommands").and_then(Value::as_array) { - for command in commands { - match process_command(&mut state, &store, &config, command) { - Ok(processed) => { - client.send_json(&command_result_envelope(&state, &processed))? - } - Err(error) => { - state.record_diagnostic(error.to_string()); - store.save(&state)?; + if state.lifecycle != "revoked" { + if let Some(commands) = payload.get("pendingCommands").and_then(Value::as_array) { + for command in commands { + match process_command(&mut state, &store, &config, command) { + Ok(processed) => { + client.send_json(&command_result_envelope(&state, &processed))? + } + Err(error) => { + state.record_diagnostic(error.to_string()); + store.save(&state)?; + } } } } @@ -3119,7 +3126,7 @@ mod tests { } #[test] - fn failed_revocation_flush_requires_operator_recovery() { + fn failed_revocation_flush_requires_a_fresh_bootstrap() { let root = temporary_root("revocation-flush-failure"); let config = config(&root); let store = DurableStateStore::new(&root).unwrap(); @@ -3139,14 +3146,15 @@ mod tests { let error = fail_revocation_flush(&mut state, &store, "socket closed").unwrap_err(); - assert!(error.to_string().contains("operator recovery")); + assert!(error.to_string().contains("fresh bootstrap")); let (restored, recovered) = store.load_or_create(&config).unwrap(); assert!(recovered); assert_eq!(restored.lifecycle, "revoked"); assert_eq!( - restored.unrecoverable_outcome.as_deref(), - Some("revocation_flush_incomplete") + restored.recoverable_failure.as_deref(), + Some("revocation_flush_requires_bootstrap") ); + assert_eq!(restored.unrecoverable_outcome, None); assert_eq!(restored.outbox.len(), 1); assert!(restored .diagnostics diff --git a/packages/paperclip-runner/src/mock-core/phase3-recovery.test.ts b/packages/paperclip-runner/src/mock-core/phase3-recovery.test.ts index 395a90396f0..99e70b61dd4 100644 --- a/packages/paperclip-runner/src/mock-core/phase3-recovery.test.ts +++ b/packages/paperclip-runner/src/mock-core/phase3-recovery.test.ts @@ -700,7 +700,7 @@ describe.sequential("Phase 3 durable transport and recovery", () => { expect(trace.assertions.sourceCursorContinuous).toBe(true); }); - it("keeps a restarted revoked runner network-silent and byte-stable with a fresh ticket", async () => { + it("uses a fresh bootstrap only to flush a restarted revoked runner's durable outbox", async () => { const scratch = process.env.PAPERCLIP_RUN_SCRATCH_DIR ?? process.env.PAPERCLIP_SCRATCH_DIR ?? @@ -720,6 +720,10 @@ describe.sequential("Phase 3 durable transport and recovery", () => { sourceSeq: 4, eventType: "run.terminal", priority: 0, + sourceInstanceId: identity.runnerInstanceId, + runId: identity.runId, + normalizedSessionId: identity.normalizedSessionId, + turnId: identity.turnId, payload: { status: "succeeded" }, }, }; @@ -764,12 +768,13 @@ describe.sequential("Phase 3 durable transport and recovery", () => { ), { mode: 0o600 }, ); - const before = readFileSync(statePath); const core = new Phase3MockCore({ stateDirectory: resolve(root, "mock-core"), identity, fault: "none", }); + core.store.state.ackedSourceSeq = 3; + core.store.save(); try { await core.start(); const ticket = core.issueBootstrapTicket(); @@ -783,11 +788,23 @@ describe.sequential("Phase 3 durable transport and recovery", () => { }); expect(result.code, result.stderr).toBe(0); - expect(core.store.state.connectionCount).toBe(connectionsBefore); - expect(Object.values(core.store.state.tickets).map((record) => record.usedAt)).toEqual([ - null, + expect(core.store.state.connectionCount).toBe(connectionsBefore + 1); + expect(Object.values(core.store.state.tickets).every((record) => record.usedAt !== null)).toBe( + true, + ); + expect(core.store.state.committedEvents).toEqual([ + expect.objectContaining({ + sourceSeq: 4, + sourceEventId: `event_${identity.runnerInstanceId}_000004`, + eventType: "run.terminal", + }), ]); - expect(readFileSync(statePath)).toEqual(before); + expect(JSON.parse(readFileSync(statePath, "utf8"))).toMatchObject({ + lifecycle: "revoked", + ackedSourceSeq: 4, + outbox: [], + recoverableFailure: null, + }); } finally { await core.stop(); rmSync(root, { recursive: true, force: true });