fix: acknowledge durable submissions before run preparation

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
+    /// Durable Submit acceptance. A `Queued` receipt follows the durable FIFO
-    /// UserInput commit; a `Queued` receipt follows the durable FIFO checkpoint.
+    /// 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 {

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<Vec<WorkingDirectoryRepositoryAccessRequest>>,
         repository_access_available: AtomicBool,
         working_directory_requests: Mutex<Vec<WorkingDirectoryRequest>>,
-        preserve_commit_ack_submission_id: AtomicBool,
+        preserve_submission_acknowledgement_id: AtomicBool,
         #[cfg(feature = "ws-server")]
         snapshots: Mutex<BTreeMap<WorkerId, protocol::Event>>,
     }
@@ -4236,8 +4236,8 @@ mod tests {
             *self.stop_result.lock().unwrap() = Some(result);
         }
 
-        fn preserve_commit_ack_submission_id(&self) {
+        fn preserve_submission_acknowledgement_id(&self) {
-            self.preserve_commit_ack_submission_id
+            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,

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();
+            .clone();
-        assert!(entries.iter().any(|entry| {
+        let deadline = std::time::Instant::now() + Duration::from_secs(2);
-            matches!(
+        let entries = loop {
-                entry,
+            let entries = handle.committed_entries();
-                LogEntry::AnnotatedUserInput { segments, .. }
+            if entries.iter().any(|entry| {
-                    if segments == &vec![Segment::text("start the ticket")]
+                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| {

crates/worker/src/controller.rs

@@ -1724,14 +1724,13 @@ async fn controller_loop<C, St>(
                     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<C, St>(
                     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(),
                             });

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());

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
+// Runtime creation can spend up to 60s bootstrapping; durable Submit
-// durable initial-input acknowledgement, and 5s confirming shutdown.
+// acceptance is acknowledged before the potentially long run preparation.
-const REMOTE_WORKER_CREATE_TIMEOUT: Duration = Duration::from_secs(195);
+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 {