From 5f61392cd634db1138ea8748bd3922660eb72d1a Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Thu, 5 Nov 2026 08:03:16 -0800 Subject: [PATCH] Reconcile dormant Codex predecessors through ordinary recovery (#19132) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The native Runner keeps provider and controller journals between turns. > - A process can stop after a completed turn while those journals remain valid. > - The next owner must reconcile that source before it starts new work. > - This pull request reuses ordinary recovery and authenticated run attachment. > - Existing journals and owner admission keep retained work under its source. > - The benefit is conversation recovery without another snapshot format or native mode. ## Linked Issues or Issue Description Refs #19057. Built on merged #19129. This is the Codex recovery transport foundation. Process selection and controller scheduling remain separate changes. On normal startup, `paperclip-runnerd` selects durable transport from its connection arguments. With `--connect-url`, the Runner dials the controller. With `--listen-address 0.0.0.0 --listen-path /api/runner/v1/connect/` and optional `--listen-port` (default `43127`), the controller dials the Runner. Both use the same durable entry point for fresh launches and recovery. The bootstrap ticket authenticates startup, and `recoveryIntegrityVersion` advertises a recovery capability. ## What Changed - Observe the existing recovered command journal before provider restoration. A saved dormant Codex session with no uncertain effectful command must still be dormant after restore. Ordinary active same-run recovery keeps its existing behavior. - Preserve prepared providers during snapshots, drain and cleanup. New completed-predecessor admission excludes prepared state. Existing cleanup and ordinary continuation keep their established behavior. - Advertise `recoveryIntegrityVersion: 1`. Reject the removed `--source-reconcile` flag. No separate native source mode or command whitelist remains. - Derive controller admission from the authenticated `authorityRunId` and actual Runner identity. Reuse retain-only settlement through attachment, authentication and checkpoint persistence. Preserve dynamic ownership across home configuration and later warm reuse. - Require the selected execution driver's trusted `sourceRecoveryVersion: 1` prelaunch attestation for a stopped predecessor. Binary metadata alone cannot authorize that launch. Ordinary live adoption keeps its existing path. - Support fresh authorized preparation from `recoveryLaunch: { requirements }` without replacing the source's ordinary cache. Require an authenticated owner throughout that transport's lifetime. Preserve immutable recorded policy through target preparation and later warm attachment; compare explicit settings using the provider factory's effective defaults. - Complete interrupted attachment at the existing fresh-snapshot boundary: persist the checkpoint, recheck current ownership, complete the external recovery callback, then release a real predecessor dispatch hold. Keep callback/checkpoint failures retryable and cancellation cleanup retained. - Preserve completed-session event delivery after target attachment, exact-thread history projections, source registration identity, exact command replay and nullable usage redaction. - Recheck current ownership and the queued run when a slow notification consumer presents a permission request. Suppress stale presentation while preserving the committed event and pending request for source-owner settlement. ## Verification Current head `e90bb379178db75f78e440c92e108f0bbe01e073` merges master `499712c68b3c` and is fully verified. [PR CI](https://github.com/paperclipai/paperclip/actions/runs/42058217937), [Runner Windows](https://github.com/paperclipai/paperclip/actions/runs/42058217814), and [Docker Runner](https://github.com/paperclipai/paperclip/actions/runs/42058217365) all passed. [Standard Greptile review 18](https://github.com/paperclipai/paperclip/pull/19132#issuecomment-6645302809) scores this exact head 5/5 with no unresolved threads. The earlier Apex review 16 applies to `c09ebb4a2a6c`, as recorded below; no new Apex review was requested after the documentation or master-forward changes. The conflict integration preserves upstream external cold-state restoration before completed-handoff inspection, predecessor ownership and pending-command checks, and the lost-source early-read exclusion. Live adoption still does not restore a backup. All 43 focused process cases passed across cold restore (including denied/foreign state), staged handoff, ordinary adoption and source recovery. Runner `tsc --noEmit`, parent-relative diff checks and added-line secret/personal-data scan passed. [Read the rewritten guide](https://github.com/paperclipai/paperclip/blob/e90bb379178db75f78e440c92e108f0bbe01e073/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md#recovering-codex-between-turns): it explains dial and listener startup, then follows completed run A through ordinary recovery and authenticated attachment to run B. The guide was rewritten in documentation commit `75db893fb9ab` and is unchanged by the master-forward. Its listener CLI test and both goal-update process tests passed. Earlier feature head `c09ebb4a2a` includes a process regression for a goal notification arriving between the restored goal response and the final thread response. Before the fix the Runner admitted the provider. The new tests prove that an active goal refuses admission and reaps the provider, and that a paused goal is delivered normally. Both process tests and five adjacent ordinary recovery cases pass locally. GitHub CI passed on the first attempt, including the main, Windows, and Docker workflows. Standard review 15 and Apex review 16 each scored this commit 5/5, with zero unresolved threads. [GitHub CI](https://github.com/paperclipai/paperclip/actions/runs/41991764330) · [Greptile review](https://github.com/paperclipai/paperclip/pull/19132#issuecomment-6645302809) - The final ownership-race checks pass 7 focused cases, including owner changes during checkpoint persistence, handler publication and saved receipt recovery. - Native `codex_provider ordinary_recovery`: 6 passed, including saved/live mismatch, pending and indeterminate same-run goals, settled idle restoration and prepared reads/cleanup without a provider restart. - Native `paperclip-runnerd tests::`: 9 passed, including metadata and legacy flag rejection. - Native `interrupted_result_delivery_reconciles_exact_receipts_without_reexecuting_operations`: passed. - Final focused compatibility run: 24 process cases passed, including all 19 retained-settlement rows with queued snapshots, both home variants of A-to-B-to-C, ordinary attachment/suspension and environment-prepared close. - Existing `recovers an active second warm business run after deleting its controller context assets`: passed. - Runner `tsc --noEmit`, `git diff --check`, and added-line secret and personal-data scans passed. - Real native processes with the maintained fake Codex provider prove abrupt A exit, ordinary recover A, same-process attach B, a second successful turn, ordinary warm attach C and a third successful turn on the same thread. Both isolated and preconfigured homes pass. The isolated case also exercises environment option copying. - Additional process cases cover cacheless preparation, same-PID controller restart adoption, revoked preparation, retained tool-input admission, direct attachment, ordinary suspension and unexpected live goals. - A prior revised-head CI run exposed 8 retained-cleanup failures caused by prepared auto-restoration. That new behavior and its journal-binding code are removed. All 19 affected settlement variants now pass locally. The same CI run also had one environment-prepared close failure (focused check passes) and two unchanged Cloud sign-in E2E heading timeouts. Original logs remain recorded. Full suites and platform checks run in CI. Previous head `16772ad3e1abfdae9abcc64f8221fa3644ecee58`: [PR workflow 41981127140](https://github.com/paperclipai/paperclip/actions/runs/41981127140), [Runner Windows 41981126926](https://github.com/paperclipai/paperclip/actions/runs/41981126926), and [Docker Runner 41981126975](https://github.com/paperclipai/paperclip/actions/runs/41981126975) passed. Greptile review 14 scores this exact head 5/5 with no unresolved findings. The PR remains draft. CI evidence: an unchanged `idle-completed-history` local HTTP readiness case failed in the initial server 11/12 job (127092367616) and its first targeted retry (127096952128). The isolated case and the full 75-case file passed locally. The same-head debug retry (127103567635) passed, and the full workflow is green. Original failures remain recorded. No fixture diagnostic, assertion change, or timeout change was pushed. Focused review-fix validation: - 57 process cases passed across recovery, attachment and all 19 retained-cleanup variants. - Final staged matrix: 5 cases, each exercising queued/prepared/activated/completed boundaries, across owned connect/listen, legacy no-owner connect/listen, and retained cancellation cleanup. - Four completion-failure cases pass: synchronous/asynchronous callback failure, checkpoint failure before any callback, and owner rebind during persistence before any callback. Fresh-snapshot retry succeeds without duplicate provider work. - Saved-policy positive and explicit-conflict negative pass, including an authenticated compatible target grant and B-to-C continuity with omitted live settings. - Factory/Core checks: 34 passed, including effective ACPX defaults, real policy changes, already-dispatching target work and predecessor admission. - Ordinary adoption compatibility checks: 11 passed. Cacheless owner construction/revocation and policy checks: 6 passed. - Deferred permission ordering/forwarding: 30 passed. The new regressions reproduce owner changes before queueing and before slow-consumer delivery; matching-owner presentation and later source-owner settlement remain intact. - Runner `tsc --noEmit`, `git diff --check`, and full-PR added-line secret/personal-data scan passed at the corrected head. New tests reproduced both review findings and premature completion on failed persistence/changed ownership before the fixes. Full suites remain assigned to CI. ## Risks - Fake-provider tests do not prove vendor restore inertness or rollout fsync. - The process owner must prove stopped ownership, no successor and strict dormant provider admission before a predecessor launch. This PR does not select or stop host tasks. - Ordinary suspended active-goal continuation remains distinct from completed-predecessor admission. - Live dormant checks use bounded goal and thread reads. Unsupported or conflicting responses fail closed. - Prepared provider state is outside new predecessor admission. Reads and cleanup retain their baseline behavior. The feature targets completed between-turn session-open or exited sessions. - No tenant changes or image publication are included. ## Model Used OpenAI GPT-6.1 Sol (`gpt-6.1-sol`), high reasoning, through Codex with local code and command tools. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used - [x] I have checked ROADMAP.md and confirmed this does not duplicate planned work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have linked the related public issue or PR - [x] I have not referenced internal issues or instance links - [x] My branch name describes the change - [x] Focused local tests pass - [x] I have added tests and updated documentation - [x] I have documented risks - [x] All Paperclip CI gates are green at the revised head - [x] Greptile is 5/5 with no unresolved comments at the revised head - [x] I will address review findings before requesting merge --------- Co-authored-by: Paperclip --- .../runner/DURABLE_TRANSPORT.md | 64 +++- .../src/bin/fake-codex-app-server.rs | 11 + .../runner-core/src/bin/paperclip-runnerd.rs | 19 + .../crates/runner-core/src/codex_provider.rs | 22 ++ .../crates/runner-core/src/durable/runner.rs | 7 + .../crates/runner-core/src/durable/state.rs | 28 +- .../src/native_provider_backend.rs | 14 + .../runner-core/src/provider_backend.rs | 65 ++++ .../runner-core/tests/codex_provider.rs | 320 ++++++++++++++++- .../durable-prp-control-plane.test.ts | 66 +++- .../durable-prp-control-plane.ts | 57 ++- .../control-plane/runner-execution-driver.ts | 4 + .../drivers/codex/codex-harness-session.ts | 8 + .../src/drivers/codex/codex-session-state.ts | 5 + .../src/live/runner-provider-launch.test.ts | 10 +- .../src/live/runner-provider-launch.ts | 17 +- .../runnerd-acpx-permission-ordering.test.ts | 38 +- ...odex-transport-attachment-recovery.test.ts | 41 ++- .../src/live/runnerd-codex-transport.test.ts | 20 +- .../src/live/runnerd-codex-transport.ts | 181 ++++++++-- .../runnerd-source-reconciliation.test.ts | 333 ++++++++++++++++++ .../src/live/runnerd-staged-handoff.test.ts | 20 +- 22 files changed, 1275 insertions(+), 75 deletions(-) create mode 100644 packages/paperclip-runner/src/live/runnerd-source-reconciliation.test.ts diff --git a/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md b/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md index 57f40c22111..ea9c726a09d 100644 --- a/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md +++ b/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md @@ -1,14 +1,20 @@ # Durable PRP transport -This layer gives `paperclip-runnerd` a provider-neutral PRP v1 transport. The -Paperclip server invokes durable mode only for a selected, flag-enabled -`paperclip_runner` agent or recovery of its persisted native run. Codex is the -only installed provider; other providers remain unavailable. +Core uses durable PRP v1 for a selected, flag-enabled `paperclip_runner` +agent or recovery of its persisted run. On a fresh launch or restart, +`paperclip-runnerd` selects durable mode from its connection arguments. + +- With `--connect-url `, the Runner dials Core. +- With `--listen-address 0.0.0.0 --listen-path /api/runner/v1/connect/`, + the Runner listens for Core. `--listen-port` defaults to `43127`. + +Choose one form. Both require `PAPERCLIP_RUNNER_BOOTSTRAP_TICKET`. +Recovery uses the same entry point. The Runner rejects `--source-reconcile`. ## Trust boundary -- The runner accepts only `ws://` destinations whose complete DNS result is - loopback. Resolution happens once and reconnects reuse the pinned addresses. +- For outbound `ws://`, every resolved address must be loopback. The Runner + resolves the destination once and reuses the pinned addresses on reconnect. - A bootstrap ticket is read from `PAPERCLIP_RUNNER_BOOTSTRAP_TICKET`, removed from the environment immediately, and never sent over the socket. Both peers prove possession through HMAC-SHA-256. @@ -63,13 +69,41 @@ the P0 reserve. When the soft limit is reached, the runner enters backpressure and rejects the new non-P0 event rather than silently losing it. Exhausting the P0 reserve is an explicit unrecoverable condition. -## Current boundary +## Recovering Codex between turns -Durable mode is selected only when `paperclip-runnerd` receives -`--connect-url`. Its executor accepts a Codex app-server descriptor and bound -completion contract through `run.prepare`, owns the provider process group, -resumes the persisted Codex thread after runner restart, and translates -provider notifications to PRP events. The server supplies the bootstrap ticket -only through the child environment, stores the process identity for bounded -cancellation, and waits for the durable result and terminal pair. The existing -local fake-runner mode remains unchanged. +Suppose run A finishes a turn, its Runner exits, and run B wants to continue +the conversation. Core uses ordinary recovery for A, then attaches B with +authenticated `run.attach`. If the original process is alive, its owner can +verify and adopt it instead. + +**Is A idle?** Before launch, the host driver (`sourceRecoveryVersion: 1`) +checks that A’s process stopped and its saved session was idle. After launch, +the Runner (`recoveryIntegrityVersion: 1`) verifies that Codex restored an idle +session when no command effect is uncertain. It reads the goal, then the +thread. An unexpected turn or active goal blocks recovery, even if the goal +changes during the thread read. Accepted updates stay queued for delivery. + +Active runs use same-run recovery. Indeterminate commands follow the replay +rules above. A prepared session can continue normally, but this handoff +requires a completed turn. Reads and cleanup leave a prepared provider unopened. + +**Who owns the session?** Core matches A’s Runner instance, environment lease, +run, normalized session, turn and item. History reads require A’s thread. A +resume request returns saved state without asking Codex to resume twice. Retained events and usage +belong to A. The authenticated `authorityRunId` identifies the current owner +(A for same-run recovery). Losing that owner blocks new commands and dispatch. + +If Core loses its cached preparation, the preparer uses A’s frozen +`recoveryLaunch.requirements`, exact identity and authenticated owner. It +leaves A’s configuration in place and the cache empty. B inherits A’s recorded +launch policy on later attachments. Explicit settings must agree after +provider defaults are applied. Authenticated context and instruction grants +keep their usual precedence and compatibility checks. + +**When can B start work?** After authenticated attachment, Core persists B’s +fresh snapshot and checkpoint and runs any recovery callback. B can then +start work. If a write or callback fails, Core can retry the handoff. +Cancellation cleanup only settles retained work. + +Live qualification still needs to verify that restoring a provider starts no +work, deployed storage honors fsync, and the old process has exited. diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs index fad7b860c79..b97440c45e9 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs @@ -889,6 +889,7 @@ fn run() -> Result<(), Box> { let pre_response_notification = args .iter() .any(|value| value == "--notification-before-response"); + let mut goal_update_on_next_thread_read: Option = None; let goal_policy_disabled = args.iter().any(|value| value == "--goal-policy-disabled"); let goal_autostart = args.iter().any(|value| value == "--goal-autostart"); let goal_autocontinue = args.iter().any(|value| value == "--goal-autocontinue"); @@ -1457,6 +1458,14 @@ fn run() -> Result<(), Box> { { return Err("thread/read must not hydrate turns".into()); } + if let Some(status) = goal_update_on_next_thread_read.take() { + state.goal = Some(json!({"objective":"Final read goal", "status":status})); + save_state(&state_path, &state)?; + log_call(call_log.as_deref(), "late-goal-on-thread-read")?; + send( + json!({"method":"thread/goal/updated", "params":{"threadId":state.thread_id,"goal":state.goal}}), + )?; + } let turns = state .active_turn_id .as_ref() @@ -1478,6 +1487,8 @@ fn run() -> Result<(), Box> { }))?, "thread/goal/get" => { send(json!({"id": id, "result": {"goal": state.goal}}))?; + goal_update_on_next_thread_read = argument(&args, "--goal-on-thread-read-file") + .and_then(|path| fs::read_to_string(path).ok()); if args .iter() .any(|value| value == "--idle-protocol-failure-on-goal-probe") diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs b/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs index 0362b8a554a..9b0d056156d 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs @@ -121,6 +121,7 @@ fn build_metadata() -> serde_json::Value { "packageName": "@paperclipai/paperclip-runner", "packageVersion": env!("CARGO_PKG_VERSION"), "binaryContractVersion": 2, + "recoveryIntegrityVersion": 1, "durableSessionCapabilities": ["unlimited_runtime", "connection_lease_renewal"], "externalProviderCapabilities": ["openai_dot_mcp", "grok_bot_mcp", "muse_external_v1"], "nativeExecutionVersion": 1, @@ -521,6 +522,11 @@ fn run(args: &[String]) -> Result<(), LocalRunnerError> { println!("{}", metadata_with_provider_packs(args)?); return Ok(()); } + if args.iter().any(|arg| arg == "--source-reconcile") { + return Err(LocalRunnerError::invalid( + "--source-reconcile is no longer supported; use ordinary durable recovery", + )); + } apply_oom_score_adj(args)?; apply_task_memory_cgroup(args)?; if args.iter().any(|argument| argument == "--connect-url") @@ -652,9 +658,22 @@ mod tests { assert!(durable_connect_url(&mixed, "run_1").is_err()); } + #[test] + fn legacy_source_mode_is_rejected_before_any_startup() { + let error = run(&[ + "--listen-port".to_owned(), + "1234".to_owned(), + "--source-reconcile".to_owned(), + ]) + .unwrap_err(); + assert!(error.to_string().contains("no longer supported")); + } + #[test] fn build_metadata_advertises_the_remote_transport_contract() { let metadata = build_metadata(); + assert_eq!(metadata["recoveryIntegrityVersion"], 1); + assert!(metadata.get("sourceRecoveryVersion").is_none()); assert_eq!(metadata["schema"], RUNNERD_BUILD_METADATA_SCHEMA); assert_eq!(metadata["binaryContractVersion"], 2); assert_eq!( diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs b/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs index a5bd41f72d0..db5efd7226d 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs @@ -2239,6 +2239,28 @@ impl CodexProvider { Ok(()) } + pub(crate) fn read_thread_after_recovery_goal( + &mut self, + goal_is_dormant: fn(&Value) -> bool, + ) -> Result { + // The preceding goal reply supersedes earlier notifications. Inspect only + // frames received by this read, and leave them queued for normal delivery. + let first = self.pending_messages.len(); + let snapshot = self.read_thread()?; + for message in self.pending_messages.iter().skip(first) { + if message.value["method"] == "thread/goal/updated" { + let params = &message.value["params"]; + validate_notification_binding(&self.thread_id, None, params)?; + if !params["goal"].is_object() || !goal_is_dormant(params) { + return Err(LocalRunnerError::invalid( + "saved dormant Codex session resumed an active or unknown goal", + )); + } + } + } + Ok(snapshot) + } + pub fn read_thread(&mut self) -> Result { // Probing provider state does not supersede an authoritative terminal. // A later replacement turn must still establish its own identity. diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs index 97bf9644745..98f9967e400 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs @@ -308,6 +308,10 @@ pub enum TerminalDeliveryReconciliation { pub trait CommandExecutor { fn execute(&mut self, command: &Command) -> Result; + fn observe_recovery_state(&mut self, _state: &DurableState) -> Result<(), DurableRunnerError> { + Ok(()) + } + /// Opt in only when semantic result delivery validates an exact durable /// call receipt and cannot dispatch the underlying business operation. /// Ordinary commands and providers without this proof remain indeterminate. @@ -462,6 +466,9 @@ pub fn run_durable_runner( { return Ok(()); } + if recovered { + executor.observe_recovery_state(&state)?; + } if recovered && state.warm_transition.is_none() { state.reconnect_count = state.reconnect_count.saturating_add(1); state.record_diagnostic("runner restored its durable identity after process recovery"); diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs index 1741d687db2..bf25c7cce29 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs @@ -359,6 +359,24 @@ pub struct DurableState { } impl DurableState { + pub(crate) fn recovery_pending_effects(&self) -> usize { + self.processed_commands + .values() + .filter(|command| { + matches!(command.status.as_str(), "pending" | "indeterminate") + && !matches!( + command.command_type.as_str(), + "session.snapshot" + | "session.goal.get" + | "runner.drain" + | "runner.suspend" + | "runner.shutdown" + | "turn.stop" + ) + }) + .count() + } + pub(crate) fn new(config: &DurableRunnerConfig) -> Self { Self { identity_output: IdentityOutputBuffer::default(), @@ -1541,7 +1559,7 @@ fn sensitive_key(key: &str, value: &Value) -> bool { if normalized == "tokenbudgetcontrol" { return !value.is_boolean(); } - if normalized == "tokenbudget" { + if matches!(normalized.as_str(), "tokenbudget" | "tokensused") { return !value.is_number() && !value.is_null(); } if matches!( @@ -1558,7 +1576,6 @@ fn sensitive_key(key: &str, value: &Value) -> bool { | "totaltokens" | "pretokens" | "posttokens" - | "tokensused" ) { return !value.is_number(); } @@ -3690,6 +3707,13 @@ mod tests { assert_eq!(sanitized["tokenBudget"], json!(4096)); assert_eq!(sanitized["tokensUsed"], json!(128)); assert_eq!(sanitized["accessToken"], json!("[REDACTED]")); + let unknown = sanitize_value(&json!({"tokenBudget":null, "tokensUsed":null})); + assert!(unknown["tokenBudget"].is_null()); + assert!(unknown["tokensUsed"].is_null()); + assert_eq!( + sanitize_value(&json!({"tokensUsed":"synthetic credential"}))["tokensUsed"], + json!("[REDACTED]") + ); } #[test] diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/native_provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/native_provider_backend.rs index 39b6d7400a6..2c856c2b68f 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/native_provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/native_provider_backend.rs @@ -278,6 +278,20 @@ impl NativeProviderCommandExecutor { } impl CommandExecutor for NativeProviderCommandExecutor { + fn observe_recovery_state( + &mut self, + state: &crate::durable::DurableState, + ) -> Result<(), DurableRunnerError> { + if state.warm_transition.is_some() { + return Ok(()); + } + self.select_recovery()?; + if let Some(SelectedExecutor::LocalFacade(executor)) = self.selected.as_mut() { + executor.observe_recovery_state(state); + } + Ok(()) + } + fn can_reconcile_result_delivery(&mut self) -> Result { self.select_recovery()?; self.selected diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs index 82a22d33c90..ef0b777c2e2 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs @@ -1102,6 +1102,15 @@ fn normalize_codex_goal(value: &Value, working_now: bool) -> Option bool { + [value.get("workingNow"), value.pointer("/goal/workingNow")] + .into_iter() + .flatten() + .all(|working| working.as_bool() == Some(false)) + && (value.get("goal") == Some(&Value::Null) + || normalize_codex_goal(value, false).is_some_and(|goal| goal.status != "active")) +} + fn goal_event_payload( goal: Option<&SessionGoalSnapshot>, capability: Option<&SessionGoalCapability>, @@ -1856,6 +1865,7 @@ pub struct CodexCommandExecutor { cursor_cloud_launch_profile: Option, startup_command: Option, startup_evidence_error: Option, + recovery_pending_effects: Option, } impl CodexCommandExecutor { @@ -1871,6 +1881,7 @@ impl CodexCommandExecutor { cursor_cloud_launch_profile: None, startup_command: None, startup_evidence_error: None, + recovery_pending_effects: None, } } @@ -1882,6 +1893,10 @@ impl CodexCommandExecutor { executor } + pub(crate) fn observe_recovery_state(&mut self, state: &crate::durable::DurableState) { + self.recovery_pending_effects = Some(state.recovery_pending_effects()); + } + fn bind_opencode_launch_profile( &self, config: &CodexProviderConfig, @@ -2221,6 +2236,21 @@ impl CodexCommandExecutor { { return Ok(()); } + let expected_dormant = state.config.provider == "codex" + && self + .recovery_pending_effects + .as_ref() + .is_some_and(|count| *count == 0) + && state.completed_turn_authoritative + && state.active_provider_turn_id.is_none() + && !state.ambiguous_turn_start_pending + && state.startup_attempt.is_none() + && !state.receipt_limit_interrupt_pending + && state.tool_bridge.pending_calls().next().is_none() + && state + .goal + .as_ref() + .is_none_or(|goal| goal.status != "active" && !goal.working_now); let provider_label = state.config.provider.clone(); let provider_name = facade_provider_name(&provider_label); let provider_had_exited = state.lifecycle == "provider_exited"; @@ -2306,6 +2336,39 @@ impl CodexCommandExecutor { return Ok(()); } let recovered_active_turn_id = provider.active_provider_turn_id().map(str::to_owned); + if expected_dormant && recovered_active_turn_id.is_some() { + return Err(DurableRunnerError::invalid( + "saved dormant Codex session resumed unexpected active work", + )); + } + if expected_dormant { + let live_goal = provider.get_goal().map_err(|error| { + DurableRunnerError::invalid(format!( + "dormant recovery goal probe failed: {error}" + )) + })?; + if !codex_goal_is_dormant(&live_goal) { + return Err(DurableRunnerError::invalid( + "saved dormant Codex session resumed an active or unknown goal", + )); + } + let live_thread = provider + .read_thread_after_recovery_goal(codex_goal_is_dormant) + .map_err(|error| { + DurableRunnerError::invalid(format!( + "dormant recovery thread probe failed: {error}" + )) + })?; + if live_thread + .pointer("/thread/status/type") + .and_then(Value::as_str) + != Some("idle") + { + return Err(DurableRunnerError::invalid( + "saved dormant Codex session resumed unexpected active work", + )); + } + } let recovered_turn_ended_with_result = active_provider_result_authoritative && previous_active_turn_id.is_some() && recovered_active_turn_id.is_none(); @@ -2610,6 +2673,7 @@ impl CodexCommandExecutor { { self.fail_started_provider(&mut provider); } else { + self.recovery_pending_effects = None; self.provider = Some(provider); } admission @@ -5295,6 +5359,7 @@ impl CommandExecutor for CodexCommandExecutor { } fn rotate_authority(&mut self, config: &DurableRunnerConfig) { + self.recovery_pending_effects = None; self.event_identity = Some(ProviderEventIdentity::from_config(config)); } diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs index 840883c58f1..11b3dd1f575 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs @@ -7,8 +7,8 @@ use paperclip_runner_core::codex_provider::{ CodexProvider, CodexProviderConfig, CodexProviderEvent, }; use paperclip_runner_core::durable::{ - Command, CommandExecutor, DurableRunnerConfig, DurableRunnerError, PolledEvent, - TerminalDeliveryReconciliation, + Command, CommandDisposition, CommandExecutor, DurableRunnerConfig, DurableRunnerError, + DurableStateStore, EventPriority, PolledEvent, TerminalDeliveryReconciliation, }; use paperclip_runner_core::native_provider_backend::NativeProviderCommandExecutor; use paperclip_runner_core::provider_backend::CodexCommandExecutor; @@ -76,6 +76,311 @@ fn provider_config(directory: &Path, switches: &[&str]) -> CodexProviderConfig { } } +fn dormant_source_fixture(directory: &Path) -> (DurableRunnerConfig, Value) { + dormant_source_fixture_with_switches(directory, &[]) +} + +fn dormant_source_fixture_with_switches( + directory: &Path, + switches: &[&str], +) -> (DurableRunnerConfig, Value) { + let mut switches_with_defaults = vec!["--record-process-start", "--durable-turn-ids"]; + switches_with_defaults.extend_from_slice(switches); + let config = durable_config(directory); + let mut executor = CodexCommandExecutor::with_runner_config(directory, &config); + executor + .execute(&command( + "prepare-source", + 1, + "run.prepare", + json!({"provider":provider_config(directory, &switches_with_defaults)}), + )) + .unwrap(); + executor + .execute(&command("open-source", 2, "session.open", json!({}))) + .unwrap(); + executor + .execute(&command( + "source-turn", + 3, + "turn.start", + json!({"text":"Complete the source turn."}), + )) + .unwrap(); + wait_for_executor_event(&mut executor, "turn.completed"); + while !poll_and_ack(&mut executor).unwrap().is_empty() {} + executor.shutdown().unwrap(); + let provider = + serde_json::from_slice(&fs::read(directory.join("codex-provider-state.json")).unwrap()) + .unwrap(); + (config, provider) +} + +#[test] +fn ordinary_recovery_restores_a_completed_idle_session() { + let directory = temporary_directory("dormant-recovery"); + let (config, _) = dormant_source_fixture(&directory); + let runner = DurableStateStore::new(&directory) + .unwrap() + .load_or_create(&config) + .unwrap() + .0; + let mut executor = NativeProviderCommandExecutor::with_runner_config(&directory, &config); + executor.observe_recovery_state(&runner).unwrap(); + executor + .execute(&command("restore", 1, "session.snapshot", json!({}))) + .unwrap(); + while !poll_and_ack(&mut executor).unwrap().is_empty() {} + let snapshot = executor + .execute(&command("snapshot", 2, "session.snapshot", json!({}))) + .unwrap(); + assert_eq!( + snapshot.result["warmAttachReady"], true, + "{}", + snapshot.result + ); + assert_eq!(call_count(&directory, "process-start"), 2); + assert_eq!(call_count(&directory, "turn/start"), 1); + executor.shutdown().unwrap(); + fs::remove_dir_all(directory).unwrap(); +} + +#[test] +fn ordinary_recovery_checks_saved_dormant_state_against_live_work() { + for fault in ["active", "goal"] { + let directory = temporary_directory("dormant-live-recovery"); + let (config, _) = dormant_source_fixture(&directory); + let path = directory.join("fake-state.json"); + let mut live: Value = serde_json::from_slice(&fs::read(&path).unwrap()).unwrap(); + if fault == "active" { + live["activeTurnId"] = json!("unexpected-provider-turn"); + } else { + live["goal"] = json!({"objective":"Unexpected autonomous work", "status":"active"}); + } + fs::write(&path, serde_json::to_vec(&live).unwrap()).unwrap(); + let runner = DurableStateStore::new(&directory) + .unwrap() + .load_or_create(&config) + .unwrap() + .0; + let mut recovered = NativeProviderCommandExecutor::with_runner_config(&directory, &config); + recovered.observe_recovery_state(&runner).unwrap(); + assert_eq!(call_count(&directory, "process-start"), 1); + let error = recovered + .execute(&command("snapshot", 1, "session.snapshot", json!({}))) + .unwrap_err(); + assert!( + error.to_string().contains("saved dormant Codex"), + "{fault}: {error}" + ); + assert_eq!(call_count(&directory, "process-start"), 2); + drop(recovered); + fs::remove_dir_all(directory).unwrap(); + } +} + +#[test] +fn dormant_recovery_rejects_goal_activation_during_the_final_thread_read() { + let directory = temporary_directory("dormant-goal-during-read"); + let trigger = directory.join("activate-goal"); + let (config, _) = dormant_source_fixture_with_switches( + &directory, + &["--goal-on-thread-read-file", trigger.to_str().unwrap()], + ); + fs::write(&trigger, b"active").unwrap(); + let runner = DurableStateStore::new(&directory) + .unwrap() + .load_or_create(&config) + .unwrap() + .0; + let mut recovered = NativeProviderCommandExecutor::with_runner_config(&directory, &config); + recovered.observe_recovery_state(&runner).unwrap(); + let result = recovered.execute(&command("snapshot", 1, "session.snapshot", json!({}))); + assert_eq!(call_count(&directory, "late-goal-on-thread-read"), 1); + let error = result.expect_err("late goal activation must refuse dormant recovery"); + assert!(error.to_string().contains("saved dormant Codex"), "{error}"); + let state: Value = + serde_json::from_slice(&fs::read(directory.join("codex-provider-state.json")).unwrap()) + .unwrap(); + assert_eq!(state["startupAttempt"]["directChildExitObserved"], true); + #[cfg(unix)] + assert!( + !std::process::Command::new("kill") + .args([ + "-0", + &state["startupAttempt"]["processId"] + .as_u64() + .unwrap() + .to_string() + ]) + .output() + .unwrap() + .status + .success(), + "provider must be reaped before refusal returns" + ); + assert_eq!(call_count(&directory, "process-start"), 2); + assert_eq!(call_count(&directory, "turn/start"), 1); + drop(recovered); + fs::remove_dir_all(directory).unwrap(); +} + +#[test] +fn dormant_recovery_delivers_a_paused_goal_observed_during_the_final_thread_read() { + let directory = temporary_directory("paused-goal-during-read"); + let trigger = directory.join("paused-goal"); + let (config, _) = dormant_source_fixture_with_switches( + &directory, + &["--goal-on-thread-read-file", trigger.to_str().unwrap()], + ); + fs::write(&trigger, b"paused").unwrap(); + let runner = DurableStateStore::new(&directory) + .unwrap() + .load_or_create(&config) + .unwrap() + .0; + let mut recovered = NativeProviderCommandExecutor::with_runner_config(&directory, &config); + recovered.observe_recovery_state(&runner).unwrap(); + recovered + .execute(&command("snapshot", 1, "session.snapshot", json!({}))) + .unwrap(); + assert_eq!(call_count(&directory, "late-goal-on-thread-read"), 1); + let update = wait_for_executor_event(&mut recovered, "session.goal.updated"); + assert_eq!(update.payload["goal"]["status"], "paused"); + drop(recovered); + fs::remove_dir_all(directory).unwrap(); +} + +#[test] +fn ordinary_recovery_preserves_pending_same_run_goal_effects() { + for status in ["pending", "indeterminate"] { + let directory = temporary_directory("pending-goal-recovery"); + let (config, _) = dormant_source_fixture(&directory); + let path = directory.join("fake-state.json"); + let mut live: Value = serde_json::from_slice(&fs::read(&path).unwrap()).unwrap(); + live["goal"] = json!({"objective":"Already authorized same-run goal", "status":"active"}); + fs::write(&path, serde_json::to_vec(&live).unwrap()).unwrap(); + let store = DurableStateStore::new(&directory).unwrap(); + let mut runner = store.load_or_create(&config).unwrap().0; + runner + .begin_command(&command( + "goal-in-flight", + 1, + "session.goal.set", + json!({"objective":"Already authorized same-run goal", "status":"active"}), + )) + .unwrap(); + if status == "indeterminate" { + store.save(&runner).unwrap(); + runner = store.load_or_create(&config).unwrap().0; + } + assert_eq!(runner.processed_commands["goal-in-flight"].status, status); + let mut recovered = NativeProviderCommandExecutor::with_runner_config(&directory, &config); + recovered.observe_recovery_state(&runner).unwrap(); + recovered + .execute(&command("snapshot", 2, "session.snapshot", json!({}))) + .unwrap(); + let goal = recovered + .execute(&command("goal-read", 3, "session.goal.get", json!({}))) + .unwrap(); + assert_eq!(goal.result["goal"]["status"], "active"); + assert_eq!(call_count(&directory, "turn/start"), 1); + recovered.shutdown().unwrap(); + fs::remove_dir_all(directory).unwrap(); + } +} + +#[test] +fn ordinary_recovery_does_not_restore_prepared_during_reads_or_cleanup() { + let directory = temporary_directory("prepared-journal-recovery"); + let (config, mut provider) = dormant_source_fixture(&directory); + provider["lifecycle"] = json!("prepared"); + provider["completedTurnAuthoritative"] = json!(false); + provider["completedTurnProcessGeneration"] = Value::Null; + provider["completedProviderTurnId"] = Value::Null; + fs::write( + directory.join("codex-provider-state.json"), + serde_json::to_vec(&provider).unwrap(), + ) + .unwrap(); + let runner = DurableStateStore::new(&directory) + .unwrap() + .load_or_create(&config) + .unwrap() + .0; + let mut executor = NativeProviderCommandExecutor::with_runner_config(&directory, &config); + executor.observe_recovery_state(&runner).unwrap(); + let snapshot = executor + .execute(&command("snapshot", 1, "session.snapshot", json!({}))) + .unwrap(); + assert_eq!(snapshot.result["warmAttachReady"], false); + executor + .execute(&command("drain", 2, "runner.drain", json!({}))) + .unwrap(); + executor + .execute(&command("stop", 3, "turn.stop", json!({}))) + .unwrap(); + assert_eq!(call_count(&directory, "process-start"), 1); + executor.shutdown().unwrap(); + fs::remove_dir_all(directory).unwrap(); +} + +#[test] +fn ordinary_recovery_preserves_command_replay_and_collision_checks() { + let directory = temporary_directory("recovery-command-replay"); + let config = durable_config(&directory); + let mut runner = DurableStateStore::new(&directory) + .unwrap() + .load_or_create(&config) + .unwrap() + .0; + let recorded = command( + "recorded-attach", + 1, + "run.attach", + json!({"paperclipNextAuthority":{}}), + ); + assert!(matches!( + runner.begin_command(&recorded).unwrap(), + CommandDisposition::Execute + )); + runner + .complete_command(&recorded, json!({"status":"attached"})) + .unwrap(); + runner + .enqueue_event( + &config, + "runner.diagnostic", + EventPriority::P0, + json!({"message":"unacknowledged fact"}), + ) + .unwrap(); + assert!( + matches!(runner.begin_command(&recorded).unwrap(), CommandDisposition::Replay(result) if result.status == "completed") + ); + let mut collision = recorded.clone(); + collision.payload["changed"] = json!(true); + assert!(runner + .begin_command(&collision) + .unwrap_err() + .to_string() + .contains("different command data")); + let pending = command( + "pending-goal", + 2, + "session.goal.set", + json!({"objective":"Original effect"}), + ); + assert!(matches!( + runner.begin_command(&pending).unwrap(), + CommandDisposition::Execute + )); + assert!( + matches!(runner.begin_command(&pending).unwrap(), CommandDisposition::Replay(result) if result.status == "pending") + ); + fs::remove_dir_all(directory).unwrap(); +} + #[test] fn delegates_command_isolation_to_an_explicit_external_sandbox() { let directory = temporary_directory("external-sandbox"); @@ -190,7 +495,12 @@ fn durable_config(directory: &Path) -> DurableRunnerConfig { fn command(id: &str, sequence: u64, command_type: &str, payload: Value) -> Command { Command { - schema: "paperclip.prp.command.v1".to_owned(), + schema: if command_type.starts_with("session.goal.") { + "paperclip.prp.command.v2" + } else { + "paperclip.prp.command.v1" + } + .to_owned(), command_id: id.to_owned(), controller_seq: sequence, command_type: command_type.to_owned(), @@ -428,7 +738,7 @@ fn recorded_tool_responses(directory: &Path) -> Vec { } fn poll_and_ack( - executor: &mut CodexCommandExecutor, + executor: &mut impl CommandExecutor, ) -> Result, DurableRunnerError> { let events = executor.poll_events()?; executor.acknowledge_events(events.len())?; @@ -472,7 +782,7 @@ fn wait_for_provider_error(provider: &mut CodexProvider) -> String { } fn wait_for_executor_event( - executor: &mut CodexCommandExecutor, + executor: &mut impl CommandExecutor, expected_event_type: &str, ) -> PolledEvent { let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); diff --git a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts index dc2ccd7140e..0e858fa1955 100644 --- a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts +++ b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts @@ -59,6 +59,40 @@ const identity: DurableRecoveryIdentity = { const expectedRunnerVersion = "0.3.0"; const expectedRunnerDigest = `sha256:${"a".repeat(64)}`; +it("admits only source maintenance while the authenticated owner differs", async () => { + const root = mkdtempSync(resolve(tmpdir(), "runner-authority-admission-")); + let owner = identity.runId; + const core = new DurablePrpControlPlane({ stateDirectory: root, identity, expectedRunnerVersion, expectedRunnerDigest, + get authorityRunId() { return owner; } }); + try { + const prior = core.queueCommand("turn.start", { text: "Owned same-run work" }, "same-run-command"); + expect(prior.type).toBe("turn.start"); + owner = "new-owner"; + const count = core.store.state.commands.length; + for (const type of ["turn.start", "session.goal.set", "semantic_tool.result", "session.open"]) { + expect(() => core.queueCommand(type, {})).toThrow("source_attachment_pending"); + } + expect(core.store.state.commands).toHaveLength(count); + expect(core.queueCommand("session.snapshot").type).toBe("session.snapshot"); + expect(core.queueCommand("turn.stop").type).toBe("turn.stop"); + expect(() => core.enableSemanticInputDispatch(identity)).toThrow("source_attachment_pending"); + } finally { await core.stop(); rmSync(root, { recursive: true, force: true }); } +}); + +it("keeps already-dispatching target work independent of predecessor completion", async () => { + const root = mkdtempSync(resolve(tmpdir(), "runner-target-dispatch-")); + const core = new DurablePrpControlPlane({ stateDirectory: root, identity, expectedRunnerVersion, expectedRunnerDigest, + authorityRunId: identity.runId }); + try { + core.queueCommand("turn.start", { text: "Current target work" }); + const settled = vi.spyOn(core, "semanticToolResultsSettled").mockReturnValue(false); + expect(() => core.enableSemanticInputDispatch(identity)).not.toThrow(); + expect(settled).not.toHaveBeenCalled(); + expect(core.queueCommand("session.goal.set", { status: "active" }).type).toBe("session.goal.set"); + expect(() => core.enableSemanticInputDispatch({ ...identity, runId: "foreign" })).toThrow("source_attachment_pending"); + } finally { await core.stop(); rmSync(root, { recursive: true, force: true }); } +}); + it("recognizes a complete unclaimed bootstrap through cold checkpoint restoration", async () => { const root = mkdtempSync(resolve(tmpdir(), "runner-unclaimed-bootstrap-")); const core = new DurablePrpControlPlane({ stateDirectory: root, identity, expectedRunnerVersion, expectedRunnerDigest }); @@ -595,6 +629,30 @@ it("persists authentication and commands before the runner can observe them", as } }); +it("does not dispatch after the authenticated owner changes during checkpoint persistence", async () => { + const root = mkdtempSync(resolve(tmpdir(), "runner-tool-owner-race-")); + let owner = identity.runId, entered = false, release!: () => void; + const barrier = new Promise(resolveWrite => { release = resolveWrite; }); + const dispatch = vi.fn(async () => ({ result: "must not run" })); + const core = new DurablePrpControlPlane({ stateDirectory: root, identity, expectedRunnerVersion, expectedRunnerDigest, + get authorityRunId() { return owner; }, onSemanticToolInput: dispatch }); + core.setCheckpointPersistence(async checkpoint => { + if (JSON.parse(checkpoint.records.get("cursor")!).ackedSourceSeq > 0) { entered = true; await barrier; } + }); + try { + await core.start(); + const client = (await authenticate(core, core.issueBootstrapTicket()))!; + sendSecure(client, semanticInputEvent()); + await vi.waitFor(() => expect(entered).toBe(true)); + owner = "replacement-owner"; release(); + const ack = await receiveSecure(client); + expect(dispatch).not.toHaveBeenCalled(); + expect(ack!.kind).toBe("ack"); + expect(core.store.state.commands).toEqual([]); + client.socket.destroy(); + } finally { release(); await core.stop(); rmSync(root, { recursive: true, force: true }); } +}); + it("does not dispatch a semantic effect when external journal persistence fails", async () => { const root = mkdtempSync(resolve(tmpdir(), "runner-external-journal-failure-")); const dispatch = vi.fn(async () => ({ result: "written" })); @@ -804,8 +862,9 @@ it.each([true, false])("reconciles a saved semantic input only from an authorita } }); -it("resolves a saved semantic input only once across overlapping recovery calls", async () => { +it.each([false, true])("resolves a saved semantic input only once across overlapping recovery calls (owner changed: %s)", async changedOwner => { const root = mkdtempSync(resolve(tmpdir(), "runner-saved-semantic-race-")); + let owner = identity.runId; const options = { stateDirectory: root, identity, expectedRunnerVersion, expectedRunnerDigest }; const original = new DurablePrpControlPlane({ ...options, onSemanticToolInput: () => new Promise<{ result: unknown }>(() => undefined) }); @@ -821,17 +880,18 @@ it("resolves a saved semantic input only once across overlapping recovery calls" expect((await receiveSecure(client))!.kind).toBe("ack"); client.socket.destroy(); await original.stop(); - replacement = new DurablePrpControlPlane({ ...options, recoverSemanticToolInput: recover }); + replacement = new DurablePrpControlPlane({ ...options, get authorityRunId() { return owner; }, recoverSemanticToolInput: recover }); pending.push(replacement.recoverPendingSemanticInputs()); await vi.waitFor(() => expect(recover).toHaveBeenCalledOnce()); pending.push(replacement.recoverPendingSemanticInputs()); await new Promise(resolveTurn => setImmediate(resolveTurn)); expect(recover).toHaveBeenCalledOnce(); + if (changedOwner) owner = "replacement-owner"; resolveOutcome({ result: { receipt: "one-authoritative-outcome" } }); await Promise.all(pending); await replacement.recoverPendingSemanticInputs(); expect(recover).toHaveBeenCalledOnce(); - expect(replacement.store.state.commands.filter(command => command.type === "semantic_tool.result")).toHaveLength(1); + expect(replacement.store.state.commands.filter(command => command.type === "semantic_tool.result")).toHaveLength(changedOwner ? 0 : 1); } finally { resolveOutcome({ result: { receipt: "one-authoritative-outcome" } }); await Promise.allSettled(pending); diff --git a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts index 218ea36aeec..cd29daad233 100644 --- a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts +++ b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts @@ -76,6 +76,10 @@ const maxStateBytes = 256 * 1024 * 1024; const authChallengeTtlMs = 5_000; const stableIdPattern = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,239}$/; const runnerDigestPattern = /^sha256:[0-9a-f]{64}$/; +export function isRunnerRecoveryMaintenanceCommand(type: string): boolean { + return ["session.snapshot", "session.goal.get", "runner.drain", "runner.suspend", "runner.shutdown", "turn.stop"].includes(type); +} + const commandTypes = new Set([ "run.prepare", "run.attach", @@ -269,6 +273,7 @@ export class SemanticToolNotDispatchedError extends Error { export interface DurablePrpControlPlaneOptions { stateDirectory: string; identity: DurableRecoveryIdentity; + authorityRunId?: string; expectedRunnerVersion: string; expectedRunnerDigest: string; /** Complete caller-owned admission before consuming a credential or releasing commands. */ @@ -2010,6 +2015,8 @@ export class DurablePrpControlPlane { }> = []; #port: number | null = null; #semanticInputMode: "dispatch" | "retain_only"; + #semanticInputSourceRunId: string | null; + readonly #authorityRunId: () => string | undefined; #onSemanticToolInput?: DurablePrpControlPlaneOptions["onSemanticToolInput"]; #onCommittedEvent?: DurablePrpControlPlaneOptions["onCommittedEvent"]; #recoverSemanticToolInput?: DurablePrpControlPlaneOptions["recoverSemanticToolInput"]; @@ -2064,7 +2071,10 @@ export class DurablePrpControlPlane { "Warm run transition requires its exact approved runner artifact.", ); } - this.#semanticInputMode = options.semanticInputMode ?? "dispatch"; + this.#authorityRunId = () => options.authorityRunId; + this.#semanticInputMode = options.authorityRunId !== undefined && options.authorityRunId !== options.identity.runId + ? "retain_only" : options.semanticInputMode ?? "dispatch"; + this.#semanticInputSourceRunId = this.#semanticInputMode === "retain_only" ? options.identity.runId : null; this.#onSemanticToolInput = options.onSemanticToolInput; this.#onCommittedEvent = options.onCommittedEvent; this.#recoverSemanticToolInput = options.recoverSemanticToolInput; @@ -2215,10 +2225,30 @@ export class DurablePrpControlPlane { ); } + #retainsSourceInputs(): boolean { + const owner = this.#authorityRunId(); + return this.#semanticInputMode === "retain_only" || (owner !== undefined && owner !== this.#identity.runId); + } + + enableSemanticInputDispatch(identity: DurableRecoveryIdentity): void { + if ((this.#authorityRunId() !== undefined && this.#authorityRunId() !== identity.runId) + || canonicalJson(this.#identity) !== canonicalJson(identity) || this.#store.state.warmTransition) { + throw new Error("native_runner_source_attachment_pending"); + } + if (this.#semanticInputMode === "dispatch") return; + if (this.#identity.runId === this.#semanticInputSourceRunId + || this.activeRunnerConnectionCount() !== 1 || !this.semanticToolResultsSettled()) { + throw new Error("native_runner_source_attachment_pending"); + } + this.#semanticInputMode = "dispatch"; + this.#semanticInputSourceRunId = null; + } + async recoverPendingSemanticInputs(): Promise { - if (this.#semanticInputMode === "retain_only" || !this.#recoverSemanticToolInput) return; + if (this.#retainsSourceInputs() || !this.#recoverSemanticToolInput) return; const identity = this.#identity; for (const saved of [...this.#store.state.committedEvents]) { + if (this.#retainsSourceInputs() || this.#identity !== identity || this.#semanticCallbacksRetired) return; if (!unsettledSemanticInput(saved, this.#store.state)) continue; const event = saved.envelope.payload as PrpEvent; const semantic = (event.payload as Record).semantic_tool as Record; @@ -2233,7 +2263,7 @@ export class DurablePrpControlPlane { correlation: { runId: identity.runId, normalizedSessionId: identity.normalizedSessionId, turnId: identity.turnId, itemId: event.itemId ?? identity.itemId } }; const outcome = await this.#recoverSemanticToolInput(call); - if (this.#identity !== identity || this.#semanticCallbacksRetired) return; + if (this.#identity !== identity || this.#semanticCallbacksRetired || this.#retainsSourceInputs()) return; if (!outcome || this.#getCommand(commandId)) continue; const { input, ...metadata } = call; this.queueCommand("semantic_tool.result", { ...metadata, inputDigest: canonicalDigest(input), @@ -2467,6 +2497,10 @@ export class DurablePrpControlPlane { ) { throw new Error("Durable PRP run identity rotation is invalid."); } + if (this.#authorityRunId() !== undefined && this.#authorityRunId() !== this.#identity.runId) { + this.#semanticInputMode = "retain_only"; + this.#semanticInputSourceRunId ??= this.#identity.runId; + } this.disconnectActiveRunner(); const leases = Object.fromEntries( Object.entries(this.#store.state.leases).map(([key, lease]) => [ @@ -2687,6 +2721,17 @@ export class DurablePrpControlPlane { deliverImmediately = false, ): DurableRecoveryCoreCommand { this.#store.assertWritable(); + const authorityRunId = this.#authorityRunId(); + const retainedSource = this.#semanticInputSourceRunId ?? this.#identity.runId; + const prior = commandId ? this.#getCommand(commandId) : undefined; + if (authorityRunId !== undefined && retainedSource !== authorityRunId + && !isRunnerRecoveryMaintenanceCommand(type) + && !(type === "run.attach" && isRecord(payload.paperclipNextAuthority) + && isRecord(payload.paperclipNextAuthority.identity) + && payload.paperclipNextAuthority.identity.runId === authorityRunId) + && !(prior && prior.status !== "pending" && prior.type === type && canonicalJson(prior.payload) === canonicalJson(payload))) { + throw new Error("native_runner_source_attachment_pending"); + } const staged = this.#store.state.stagedRunAttachment; const retireStaged = stagedTargetActivated(this.#store.state) && (type === "turn.start" || (this.#stagedTargetCheckpointPublished === canonicalJson(this.#store.state.identity) @@ -3974,7 +4019,7 @@ export class DurablePrpControlPlane { eventType === "semantic_tool.input" || eventType === "mcp_app.tool_input"; if ( isSemanticInput && - ((this.#onSemanticToolInput === undefined && this.#semanticInputMode !== "retain_only") || + ((this.#onSemanticToolInput === undefined && !this.#retainsSourceInputs()) || semantic?.phase !== "input" || typeof semantic.callId !== "string" || typeof semantic.operationId !== "string" || @@ -4114,7 +4159,7 @@ export class DurablePrpControlPlane { } if (!isCurrentOwner()) return; - const liveCall = this.#semanticInputMode !== "retain_only" && isSemanticInput && this.#onSemanticToolInput && semantic !== undefined + const liveCall = !this.#retainsSourceInputs() && isSemanticInput && this.#onSemanticToolInput && semantic !== undefined && typeof semantic.callId === "string" && typeof semantic.operationId === "string" ? { callId: semantic.callId, operationId: semantic.operationId, input: semantic.input, sourceEventId, sourceEventType: eventType, @@ -4132,7 +4177,7 @@ export class DurablePrpControlPlane { try { await this.flushCheckpoint(); if (this.#eventOwner !== connection || this.#identity !== identity) return; - if (ownsLiveDispatch && liveCall && liveCommandId && this.#onSemanticToolInput && !this.#getCommand(liveCommandId)) { + if (!this.#retainsSourceInputs() && ownsLiveDispatch && liveCall && liveCommandId && this.#onSemanticToolInput && !this.#getCommand(liveCommandId)) { const call = liveCall; const commandId = liveCommandId; const recordFailure = (stage: "persist_result" | "dispatch", error: unknown) => { diff --git a/packages/paperclip-runner/src/control-plane/runner-execution-driver.ts b/packages/paperclip-runner/src/control-plane/runner-execution-driver.ts index 526d628d9f7..681bf7234fb 100644 --- a/packages/paperclip-runner/src/control-plane/runner-execution-driver.ts +++ b/packages/paperclip-runner/src/control-plane/runner-execution-driver.ts @@ -9,6 +9,10 @@ import type { DurableRecoveryIdentity } from "../contracts/durable-recovery.js"; * specifications stay private to the driver; plugins receive versioned requirements. */ export interface RunnerExecutionDriver { + /** Attests stopped ownership and dormant prelaunch admission for the selected provider. */ + sourceRecoveryVersion?: 1; + /** Current authenticated lease/checkpoint-writer owner; may change during warm reuse. */ + authorityRunId?: string; /** Own provider configuration and file preparation in the execution environment. */ prepareLaunch?: RunnerLaunchPreparer; /** Reauthorize retained resources before adoption without preparing them again. diff --git a/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts b/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts index 1c60274c6af..e3d46a188bd 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts @@ -60,6 +60,8 @@ export class CodexHarnessSession extends CodexSessionState implements HarnessSession { + #deferredNotificationPump = false; + constructor(input: CodexSessionStateInput) { super(input); this.transport.setServerRequestHandler((request) => @@ -77,6 +79,7 @@ export class CodexHarnessSession )); } if (this.terminal) { + this.#deferredNotificationPump = true; this.eventQueue.close(); } else { void pumpNotifications(this); @@ -150,6 +153,11 @@ export class CodexHarnessSession this.protocolFailed = false; this.protocolFailureCode = null; this.protocolFailureMessage = null; + if (this.#deferredNotificationPump) { + this.eventQueue.reopen(); + this.#deferredNotificationPump = false; + void pumpNotifications(this); + } this.emit("run.attached", { runId: input.runId, sameSession: true }); } finally { this.pendingRunAttachment = null; diff --git a/packages/paperclip-runner/src/drivers/codex/codex-session-state.ts b/packages/paperclip-runner/src/drivers/codex/codex-session-state.ts index f2bc252abe0..3113d182f7e 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-session-state.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-session-state.ts @@ -56,6 +56,11 @@ class AsyncQueue implements AsyncIterable { waiter.resolve({ value: undefined, done: true }); } + reopen(): void { + if (this.#failure !== null) throw this.#failure; + this.#closed = false; + } + fail(error: Error): void { this.#failure ??= error; this.#closed = true; diff --git a/packages/paperclip-runner/src/live/runner-provider-launch.test.ts b/packages/paperclip-runner/src/live/runner-provider-launch.test.ts index d5c1fd22aa0..62183f9a5e8 100644 --- a/packages/paperclip-runner/src/live/runner-provider-launch.test.ts +++ b/packages/paperclip-runner/src/live/runner-provider-launch.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "vitest"; -import { createRunnerProviderLaunch, preparedRunnerLaunchWorkingDirectory, runnerLaunchRequirementsDigest, validatePreparedRunnerLaunch, +import { createRunnerProviderLaunch, preparedRunnerLaunchWorkingDirectory, runnerLaunchRequirementsDigest, runnerLaunchSettingsMatch, validatePreparedRunnerLaunch, runnerProviderLaunchRequirements, type RunnerProviderLaunchEnvironment, type RunnerProviderLaunchRequirements } from "./runner-provider-launch.js"; const requirements: RunnerProviderLaunchRequirements = { @@ -19,6 +19,14 @@ const environment: RunnerProviderLaunchEnvironment = { }; describe("built-in provider launch preparation", () => { + it("compares effective ACPX defaults without changing the saved requirements", () => { + const frozen = { ...requirements, provider: "acpx" as const, model: "gpt-5.4", settings: {} }; + expect(runnerLaunchSettingsMatch(frozen, { acpxAgent: "codex", acpxPermissionMode: "approve-all", acpxPermissionModePinned: true })).toBe(true); + expect(runnerLaunchSettingsMatch(frozen, { acpxPermissionMode: "approve-reads" })).toBe(false); + expect(runnerLaunchSettingsMatch(frozen, { acpxPermissionModePinned: false })).toBe(false); + expect(frozen.settings).toEqual({}); + }); + it("preserves probe project-instruction suppression through a prepared launch", () => { const selected = runnerProviderLaunchRequirements({ provider: "codex", disableCodexProjectInstructions: true, disableCodexWebSearch: true }, requirements.identity, diff --git a/packages/paperclip-runner/src/live/runner-provider-launch.ts b/packages/paperclip-runner/src/live/runner-provider-launch.ts index a7afb0b0f5f..cff4b9a4154 100644 --- a/packages/paperclip-runner/src/live/runner-provider-launch.ts +++ b/packages/paperclip-runner/src/live/runner-provider-launch.ts @@ -100,6 +100,18 @@ export function runnerLaunchRequirementsDigest(requirements: RunnerProviderLaunc return `sha256:${createHash("sha256").update(canonical).digest("hex")}`; } +function runnerLaunchPolicy(requirements: RunnerProviderLaunchRequirements): Record { + return createRunnerProviderLaunch(requirements, { + options: { codexArgs: [] }, root: "/", cwd: "/", codexHome: "/", + providerNodeCommand: "", opencodeProxyPath: "", acpxSidecarPath: "", codexCommand: "", + }); +} + +/** Compare effective provider policy, including the factory's defaults. */ +export function runnerLaunchSettingsMatch(requirements: RunnerProviderLaunchRequirements, settings: RunnerProviderLaunchRequirements["settings"]): boolean { + return JSON.stringify(runnerLaunchPolicy(requirements)) === JSON.stringify(runnerLaunchPolicy({ ...requirements, settings })); +} + /** The environment may resolve paths, but it cannot change Core's selected policy. */ export function validatePreparedRunnerLaunch( requirements: RunnerProviderLaunchRequirements, @@ -138,10 +150,7 @@ export function validatePreparedRunnerLaunch( || descriptor.runId !== requirements.identity.runId || descriptor.normalizedSessionId !== requirements.identity.normalizedSessionId) reject(); } - const expected = createRunnerProviderLaunch(requirements, { - options: { codexArgs: [] }, root: "/", cwd: "/", codexHome: "/", - providerNodeCommand: "", opencodeProxyPath: "", acpxSidecarPath: "", codexCommand: "", - }); + const expected = runnerLaunchPolicy(requirements); const hostFields = new Set(["command", "args", "sidecarCommand", "sidecarArgs", "runtimeDirectory", "cwd", "preconfiguredHome", "externallySandboxed", "runtimeContext", "instructions"]); for (const [key, value] of Object.entries(expected)) { diff --git a/packages/paperclip-runner/src/live/runnerd-acpx-permission-ordering.test.ts b/packages/paperclip-runner/src/live/runnerd-acpx-permission-ordering.test.ts index 4f38d7edf40..4800a50f5e4 100644 --- a/packages/paperclip-runner/src/live/runnerd-acpx-permission-ordering.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-acpx-permission-ordering.test.ts @@ -124,16 +124,18 @@ describe("permission and tool order through the actual transport and harness", ( }); }); -async function bareTransport() { +async function bareTransport(authorityRunId?: () => string | undefined) { const root = mkdtempSync(join(tmpdir(), "permission-queue-lifecycle-")); const binary = join(root, "unexecuted-runner"); writeFileSync(binary, "not executable: process boundary is mocked"); const bundle = createCapabilityRunnerdCodexTransport({ runnerBinary: binary, stateDirectory: root, + get authorityRunId() { return authorityRunId?.(); }, sourceCodexHome: join(root, "absent-source-home"), environment: { HOME: root, PATH: "/usr/bin:/bin" }, }); await bundle.transport.request("initialize", {}); await bundle.transport.request("thread/start", { cwd: root }); - await bundle.transport.request("turn/start", { input: [{ type: "text", text: "Fixture" }] }); + const started = await bundle.transport.request("turn/start", { input: [{ type: "text", text: "Fixture" }] }); + const turnId = (started.turn as { id: string }).id; const iterator = bundle.transport.notifications()[Symbol.asyncIterator](); expect((await iterator.next()).value?.method).toBe("turn/started"); const emitPermission = (requestId = "permission-7", prompt = "Run command") => peer.current!.emit("runtime_request.created", { request: { @@ -142,9 +144,39 @@ async function bareTransport() { origin: { adapter: "acpx-runtime-sidecar", provider: "cursor", method: "session/request_permission" }, } }); const emitMarker = () => peer.current!.emit("item.delta", { kind: "agentMessage", text: "marker" }); - return { bundle, iterator, emitPermission, emitMarker, async close() { await bundle.detachControllerForRestart(); rmSync(root, { recursive: true, force: true }); } }; + return { bundle, iterator, turnId, emitPermission, emitMarker, async close() { await bundle.detachControllerForRestart(); rmSync(root, { recursive: true, force: true }); } }; } +it.each(["same-owner", "rebound-before-queue", "rebound-before-delivery"] as const)("fences deferred permission presentation and preserves settlement: %s", async scenario => { + let owner: string | undefined; + const fixture = await bareTransport(() => owner); const handler = vi.fn(async () => ({})); + fixture.bundle.transport.setServerRequestHandler(handler); + const sourceRunId = peer.current!.store.state.identity.runId; + owner = scenario === "rebound-before-queue" ? "replacement-owner" : sourceRunId; + try { + fixture.emitMarker(); fixture.emitPermission(); fixture.emitMarker(); + expect((await fixture.iterator.next()).value?.method).toBe("item/agentMessage/delta"); + const request = structuredClone(peer.current!.store.state.committedEvents.find(event => event.eventType === "runtime_request.created")); + expect(request).toBeDefined(); + expect(handler).not.toHaveBeenCalled(); + if (scenario === "rebound-before-delivery") owner = "replacement-owner"; + expect((await fixture.iterator.next()).value?.method).toBe("item/agentMessage/delta"); + expect(handler).toHaveBeenCalledTimes(scenario === "same-owner" ? 1 : 0); + const resolve = () => fixture.bundle.transport.resolveRuntimeRequest!({requestId:"permission-7",turnId:fixture.turnId,resolution:{action:"cancel"}}); + if (scenario !== "same-owner") await expect(resolve()).rejects.toThrow("source_attachment_pending"); + expect(peer.current!.store.state.committedEvents.find(event => event.eventType === "runtime_request.created")).toEqual(request); + // Suppressing presentation must retain the pending request. Its source + // owner can still settle it; the later durable receipt retires the binding. + owner = sourceRunId; + await resolve(); + expect(peer.current!.store.state.commands.filter(command => command.type === "request.resolve")).toHaveLength(1); + peer.current!.emit("runtime_request.resolved", {requestId:"permission-7"}); fixture.emitMarker(); + expect((await fixture.iterator.next()).value?.method).toBe("item/agentMessage/delta"); + await expect(resolve()).rejects.toThrow("no longer pending"); + expect(handler).toHaveBeenCalledTimes(scenario === "same-owner" ? 1 : 0); + } finally { await fixture.close(); } +}); + it("discards a queued permission on close before the notification consumer asks for next", async () => { const fixture = await bareTransport(); const handler = vi.fn(async () => ({})); fixture.bundle.transport.setServerRequestHandler(handler); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport-attachment-recovery.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport-attachment-recovery.test.ts index 72b95a8206c..d672d9bb2e3 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport-attachment-recovery.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport-attachment-recovery.test.ts @@ -84,8 +84,10 @@ it.each([ "wrong_thread", "callback_failure", "callback_async_failure", + "callback_checkpoint_failure", + "callback_checkpoint_owner_change", ].map((ordinaryFollowup) => ({ - lossPoint: "after-confirmation", + lossPoint: ordinaryFollowup.startsWith("callback_") ? "after-result" : "after-confirmation", routed: true, recoveryFault: "none", ordinaryFollowup, @@ -116,6 +118,7 @@ it.each([ let recovering = false; let recoveryClaimCurrent = true; let recoveryFenceActive = true; + let recoveredOwner: string; const completionSnapshotIds: string[] = []; let rejectHeldCompletion: ((error: Error) => void) | undefined; let markHeldCompletionEntered!: () => void; @@ -183,6 +186,26 @@ it.each([ ) { const core = new OriginalCore(options); cores.push(core); + if (recovering && (ordinaryFollowup === "callback_checkpoint_failure" || ordinaryFollowup === "callback_checkpoint_owner_change")) { + const getCommand = core.getCommand.bind(core); + let injected = false; + vi.spyOn(core, "getCommand").mockImplementation(id => { + const command = getCommand(id); + if (!injected && command?.type === "session.snapshot" && command.status === "completed") { + injected = true; + if (ordinaryFollowup === "callback_checkpoint_failure") { + vi.spyOn(core, "flushCheckpoint").mockRejectedValueOnce(new Error("fixture recovery completion callback failed")); + } else { + const flush = core.flushCheckpoint.bind(core); + vi.spyOn(core, "flushCheckpoint").mockImplementationOnce(async () => { + await flush(); + recoveredOwner = "replacement-owner"; + }); + } + } + return command; + }); + } if (recovering && snapshotFault) { const getCommand = core.getCommand.bind(core); snapshotObserverSpy = vi @@ -694,8 +717,10 @@ it.each([ const beforeProviderBytes = await readFile( join(stateDirectory, "runner", "codex-provider-state.json"), ); + recoveredOwner = desired.runId; resumed = createCapabilityRunnerdCodexTransport({ ...options, + get authorityRunId() { return ordinaryFollowup ? recoveredOwner : undefined; }, ...(externalArtifact ? { runnerBinary: "/controller-missing-runner", runnerArtifact: readRunnerdArtifactBinding(options.runnerBinary), @@ -891,7 +916,9 @@ it.each([ if ( snapshotFault || ordinaryFollowup === "callback_failure" || - ordinaryFollowup === "callback_async_failure" + ordinaryFollowup === "callback_async_failure" || + ordinaryFollowup === "callback_checkpoint_failure" || + ordinaryFollowup === "callback_checkpoint_owner_change" ) { const firstRead = within(resumed.transport.request("thread/read", {})); void firstRead.catch(() => undefined); @@ -932,10 +959,10 @@ it.each([ ? ordinaryFollowup === "rejected_snapshot" ? "snapshot observation rejected" : "completion_unproven" - : "recovery completion callback failed", + : ordinaryFollowup === "callback_checkpoint_owner_change" ? "source_attachment_pending" : "recovery completion callback failed", ); expect(recoveryFenceActive).toBe(true); - expect(completionSnapshotIds).toHaveLength(snapshotFault ? 0 : 1); + expect(completionSnapshotIds).toHaveLength(snapshotFault || ordinaryFollowup === "callback_checkpoint_failure" || ordinaryFollowup === "callback_checkpoint_owner_change" ? 0 : 1); const beforeDeniedWork = cores[1]!.store.state.commands.map( (command) => command.commandId, ); @@ -962,18 +989,21 @@ it.each([ beforeDeniedProviderCalls, ); snapshotObserverSpy?.mockRestore(); + recoveredOwner = desired.runId; const providerCallsBeforeRetry = await readFile(callsPath, "utf8"); expect( (await within(resumed.transport.request("thread/read", {}))).thread, ).toMatchObject(thread); expect(recoveryFenceActive).toBe(false); - expect(completionSnapshotIds).toHaveLength(snapshotFault ? 1 : 2); + expect(completionSnapshotIds).toHaveLength(snapshotFault || ordinaryFollowup === "callback_checkpoint_failure" || ordinaryFollowup === "callback_checkpoint_owner_change" ? 1 : 2); expect(new Set(completionSnapshotIds).size).toBe( completionSnapshotIds.length, ); expect(await readFile(callsPath, "utf8")).toBe( providerCallsBeforeRetry, ); + await within(resumed.transport.request("turn/start", { input: [{ type: "text", text: "Work after completion retry" }] })); + for await (const event of resumed.transport.notifications()) if (event.method === "turn/completed") break; return; } const read = await within( @@ -1016,6 +1046,7 @@ it.each([ (await within(resumed.transport.request("thread/read", {}))).thread, ).toMatchObject(thread); const thirdRunId = randomUUID(); + recoveredOwner = thirdRunId; await within( resumed.transport.attachRun!({ runId: thirdRunId, diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts index 4db05f6b55a..fdcb5a2cc54 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts @@ -1173,7 +1173,7 @@ it("installs required accounting acceptance before the Runner acknowledgement ca } }); -it("uses the same API result envelope for live dispatch and recovered receipts", async () => { +it.each([false, true])("uses the same API result envelope and fences handler publication (owner changed: %s)", async changedOwner => { const stateDirectory = await mkdtemp(join(tmpdir(), "runnerd-recovered-api-")); const OriginalCore = durableControlPlane.DurablePrpControlPlane; let callbacks!: ConstructorParameters[0]; @@ -1185,17 +1185,31 @@ it("uses the same API result envelope for live dispatch and recovered receipts", let outcome: { result: unknown; isError?: boolean } | null = { result: { ok: true, apiOperationId: "POST /api/example", data: { saved: true } }, isError: false, }; + let owner: string | undefined; const bundle = createCapabilityRunnerdCodexTransport({ + get authorityRunId() { return owner; }, runnerBinary: defaultCapabilityRunnerdBinary(), codexCommand: fakeCodex, codexArgs: fakeCodexArgs(stateDirectory), stateDirectory, recoverSemanticToolInput: async () => outcome, }); - bundle.transport.setServerRequestHandler(async () => ({ success: !outcome?.isError, - contentItems: [{ type: "inputText", text: JSON.stringify(outcome?.result) }] })); + const handler = vi.fn(async () => ({ success: !outcome?.isError, + contentItems: [{ type: "inputText" as const, text: JSON.stringify(outcome?.result) }] })); + if (!changedOwner) bundle.transport.setServerRequestHandler(handler); try { await bundle.transport.request("thread/start", { cwd: tmpdir() }); const call = { callId: "recovered-api", operationId: "call_api", input: { operationId: "POST /api/example" }, sourceEventId: "saved-event", sourceEventType: "semantic_tool.input", correlation: core.store.state.identity }; + if (changedOwner) { + owner = core.store.state.identity.runId; + const pending = callbacks.onSemanticToolInput!(call); + await new Promise(resolveTurn => setImmediate(resolveTurn)); + owner = "replacement-owner"; + bundle.transport.setServerRequestHandler(handler); + await expect(pending).rejects.toBeInstanceOf(durableControlPlane.SemanticToolNotDispatchedError); + expect(handler).not.toHaveBeenCalled(); + expect(await callbacks.recoverSemanticToolInput!(call)).toBeNull(); + return; + } const live = await callbacks.onSemanticToolInput!(call); expect(live).toMatchObject({ isError: false, result: { ok: true, operationId: "call_api", callId: call.callId, result: outcome!.result } }); expect(await callbacks.recoverSemanticToolInput!(call)).toMatchObject({ result: live.result, isError: live.isError }); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index f699208da3d..7452a2e5944 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -4,7 +4,7 @@ import { resolveGrokDisableWebSearch, admittedGrokWebSearchPolicy } from "../dri import { PRP_COMPLETION_TOOL_NAME, PRP_COMPLETION_RESULT_BRIDGE_INPUT_SCHEMA } from "../contracts/completion-result.js"; import { RunnerLaunchPreparationStore } from "./runner-launch-preparation-store.js"; import type { PreparedRunnerProviderLaunch, RunnerProviderLaunchRequirements } from "./runner-provider-launch.js"; -import { preparedRunnerLaunchWorkingDirectory, validatePreparedRunnerLaunch, runnerProviderLaunchRequirements, createRunnerProviderLaunch, createRunnerdCodexAppServerArgs, trustedRuntimeReadOnlyRoots } from "./runner-provider-launch.js"; +import { preparedRunnerLaunchWorkingDirectory, validatePreparedRunnerLaunch, runnerLaunchSettingsMatch, runnerProviderLaunchRequirements, createRunnerProviderLaunch, createRunnerdCodexAppServerArgs, trustedRuntimeReadOnlyRoots } from "./runner-provider-launch.js"; export { createRunnerdCodexAppServerArgs, resolveRunnerdAcpxPermissionMode, trustedRuntimeReadOnlyRoots, withCodexCollaborationRuntimeInstructions } from "./runner-provider-launch.js"; import { applyProviderObservabilityEnvironment, redactObservabilityDiagnostic } from "../drivers/agent-observability.js"; import type { RunnerExecutionDriver } from "../control-plane/runner-execution-driver.js"; @@ -74,6 +74,7 @@ import { SemanticToolNotDispatchedError, durableRecoveryInternals, inspectWarmRunTransition, + isRunnerRecoveryMaintenanceCommand, spawnRunner, waitForProcess, type RunnerProcessHandle, @@ -1255,6 +1256,8 @@ export interface CapabilityRunnerdCodexTransportOptions extends RunnerExecutionD cursorCloudProxyPath?: string; cursorCloudProxySha256?: string; recoveryPurpose?: "execution" | "stop_cleanup"; + /** Cacheless preparation requires an authenticated authorityRunId throughout this transport's lifetime. */ + recoveryLaunch?: { requirements: RunnerProviderLaunchRequirements }; opencodePermissionMode?: NativeOpenCodePermissionMode; acpxAgent?: QualifiedAcpxAgent; /** Explicit evaluation-only candidate selection, never derived from persisted session input. */ @@ -3660,6 +3663,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { #rejectFailureSignal!: (error: Error) => void; #runnerRecoveryInProgress = false; #startupComplete = false; + readonly #authorityRunId: () => string | undefined; #startupFailureCode = "native_runner_process_exited"; #controlPlaneCheckpoint: ((settlement: "settled" | "unsettled") => Promise | void) | null = @@ -3674,6 +3678,19 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { readonly #bridgedRuntimeInputs = new Map(); constructor(readonly options: CapabilityRunnerdCodexTransportOptions) { + this.#authorityRunId = options.recoveryLaunch ? () => { + const owner = options.authorityRunId; + if (typeof owner !== "string" || !owner.trim()) throw new Error("native_runner_recovery_launch_requirements_invalid"); + return owner; + } : () => options.authorityRunId; + if (options.recoveryLaunch) this.#authorityRunId(); + const requirements = options.recoveryLaunch?.requirements; + if (options.recoveryLaunch && (!options.prepareLaunch || !options.resumeProviderSession?.driverSessionId + || requirements?.schema !== "paperclip.runner.launch-requirements.v1" + || requirements.provider !== (options.provider ?? "codex") || !options.prpIdentity + || commandDigest(requirements.identity) !== commandDigest(options.prpIdentity))) { + throw new Error("native_runner_recovery_launch_requirements_invalid"); + } if (options.adoptExistingRunner && options.authorizeRunnerAdoption && !options.prepareLaunch) { throw new Error("runner_adoption_authorization_requires_launch_preparation"); } @@ -3811,6 +3828,12 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { }; } + #retainsSourceInputs(): boolean { + const identity = this.#eventIdentity ?? this.options.prpIdentity; + return this.#authorityRunId() !== undefined && identity !== undefined && identity !== null + && this.#authorityRunId() !== identity.runId; + } + evidence(): CapabilityRunnerdProcessEvidence { return structuredClone(this.#evidence); } @@ -3821,6 +3844,11 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { ): Promise> { if (this.#closed) throw new Error("PRP Codex transport is closed"); this.#throwIfFailed(); + const expectedThread = this.#threadId || this.options.resumeProviderSession?.driverSessionId; + if (expectedThread && params.threadId !== undefined && params.threadId !== expectedThread + && ["thread/read", "thread/resume", "thread/turns/list", "thread/items/list", "thread/goal/get"].includes(method)) { + throw new Error("codex_history_identity_mismatch"); + } if (this.#sourceRecoveryIdentity && !["initialize", "collaborationMode/list"].includes(method)) { throw new Error("native_runner_source_attachment_pending"); } @@ -4119,6 +4147,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { takeRestoredRuntimeRequests(): CodexRpcServerRequest[] { this.#throwIfFailed(); + if (this.#retainsSourceInputs()) return []; const requests = this.#restoredRuntimeRequests; this.#restoredRuntimeRequests = []; return requests.filter(request => this.#bridgedRuntimeInputs.has(String(request.id)) @@ -4133,7 +4162,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { } #recoverPendingSemanticInputsWhenReady(): void { - if (this.options.recoveryPurpose !== "stop_cleanup" && this.#startupComplete && this.#handlerInstalled && this.#sourceRecoveryIdentity === null && this.#pendingWarmRecoveryCompletion === null) { + if (!this.#retainsSourceInputs() && this.options.recoveryPurpose !== "stop_cleanup" && this.#startupComplete && this.#handlerInstalled && this.#sourceRecoveryIdentity === null && this.#pendingWarmRecoveryCompletion === null) { void Promise.resolve().then(() => this.#core?.recoverPendingSemanticInputs()) .catch(error => this.#failTransport(error instanceof Error ? error : new Error(String(error)))); } @@ -4238,11 +4267,18 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { if (this.#pendingWarmRecoveryCompletion !== null) { throw new Error("native_runner_warm_transition_completion_pending"); } + const retainingSource = this.#retainsSourceInputs(); const core = this.#core; if (!core || !this.#startupComplete) { throw new Error("native_runner_prp_run_rotation_unavailable"); } + if (retainingSource && this.#bridgedRuntimeInputs.size > 0) { + throw new Error("native_runner_source_runtime_reconciliation_required"); + } await this.#awaitWarmRunAttachmentReady(); + if (retainingSource && !core.semanticToolResultsSettled()) { + throw new Error("native_runner_source_semantic_reconciliation_required"); + } this.#assertStartupOpen(); const currentRunGrant = input.currentRunGrant ? projectCopilotRunGrant(input.currentRunGrant, this.options.runnerFilesystemRoot) @@ -4295,11 +4331,30 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { ); if (this.options.prepareLaunch && !staged) { const priorProvider = record(runAttachTemplate.provider); - const priorRequirements = new RunnerLaunchPreparationStore(this.#root).load(prior)?.requirements; - await this.#prepareOwnedProvider(runAttachTemplate, desired, - currentRunGrant ? currentRunGrant.runtimeContext : priorRequirements?.runtimeContext - ?? priorProvider.runtimeContext as NativeRuntimeContextSnapshot | null, - currentRunGrant?.instructions ?? priorRequirements?.instructions ?? String(priorProvider.instructions ?? "")); + const sourceRequirements = this.options.recoveryLaunch?.requirements; + const priorRequirements = new RunnerLaunchPreparationStore(this.#root).load(prior)?.requirements + ?? (sourceRequirements && recoveryIdentityMatches(sourceRequirements.identity, prior) ? sourceRequirements : undefined); + const runtimeContext = currentRunGrant ? currentRunGrant.runtimeContext : priorRequirements?.runtimeContext + ?? priorProvider.runtimeContext as NativeRuntimeContextSnapshot | null; + const instructions = currentRunGrant?.instructions ?? priorRequirements?.instructions ?? String(priorProvider.instructions ?? ""); + let recordedRequirements: RunnerProviderLaunchRequirements | undefined; + if (sourceRequirements && priorRequirements + && priorRequirements.provider === sourceRequirements.provider + && prior.runnerInstanceId === sourceRequirements.identity.runnerInstanceId + && prior.environmentLeaseId === sourceRequirements.identity.environmentLeaseId + && prior.normalizedSessionId === sourceRequirements.identity.normalizedSessionId) { + // A recovered session keeps its frozen policy through later warm attaches. + const selected = runnerProviderLaunchRequirements(this.options, desired, {}, runtimeContext, instructions); + const settings = { ...priorRequirements.settings }; + for (const key of Object.keys({ ...settings, ...selected.settings }) as Array) { + if (this.options[key] !== undefined) Object.assign(settings, { [key]: selected.settings[key] }); + } + recordedRequirements = { ...priorRequirements, identity: desired, runtimeContext, instructions }; + if (!runnerLaunchSettingsMatch(recordedRequirements, settings)) { + throw new Error("native_runner_recovery_launch_policy_changed"); + } + } + await this.#prepareOwnedProvider(runAttachTemplate, desired, runtimeContext, instructions, undefined, recordedRequirements); } this.#assertStartupOpen(); this.#runAttachTemplate = structuredClone(runAttachTemplate); @@ -4375,6 +4430,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { await new Promise((resolveWait) => setTimeout(resolveWait, 10)); } this.#assertStartupOpen(); + if (retainingSource) await core.flushCheckpoint(); this.#eventIdentity = structuredClone(desired); this.#eventSourceSeq = 0; this.#deferredTurnStartEvents = []; @@ -4382,6 +4438,10 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { if (!previousReleased) await previousRelease?.(); previousReleased = true; this.#controlPlaneRelease = registration?.release ?? null; + if (retainingSource) { + core.enableSemanticInputDispatch(desired); + this.#recoverPendingSemanticInputsWhenReady(); + } } catch (error) { const failure = error instanceof Error ? error : new Error(String(error)); // The future route is ours from registration onward, including failures @@ -4412,6 +4472,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { turnId: string; resolution: HarnessRuntimeRequestResolution; }): Promise { + if (this.#retainsSourceInputs()) throw new Error("native_runner_source_attachment_pending"); if (this.#pendingWarmRecoveryCompletion !== null) { throw new Error("native_runner_warm_transition_completion_pending"); } @@ -5040,8 +5101,12 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { this.#recordPreparedWorkingDirectory(prepared); this.#assertStartupOpen(); template.provider = prepared.provider; - new RunnerLaunchPreparationStore(this.#root).save(requirements, prepared, this.#core!.store.state.identity); - this.#core!.persistPreparedLaunchConfiguration(identity, prepared.configurationId); + const sourceLaunch = this.options.recoveryLaunch + && recoveryIdentityMatches(identity, this.options.recoveryLaunch.requirements.identity); + if (!sourceLaunch) { + new RunnerLaunchPreparationStore(this.#root).save(requirements, prepared, this.#core!.store.state.identity); + this.#core!.persistPreparedLaunchConfiguration(identity, prepared.configurationId); + } return prepared; } @@ -5320,19 +5385,22 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { const dynamicTools = Array.isArray(params.dynamicTools) ? params.dynamicTools.map(record) : []; + const authorityRunId = this.#authorityRunId; const core = new DurablePrpControlPlane({ stateDirectory: resolve(this.#root, "control-plane"), identity, expectedRunnerVersion: runnerArtifact.version, expectedRunnerDigest: runnerArtifact.digest, onProtocolIntegrityError: (error) => this.#failTransport(error), - semanticInputMode: this.options.recoveryPurpose === "stop_cleanup" ? "retain_only" : "dispatch", + get authorityRunId() { return authorityRunId(); }, + semanticInputMode: this.#retainsSourceInputs() || this.options.recoveryPurpose === "stop_cleanup" ? "retain_only" : "dispatch", onSemanticToolInput: this.options.recoveryPurpose === "stop_cleanup" ? undefined : (call) => this.#handleSemanticToolInput(call), recoverSemanticToolInput: this.options.recoveryPurpose === "stop_cleanup" ? undefined : call => this.#recoverSemanticToolInput(call), onCommittedEvent: async (event) => { - if (this.#sourceRecoveryIdentity && event.runId === this.#sourceRecoveryIdentity.runId) { + const source = this.#sourceRecoveryIdentity ?? (this.#retainsSourceInputs() ? this.#eventIdentity ?? this.options.prpIdentity : undefined); + if (source && event.runId === source.runId) { if (!this.options.onRecoveryAccountingEvent && this.options.onAccountingEvent) throw new Error("native_runner_source_accounting_unavailable"); - await this.options.onRecoveryAccountingEvent?.(event, this.#sourceRecoveryIdentity); + await this.options.onRecoveryAccountingEvent?.(event, source); } else await this.options.onAccountingEvent?.(event); if (event.eventType === "auth.credentials.updated") { await this.options.onCredentialsUpdated?.(event); @@ -5526,6 +5594,21 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { let identity = controlPlaneState ? controlPlaneIdentity(controlPlaneState) : desiredIdentity; + const predecessor = this.#authorityRunId() !== undefined + && this.#authorityRunId() !== identity.runId && recoveryIdentityMatches(identity, desiredIdentity); + const lostPredecessor = predecessor && !this.options.adoptExistingRunner; + if (predecessor && (!controlPlaneState || !recoveryIdentityMatches(identity, desiredIdentity) + || controlPlaneState.warmTransition || controlPlaneState.stagedRunAttachment + || (Array.isArray(controlPlaneState.commands) && controlPlaneState.commands.some(entry => { + const command = record(entry); + return command.status === "pending" && !isRunnerRecoveryMaintenanceCommand(String(command.type)); + })))) { + throw new Error("native_runner_source_reconciliation_requires_exact_authority"); + } + if (lostPredecessor && (this.options.provider ?? "codex") !== "codex") throw new Error("native_runner_recovery_integrity_unsupported"); + if (lostPredecessor && this.options.sourceRecoveryVersion !== 1) { + throw new Error("native_runner_recovery_integrity_unsupported"); + } const hasControlPlaneWarmBoundary = controlPlaneState?.warmTransition !== undefined || controlPlaneState?.schema === @@ -5541,7 +5624,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { ); }); if ( - this.options.readRunnerState && + !lostPredecessor && this.options.readRunnerState && controlPlaneState && (hasControlPlaneWarmBoundary || controlPlaneState.completedWarmTransition !== undefined || @@ -5834,19 +5917,22 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { opencodeExecutable!, ) : undefined; + const authorityRunId = this.#authorityRunId; const core = new DurablePrpControlPlane({ stateDirectory: controlPlaneDirectory, identity, expectedRunnerVersion: runnerArtifact.version, expectedRunnerDigest: runnerArtifact.digest, onProtocolIntegrityError: (error) => this.#failTransport(error), - semanticInputMode: this.options.recoveryPurpose === "stop_cleanup" ? "retain_only" : "dispatch", + get authorityRunId() { return authorityRunId(); }, + semanticInputMode: this.#retainsSourceInputs() || this.options.recoveryPurpose === "stop_cleanup" ? "retain_only" : "dispatch", onSemanticToolInput: this.options.recoveryPurpose === "stop_cleanup" ? undefined : (call) => this.#handleSemanticToolInput(call), recoverSemanticToolInput: this.options.recoveryPurpose === "stop_cleanup" ? undefined : call => this.#recoverSemanticToolInput(call), onCommittedEvent: async (event) => { - if (this.#sourceRecoveryIdentity && event.runId === this.#sourceRecoveryIdentity.runId) { + const source = this.#sourceRecoveryIdentity ?? (this.#retainsSourceInputs() ? this.#eventIdentity ?? this.options.prpIdentity : undefined); + if (source && event.runId === source.runId) { if (!this.options.onRecoveryAccountingEvent && this.options.onAccountingEvent) throw new Error("native_runner_source_accounting_unavailable"); - await this.options.onRecoveryAccountingEvent?.(event, this.#sourceRecoveryIdentity); + await this.options.onRecoveryAccountingEvent?.(event, source); } else await this.options.onAccountingEvent?.(event); if (event.eventType === "auth.credentials.updated") { await this.options.onCredentialsUpdated?.(event); @@ -5877,7 +5963,8 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { // Reusing resources still requires current environment authorization. The // optional hook avoids re-preparation after controller storage is lost. // Existing consumers retain the preparation callback's authorization path. - const authorizeAdoption = this.options.adoptExistingRunner && this.options.authorizeRunnerAdoption; + const authorizeAdoption = !this.options.recoveryLaunch + && this.options.adoptExistingRunner && this.options.authorizeRunnerAdoption; if (this.options.prepareLaunch && authorizeAdoption) { const configurationId = core.preparedLaunchConfigurationId(desiredIdentity); if (!configurationId) throw new Error("runner_launch_recovery_configuration_unavailable"); @@ -5886,7 +5973,14 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { this.#recordPreparedWorkingDirectory({ provider: record(record(controlPlaneState!.runAttachTemplate).provider) }); } let environmentPrepared: PreparedRunnerProviderLaunch | undefined; - if (this.options.prepareLaunch && !authorizeAdoption) { + if (this.options.recoveryLaunch) { + const requirements = this.options.recoveryLaunch.requirements; + const template = rotatedRunAttachPayload(controlPlaneState!, desiredIdentity, this.#authorizedTools, + this.options.resumeCompletionContract); + environmentPrepared = await this.#prepareOwnedProvider(template, desiredIdentity, + requirements.runtimeContext, requirements.instructions, undefined, requirements); + this.#runAttachTemplate = structuredClone(template); + } else if (this.options.prepareLaunch && !authorizeAdoption) { const preparations = new RunnerLaunchPreparationStore(this.#root); const saved = preparations.load(desiredIdentity); const previous = preparations.load(controlPlaneIdentity(controlPlaneState!)); @@ -6035,6 +6129,18 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { : "restored provider identity from the exact durable checkpoint; awaiting live confirmation", ); } + if (queuedSourceAttachment && this.#retainsSourceInputs() && this.options.recoveryPurpose !== "stop_cleanup" + && this.#checkpointProviderIdentityExpectation === null) { + if (!this.#threadId || !this.#sessionId) throw new Error("native_runner_warm_transition_completion_unproven"); + const saved = this.options.resumeProviderSession; + this.#checkpointProviderIdentityExpectation = { + driverSessionId: saved?.driverSessionId ?? this.#threadId, + providerSessionId: saved?.providerSessionId ?? this.#sessionId, + providerIdentity: saved?.providerIdentity === undefined ? this.#providerIdentity : record(saved.providerIdentity), + }; + this.#confirmCheckpointProviderIdentity({ threadId: this.#threadId, sessionId: this.#sessionId, + providerIdentity: this.#providerIdentity }, "committed queued attachment identity"); + } if (staged && !sourceAttachmentTarget) { await this.options.onStagedRunAttachment?.(core, { sourceIdentity: staged.sourceIdentity, targetIdentity: staged.targetIdentity, templateDigest: staged.templateDigest }); @@ -6080,7 +6186,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { ? recoveryIdentityMatches(warmRecovery?.runnerIdentity ?? queuedSourceAttachment!.sourceIdentity, recoverySource!) ? oldTransitionRegistration : newTransitionRegistration : this.options.controlPlaneRegistration - ? await this.options.controlPlaneRegistration(core, this.#sourceRecoveryIdentity ?? undefined) : null; + ? await this.options.controlPlaneRegistration(core, this.#retainsSourceInputs() ? identity : this.#sourceRecoveryIdentity ?? undefined) : null; this.#startupFailureCode = registration?.startupFailureCode ?? "runner_local_connect_failed"; if (registration === null) { @@ -6285,6 +6391,15 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { if (core.getCommand(warmRecovery?.commandId ?? queuedSourceAttachment!.commandId!)?.status !== "completed") { throw new Error("native_runner_warm_transition_result_unproven"); } + if (queuedSourceAttachment && this.#retainsSourceInputs() && this.options.recoveryPurpose !== "stop_cleanup") { + const receipt = core.store.state.completedWarmTransition?.receipt; + if (!receipt || receipt.commandId !== queuedSourceAttachment.commandId + || !recoveryIdentityMatches(receipt.oldIdentity, queuedSourceAttachment.sourceIdentity) + || !recoveryIdentityMatches(receipt.newIdentity, recoveryTarget!)) { + throw new Error("native_runner_warm_transition_completion_unproven"); + } + this.#pendingWarmRecoveryCompletion = { transitionId: receipt.transitionId, identity: structuredClone(recoveryTarget!) }; + } desiredIdentity = recoveryTarget!; this.#durableTurnId = desiredIdentity.turnId; this.#eventIdentity = structuredClone(desiredIdentity); @@ -6363,6 +6478,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { } async #recoverSemanticToolInput(call: Parameters>[0]) { + if (this.#retainsSourceInputs() || (this.#authorityRunId() !== undefined && call.correlation.runId !== this.#authorityRunId())) return null; if (this.#sourceRecoveryIdentity) throw new Error("native_runner_source_semantic_reconciliation_required"); if (call.operationId === "paperclip_finish" || call.operationId === "paperclip_block") { return this.#handleSemanticToolInput(call); @@ -6414,6 +6530,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { } if ( !accepted || + this.#retainsSourceInputs() || this.#closed || epoch !== this.#turnStartResponseEpoch || threadId !== this.#threadId || @@ -6709,14 +6826,24 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { // final activation ACK. Callback failure retains the completion gate; // retry observes a fresh snapshot, never repeats a provider turn. const completing = (this.#warmRecoveryCompletionInFlight ??= - Promise.resolve().then(() => - this.options.onWarmTransitionRecoveryCompleted?.({ + Promise.resolve().then(async () => { + this.#assertStartupOpen(); + await core.flushCheckpoint(); + this.#assertStartupOpen(); + const owner = this.#authorityRunId(); + if (owner !== undefined && owner !== completion.identity.runId) throw new Error("native_runner_source_attachment_pending"); + if (!recoveryIdentityMatches(core.store.state.identity, completion.identity) || core.store.state.warmTransition !== undefined) { + throw new Error("native_runner_warm_transition_completion_unproven"); + } + await this.options.onWarmTransitionRecoveryCompleted?.({ transitionId: completion.transitionId, - }), - )); + }); + })); try { await completing; + this.#assertStartupOpen(); if (this.#pendingWarmRecoveryCompletion === completion) { + if (this.options.recoveryPurpose !== "stop_cleanup") core.enableSemanticInputDispatch(completion.identity); this.#pendingWarmRecoveryCompletion = null; this.#recoverPendingSemanticInputsWhenReady(); } @@ -6985,10 +7112,14 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { }, }; const threadId = this.#threadId, turnId = this.#turnId, handler = this.#handler; + const runId = core.store.state.identity.runId; this.#queue.pushControl(request, () => { - // A detached controller, replaced turn or settled durable request - // cannot acquire approval authority when a slow consumer resumes. + // Recheck authority when the consumer resumes. Retain the durable + // fact and pending binding while stale source presentation is held. + const owner = this.#authorityRunId(); if (this.#closed || this.#failure || this.#core !== core || this.#handler !== handler + || this.#retainsSourceInputs() + || core.store.state.identity.runId !== runId || owner !== undefined && owner !== runId || this.#threadId !== threadId || this.#turnId !== turnId || this.#bridgedRuntimeInputs.get(requestId) !== binding) return; try { diff --git a/packages/paperclip-runner/src/live/runnerd-source-reconciliation.test.ts b/packages/paperclip-runner/src/live/runnerd-source-reconciliation.test.ts new file mode 100644 index 00000000000..603f9123d63 --- /dev/null +++ b/packages/paperclip-runner/src/live/runnerd-source-reconciliation.test.ts @@ -0,0 +1,333 @@ +import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { createHash } from "node:crypto"; +import { spawn } from "node:child_process"; +import { createServer } from "node:http"; +import type { DurablePrpControlPlane } from "../control-plane/durable-prp-control-plane.js"; +import { expect, it, vi } from "vitest"; +import { createCapabilityRunnerdCodexTransport, type CapabilityRunnerdCodexTransportOptions } from "./runnerd-codex-transport.js"; +import { createRunnerProviderLaunch, runnerLaunchRequirementsDigest, runnerProviderLaunchRequirements, type RunnerProviderLaunchRequirements } from "./runner-provider-launch.js"; +import { defaultCapabilityRunnerdBinary, fakeCodex, fakeCodexArgs } from "../../test/support/runnerd-transport.js"; +import { createNativeSessionBackend } from "../backends/native-backend-factory.js"; +import type { NativeExecutionInput } from "../contracts/native-execution.js"; +import type { PersistedNativeSession } from "../contracts/native-session-backend.js"; + +type Bundle = ReturnType; +const read = async (path: string) => JSON.parse(await readFile(path, "utf8")); +const absent = (pid: number) => { try { process.kill(pid, 0); return false; } catch (error) { return (error as NodeJS.ErrnoException).code === "ESRCH"; } }; +const kill = (pid: number | null) => { if (pid) { try { process.kill(-pid, "SIGKILL"); } catch (error) { if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error; } } }; + +async function sourceFixture(registration?: CapabilityRunnerdCodexTransportOptions["controlPlaneRegistration"], nativeSource = false, homeMode: "isolated" | "preconfigured" = "isolated", disableCodexWebSearch = false) { + const root = await mkdtemp(join(tmpdir(), "runner-source-reconciliation-")); + const home = join(root, "empty-codex-home"); await mkdir(home); + const report = join(root, "completion.json"); + if (nativeSource) await writeFile(report, JSON.stringify({ schema: "paperclip.run_result.v1", reportedWorkDisposition: "done", summary: "Completed fake work", + completionClaim: { contractRevision: "1", objectiveSatisfied: true, criteria: [{ criterionId: "objective", status: "satisfied", evidenceRefs: [] }], remainingWork: [] }, + evidence: [], verification: [], attentionRequests: [], artifacts: [] })); + const identity = { runnerInstanceId: "runner-source", environmentLeaseId: "lease-source", runId: "run-source", normalizedSessionId: "session-source", turnId: "turn-source", itemId: "item-source" }; + const options = { + runnerBinary: defaultCapabilityRunnerdBinary(), codexCommand: fakeCodex, + codexArgs: fakeCodexArgs(root, "--call-log", join(root, "calls"), "--record-process-start", "--durable-turn-ids", ...(nativeSource ? ["--terminal-result-file", report] : [])), + stateDirectory: root, sourceCodexHome: home, prpIdentity: identity, homeMode, + ...(disableCodexWebSearch ? { disableCodexWebSearch: true } : {}), + ...(homeMode === "preconfigured" ? { environment: { ...process.env, HOME: home, CODEX_HOME: home } } : {}), + lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 60_000 }, runnerProcessReconnectGraceMs: null, runnerReconnectGraceMs: 3_000, + ...(registration ? { controlPlaneRegistration: registration } : {}), + }; + const bundles: Bundle[] = []; + const first = createCapabilityRunnerdCodexTransport(options); bundles.push(first); + const runnerFile = join(root, "runner", "runner-state.json"), providerFile = join(root, "runner", "codex-provider-state.json"), coreFile = join(root, "control-plane", "control-plane-state.json"); + const calls = async (method: string) => (await readFile(join(root, "calls"), "utf8")).split("\n").filter(line => line === method).length; + const settled = async (completed = true) => vi.waitFor(async () => { + const runner = await read(runnerFile), provider = await read(providerFile), core = await read(coreFile); + expect(runner.outbox).toEqual([]); expect(runner.ackedSourceSeq).toBe(runner.nextSourceSeq - 1); + expect(provider.pendingEvents).toEqual([]); expect(provider.queuedEvents).toEqual([]); + expect(provider.activeProviderTurnId).toBeNull(); expect(provider.completedTurnAuthoritative).toBe(completed); + expect(core.commands.some((command: { status: string }) => command.status === "pending")).toBe(false); + }, { timeout: 5_000, interval: 25 }); + const complete = async (bundle: Bundle, text: string) => { + await bundle.transport.request("turn/start", { input: [{ type: "text", text }] }); + for await (const notification of bundle.transport.notifications()) if (notification.method === "turn/completed") return; + throw new Error("missing terminal turn"); + }; + const dispose = async () => { + for (const bundle of bundles.reverse()) { await bundle.transport.close().catch(() => undefined); const evidence = bundle.evidence(); kill(evidence.runnerPid); kill(evidence.codexPid); } + await rm(root, { recursive: true, force: true }); + }; + const nativeIdentity = { companyId: "company", agentId: "agent", issueId: "issue", runId: identity.runId, sessionId: identity.normalizedSessionId }; + const execution: NativeExecutionInput = { schema: "paperclip.native-execution-input.v1", + binding: { companyId: nativeIdentity.companyId, agentId: nativeIdentity.agentId, issueId: nativeIdentity.issueId, + runId: identity.runId, executionWorkspaceId: "workspace" }, + task: { identifier: "TASK-1", title: "Complete fake work", description: null, prompt: "Complete fake work", workMode: "standard" }, + workspace: { cwd: root, repoUrl: null, repoRef: null, branchName: null }, + session: { normalizedSessionId: identity.normalizedSessionId, driverKind: "codex_app_server", protocolVersion: 1 }, + provider: { kind: "codex", model: null, approvalPolicy: "never" }, + completionContract: { id: "contract", sha256: `sha256:${"a".repeat(64)}`, schemaVersion: "paperclip.completion-contract.v1", + contract: { revision: "1", objective: "Complete fake work", criteria: [{ id: "objective", requirement: "Complete fake work" }] } }, interactionResponses: [], credentialBindings: [] }; + try { + let nativeSnapshot: PersistedNativeSession | undefined; + let thread: { id: string; sessionId: string }; + if (nativeSource) { + const session = await createNativeSessionBackend(execution, { codexTransportFactory: () => first.transport, + environment: { PAPERCLIP_WORKSPACE_CWD: root }, dynamicTools: [] }).openSession({ identity: nativeIdentity, workingDirectory: root }); + await session.startTurn({ message: { role: "user", text: "Complete before abrupt loss." } }); + for await (const event of session.events()) if (event.eventType === "turn.completed") break; + nativeSnapshot = await session.snapshot(); + expect(nativeSnapshot.terminal?.runTerminalState).toBe("succeeded"); + thread = { id: nativeSnapshot.sessionId, sessionId: nativeSnapshot.providerSessionId! }; + } else { + const opened = await first.transport.request("thread/start", { cwd: root, dynamicTools: [] }); + const originalThread = opened.thread as { id: string; sessionId: string }; + thread = { id: originalThread.id, sessionId: originalThread.sessionId }; + await complete(first, "Complete before abrupt loss."); + } + await settled(); + const evidence = first.evidence(); + await first.detachControllerForRestart(); + expect(absent(evidence.runnerPid!)).toBe(false); expect(absent(evidence.codexPid!)).toBe(false); + kill(evidence.runnerPid); kill(evidence.codexPid); + await vi.waitFor(() => { expect(absent(evidence.runnerPid!)).toBe(true); expect(absent(evidence.codexPid!)).toBe(true); }); + expect((await read(runnerFile)).lifecycle).toBe("ready"); + const recover = (overrides: Partial = {}) => { + const bundle = createCapabilityRunnerdCodexTransport({ ...options, sourceRecoveryVersion: 1, resumeProviderSession: { driverSessionId: thread.id, providerSessionId: thread.sessionId }, onRecoveryAccountingEvent: async () => {}, ...overrides, get authorityRunId() { return Object.hasOwn(overrides, "authorityRunId") ? overrides.authorityRunId : "run-target"; } }); + bundles.push(bundle); return bundle; + }; + return { root, options, identity, thread, runnerFile, providerFile, coreFile, calls, complete, settled, recover, dispose, nativeIdentity, execution, nativeSnapshot }; + } catch (error) { await dispose(); throw error; } +} + +it.each(["isolated", "preconfigured"] as const)("recovers the real native backend before attaching the next run with %s home", async homeMode => { + const f = await sourceFixture(undefined, true, homeMode); + try { + expect(f.nativeSnapshot!.sessionId).toBe(f.thread.id); + let owner = "run-target"; + const source = f.recover({ ...(homeMode === "isolated" ? { environment: { ...process.env, PAPERCLIP_RUNNER_HOME_MODE: "preconfigured" } } : {}), get authorityRunId() { return owner; } }); + await expect(source.transport.request("thread/read", { threadId: "foreign-thread" })).rejects.toThrow("history_identity_mismatch"); + const recovered = await createNativeSessionBackend(f.execution, { codexTransportFactory: () => source.transport, + environment: { PAPERCLIP_WORKSPACE_CWD: f.root }, dynamicTools: [] }).recoverSession!(f.nativeSnapshot!, { signal: new AbortController().signal }); + expect(recovered.recovered).toBe(true); + expect(await f.calls("turn/start")).toBe(1); expect(await f.calls("thread/resume")).toBe(1); + const before = source.evidence(); + await expect(source.transport.request("thread/resume", { threadId: "foreign-thread" })).rejects.toThrow("history_identity_mismatch"); + await expect(source.transport.request("thread/turns/list", { threadId: "foreign-thread" })).rejects.toThrow("history_identity_mismatch"); + await expect(recovered.session!.startTurn({ message: { role: "user", text: "Not yet authorized" } })).rejects.toThrow(); + expect(await f.calls("turn/start")).toBe(1); + await expect(source.transport.request("thread/goal/set", { objective: "Unowned source work", status: "active" })).rejects.toThrow("source_attachment_pending"); + expect((await read(f.coreFile)).commands.some((command: { type: string }) => command.type === "session.goal.set")).toBe(false); + for (const runId of ["run-target", "run-warm-third"]) { + owner = runId; + await recovered.session!.attachRun!({ identity: { ...f.nativeIdentity, runId } }); + expect(source.evidence().runnerPid).toBe(before.runnerPid); + await recovered.session!.startTurn({ message: { role: "user", text: "Continue under the current run." } }); + let completed = false; + for await (const event of recovered.session!.events()) { + if (event.eventType === "turn.completed" && event.runId === runId) { completed = true; break; } + } + expect(completed).toBe(true); await f.settled(); + const snapshot = await recovered.session!.snapshot(); + expect(snapshot.sessionId).toBe(f.thread.id); + expect(snapshot.terminal).toMatchObject({ runTerminalState: "succeeded" }); + expect(snapshot.semanticResult?.reportedWorkDisposition).toBe("done"); + } + expect(await f.calls("turn/start")).toBe(3); expect(await f.calls("process-start")).toBe(2); + expect(await f.calls("thread/resume")).toBe(1); + await recovered.session!.close({ reason: "test complete" }); + } finally { await f.dispose(); } +}, 30_000); + +it.each(["attach", "suspend"] as const)("reconciles killed source then uses ordinary %s without a cold receipt", async mode => { + const fixture = await sourceFixture(); + try { + const resumed = fixture.recover(); + expect((await resumed.transport.request("thread/read", {})).thread).toMatchObject(fixture.thread); + await fixture.settled(); + for (const method of ["turn/start", "turn/steer", "thread/goal/set"]) { + await expect(resumed.transport.request(method, method === "turn/steer" ? { input: [{ type: "text", text: "Unowned" }] } : {})).rejects.toThrow("source_attachment_pending"); + } + expect(await fixture.calls("turn/start")).toBe(1); + const target = { ...fixture.identity, runId: "run-target", turnId: "turn-target", itemId: "item-target" }; + if (mode === "suspend") { + await resumed.transport.close(); + expect((await read(fixture.runnerFile)).lifecycle).toBe("suspended"); + expect(await fixture.calls("process-start")).toBe(2); + } else { + const before = resumed.evidence(); + await resumed.transport.attachRun!(target); + expect(resumed.evidence()).toMatchObject({ runnerPid: before.runnerPid, codexPid: before.codexPid }); + await fixture.complete(resumed, "Continue under new authority."); await fixture.settled(); + expect((await read(fixture.coreFile)).identity).toEqual(target); + expect((await read(fixture.providerFile)).threadId).toBe(fixture.thread.id); + expect(await fixture.calls("turn/start")).toBe(2); + expect(await fixture.calls("process-start")).toBe(2); + } + } finally { await fixture.dispose(); } +}, 30_000); + +it.each([["launch", "default"], ["adopt", "default"], ["adopt-denied", "default"], ["launch", "saved"], ["launch", "conflict"]] as const)("prepares a cacheless source %s with %s policy without replacing the ordinary source configuration", async (mode, policy) => { + let authority: DurablePrpControlPlane | null = null; + const server = createServer(); + server.on("upgrade", (request, socket, head) => { + if (authority) authority.handleUpgrade(request, socket, "/runner", head); + else socket.destroy(); + }); + await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("missing source listener"); + const registeredIdentities: unknown[] = []; + const registration: NonNullable = async (next, identity) => { + registeredIdentities.push(identity); + authority = next; + return { connectUrl: `ws://127.0.0.1:${address.port}/runner`, release: () => { if (authority === next) authority = null; } }; + }; + let cleanup = async () => {}; + try { + const fixture = await sourceFixture(registration, false, "isolated", policy !== "default"); cleanup = fixture.dispose; + const before = await read(fixture.coreFile), provider = before.runAttachTemplate.provider; + before.preparedLaunchConfigurations = [{ identity: fixture.identity, configurationId: "old-source-configuration" }]; + await writeFile(fixture.coreFile, JSON.stringify(before)); + const requirements = runnerProviderLaunchRequirements(fixture.options, fixture.identity, { + model: provider.model, approvalPolicy: provider.approvalPolicy, + permissions: provider.readOnly ? "paperclip-runner-workspace-read-only" : "paperclip-runner-workspace-only", + conversationMode: provider.conversationMode, + config: { include_collaboration_mode_instructions: provider.includeCollaborationModeInstructions !== false }, + }, null, "You are a Paperclip agent."); + const prepare = vi.fn(async (input: RunnerProviderLaunchRequirements) => ({ configurationId: input.identity.runId === fixture.identity.runId ? "new-source-incarnation" : "target-configuration", requirementsDigest: runnerLaunchRequirementsDigest(input), + provider: createRunnerProviderLaunch(input, { options: fixture.options, root: fixture.root, cwd: provider.cwd, + codexHome: join(fixture.root, "codex-home"), codexCommand: fakeCodex, providerNodeCommand: "", opencodeProxyPath: "", acpxSidecarPath: "" }) })); + expect((await prepare(requirements)).provider).toEqual(provider); prepare.mockClear(); + const binary = fixture.options.runnerBinary; + let owner: string | undefined = "prepared-target"; + const sourceOptions: Partial = { sourceRecoveryVersion: 1, get authorityRunId() { return owner; }, + recoveryLaunch: { requirements }, prepareLaunch: prepare, + disableCodexWebSearch: policy === "conflict" ? false : undefined, + runnerArtifact: { version: "0.3.0", digest: `sha256:${createHash("sha256").update(await readFile(binary)).digest("hex")}` }, + controlPlaneRegistration: registration, + runnerProcessLauncher: spec => { + expect(spec.nativeRunner).not.toHaveProperty("recoveryPurpose"); + expect(spec.args).not.toContain("--source-reconcile"); + const child = spawn(binary, [...spec.args], { env: spec.environment, stdio: "ignore", detached: process.platform !== "win32" }); + return { child, completion: new Promise((resolve, reject) => { + child.once("error", reject); child.once("exit", (code, signal) => resolve({ code, signal, stdout: "", stderr: "" })); + }) }; + }, + }; + if (mode === "launch") { + const launch = { ...fixture.options, ...sourceOptions, + resumeProviderSession: { driverSessionId: fixture.thread.id, providerSessionId: fixture.thread.sessionId } }; + for (const authorityRunId of [undefined, "", " "]) { + expect(() => createCapabilityRunnerdCodexTransport({ ...launch, authorityRunId })) + .toThrow("recovery_launch_requirements_invalid"); + } + expect(() => createCapabilityRunnerdCodexTransport({ ...launch, authorityRunId: fixture.identity.runId })).not.toThrow(); + expect(prepare).not.toHaveBeenCalled(); + } + let resumed = fixture.recover(sourceOptions); + expect((await resumed.transport.request("thread/read", {})).thread).toMatchObject(fixture.thread); + expect(prepare).toHaveBeenCalledExactlyOnceWith(requirements, undefined); + expect(registeredIdentities.at(-1)).toEqual(fixture.identity); + expect((await read(fixture.coreFile)).preparedLaunchConfigurations).toEqual(before.preparedLaunchConfigurations); + await expect(readFile(join(fixture.root, "launch-preparation.json"))).rejects.toHaveProperty("code", "ENOENT"); + if (mode !== "launch") { + const evidence = resumed.evidence(), startedAt = resumed.transport.processInfo().startedAt; + await resumed.detachControllerForRestart(); + prepare.mockClear(); + if (mode === "adopt-denied") prepare.mockRejectedValueOnce(new Error("source preparation authorization revoked")); + const authorizeOldConfiguration = vi.fn(async () => { throw new Error("old configuration must not authorize source adoption"); }); + const launch = vi.fn(() => { throw new Error("duplicate source process"); }); + const adoption = { pid: evidence.runnerPid, processGroupId: evidence.runnerPid, startedAt, + isAlive: () => !absent(evidence.runnerPid!), signal: () => false }; + resumed = fixture.recover({ ...sourceOptions, runnerProcessLauncher: launch, authorizeRunnerAdoption: authorizeOldConfiguration, + adoptExistingRunner: adoption }); + if (mode === "adopt-denied") { + await expect(resumed.transport.request("thread/read", {})).rejects.toThrow("source preparation authorization revoked"); + expect(absent(evidence.runnerPid!)).toBe(false); + expect(authorizeOldConfiguration).not.toHaveBeenCalled(); expect(launch).not.toHaveBeenCalled(); + expect(await fixture.calls("turn/start")).toBe(1); expect(await fixture.calls("process-start")).toBe(2); + return; + } + expect((await resumed.transport.request("thread/read", {})).thread).toMatchObject(fixture.thread); + expect(resumed.evidence().runnerPid).toBe(evidence.runnerPid); + expect(registeredIdentities.at(-1)).toEqual(fixture.identity); + expect(prepare).toHaveBeenCalledExactlyOnceWith(requirements, undefined); + expect(authorizeOldConfiguration).not.toHaveBeenCalled(); expect(launch).not.toHaveBeenCalled(); + expect((await read(fixture.coreFile)).preparedLaunchConfigurations).toEqual(before.preparedLaunchConfigurations); + await expect(readFile(join(fixture.root, "launch-preparation.json"))).rejects.toHaveProperty("code", "ENOENT"); + } + const target = { ...fixture.identity, runId: "prepared-target", turnId: "prepared-turn", itemId: "prepared-item" }; + const attachedAuthority = authority!; + const grant = policy === "saved" ? { runtimeContext: null, instructions: requirements.instructions } : undefined; + if (policy === "conflict") { + const preparations = prepare.mock.calls.length; + await expect(resumed.transport.attachRun!(target)).rejects.toThrow("recovery_launch_policy_changed"); + expect(prepare).toHaveBeenCalledTimes(preparations); + expect(attachedAuthority.store.state.identity).toEqual(fixture.identity); + expect(await fixture.calls("turn/start")).toBe(1); + return; + } + await resumed.transport.attachRun!({ ...target, ...(grant ? { currentRunGrant: grant } : {}) }); + expect(prepare.mock.calls.at(-1)?.[0].instructions).toBe(grant?.instructions ?? requirements.instructions); + if (policy === "saved") expect(prepare.mock.calls.at(-1)?.[0].settings.disableCodexWebSearch).toBe(true); + expect((await read(fixture.coreFile)).preparedLaunchConfigurations).toContainEqual({ identity: target, configurationId: "target-configuration" }); + expect((await read(join(fixture.root, "launch-preparation.json"))).entries).toContainEqual(expect.objectContaining({ prepared: expect.objectContaining({ configurationId: "target-configuration" }) })); + if (mode === "launch") { + expect(attachedAuthority.store.state.identity).toEqual(target); + const commands = attachedAuthority.store.state.commands.length; + owner = undefined; + await expect(resumed.transport.request("turn/start", { input: [{ type: "text", text: "Unowned work" }] })) + .rejects.toThrow("recovery_launch_requirements_invalid"); + expect(() => attachedAuthority.queueCommand("turn.start", { text: "Unowned work" })) + .toThrow("recovery_launch_requirements_invalid"); + expect(attachedAuthority.store.state.commands).toHaveLength(commands); + expect(await fixture.calls("turn/start")).toBe(1); + owner = "prepared-target"; + } + await fixture.complete(resumed, "Continue after source preparation."); await fixture.settled(); + expect(await fixture.calls("turn/start")).toBe(2); expect(await fixture.calls("process-start")).toBe(2); + if (policy === "saved") { + owner = "warm-third"; + await resumed.transport.attachRun!({ runId: owner, turnId: "third-turn", itemId: "third-item", currentRunGrant: { runtimeContext: null, instructions: requirements.instructions } }); + expect(prepare.mock.calls.at(-1)?.[0]).toMatchObject({ identity: { runId: owner }, settings: { disableCodexWebSearch: true }, instructions: requirements.instructions }); + await fixture.complete(resumed, "Continue with frozen launch policy."); await fixture.settled(); + expect(await fixture.calls("turn/start")).toBe(3); expect(await fixture.calls("process-start")).toBe(2); + } + await resumed.transport.close(); + expect((await read(fixture.runnerFile)).lifecycle).toBe("suspended"); + expect(() => fixture.recover({ recoveryLaunch: { requirements: { ...requirements, identity: { ...fixture.identity, runId: "foreign-source" } } }, prepareLaunch: prepare })).toThrow("recovery_launch_requirements_invalid"); + expect(() => fixture.recover({ recoveryLaunch: { requirements }, prepareLaunch: undefined })).toThrow("recovery_launch_requirements_invalid"); + } finally { await cleanup(); await new Promise((resolve, reject) => server.close(error => error ? reject(error) : resolve())); } +}, 30_000); + +it("rejects unexpected live goal after ordinary dormant recovery", async () => { + const fixture = await sourceFixture(); + try { + const file = join(fixture.root, "fake-codex-state.json"); + const actual = await read(file); + actual.goal = { objective: "Unexpected autonomous work", status: "active", tokenBudget: null, tokensUsed: 0, elapsedSeconds: 0, iterations: 0, createdAt: null, updatedAt: null, workingNow: false }; + await writeFile(file, JSON.stringify(actual)); + await expect(fixture.recover().transport.request("thread/read", {})).rejects.toThrow("saved dormant Codex session resumed an active or unknown goal"); + expect(await fixture.calls("process-start")).toBe(2); expect(await fixture.calls("turn/start")).toBe(1); + } finally { await fixture.dispose(); } +}, 30_000); + +it("retains replayed source tool inputs without dispatching business work or attaching the target", async () => { + const fixture = await sourceFixture(); + try { + const provider = await read(fixture.providerFile); + provider.pendingEvents.push({ executorEventId: `codex_provider_${String(provider.nextProviderEventSeq++).padStart(16, "0")}`, eventType: "semantic_tool.input", priority: "p0", payload: { semantic_tool: { + schema: "paperclip.prp.semantic_tool.v1", schemaVersion: 1, phase: "input", callId: "retained-source-call", operationId: "get_task_context", input: {}, idempotencyKey: null, + correlation: { runId: fixture.identity.runId, normalizedSessionId: fixture.identity.normalizedSessionId, turnId: fixture.identity.turnId, itemId: fixture.identity.itemId }, + content: { digest: `sha256:${createHash("sha256").update("{}").digest("hex")}`, redactionDisposition: "digest_only", references: [] }, + } } }); + await writeFile(fixture.providerFile, JSON.stringify(provider)); + const resumed = fixture.recover(), dispatch = vi.fn(async () => ({ success: true, contentItems: [] })); + resumed.transport.setServerRequestHandler(dispatch); + await resumed.transport.request("thread/read", {}); + await vi.waitFor(async () => expect((await read(fixture.coreFile)).committedEvents.some((event: { eventType: string }) => event.eventType === "semantic_tool.input")).toBe(true)); + expect(dispatch).not.toHaveBeenCalled(); + await expect(resumed.transport.attachRun!({ runId: "run-target", turnId: "turn-target", itemId: "item-target" })).rejects.toThrow(); + expect(dispatch).not.toHaveBeenCalled(); + expect((await read(fixture.coreFile)).commands.some((command: { type: string }) => command.type === "run.attach")).toBe(false); + expect(await fixture.calls("turn/start")).toBe(1); + } finally { await fixture.dispose(); } +}, 30_000); diff --git a/packages/paperclip-runner/src/live/runnerd-staged-handoff.test.ts b/packages/paperclip-runner/src/live/runnerd-staged-handoff.test.ts index 7d7d89ecd80..7749ce4d927 100644 --- a/packages/paperclip-runner/src/live/runnerd-staged-handoff.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-staged-handoff.test.ts @@ -99,7 +99,7 @@ it("delivers a WebSocket message coalesced with the real HTTP upgrade response", }); // Real encrypted PRP and real Runner/provider processes; no model or server imports. -it.each(["connect", "listen"] as const)("recovers a staged %s handoff at every durable boundary without another process", async mode => { +it.each([["connect", "owned"], ["listen", "owned"], ["connect", "legacy"], ["listen", "legacy"], ["connect", "cleanup"]] as const)("recovers a staged %s %s handoff at every durable boundary without another process", async (mode, policy) => { const root = await mkdtemp(join(tmpdir(), "runner-staged-handoff-")); const shared = join(root, "shared"), calls = join(root, "calls.log"); const source = { runnerInstanceId: "runner-staged", environmentLeaseId: "lease-staged", runId: "source-run", @@ -141,6 +141,7 @@ it.each(["connect", "listen"] as const)("recovers a staged %s handoff at every d const registration = async (core: DurablePrpControlPlane, identity = core.store.state.identity) => { currentCore = core; if (!registered.has(core)) { + if (policy === "cleanup") vi.spyOn(core, "enableSemanticInputDispatch"); registered.add(core); cores.add(core); const epoch = generation; core.setCheckpointPersistence(async saved => { @@ -208,7 +209,8 @@ it.each(["connect", "listen"] as const)("recovers a staged %s handoff at every d checkpoint.restoreSessionDirectory(next); const noSpawn = vi.fn(() => { throw new Error("Duplicate Runner launch"); }); const recoveryOptions: Parameters[0] = { ...options, stateDirectory: next, - prpIdentity: target, runnerArtifact: readRunnerdArtifactBinding(options.runnerBinary), + authorityRunId: policy === "legacy" ? undefined : target.runId, + recoveryPurpose: policy === "cleanup" ? "stop_cleanup" : undefined, prpIdentity: target, runnerArtifact: readRunnerdArtifactBinding(options.runnerBinary), runnerStateDirectory: join(directory, "runner"), readRunnerState: async () => JSON.parse(await readFile(join(directory, "runner/runner-state.json"), "utf8")), prepareExternalRunnerState: async () => { throw new Error("Verified live handoff must not restore a backup"); }, runnerProcessLauncher: noSpawn, warmTransitionRegistrationMode: mode === "listen" ? "routed_same_endpoint" : "routed_connect", @@ -216,6 +218,13 @@ it.each(["connect", "listen"] as const)("recovers a staged %s handoff at every d adoptExistingRunner: { pid: runnerPid, processGroupId: runnerPid, startedAt, isAlive: () => !dead(runnerPid!), signal: signal => { if (dead(runnerPid!)) return true; process.kill(-runnerPid!, signal); return true; } } }; if (boundary === "queued") { + if (policy === "owned") { + const wrongProvider = createCapabilityRunnerdCodexTransport({ ...recoveryOptions, + resumeProviderSession: { ...recoveryOptions.resumeProviderSession!, driverSessionId: "foreign-thread" } }); + await expect(wrongProvider.transport.request("thread/read", {})).rejects.toThrow("provider_identity_mismatch"); + await wrongProvider.detachControllerForRestart().catch(() => undefined); + checkpoint.restoreSessionDirectory(next); + } const noProof = createCapabilityRunnerdCodexTransport({ ...recoveryOptions, adoptExistingRunner: undefined }); await expect(noProof.transport.request("thread/read", {})).rejects.toThrow("queued_attachment_requires_verified_live_source"); await noProof.detachControllerForRestart().catch(() => undefined); @@ -250,11 +259,16 @@ it.each(["connect", "listen"] as const)("recovers a staged %s handoff at every d await currentCore.flushCheckpoint(); expect(checkpoint).toBe(beforeMark); await within(replacement.transport.request("thread/goal/get", {}), "post-checkpoint read"); expect(currentCore.store.state.stagedRunAttachment).toBeUndefined(); + if (policy === "cleanup") expect(currentCore.enableSemanticInputDispatch).not.toHaveBeenCalled(); + else { + await within(replacement.transport.request("turn/start", { input: [{ type: "text", text: "Work after recovered attachment" }] }), "target turn"); + for await (const event of replacement.transport.notifications()) if (event.method === "turn/completed") break; + } expect(noSpawn).not.toHaveBeenCalled(); const effects = (await readFile(callLog, "utf8")).trim().split(/\r?\n/); expect(effects.filter(line => line === "process-start")).toHaveLength(1); expect(effects.filter(line => line === "thread/start")).toHaveLength(1); - expect(effects.filter(line => line === "turn/start")).toHaveLength(0); + expect(effects.filter(line => line === "turn/start")).toHaveLength(policy === "cleanup" ? 0 : 1); } catch (error) { const physical = JSON.parse(await readFile(join(directory, "runner/runner-state.json"), "utf8").catch(() => "{}")); throw new Error(`${mode}/${boundary}: ${String(error)}; ${JSON.stringify({