fix(runner): defer resumed Codex results until reattach

Persist a completed semantic result even when Codex has not reissued its JSON-RPC request yet. Deliver it when the original request arrives after thread resume, and cover the race in the restart regression.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-08-26 19:51:37 +00:00
1 parent acd412780b
commit 7a5688c6b6
3 files changed
+49 -27

No files matched your search

@@ -401,6 +401,10 @@ impl CodexProvider {
Ok(())
}
pub fn has_pending_tool_call(&self, call_id: &str) -> bool {
self.pending_tool_calls.contains_key(call_id)
}
pub fn poll(&mut self) -> Result<Option<CodexProviderEvent>, LocalRunnerError> {
let message = if let Some(message) = self.pending_messages.pop_front() {
message
@@ -825,13 +825,21 @@ impl CodexCommandExecutor {
// Persist the completed result before crossing the provider boundary.
// If delivery is interrupted, Codex reissues the same native call on
// thread resume and poll_provider redelivers this stored result.
self.ensure_provider()?
.deliver_tool_result(call_id, &result, is_error)
.map_err(|error| {
DurableRunnerError::invalid(format!("Codex tool result delivery failed: {error}"))
})?;
let provider = self.ensure_provider()?;
let delivered = if provider.has_pending_tool_call(call_id) {
provider
.deliver_tool_result(call_id, &result, is_error)
.map_err(|error| {
DurableRunnerError::invalid(format!(
"Codex tool result delivery failed: {error}"
))
})?;
true
} else {
false
};
Ok(CommandExecution::result(json!({
"status": "delivered",
"status": if delivered { "delivered" } else { "deferred" },
"callId": call_id,
})))
}
@@ -247,11 +247,28 @@ fn durable_backend_resumes_one_pending_native_tool_call_with_the_same_identity()
drop(first);
let mut recovered = CodexCommandExecutor::new(&directory);
let result = recovered
.execute(&command(
"semantic-result",
4,
"semantic_tool.result",
json!({
"callId": "semantic-call-1",
"result": {"ok": true, "value": {"task": "NCV-1"}},
"isError": false,
}),
))
.expect("persist semantic tool result before Codex reissues its request");
assert_eq!(result.result["status"], "deferred");
let mut reconciled = None;
let mut completed = false;
for _ in 0..32 {
let events = recovered
.poll_events()
.expect("poll reconciled semantic call");
completed |= events
.iter()
.any(|event| event.event_type == "turn.completed");
if let Some(event) = events
.iter()
.find(|event| event.event_type == "semantic_tool.reconciled")
@@ -268,30 +285,23 @@ fn durable_backend_resumes_one_pending_native_tool_call_with_the_same_identity()
}
let reconciled = reconciled.expect("observe native tool-call reconciliation");
assert_eq!(reconciled.payload["callId"], first_input.payload["callId"]);
assert_eq!(reconciled.payload["outcome"], "pending_call_resumed");
assert_eq!(
reconciled.payload["outcome"],
"completed_result_redelivered"
);
assert_eq!(call_count(&directory, "thread/resume"), 1);
assert_eq!(call_count(&directory, "turn/start"), 1);
recovered
.execute(&command(
"semantic-result",
4,
"semantic_tool.result",
json!({
"callId": "semantic-call-1",
"result": {"ok": true, "value": {"task": "NCV-1"}},
"isError": false,
}),
))
.expect("deliver semantic tool result to resumed Codex request");
let mut completed = false;
for _ in 0..32 {
let events = poll_and_ack(&mut recovered).expect("poll completed resumed turn");
completed |= events
.iter()
.any(|event| event.event_type == "turn.completed");
if completed {
break;
if !completed {
for _ in 0..256 {
let events = poll_and_ack(&mut recovered).expect("poll completed resumed turn");
completed |= events
.iter()
.any(|event| event.event_type == "turn.completed");
if completed {
break;
}
std::thread::sleep(std::time::Duration::from_millis(1));
}
}
assert!(completed);