From f966470d3383417a4cb45f1aec7a851390c3e611 Mon Sep 17 00:00:00 2001 From: Hare Date: Mon, 7 Sep 2026 21:38:19 +0900 Subject: [PATCH] fix: acknowledge durable submissions before run preparation --- crates/protocol/src/lib.rs | 5 +-- crates/worker-runtime/src/runtime.rs | 18 +++++----- crates/worker-runtime/src/worker_backend.rs | 37 +++++++++++++-------- crates/worker/src/controller.rs | 32 ++++++------------ crates/worker/tests/controller_test.rs | 34 +++++++++++++++++++ crates/workspace-server/src/hosts.rs | 8 ++--- 6 files changed, 84 insertions(+), 50 deletions(-) diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index d54c9232..9cbd089d 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -932,8 +932,9 @@ pub struct SessionToolAttachment { #[cfg_attr(feature = "typescript", derive(ts_rs::TS))] #[serde(tag = "event", content = "data", rename_all = "snake_case")] pub enum Event { - /// Durable Submit acceptance. A `Started` receipt follows the atomic - /// UserInput commit; a `Queued` receipt follows the durable FIFO checkpoint. + /// Durable Submit acceptance. A `Queued` receipt follows the durable FIFO + /// checkpoint. A `Started` receipt follows the pending activation checkpoint; + /// the corresponding [`Event::UserMessage`] marks the later history commit. /// Repeating the same request id and exact payload returns the same receipt /// without appending or activating twice. SubmissionAccepted { diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 7cb4367a..77d72b11 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -929,15 +929,15 @@ impl Runtime { result: dispatch_result, }); } - let has_commit_ack = dispatch_result + let has_durable_acceptance = dispatch_result .submission .as_ref() .is_some_and(|ack| ack.submission_request_id == expected_submission_id); - if !has_commit_ack { + if !has_durable_acceptance { self.cleanup_connected_failed_create(&backend, &worker_ref, &handle)?; let result = WorkerExecutionResult::rejected( WorkerExecutionOperation::Input, - "execution backend accepted initial input without a durable session commit acknowledgement", + "execution backend accepted initial input without a durable submission acknowledgement", ); return Err(RuntimeError::WorkerExecutionRejected { worker_id: worker_ref.worker_id.clone(), @@ -4222,7 +4222,7 @@ mod tests { repository_accesses: Mutex>, repository_access_available: AtomicBool, working_directory_requests: Mutex>, - preserve_commit_ack_submission_id: AtomicBool, + preserve_submission_acknowledgement_id: AtomicBool, #[cfg(feature = "ws-server")] snapshots: Mutex>, } @@ -4236,8 +4236,8 @@ mod tests { *self.stop_result.lock().unwrap() = Some(result); } - fn preserve_commit_ack_submission_id(&self) { - self.preserve_commit_ack_submission_id + fn preserve_submission_acknowledgement_id(&self) { + self.preserve_submission_acknowledgement_id .store(true, Ordering::SeqCst); } @@ -4382,7 +4382,7 @@ mod tests { ) }); if !self - .preserve_commit_ack_submission_id + .preserve_submission_acknowledgement_id .load(Ordering::SeqCst) && let (Some(ack), Some(submission_id)) = (result.submission.as_mut(), submission_id) @@ -5202,7 +5202,7 @@ mod tests { #[test] fn create_worker_rejects_mismatched_submission_acknowledgement() { let (runtime, backend) = runtime_and_backend(); - backend.preserve_commit_ack_submission_id(); + backend.preserve_submission_acknowledgement_id(); backend.set_dispatch_result(WorkerExecutionResult::accepted_submission( WorkerExecutionOperation::Input, "request-test", @@ -5225,7 +5225,7 @@ mod tests { } #[test] - fn create_worker_rejects_initial_input_without_commit_acknowledgement() { + fn create_worker_rejects_initial_input_without_durable_submission_acknowledgement() { let (runtime, backend) = runtime_and_backend(); backend.set_dispatch_result(WorkerExecutionResult::accepted( WorkerExecutionOperation::Input, diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index c81f5523..a413c45b 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -90,11 +90,11 @@ use worker::{ const DEFAULT_BACKEND_ID: &str = "worker-crate"; const RUNTIME_TASK_TIMEOUT: Duration = Duration::from_secs(10); const SPAWN_RESTORE_TASK_TIMEOUT: Duration = Duration::from_secs(60); -const USER_INPUT_TASK_TIMEOUT: Duration = Duration::from_secs(125); +const USER_INPUT_TASK_TIMEOUT: Duration = Duration::from_secs(10); const WORKSPACE_CONFIG_HTTP_TIMEOUT: Duration = Duration::from_secs(8); const MAX_WORKSPACE_CONFIG_RESPONSE_BYTES: usize = 72 * 1024 * 1024; // Leave adapter cancellation margin after the durable submission deadline. -const USER_INPUT_COMMIT_TIMEOUT: Duration = Duration::from_secs(120); +const USER_INPUT_COMMIT_TIMEOUT: Duration = Duration::from_secs(9); pub struct RuntimeWorkerController { pub handle: WorkerHandle, @@ -2546,7 +2546,7 @@ mod tests { assert!(USER_INPUT_TASK_TIMEOUT > USER_INPUT_COMMIT_TIMEOUT); assert_eq!( USER_INPUT_TASK_TIMEOUT - USER_INPUT_COMMIT_TIMEOUT, - Duration::from_secs(5) + Duration::from_secs(1) ); } @@ -3562,7 +3562,7 @@ mod tests { } #[test] - fn create_with_initial_input_returns_after_session_commit() { + fn create_with_initial_input_returns_after_durable_submission_acceptance() { let client = MockClient::new(simple_text_events()); let runtime_base = tempfile::tempdir().unwrap(); let cwd = tempfile::tempdir().unwrap(); @@ -3586,21 +3586,32 @@ mod tests { let detail = runtime.create_worker(request).unwrap(); - let entries = backend + let handle = backend .workers .lock() .unwrap() .get(&detail.worker_ref) .expect("live Worker execution") .handle - .committed_entries(); - assert!(entries.iter().any(|entry| { - matches!( - entry, - LogEntry::AnnotatedUserInput { segments, .. } - if segments == &vec![Segment::text("start the ticket")] - ) - })); + .clone(); + let deadline = std::time::Instant::now() + Duration::from_secs(2); + let entries = loop { + let entries = handle.committed_entries(); + if entries.iter().any(|entry| { + matches!( + entry, + LogEntry::AnnotatedUserInput { segments, .. } + if segments == &vec![Segment::text("start the ticket")] + ) + }) { + break entries; + } + assert!( + std::time::Instant::now() < deadline, + "durably accepted initial input must eventually commit to history" + ); + std::thread::sleep(Duration::from_millis(10)); + }; let submission_id = entries .iter() .find_map(|entry| { diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 1655a387..26e8a77c 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -1724,14 +1724,13 @@ async fn controller_loop( true, ) { Ok(acceptance) => { + let _ = working_event_tx.send(Event::SubmissionAccepted { + submission_request_id: acceptance.submission_request_id.clone(), + submission_id: acceptance.submission_id.clone(), + disposition: acceptance.disposition, + }); if let Some(activation) = acceptance.activation { pending = Some(PendingRun::Submit(activation)); - } else { - let _ = working_event_tx.send(Event::SubmissionAccepted { - submission_request_id: acceptance.submission_request_id, - submission_id: acceptance.submission_id, - disposition: acceptance.disposition, - }); } } Err(error) => { @@ -1758,14 +1757,13 @@ async fn controller_loop( true, ) { Ok(acceptance) => { + let _ = working_event_tx.send(Event::SubmissionAccepted { + submission_request_id: acceptance.submission_request_id.clone(), + submission_id: acceptance.submission_id.clone(), + disposition: acceptance.disposition, + }); if let Some(activation) = acceptance.activation { pending = Some(PendingRun::Submit(activation)); - } else { - let _ = working_event_tx.send(Event::SubmissionAccepted { - submission_request_id: acceptance.submission_request_id, - submission_id: acceptance.submission_id, - disposition: acceptance.disposition, - }); } } Err(error) => { @@ -2481,11 +2479,6 @@ where if committed.is_ok() { if let Some(submission) = submission { pending_submissions.finish_activation(&submission.submission_id); - let _ = working_event_tx.send(Event::SubmissionAccepted { - submission_request_id: submission.submission_request_id, - submission_id: submission.submission_id, - disposition: protocol::SubmissionDisposition::Started, - }); let _ = working_event_tx.send(Event::PendingSubmissionsChanged { pending: pending_submissions.snapshot(), }); @@ -2506,11 +2499,6 @@ where match receiver.try_recv() { Ok(()) => { pending_submissions.finish_activation(&submission.submission_id); - let _ = working_event_tx.send(Event::SubmissionAccepted { - submission_request_id: submission.submission_request_id, - submission_id: submission.submission_id, - disposition: protocol::SubmissionDisposition::Started, - }); let _ = working_event_tx.send(Event::PendingSubmissionsChanged { pending: pending_submissions.snapshot(), }); diff --git a/crates/worker/tests/controller_test.rs b/crates/worker/tests/controller_test.rs index ba1fada3..038d99e9 100644 --- a/crates/worker/tests/controller_test.rs +++ b/crates/worker/tests/controller_test.rs @@ -1037,6 +1037,40 @@ permission = "write" ); } +#[tokio::test] +async fn started_submit_emits_one_durable_acceptance_receipt() { + let worker = make_worker(MockClient::new(simple_text_events())).await; + let handle = spawn_controller(worker).await; + let mut events = handle.subscribe(); + let submission_request_id = protocol::new_submission_request_id(); + handle + .send(Method::submit_text(submission_request_id.clone(), "start")) + .await + .unwrap(); + + let mut receipts = Vec::new(); + tokio::time::timeout(std::time::Duration::from_secs(2), async { + loop { + match events.recv().await.unwrap() { + Event::SubmissionAccepted { + submission_request_id: received_request_id, + submission_id, + disposition, + } if received_request_id == submission_request_id => { + receipts.push((submission_id, disposition)); + } + Event::TurnEnd { .. } => break, + _ => {} + } + } + }) + .await + .expect("submitted turn completes"); + + assert_eq!(receipts.len(), 1); + assert_eq!(receipts[0].1, protocol::SubmissionDisposition::Started); +} + #[tokio::test] async fn run_end_returns_to_idle_without_busy_status() { let client = MockClient::new(simple_text_events()); diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index bc4a7656..27d1c4e6 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -69,9 +69,9 @@ const EMBEDDED_HOST_KIND: &str = "embedded-worker-runtime-host"; const REMOTE_HOST_KIND: &str = "remote-worker-runtime-host"; const MAX_DIAGNOSTICS: usize = 16; const MAX_RUNTIME_PING_RESPONSE_BYTES: usize = 8 * 1024; -// Runtime creation can spend up to 60s bootstrapping, 125s waiting for the -// durable initial-input acknowledgement, and 5s confirming shutdown. -const REMOTE_WORKER_CREATE_TIMEOUT: Duration = Duration::from_secs(195); +// Runtime creation can spend up to 60s bootstrapping; durable Submit +// acceptance is acknowledged before the potentially long run preparation. +const REMOTE_WORKER_CREATE_TIMEOUT: Duration = Duration::from_secs(80); const MAX_HOST_SCAN: usize = 256; const MAX_IDENTIFIER_LEN: usize = 120; const ID_DIGEST_HEX_LEN: usize = 16; @@ -4736,7 +4736,7 @@ mod tests { #[test] fn remote_worker_create_timeout_covers_runtime_phase_budgets() { - assert!(REMOTE_WORKER_CREATE_TIMEOUT > Duration::from_secs(60 + 125 + 5)); + assert!(REMOTE_WORKER_CREATE_TIMEOUT > Duration::from_secs(60 + 10 + 5)); } fn test_create_binding() -> WorkerCreateBinding {