From fe3e0f2d5fc6ebbb39404aec9b99e920fd2b5b93 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Thu, 29 Oct 2026 18:49:01 -0500 Subject: [PATCH] Preserve rejected native completion reports as failures (#18227) Keep rejected declared completion reports on the existing permanent native failure path. Preserve rejection across restoration and order permanent failure ahead of terminal goal facts, goal results, and continuation while retaining the original provider terminal. Co-Authored-By: Paperclip --- doc/run-log-events.md | 13 + .../src/bin/fake-codex-app-server.rs | 6 +- .../runner-core/src/provider_backend.rs | 308 ++++++++++++++++-- .../crates/runner-core/src/provider_events.rs | 9 +- .../runner-core/tests/codex_provider.rs | 293 +++++++++++++---- .../src/native-session-runtime.test.ts | 77 +++++ .../src/native-session-runtime.ts | 4 + 7 files changed, 632 insertions(+), 78 deletions(-) diff --git a/doc/run-log-events.md b/doc/run-log-events.md index e5fb2720f21..5221b400f19 100644 --- a/doc/run-log-events.md +++ b/doc/run-log-events.md @@ -193,6 +193,19 @@ The rejected call does not create a semantic input event or execution receipt. Validated runner diagnostics retain this scalar metadata even when the operation is an API or case tool. API request and response bodies keep their existing redaction. +If a declared native completion tool is rejected and the turn ends without an +accepted replacement report, the runner emits `session.failed` with code +`semantic_completion_rejected` and `recoverable: false` before the provider's +original turn terminal. It emits no successful result proposal or human review +request. Core uses its existing permanent native failure path and does not start +a missing-result recovery turn. This failure also precedes terminal goal updates; +Core records the original turn terminal before rejecting goal success or further +goal continuation. Recoverable goal failures retain their existing behavior. +The rejection survives same-turn restoration; +a new turn or run binding clears it. A later accepted completion report keeps +its normal authority, and ordinary turns with no completion-tool attempt keep +their existing fallback. + Before interrupting a Codex turn, the runner durably records a `harness.diagnostic` event with code `provider_interrupt_requested`, the stop reason, provider turn ID, and pending tool call IDs. A harmless identical result 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 52ef2fac6b4..47c83bc81b5 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 @@ -1072,7 +1072,11 @@ fn run() -> Result<(), Box> { finish_turn( &state_path, &mut state, - if success { "completed" } else { "failed" }, + if success || complete_after_tool_call { + "completed" + } else { + "failed" + }, )?; continue; } 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 1547ce4a076..c84f57773e8 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 @@ -531,7 +531,11 @@ fn admit_terminal_tool_authority( input: &Value, result_is_error: bool, ) -> Result<(), DurableRunnerError> { - if result_is_error || !matches!(operation_id, "paperclip_finish" | "paperclip_block") { + if result_is_error { + state.record_terminal_tool_rejection(operation_id); + return Ok(()); + } + if !matches!(operation_id, "paperclip_finish" | "paperclip_block") { return Ok(()); } // The correlated TypeScript semantic-tool handler validates the provider @@ -585,6 +589,7 @@ fn admit_terminal_tool_authority( // second, conflicting result when the provider turn terminates. state.active_provider_result_fingerprint = Some(fingerprint); state.active_provider_result_disposition = Some(disposition); + state.active_terminal_tool_rejected = false; Ok(()) } (Some(existing_fingerprint), Some(existing_disposition)) @@ -617,6 +622,7 @@ fn normalize_provider_notification( (None, None) => { state.active_provider_result_fingerprint = Some(fingerprint); state.active_provider_result_disposition = Some(disposition); + state.active_terminal_tool_rejected = false; Ok(vec![NormalizedProviderEvent { event_type: "run.result.proposed".to_owned(), priority: EventPriority::P0, @@ -639,12 +645,39 @@ fn terminal_events( event_type: &str, goal_status: Option<&str>, ) -> Vec { - if goal_status == Some("active") { + let rejected_completion = + state.active_terminal_tool_rejected && state.active_provider_result_fingerprint.is_none(); + if goal_status == Some("active") && !rejected_completion { return Vec::new(); } let Some(contract) = state.completion_contract.as_ref() else { return Vec::new(); }; + if rejected_completion { + return vec![ + NormalizedProviderEvent { + event_type: "session.failed".to_owned(), + priority: EventPriority::P0, + payload: json!({ + "provider": state.config.provider, + "code": "semantic_completion_rejected", + "message": "The turn ended without correcting its rejected completion report.", + "recoverable": false, + }), + }, + NormalizedProviderEvent { + event_type: "run.terminal".to_owned(), + priority: EventPriority::P0, + payload: json!({ + "schema": "paperclip.prp.terminal.v1", + "provider": state.config.provider, + "turnTerminalState": "failed", + "runTerminalState": "failed", + "reportedWorkDisposition": "needs_review", + }), + }, + ]; + } // Once the correlated semantic-tool result has been accepted, Paperclip's // bounded controller finalizer may interrupt the provider after the result // proposal. That provider terminal closes the exact turn; it does not @@ -1018,6 +1051,8 @@ struct CodexProviderState { active_provider_result_fingerprint: Option, #[serde(default)] active_provider_result_disposition: Option, + #[serde(default)] + active_terminal_tool_rejected: bool, last_agent_message: Option, #[serde(default)] goal_capability: Option, @@ -1098,6 +1133,7 @@ impl CodexProviderState { receipt_limit_interrupt_deadline_unix_ms: None, active_provider_result_fingerprint: None, active_provider_result_disposition: None, + active_terminal_tool_rejected: false, last_agent_message: None, goal_capability: None, goal: None, @@ -1407,10 +1443,12 @@ impl CodexProviderState { operation_id: &str, input: &Value, ) -> Result { - if let Some(result) = self - .tool_bridge - .replay_result(call_id, operation_id, input)? - { + let replay = self.tool_bridge.replay_result(call_id, operation_id, input); + let replay = replay.map_err(|error| { + self.record_terminal_tool_rejection(operation_id); + error + })?; + if let Some(result) = replay { // An exact completed replay is a transport retry. Its input and // result receipts are already durable, so recording another event // would make an otherwise idempotent replay consume bounded event @@ -1422,11 +1460,13 @@ impl CodexProviderState { .tool_bridge .pending_calls() .any(|pending| pending.call_id == call_id); - let call = self.tool_bridge.begin_call( - call_id.to_owned(), - operation_id.to_owned(), - input.clone(), - )?; + let call = + self.tool_bridge + .begin_call(call_id.to_owned(), operation_id.to_owned(), input.clone()); + let call = call.map_err(|error| { + self.record_terminal_tool_rejection(operation_id); + error + })?; if pending_replay { // The first input receipt is already durable. An exact pending // replay is only the provider re-establishing its request after a @@ -1438,7 +1478,24 @@ impl CodexProviderState { } } + fn record_terminal_tool_rejection(&mut self, operation_id: &str) { + if matches!(operation_id, "paperclip_finish" | "paperclip_block") + && self.active_provider_result_fingerprint.is_none() + && self + .tool_bridge + .authorized_tools() + .any(|tool| tool.operation_id == operation_id) + { + self.active_terminal_tool_rejected = true; + } + } + fn reconcile_active_provider_turn(&mut self, active_provider_turn_id: Option) { + if active_provider_turn_id.is_some() + && self.active_provider_turn_id != active_provider_turn_id + { + self.active_terminal_tool_rejected = false; + } self.active_provider_turn_id = active_provider_turn_id; if self.active_provider_turn_id.is_some() { // A newly discovered turn supersedes completion authority from the @@ -1903,6 +1960,7 @@ impl CodexCommandExecutor { state.receipt_limit_interrupt_deadline_unix_ms = None; state.active_provider_result_fingerprint = None; state.active_provider_result_disposition = None; + state.active_terminal_tool_rejected = false; state.last_agent_message = None; state.lifecycle = "closed".to_owned(); let _ = state.push_terminal_event(NormalizedProviderEvent { @@ -1957,6 +2015,7 @@ impl CodexCommandExecutor { state.receipt_limit_interrupt_deadline_unix_ms = None; state.active_provider_result_fingerprint = None; state.active_provider_result_disposition = None; + state.active_terminal_tool_rejected = false; state.last_agent_message = None; state.lifecycle = "closed".to_owned(); // Closing the provider is the safety boundary. Preserve that @@ -2132,7 +2191,7 @@ impl CodexCommandExecutor { }) } else { Value::Null }, }), }; - if provider_label == "codex" { + if provider_label == "codex" && !state.active_terminal_tool_rejected { // No provider result was observed. Preserve the lost // turn as a fact, but leave the run open for the // controller's bounded same-conversation recovery. @@ -2552,6 +2611,7 @@ impl CodexCommandExecutor { next_state.receipt_limit_interrupt_deadline_unix_ms = None; next_state.active_provider_result_fingerprint = None; next_state.active_provider_result_disposition = None; + next_state.active_terminal_tool_rejected = false; next_state.last_agent_message = None; let retained_provider = if let Some(provider) = self.provider.as_mut() { !runtime_launch_changed @@ -2779,6 +2839,7 @@ impl CodexCommandExecutor { state.receipt_limit_interrupt_deadline_unix_ms = None; state.active_provider_result_fingerprint = None; state.active_provider_result_disposition = None; + state.active_terminal_tool_rejected = false; state.last_agent_message = None; state.lifecycle = "closed".to_owned(); let provider_label = state.config.provider.clone(); @@ -3079,6 +3140,7 @@ impl CodexCommandExecutor { state.completed_provider_turn_id = None; state.active_provider_result_fingerprint = None; state.active_provider_result_disposition = None; + state.active_terminal_tool_rejected = false; state.last_agent_message = None; } self.save_state()?; @@ -3117,6 +3179,7 @@ impl CodexCommandExecutor { state.receipt_limit_interrupt_deadline_unix_ms = None; state.active_provider_result_fingerprint = None; state.active_provider_result_disposition = None; + state.active_terminal_tool_rejected = false; state.last_agent_message = None; state.lifecycle = "turn_active".to_owned(); let provider_label = state.config.provider.clone(); @@ -3243,6 +3306,7 @@ impl CodexCommandExecutor { state.receipt_limit_interrupt_deadline_unix_ms = None; state.active_provider_result_fingerprint = None; state.active_provider_result_disposition = None; + state.active_terminal_tool_rejected = false; state.last_agent_message = None; // Do not let the runner.drain command that follows turn.stop restore a // fresh provider process. `prepared` retains the durable thread while @@ -3533,6 +3597,7 @@ impl CodexCommandExecutor { .state .as_mut() .expect("Codex state remains available for a rejected tool call"); + state.record_terminal_tool_rejection(&operation_id); let mut payload = json!({ "provider": state.config.provider, "code": "semantic_tool_denied", @@ -3958,6 +4023,7 @@ impl CodexCommandExecutor { state.receipt_limit_interrupt_deadline_unix_ms = None; state.active_provider_result_fingerprint = None; state.active_provider_result_disposition = None; + state.active_terminal_tool_rejected = false; state.lifecycle = "closed".to_owned(); let thread_id = state.thread_id.clone(); let provider_name = state.config.provider.clone(); @@ -4257,6 +4323,7 @@ impl CodexCommandExecutor { } let normalized = normalize_provider_notification(state, &method, ¶ms)?; let normalized_event_count = normalized.len(); + let mut terminal_goal_events = Vec::new(); if terminal_event_type.is_some() { state.settle_active_provider_turn_identity()?; let settled = state @@ -4307,7 +4374,7 @@ impl CodexCommandExecutor { if let Some(goal) = state.goal.as_mut() { goal.working_now = false; state.goal_revision = state.goal_revision.saturating_add(1); - state.push_event(NormalizedProviderEvent { + terminal_goal_events.push(NormalizedProviderEvent { event_type: "session.goal.updated".to_owned(), priority: EventPriority::P0, payload: goal_event_payload( @@ -4315,7 +4382,7 @@ impl CodexCommandExecutor { state.goal_capability.as_ref(), state.goal_revision, ), - })?; + }); } } if method == "thread/goal/updated" { @@ -4350,7 +4417,7 @@ impl CodexCommandExecutor { if next_goal != state.goal { state.goal = next_goal; state.goal_revision = state.goal_revision.saturating_add(1); - state.push_event(NormalizedProviderEvent { + terminal_goal_events.push(NormalizedProviderEvent { event_type: "session.goal.snapshot".to_owned(), priority: EventPriority::P0, payload: goal_event_payload( @@ -4358,11 +4425,11 @@ impl CodexCommandExecutor { state.goal_capability.as_ref(), state.goal_revision, ), - })?; + }); } } Err(_) => { - state.push_event(NormalizedProviderEvent { + terminal_goal_events.push(NormalizedProviderEvent { event_type: "provider.notice.recorded".to_owned(), priority: EventPriority::P0, payload: json!({ @@ -4375,7 +4442,7 @@ impl CodexCommandExecutor { "userActionable": false, "summary": "Codex goal state could not be reconciled after the turn; Paperclip retained the last durable snapshot.", }), - })?; + }); } } } @@ -4383,7 +4450,11 @@ impl CodexCommandExecutor { if let Some(ref event_type) = terminal_event_type { let goal_status = state.goal.as_ref().map(|goal| goal.status.as_str()); let outcome = terminal_events(state, event_type, goal_status); - state.extend_terminal_events(with_terminal_outcome(normalized, outcome))?; + terminal_goal_events.extend(normalized); + state.extend_terminal_events(with_terminal_outcome( + terminal_goal_events, + outcome, + ))?; } else if receipt_limit_terminal_poll { for event in normalized { state.push_receipt_limit_cleanup_event(event)?; @@ -5156,6 +5227,7 @@ mod tests { #[test] fn preserves_one_verified_opencode_result_before_its_terminal() { let mut state = opencode_result_state(); + state.active_terminal_tool_rejected = true; let params = json!({ "threadId": "thread-1", "turnId": "turn-1", @@ -5173,6 +5245,7 @@ mod tests { assert_eq!(result_events[0].event_type, "run.result.proposed"); assert_eq!(result_events[0].priority, EventPriority::P0); assert!(replay_events.is_empty()); + assert!(!state.active_terminal_tool_rejected); assert_eq!(terminal.len(), 1); assert_eq!(terminal[0].event_type, "run.terminal"); assert_eq!(terminal[0].payload["reportedWorkDisposition"], "done"); @@ -5195,6 +5268,7 @@ mod tests { state.active_provider_turn_id = None; state.provider_process_generation = 1; state.lifecycle = "prepared".to_owned(); + state.active_terminal_tool_rejected = true; let writer = CodexCommandExecutor::new(&directory); writer.persist_state(&state).unwrap(); let mut executor = CodexCommandExecutor::new(&directory); @@ -5208,6 +5282,7 @@ mod tests { assert_eq!(rebound.provider_process_generation, 1); assert!(rebound.pending_events.is_empty()); assert!(rebound.queued_events.is_empty()); + assert!(!rebound.active_terminal_tool_rejected); fs::remove_dir_all(directory).unwrap(); } @@ -5296,6 +5371,200 @@ mod tests { assert!(state.validate().is_ok()); } + fn declared_finish_state() -> CodexProviderState { + let input_schema: Value = serde_json::from_slice( + &fs::read( + PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .join("../../../protocol/schemas/result.schema.json"), + ) + .unwrap(), + ) + .unwrap(); + let operation = crate::provider_bridge::AuthorizedTool { + operation_id: "paperclip_finish".to_owned(), + version: 1, + description: "Report the completed turn.".to_owned(), + input_schema, + response_schema: json!({"type": "object"}), + }; + let mut state = opencode_result_state(); + state.config.provider = "codex".to_owned(); + state.config.driver = "codex_app_server".to_owned(); + state.config.provider_version = "test".to_owned(); + state + .tool_bridge + .prepare(AuthorizedToolSet { + schema: TOOL_SET_SCHEMA.to_owned(), + schema_version: 1, + catalog_digest: authorized_tool_catalog_digest(std::slice::from_ref(&operation)) + .unwrap(), + operations: vec![operation], + }) + .unwrap(); + state + } + + #[test] + fn controller_rejected_finish_cannot_become_a_successful_terminal_fallback() { + let mut state = declared_finish_state(); + let input = valid_opencode_result(); + assert!(matches!( + state + .admit_tool_call("rejected-finish", "paperclip_finish", &input) + .unwrap(), + ToolCallAdmission::Pending(_) + )); + state + .tool_bridge + .apply_result(ToolResult { + call_id: "rejected-finish".to_owned(), + operation_id: "paperclip_finish".to_owned(), + result: json!({"error": {"code": "deliverable_missing"}}), + is_error: true, + }) + .unwrap(); + admit_terminal_tool_authority(&mut state, "paperclip_finish", &input, true).unwrap(); + state.last_agent_message = + Some("The required deliverable could not be published.".to_owned()); + + assert_rejected_finish_terminal(&state); + assert!(state.validate().is_ok()); + } + + #[test] + fn schema_rejected_finish_cannot_become_a_successful_terminal_fallback() { + let mut state = declared_finish_state(); + assert!(state + .admit_tool_call("invalid-finish", "paperclip_finish", &json!({})) + .unwrap_err() + .to_string() + .contains("JSON Schema validation")); + assert_eq!(state.tool_bridge.pending_calls().count(), 0); + + assert_rejected_finish_terminal(&state); + let active_goal_events = terminal_events(&state, "turn.completed", Some("active")); + assert_eq!(active_goal_events[0].event_type, "session.failed"); + assert_eq!(active_goal_events[0].payload["recoverable"], false); + assert!(state.validate().is_ok()); + } + + fn assert_rejected_finish_terminal(state: &CodexProviderState) { + let events = with_terminal_outcome( + vec![NormalizedProviderEvent { + event_type: "turn.completed".to_owned(), + priority: EventPriority::P0, + payload: json!({"status": "completed"}), + }], + terminal_events(state, "turn.completed", None), + ); + assert_eq!( + events + .iter() + .map(|event| event.event_type.as_str()) + .collect::>(), + vec!["session.failed", "turn.completed", "run.terminal"], + ); + assert_eq!(events[0].payload["code"], "semantic_completion_rejected"); + assert_eq!(events[0].payload["recoverable"], false); + assert_eq!(events[1].payload["status"], "completed"); + assert_eq!(events[2].payload["runTerminalState"], "failed"); + assert_eq!(events[2].payload["reportedWorkDisposition"], "needs_review"); + assert!(events + .iter() + .all(|event| event.payload.get("attentionRequests").is_none())); + } + + #[test] + fn rejected_finish_survives_snapshot_restore_and_only_a_new_turn_clears_it() { + let mut state = declared_finish_state(); + state + .admit_tool_call("invalid-finish", "paperclip_finish", &json!({})) + .unwrap_err(); + let mut restored: CodexProviderState = + serde_json::from_value(serde_json::to_value(&state).unwrap()).unwrap(); + assert_rejected_finish_terminal(&restored); + restored.reconcile_active_provider_turn(Some("turn-1".to_owned())); + assert_rejected_finish_terminal(&restored); + + restored.reconcile_active_provider_turn(Some("turn-2".to_owned())); + + assert_eq!( + terminal_events(&restored, "turn.completed", None)[0].payload + ["reportedWorkDisposition"], + "done" + ); + assert!(restored.validate().is_ok()); + } + + #[test] + fn legacy_snapshot_without_a_rejected_finish_marker_keeps_its_fallback() { + let mut encoded = serde_json::to_value(declared_finish_state()).unwrap(); + assert!(encoded + .as_object_mut() + .unwrap() + .remove("activeTerminalToolRejected") + .is_some()); + let restored: CodexProviderState = serde_json::from_value(encoded).unwrap(); + + assert_eq!( + terminal_events(&restored, "turn.completed", None)[0].payload + ["reportedWorkDisposition"], + "done" + ); + assert!(restored.validate().is_ok()); + } + + #[test] + fn an_undeclared_finish_rejection_does_not_change_terminal_authority() { + let mut state = opencode_result_state(); + state + .admit_tool_call("undeclared-finish", "paperclip_finish", &json!({})) + .unwrap_err(); + + assert_eq!( + terminal_events(&state, "turn.completed", None)[0].payload["reportedWorkDisposition"], + "done" + ); + assert!(state.validate().is_ok()); + } + + #[test] + fn rejected_finish_allows_a_later_accepted_finish_in_the_same_turn() { + let mut state = declared_finish_state(); + let input = valid_opencode_result(); + admit_terminal_tool_authority(&mut state, "paperclip_finish", &input, true).unwrap(); + admit_terminal_tool_authority(&mut state, "paperclip_finish", &input, false).unwrap(); + + let events = terminal_events(&state, "turn.completed", None); + + assert_eq!(events.len(), 1); + assert_eq!(events[0].event_type, "run.terminal"); + assert_eq!(events[0].payload["reportedWorkDisposition"], "done"); + assert_eq!(events[0].payload["runTerminalState"], "succeeded"); + assert!(state.validate().is_ok()); + } + + #[test] + fn ordinary_final_message_without_a_finish_attempt_keeps_its_fallback() { + let mut state = declared_finish_state(); + state.last_agent_message = Some("The requested answer is ready.".to_owned()); + + let events = terminal_events(&state, "turn.completed", None); + + assert_eq!(events[0].payload["reportedWorkDisposition"], "done"); + assert_eq!( + events[0].payload["completionClaim"]["objectiveSatisfied"], + true + ); + assert_eq!( + events[0].payload["summary"], + "The requested answer is ready." + ); + assert_eq!(events[1].payload["runTerminalState"], "succeeded"); + assert!(terminal_events(&state, "turn.completed", Some("active")).is_empty()); + assert!(state.validate().is_ok()); + } + #[test] fn accepted_terminal_tool_preserves_an_explicit_response_wait() { let mut state = opencode_result_state(); @@ -5645,6 +5914,7 @@ mod tests { receipt_limit_interrupt_deadline_unix_ms: None, active_provider_result_fingerprint: None, active_provider_result_disposition: None, + active_terminal_tool_rejected: false, last_agent_message: None, goal_capability: None, goal: None, diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs b/packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs index 755ad5402d5..b1d76fef47a 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs @@ -27,9 +27,12 @@ pub(crate) fn with_terminal_outcome( provider_events: Vec, outcome_events: Vec, ) -> Vec { - let (results, terminals): (Vec<_>, Vec<_>) = outcome_events - .into_iter() - .partition(|event| event.event_type == "run.result.proposed"); + let (results, terminals): (Vec<_>, Vec<_>) = outcome_events.into_iter().partition(|event| { + matches!( + event.event_type.as_str(), + "run.result.proposed" | "session.failed" + ) + }); results .into_iter() .chain(provider_events) 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 f4f05edc76d..ead00ee0f33 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 @@ -3603,62 +3603,102 @@ fn durable_backend_replays_pending_tool_calls_without_mutating_the_event_queue() } #[test] -fn restart_lost_turn_retains_a_recoverable_cause_without_inventing_a_result() { - let directory = temporary_directory("restart-lost-turn"); - let config = provider_config( - &directory, - &["--durable-turn-ids", "--linger-after-turn-start"], - ); - let mut first = CodexCommandExecutor::new(&directory); - first - .execute(&command( - "prepare", - 1, - "run.prepare", - json!({"provider": config}), - )) +fn restart_lost_turn_preserves_rejected_completion_without_changing_ordinary_recovery() { + for rejected in [false, true] { + let directory = temporary_directory("restart-lost-turn"); + let config = provider_config( + &directory, + &["--durable-turn-ids", "--linger-after-turn-start"], + ); + let operation = AuthorizedTool { + operation_id: "paperclip_finish".to_owned(), + version: 1, + description: "Report completed work.".to_owned(), + input_schema: json!({"type": "object"}), + response_schema: json!({"type": "object"}), + }; + let authorized = AuthorizedToolSet { + schema: TOOL_SET_SCHEMA.to_owned(), + schema_version: 1, + catalog_digest: authorized_tool_catalog_digest(std::slice::from_ref(&operation)) + .unwrap(), + operations: vec![operation], + }; + let mut first = CodexCommandExecutor::new(&directory); + first.execute(&command("prepare", 1, "run.prepare", json!({"provider": config, + "authorizedTools": authorized, "completionContract": {"revision": "restore-contract", "criterionIds": ["output"]}}))).unwrap(); + first + .execute(&command("open", 2, "session.open", json!({}))) + .unwrap(); + first + .execute(&command( + "start", + 3, + "turn.start", + json!({"text": "Watch the existing check run."}), + )) + .unwrap(); + drop(first); + if rejected { + let state_path = directory.join("codex-provider-state.json"); + let mut saved: Value = serde_json::from_slice(&fs::read(&state_path).unwrap()).unwrap(); + saved["activeTerminalToolRejected"] = json!(true); + fs::write(state_path, serde_json::to_vec(&saved).unwrap()).unwrap(); + } + fs::write( + directory.join("fake-state.json"), + serde_json::to_vec(&json!({ + "threadId": "codex-thread-1", "activeTurnId": null, "nextTurn": 1, + })) + .unwrap(), + ) .unwrap(); - first - .execute(&command("open", 2, "session.open", json!({}))) - .unwrap(); - first - .execute(&command( - "start", - 3, - "turn.start", - json!({"text": "Watch the existing check run."}), - )) - .unwrap(); - // The process disappeared while its command was in flight. Codex restores - // the conversation, but cannot restore the running turn. - drop(first); - fs::write( - directory.join("fake-state.json"), - serde_json::to_vec(&json!({ - "threadId": "codex-thread-1", "activeTurnId": null, "nextTurn": 1, - })) - .unwrap(), - ) - .unwrap(); - let mut restored = CodexCommandExecutor::new(&directory); - restored - .execute(&command("snapshot", 4, "session.snapshot", json!({}))) - .unwrap(); - let events = poll_and_ack(&mut restored).unwrap(); - let interrupted = events - .iter() - .find(|event| event.event_type == "turn.failed") - .unwrap(); - assert_eq!( - interrupted.payload["error"]["code"], - "provider_turn_lost_on_restore" - ); - assert_eq!(interrupted.payload["error"]["recoverable"], true); - assert!(!events - .iter() - .any(|event| event.event_type == "run.result.proposed")); - restored.shutdown().unwrap(); - fs::remove_dir_all(directory).unwrap(); + let mut restored = CodexCommandExecutor::new(&directory); + restored + .execute(&command("snapshot", 4, "session.snapshot", json!({}))) + .unwrap(); + let events = poll_and_ack(&mut restored).unwrap(); + let turn = events + .iter() + .position(|event| event.event_type == "turn.failed") + .unwrap(); + assert_eq!( + events[turn].payload["error"]["code"], + "provider_turn_lost_on_restore" + ); + assert!(!events + .iter() + .any(|event| event.event_type == "run.result.proposed")); + if rejected { + let failure = events + .iter() + .position(|event| event.event_type == "session.failed") + .unwrap(); + assert!(failure < turn); + assert_eq!( + events[failure].payload["code"], + "semantic_completion_rejected" + ); + assert_eq!(events[failure].payload["recoverable"], false); + assert_eq!( + events + .iter() + .find(|event| event.event_type == "run.terminal") + .unwrap() + .payload["runTerminalState"], + "failed" + ); + } else { + assert_eq!(events[turn].payload["error"]["recoverable"], true); + assert!(!events + .iter() + .any(|event| event.event_type == "session.failed" + || event.event_type == "run.terminal")); + } + assert_eq!(call_count(&directory, "turn/start"), 1); + restored.shutdown().unwrap(); + fs::remove_dir_all(directory).unwrap(); + } } #[test] @@ -6174,6 +6214,149 @@ fn codex_completion_emits_the_bound_result_before_the_terminal_event() { fs::remove_dir_all(directory).expect("remove Codex integration-test directory"); } +#[test] +fn rejected_declared_completion_stays_failed_when_the_provider_completes() { + let schemas = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../../protocol/schemas"); + let result_schema: Value = + serde_json::from_slice(&fs::read(schemas.join("result.schema.json")).unwrap()).unwrap(); + let terminal_schema: Value = + serde_json::from_slice(&fs::read(schemas.join("terminal.schema.json")).unwrap()).unwrap(); + let stop_reason_schema: Value = + serde_json::from_slice(&fs::read(schemas.join("stop-reason.schema.json")).unwrap()) + .unwrap(); + let registry = jsonschema::Registry::new() + .add( + "https://paperclip.dev/schemas/prp/v1/stop-reason.schema.json", + stop_reason_schema, + ) + .unwrap() + .prepare() + .unwrap(); + let terminal_validator = jsonschema::options() + .with_registry(®istry) + .build(&terminal_schema) + .unwrap(); + for scenario in ["schema", "summary", "controller", "controller-goal"] { + let directory = temporary_directory(&format!("rejected-completion-{scenario}")); + let input = if scenario == "schema" { + json!({}) + } else { + json!({ + "schema": "paperclip.run_result.v1", + "reportedWorkDisposition": "done", + "summary": if scenario == "summary" { "x".repeat(12_001) } else { "Requested work is ready.".to_owned() }, + "completionClaim": { "contractRevision": "contract-1", "objectiveSatisfied": true, + "criteria": [{"criterionId": "output", "status": "satisfied", "evidenceRefs": []}], + "remainingWork": [] }, + "evidence": [], "verification": [], "attentionRequests": [], "artifacts": [], + }) + }; + let input_path = directory.join("finish-input.json"); + fs::write(&input_path, serde_json::to_vec(&input).unwrap()).unwrap(); + let mut config = provider_config(&directory, &["--complete-after-tool-call"]); + if scenario == "controller-goal" { + config.args.push("--agent-created-goal-on-open".to_owned()); + } + config.args.extend([ + "--terminal-result-file".to_owned(), + input_path.to_string_lossy().into_owned(), + ]); + let operation = AuthorizedTool { + operation_id: "paperclip_finish".to_owned(), + version: 1, + description: "Report completed work.".to_owned(), + input_schema: result_schema.clone(), + response_schema: json!({"type": "object"}), + }; + let authorized = AuthorizedToolSet { + schema: TOOL_SET_SCHEMA.to_owned(), + schema_version: 1, + catalog_digest: authorized_tool_catalog_digest(std::slice::from_ref(&operation)) + .unwrap(), + operations: vec![operation], + }; + let mut executor = + CodexCommandExecutor::with_runner_config(&directory, &durable_config(&directory)); + executor.execute(&command("prepare", 1, "run.prepare", json!({"provider": config, + "authorizedTools": authorized, "completionContract": {"revision": "contract-1", "criterionIds": ["output"]}}))).unwrap(); + executor + .execute(&command("open", 2, "session.open", json!({}))) + .unwrap(); + executor + .execute(&command( + "turn", + 3, + "turn.start", + json!({"text": "Report the local fixture result."}), + )) + .unwrap(); + let mut emitted = Vec::new(); + let mut input_calls = 0; + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + while std::time::Instant::now() < deadline { + let events = poll_and_ack(&mut executor).unwrap(); + for event in &events { + if event.event_type == "semantic_tool.input" { + input_calls += 1; + assert!(scenario.starts_with("controller")); + executor.execute(&command("reject", 4, "semantic_tool.result", json!({ + "callId": "terminal-result-call", "operationId": "paperclip_finish", "isError": true, + "result": {"error": {"code": "deliverable_missing"}} }))).unwrap(); + } + } + emitted.extend(events); + if emitted + .iter() + .any(|event| event.event_type == "run.terminal") + { + break; + } + std::thread::sleep(std::time::Duration::from_millis(1)); + } + let failure = emitted + .iter() + .position(|event| event.event_type == "session.failed") + .unwrap(); + let turn = emitted + .iter() + .position(|event| event.event_type == "turn.completed") + .unwrap(); + let terminal = emitted + .iter() + .position(|event| event.event_type == "run.terminal") + .unwrap(); + assert!(failure < turn && turn < terminal); + if scenario == "controller-goal" { + let settled_goal = emitted + .iter() + .rposition(|event| { + matches!( + event.event_type.as_str(), + "session.goal.updated" | "session.goal.snapshot" + ) && event.payload["workingNow"] == false + }) + .unwrap(); + assert!(failure < settled_goal && settled_goal < turn); + } + assert_eq!( + emitted[failure].payload["code"], + "semantic_completion_rejected" + ); + assert_eq!(emitted[failure].payload["recoverable"], false); + assert_eq!(emitted[terminal].payload["runTerminalState"], "failed"); + terminal_validator + .validate(&emitted[terminal].payload) + .unwrap(); + assert!(!emitted + .iter() + .any(|event| event.event_type == "run.result.proposed")); + assert_eq!(input_calls, usize::from(scenario.starts_with("controller"))); + assert_eq!(call_count(&directory, "terminal-result:rejected"), 1); + executor.shutdown().unwrap(); + fs::remove_dir_all(directory).unwrap(); + } +} + #[test] fn idle_integrity_failure_persists_reconciliation_and_retires_provider() { assert_idle_failure_reconciled(false); diff --git a/packages/paperclip-runner/src/native-session-runtime.test.ts b/packages/paperclip-runner/src/native-session-runtime.test.ts index caba0b66d12..1515b729ca3 100644 --- a/packages/paperclip-runner/src/native-session-runtime.test.ts +++ b/packages/paperclip-runner/src/native-session-runtime.test.ts @@ -1245,6 +1245,83 @@ describe("executeNativeSession recovery", () => { }, ); + it.each([ + { status: "active", code: "semantic_completion_rejected", recoverable: false }, + { status: "complete", code: "semantic_completion_rejected", recoverable: false }, + { status: "complete", code: "provider_output_failed", recoverable: false }, + { status: "active", code: "provider_turn_lost_on_restore", recoverable: true }, + ] as const)( + "preserves session-goal failure priority ($status, $code, recoverable=$recoverable)", + async ({ status, code, recoverable }) => { + const capabilities = { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }; + const activeGoal = { + threadId: "provider-recovery", objective: "Finish autonomous work", status: "active" as const, + tokenBudget: null, tokensUsed: 100, timeUsedSeconds: 1, + createdAt: Date.parse("2026-08-09T00:00:00.000Z"), updatedAt: Date.parse("2026-08-09T00:00:01.000Z"), + }; + const completeGoal = { ...activeGoal, status: "complete" as const }; + const goal = vi.fn(async () => completeGoal); + const startTurn = vi.fn(async () => ({ turnId: "turn-recovery" })); + const close = vi.fn(async () => {}); + const session: NativeSession = { + identity: () => identity, + async capabilities() { return capabilities; }, + async *events() { + yield runnerEvent(1, "session.goal.updated", { goal: activeGoal, workingNow: true }); + yield runnerEvent(2, "session.failed", { + code, recoverable, + message: "The turn ended without correcting its rejected completion report.", + }); + yield runnerEvent(3, "session.goal.updated", { goal: { ...activeGoal, status }, workingNow: false }); + yield runnerEvent(4, "turn.completed"); + yield runnerEvent(5, "session.goal.snapshot", { goal: completeGoal, workingNow: false }); + }, + startTurn, goal, + async result() { return null; }, + async snapshot() { + return { backendKind: "mock", sessionId: "driver-recovery", identity, + providerSessionId: "provider-recovery", cursor: null, activeTurnId: null, + pendingRuntimeRequests: [], goal: activeGoal, lineage: [] }; + }, + close, + }; + const appended: PrpEvent[] = []; + const completeRun = vi.fn(async () => {}); + const execution = executeNativeSession({ + input, + backend: { + async descriptor() { return { kind: "mock", name: "failed-goal", version: "1", capabilities }; }, + async openSession() { return session; }, + }, + controlPlane: { + async openRun() {}, async checkpointSession() {}, + async appendEvent(event) { appended.push(event); return { + cursor: event.sourceSeq, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed", + }; }, + async replayEvents() { return { events: [], highestContiguousSourceSeq: 0 }; }, + completeRun, + }, + runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery", + }); + if (recoverable) { + await expect(execution).resolves.toMatchObject({ result: { reportedWorkDisposition: "done" } }); + expect(goal).toHaveBeenCalledExactlyOnceWith({ action: "get" }); + expect(completeRun).toHaveBeenCalledOnce(); + } else { + await expect(execution).rejects.toMatchObject({ + code: "native_provider_terminal_failed", providerCode: code, recoverable: false, + }); + expect(appended.map(event => event.eventType)).toEqual([ + "session.goal.updated", "session.failed", "session.goal.updated", "turn.completed", + ]); + expect(goal).not.toHaveBeenCalled(); + expect(completeRun).not.toHaveBeenCalled(); + } + expect(startTurn).toHaveBeenCalledOnce(); + expect(close).toHaveBeenCalledOnce(); + }, + ); + it("keeps governed-wait discovery synchronous", () => { type GovernedWaitResolver = NonNullable< ExecuteNativeSessionOptions["resolveGovernedWait"] diff --git a/packages/paperclip-runner/src/native-session-runtime.ts b/packages/paperclip-runner/src/native-session-runtime.ts index c9ef9feed4d..d950524c265 100644 --- a/packages/paperclip-runner/src/native-session-runtime.ts +++ b/packages/paperclip-runner/src/native-session-runtime.ts @@ -1198,11 +1198,15 @@ async function consumeTurn( ); } if (sessionGoalObserved) { + if (isTurnTerminal(event) && providerFailure && !providerFailure.recoverable) { + throw providerFailure; + } if (sessionGoal && payload.requestId === sessionGoal.requestId) { goalControlObserved = true; } if ( goalControlObserved && + (providerFailure === null || providerFailure.recoverable) && eventGoal !== undefined && goalStatus(eventGoal) !== "active" && payload.workingNow !== true