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 e84d343e00e..e8dcb86f91d 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 @@ -1781,6 +1781,20 @@ fn run() -> Result<(), Box> { "id": id, "result": {"turn": {"id": provider_turn_id, "status": "inProgress"}} }))?; + // A resumed Codex thread can report the cumulative usage of the + // turn lost to a restart after it accepts the next turn. + if durable_turn_ids + && state.next_turn > 1 + && args + .iter() + .any(|value| value == "--stale-usage-after-turn-start") + { + send(json!({"method": "thread/tokenUsage/updated", "params": { + "threadId": state.thread_id, + "turnId": format!("provider-turn-{}", state.next_turn - 1), + "tokenUsage": {"total": {"inputTokens": 100, "outputTokens": 20}, "last": {"inputTokens": 50, "outputTokens": 10}} + }}))?; + } // A loaded host can accept a turn long before it starts it. if let Some(delay) = argument(&args, "--turn-started-delay-ms") { thread::sleep(Duration::from_millis(delay.parse()?)); 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 b195954f2f9..8e71e9662f4 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 @@ -2631,6 +2631,26 @@ impl CodexCommandExecutor { )); } } + if let Some(ended_turn_id) = previous_active_turn_id + .as_ref() + .filter(|_| recovered_active_turn_id.is_none() && !provider_epoch_requires_rollover) + { + // Recovery settles the turn that ended while the runner was + // offline in the durable ledger below. The live provider must + // know it too: a resumed thread can report that turn's usage + // after it accepts the restart continuation, which is history, + // not a foreign turn that fails the continuation. + provider + .restore_settled_turn_identities( + std::iter::once(ended_turn_id.clone()), + settled_provider_turn_filter.clone(), + ) + .map_err(|error| { + DurableRunnerError::invalid(format!( + "failed to restore the ended provider turn identity: {error}" + )) + })?; + } provider .restore_completed_turn_authority( (completed_turn_authoritative || recovered_turn_ended_with_result) 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 9a83c92d94f..7de236dfd31 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 @@ -4383,6 +4383,98 @@ fn restart_lost_turn_preserves_rejected_completion_without_changing_ordinary_rec } } +#[test] +fn restart_continuation_ignores_late_usage_for_the_turn_lost_on_restore() { + // RC-20c: a resumed Codex thread reported the cumulative usage of the turn + // lost to the restart after it accepted the continuation turn. The runner + // must treat that usage as history of a settled turn, not as a foreign + // event that fails the continuation with turn_binding_mismatch. + let directory = temporary_directory("restart-continuation-stale-usage"); + let config = provider_config( + &directory, + &[ + "--durable-turn-ids", + "--linger-after-turn-start", + "--stale-usage-after-turn-start", + ], + ); + let mut first = CodexCommandExecutor::new(&directory); + first + .execute(&command( + "prepare", + 1, + "run.prepare", + json!({"provider": config}), + )) + .unwrap(); + first + .execute(&command("open", 2, "session.open", json!({}))) + .unwrap(); + first + .execute(&command( + "start", + 3, + "turn.start", + json!({"text": "Run the long benchmark."}), + )) + .unwrap(); + 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 lost = poll_and_ack(&mut restored).unwrap(); + assert!(lost.iter().any(|event| event.event_type == "turn.failed" + && event.payload["error"]["code"] == "provider_turn_lost_on_restore")); + let accepted = restored + .execute(&command( + "continue", + 5, + "turn.start", + json!({"text": "Continue after the restart."}), + )) + .unwrap(); + assert_eq!(accepted.result["providerTurnId"], "provider-turn-2"); + let mut events = Vec::new(); + for _ in 0..50 { + events.extend(poll_and_ack(&mut restored).unwrap()); + if events + .iter() + .any(|event| event.event_type == "turn.started") + { + break; + } + std::thread::sleep(std::time::Duration::from_millis(20)); + } + assert!( + !events + .iter() + .any(|event| event.payload["code"] == "turn_binding_mismatch" + || event.payload["error"]["code"] == "turn_binding_mismatch"), + "late usage for the lost turn must not fail the continuation: {:?}", + events + .iter() + .map(|event| (&event.event_type, &event.payload)) + .collect::>() + ); + assert!(!events + .iter() + .any(|event| event.event_type == "run.terminal" || event.event_type == "turn.failed")); + assert!(events.iter().any(|event| event.event_type == "turn.started" + && event.payload["providerTurnId"] == "provider-turn-2")); + assert_eq!(call_count(&directory, "turn/start"), 2); + restored.shutdown().unwrap(); + fs::remove_dir_all(directory).unwrap(); +} + #[test] fn durable_backend_settles_pending_tools_when_recovery_finds_the_turn_ended() { let directory = temporary_directory("durable-tool-ended-offline");