diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index b53729543..e616231c4 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -205,7 +205,9 @@ impl From for WorkerStateSnapshot { #[cfg_attr(feature = "typescript", derive(ts_rs::TS))] #[serde(tag = "method", content = "params", rename_all = "snake_case")] pub enum Method { - /// Durably accept typed input for immediate activation or the session FIFO. + /// Durably accept typed input for immediate activation while the Worker is Idle. + /// Running or Paused Workers reject Submit; use Notify for advisory information + /// that should be incorporated into work already in progress. /// /// `submission_request_id` is generated by the authenticated caller and is /// used only for idempotent retry. Worker allocates the durable @@ -724,6 +726,8 @@ pub enum ToolResultDisposition { #[serde(rename_all = "snake_case")] pub enum SubmissionDisposition { Started, + /// Compatibility value for durable receipts created before Submit became + /// Idle-only. New submissions are never accepted with this disposition. Queued, } @@ -859,9 +863,10 @@ 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 `Queued` receipt follows the durable FIFO - /// checkpoint. A `Started` receipt follows the pending activation checkpoint; - /// the corresponding [`Event::UserMessage`] marks the later history commit. + /// Durable Submit acceptance. New submissions are accepted only while Idle, + /// and a `Started` receipt follows the pending activation checkpoint; the + /// corresponding [`Event::UserMessage`] marks the later history commit. + /// `Queued` remains readable for receipts persisted by older runtimes. /// Repeating the same request id and exact payload returns the same receipt /// without appending or activating twice. SubmissionAccepted { diff --git a/crates/tui/src/app.rs b/crates/tui/src/app.rs index 86cd7cbef..0c0147387 100644 --- a/crates/tui/src/app.rs +++ b/crates/tui/src/app.rs @@ -759,6 +759,12 @@ impl App { } return None; } + if self.worker_status != WorkerStatus::Idle { + self.push_error( + "Submit requires an idle Worker; use Alt-n Notify for work already in progress.", + ); + return None; + } self.record_input_history(segments.clone()); self.input.clear(); Some(self.method_for_run(segments)) @@ -773,7 +779,9 @@ impl App { .iter() .any(|segment| matches!(segment, Segment::UploadedFile { .. })) { - self.push_error("Notify accepts text only; remove attachments or queue a Submit."); + self.push_error( + "Notify accepts text only; remove attachments or wait until the Worker is idle to Submit.", + ); return None; } let message = Segment::flatten_to_text(&segments); @@ -3514,16 +3522,22 @@ mod completion_flow_tests { } #[test] - fn running_submit_is_sent_to_the_worker_and_not_queued_locally() { + fn running_submit_is_rejected_locally_and_preserves_the_draft() { let mut app = App::new("test".into()); app.set_worker_status(WorkerStatus::Running); - insert_text(&mut app, "queued turn"); + insert_text(&mut app, "advisory context"); let method = app.submit_input(); - assert!(matches!(method, Some(Method::Submit { .. }))); + assert!(method.is_none()); assert_eq!(app.queued_input_count(), 0); - assert_eq!(input_text(&app), ""); + assert_eq!(input_text(&app), "advisory context"); + assert!(app.blocks.iter().any(|block| matches!( + block, + Block::Alert { message, .. } + if message.contains("requires an idle Worker") + && message.contains("Alt-n Notify") + ))); } #[test] @@ -3554,6 +3568,17 @@ mod completion_flow_tests { assert_eq!(app.queued_input_count(), 1); } + #[test] + fn paused_non_empty_submit_is_rejected_locally_and_preserves_the_draft() { + let mut app = App::new("test".into()); + app.set_worker_status(WorkerStatus::Paused); + insert_text(&mut app, "different request"); + + assert!(app.submit_input().is_none()); + assert_eq!(input_text(&app), "different request"); + assert_eq!(app.queued_input_count(), 0); + } + #[test] fn paused_empty_submit_still_resumes_immediately() { let mut app = App::new("test".into()); @@ -4609,13 +4634,18 @@ mod completion_flow_tests { } #[test] - fn input_history_records_running_submits_and_suppresses_consecutive_duplicates() { + fn input_history_records_only_an_idle_submit_after_running_rejection() { let mut app = App::new("test".into()); - app.running = true; + app.set_worker_status(WorkerStatus::Running); for c in "repeat".chars() { app.insert_char(c); } + assert!(app.submit_input().is_none()); + assert_eq!(app.input_history_len(), 0); + assert_eq!(input_text(&app), "repeat"); + + app.set_worker_status(WorkerStatus::Idle); assert!(app.submit_input().is_some()); assert_eq!(app.input_history_len(), 1); assert_eq!(app.queued_input_count(), 0); diff --git a/crates/tui/src/console/mod.rs b/crates/tui/src/console/mod.rs index a3459b7c7..fa5a2b2fa 100644 --- a/crates/tui/src/console/mod.rs +++ b/crates/tui/src/console/mod.rs @@ -1682,8 +1682,8 @@ mod tests { assert!(connection.awaiting_attachment_acceptance.is_empty()); } - #[tokio::test] - async fn running_attachment_submit_failure_restores_draft_without_exiting_console() { + #[test] + fn running_attachment_submit_is_rejected_without_consuming_the_draft() { let file = UploadedFileRef { artifact_id: "artifact-queued".into(), file_name: "queued.txt".into(), @@ -1694,7 +1694,7 @@ mod tests { sha256: "a".repeat(64), source_entry_id: None, }; - let mut connection = ConsoleConnection { + let connection = ConsoleConnection { client: Client::new(FailOnceSocket { fail_next_send: true, }), @@ -1709,13 +1709,10 @@ mod tests { let mut app = App::new("worker".into()); app.set_worker_status(WorkerStatus::Running); app.input.insert_str("queued inspect"); - let method = app - .submit_input() - .expect("running Submit is sent immediately"); - - send_console_method(&mut app, &mut connection, &method) - .await - .unwrap(); + assert!( + app.submit_input().is_none(), + "running Submit must be rejected before transport" + ); assert_eq!(app.input.plain_text(), "queued inspect"); assert_eq!(connection.pending_attachments, vec![file]); @@ -2035,7 +2032,7 @@ mod tests { } #[test] - fn running_enter_sends_submit_to_worker() { + fn running_enter_rejects_submit_and_preserves_the_draft() { let mut app = App::new("agent".to_string()); app.set_worker_status(WorkerStatus::Running); for c in "queued".chars() { @@ -2048,13 +2045,10 @@ mod tests { ); } - assert!(matches!( - handle_key(&mut app, KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE)), - Some(Method::Submit { .. }) - )); + assert!(handle_key(&mut app, KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE)).is_none()); assert_eq!(app.queued_input_count(), 0); - assert_eq!(input_text(&app), ""); + assert_eq!(input_text(&app), "queued"); } #[test] diff --git a/crates/tui/src/ui.rs b/crates/tui/src/ui.rs index b7496a3e5..33d4b2756 100644 --- a/crates/tui/src/ui.rs +++ b/crates/tui/src/ui.rs @@ -1868,7 +1868,7 @@ fn draw_status(frame: &mut Frame, app: &App, area: Rect) { .add_modifier(Modifier::BOLD), )); spans.push(Span::styled( - " — Enter to resume, Ctrl-X to cancel, type to start new turn", + " — Enter to resume, Ctrl-X to cancel, Alt-n to notify", Style::default().fg(Color::DarkGray), )); } else if !app.running { diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 87c61f140..19ca2f55d 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -1866,18 +1866,20 @@ impl Runtime { mut input: WorkerInput, ) -> Result { validate_worker_input(&input)?; - let expected_submission_id = - if matches!(input.kind, WorkerInputKind::User | WorkerInputKind::Notify) { - let submission_id = input - .submission_request_id - .clone() - .filter(|request_id| !request_id.trim().is_empty()) - .unwrap_or_else(|| Uuid::now_v7().to_string()); - input.submission_request_id = Some(submission_id.clone()); - Some(submission_id) - } else { - None - }; + if matches!(input.kind, WorkerInputKind::User | WorkerInputKind::Notify) { + let request_id = input + .submission_request_id + .clone() + .filter(|request_id| !request_id.trim().is_empty()) + .unwrap_or_else(|| Uuid::now_v7().to_string()); + input.submission_request_id = Some(request_id); + } + let expected_submission_id = (input.kind == WorkerInputKind::User).then(|| { + input + .submission_request_id + .clone() + .expect("User input request id must be assigned before dispatch") + }); self.ensure_worker_execution(worker_ref)?; let (backend, handle) = { let state = self.lock()?; diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index 0cd3ad5de..157cb6a78 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -4360,7 +4360,7 @@ mod tests { } #[test] - fn running_worker_accepts_a_second_submit_as_queued() { + fn running_worker_replays_retry_but_rejects_a_new_submit() { let client = MockClient::sequential(vec![MockResponse::Hang(vec![])]); let runtime_base = tempfile::tempdir().unwrap(); let cwd = tempfile::tempdir().unwrap(); @@ -4404,13 +4404,50 @@ mod tests { let mut second_input = WorkerInput::user("second"); second_input.submission_request_id = Some("request-second".into()); - let second = runtime + let error = runtime .send_input(&detail.worker_ref, second_input) + .expect_err("a new Submit must be rejected while the Worker is running"); + assert!(error.to_string().contains("requires an idle Worker")); + assert!(error.to_string().contains("use Notify")); + } + + #[test] + fn running_worker_accepts_notify_without_a_submit_receipt() { + let client = MockClient::sequential(vec![MockResponse::Hang(vec![])]); + let runtime_base = tempfile::tempdir().unwrap(); + let cwd = tempfile::tempdir().unwrap(); + let store = tempfile::tempdir().unwrap(); + let factory = MockFactory { + client, + runtime_base: runtime_base.path().to_path_buf(), + cwd: cwd.path().to_path_buf(), + store_dir: store.path().join("sessions"), + worker_metadata_dir: store.path().join("workers"), + observed_cwds: Arc::new(Mutex::new(Vec::new())), + observed_workspace_clients: Arc::new(Mutex::new(Vec::new())), + }; + let backend = Arc::new(WorkerRuntimeExecutionBackend::new(factory).unwrap()); + let runtime = + EmbeddedRuntime::with_execution_backend(RuntimeOptions::default(), backend.clone()) + .unwrap(); + runtime.store_config_bundle(test_bundle()).unwrap(); + let detail = runtime + .create_worker(create_request("running-notify")) .unwrap(); - assert_eq!( - second.submission.as_ref().map(|ack| ack.disposition), - Some(protocol::SubmissionDisposition::Queued) - ); + + runtime + .send_input(&detail.worker_ref, WorkerInput::user("first")) + .unwrap(); + wait_for_adapter_state(&backend, &detail.worker_ref, WorkerStatus::Running); + + let mut notification = WorkerInput::notify("advisory context"); + notification.submission_request_id = Some("notification-request".into()); + let acknowledgement = runtime + .send_input(&detail.worker_ref, notification) + .expect("Running Worker must accept Notify without a Submit receipt"); + + assert!(acknowledgement.submission.is_none()); + wait_for_adapter_state(&backend, &detail.worker_ref, WorkerStatus::Running); } #[test] diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index babae8e21..ebd3c17b7 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -1477,6 +1477,51 @@ where Ok(workdir_for_view) } +const SUBMIT_REQUIRES_IDLE_MESSAGE: &str = + "Submit requires an idle Worker; use Notify for work already in progress"; + +fn reject_submit_while_not_idle( + submission_request_id: String, + working_event_tx: &broadcast::Sender, +) { + let _ = working_event_tx.send(Event::SubmissionRejected { + submission_request_id, + message: SUBMIT_REQUIRES_IDLE_MESSAGE.into(), + }); +} + +fn replay_or_reject_submit_while_not_idle( + pending_submissions: &PendingSubmissionHandle, + submission_request_id: String, + input: Vec, + source_namespace: String, + working_event_tx: &broadcast::Sender, +) where + St: session_store::Store + Clone, +{ + let request_id = submission_request_id.clone(); + match pending_submissions.replay_acceptance_from_source( + submission_request_id, + input, + source_namespace, + ) { + Ok(Some(acceptance)) => { + let _ = working_event_tx.send(Event::SubmissionAccepted { + submission_request_id: acceptance.submission_request_id, + submission_id: acceptance.submission_id, + disposition: acceptance.disposition, + }); + } + Ok(None) => reject_submit_while_not_idle(request_id, working_event_tx), + Err(error) => { + let _ = working_event_tx.send(Event::SubmissionRejected { + submission_request_id: request_id, + message: error.to_string(), + }); + } + } +} + fn durably_accept_method_while_busy( method: Method, pending_submissions: &PendingSubmissionHandle, @@ -1490,28 +1535,13 @@ where submission_request_id, input, } => { - let request_id = submission_request_id.clone(); - match pending_submissions.accept_from_source( + replay_or_reject_submit_while_not_idle( + pending_submissions, submission_request_id, input, pending_submissions.direct_client_namespace(), - session_store::LoggedSessionHistoryOrigin::LegacyUnknown, - false, - ) { - Ok(acceptance) => { - let _ = working_event_tx.send(Event::SubmissionAccepted { - submission_request_id: acceptance.submission_request_id, - submission_id: acceptance.submission_id, - disposition: acceptance.disposition, - }); - } - Err(error) => { - let _ = working_event_tx.send(Event::SubmissionRejected { - submission_request_id: request_id, - message: error.to_string(), - }); - } - } + working_event_tx, + ); None } Method::SubmitTracked { @@ -1519,30 +1549,14 @@ where input, source, } => { - let request_id = submission_request_id.clone(); - let (source_namespace, provenance) = - resolved_input_source(pending_submissions, &source); - match pending_submissions.accept_from_source( + let (source_namespace, _) = resolved_input_source(pending_submissions, &source); + replay_or_reject_submit_while_not_idle( + pending_submissions, submission_request_id, input, source_namespace, - provenance, - false, - ) { - Ok(acceptance) => { - let _ = working_event_tx.send(Event::SubmissionAccepted { - submission_request_id: acceptance.submission_request_id, - submission_id: acceptance.submission_id, - disposition: acceptance.disposition, - }); - } - Err(error) => { - let _ = working_event_tx.send(Event::SubmissionRejected { - submission_request_id: request_id, - message: error.to_string(), - }); - } - } + working_event_tx, + ); None } Method::Notify { @@ -1955,13 +1969,22 @@ async fn controller_loop( submission_request_id, input, } => { + if shared_state.catalog_status() != WorkerStatus::Idle { + replay_or_reject_submit_while_not_idle( + &pending_submissions, + submission_request_id, + input, + pending_submissions.direct_client_namespace(), + &working_event_tx, + ); + continue; + } let request_id = submission_request_id.clone(); - match pending_submissions.accept_from_source( + match pending_submissions.accept_idle_from_source( submission_request_id, input, pending_submissions.direct_client_namespace(), session_store::LoggedSessionHistoryOrigin::LegacyUnknown, - true, ) { Ok(acceptance) => { let _ = working_event_tx.send(Event::SubmissionAccepted { @@ -1986,15 +2009,24 @@ async fn controller_loop( input, source, } => { - let request_id = submission_request_id.clone(); let (source_namespace, provenance) = resolved_input_source(&pending_submissions, &source); - match pending_submissions.accept_from_source( + if shared_state.catalog_status() != WorkerStatus::Idle { + replay_or_reject_submit_while_not_idle( + &pending_submissions, + submission_request_id, + input, + source_namespace, + &working_event_tx, + ); + continue; + } + let request_id = submission_request_id.clone(); + match pending_submissions.accept_idle_from_source( submission_request_id, input, source_namespace, provenance, - true, ) { Ok(acceptance) => { let _ = working_event_tx.send(Event::SubmissionAccepted { @@ -2951,64 +2983,28 @@ where submission_request_id, input, }) => { - let request_id = submission_request_id.clone(); - match pending_submissions.accept_from_source( + replay_or_reject_submit_while_not_idle( + pending_submissions, submission_request_id, input, pending_submissions.direct_client_namespace(), - session_store::LoggedSessionHistoryOrigin::LegacyUnknown, - false, - ) { - Ok(acceptance) => { - let _ = working_event_tx.send(Event::SubmissionAccepted { - submission_request_id: acceptance.submission_request_id, - submission_id: acceptance.submission_id, - disposition: acceptance.disposition, - }); - let _ = working_event_tx.send(Event::PendingSubmissionsChanged { - pending: pending_submissions.snapshot(), - }); - } - Err(error) => { - let _ = working_event_tx.send(Event::SubmissionRejected { - submission_request_id: request_id, - message: error.to_string(), - }); - } - } + working_event_tx, + ); } Some(Method::SubmitTracked { submission_request_id, input, source, }) => { - let request_id = submission_request_id.clone(); - let (source_namespace, provenance) = + let (source_namespace, _) = resolved_input_source(pending_submissions, &source); - match pending_submissions.accept_from_source( + replay_or_reject_submit_while_not_idle( + pending_submissions, submission_request_id, input, source_namespace, - provenance, - false, - ) { - Ok(acceptance) => { - let _ = working_event_tx.send(Event::SubmissionAccepted { - submission_request_id: acceptance.submission_request_id, - submission_id: acceptance.submission_id, - disposition: acceptance.disposition, - }); - let _ = working_event_tx.send(Event::PendingSubmissionsChanged { - pending: pending_submissions.snapshot(), - }); - } - Err(error) => { - let _ = working_event_tx.send(Event::SubmissionRejected { - submission_request_id: request_id, - message: error.to_string(), - }); - } - } + working_event_tx, + ); } Some(Method::Resume { command }) => { if let Err(disposition) = @@ -3568,6 +3564,38 @@ mod tests { } } + #[test] + fn busy_submit_is_rejected_without_queueing() { + let dir = tempfile::tempdir().unwrap(); + let pending = PendingSubmissionHandle::for_test(dir.path()); + let (event_tx, mut event_rx) = broadcast::channel(4); + + assert!( + durably_accept_method_while_busy( + Method::SubmitTracked { + submission_request_id: "request-1".into(), + input: vec![protocol::Segment::text("follow-up")], + source: protocol::AuthenticatedInputSource::Backend { + operation_id: "operation-1".into(), + }, + }, + &pending, + &event_tx, + ) + .is_none() + ); + assert!(matches!( + event_rx.try_recv().unwrap(), + Event::SubmissionRejected { + submission_request_id, + message, + } if submission_request_id == "request-1" + && message == SUBMIT_REQUIRES_IDLE_MESSAGE + )); + assert!(pending.snapshot().submissions.is_empty()); + assert!(pending.persisted_entries_for_test().is_empty()); + } + #[test] fn busy_notification_is_durable_before_shutdown_can_observe_it() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates/worker/src/feature/builtin/manage_worker.rs b/crates/worker/src/feature/builtin/manage_worker.rs index 1f9ced49b..676f73963 100644 --- a/crates/worker/src/feature/builtin/manage_worker.rs +++ b/crates/worker/src/feature/builtin/manage_worker.rs @@ -827,8 +827,12 @@ impl WorkerOperation { Self::Spawn => { "Spawn a Backend/Runtime Worker session in an existing Workspace Workdir. The Workdir id is authority; filesystem paths and Runtime URLs are not accepted. `initial_submit` carries the normal typed user submission. After the Orchestrator has committed a Ticket to `inprogress`, set `ticket_id` with a Flow segment in `initial_submit` to atomically assign the new Coder Worker; the operation id is derived from the durable tool call rather than model input." } - Self::SendInput => "Send user input to a known Runtime Worker when allowed.", - Self::Notify => "Send an advisory notification to a known Runtime Worker when allowed.", + Self::SendInput => { + "Start a fresh turn by sending user input to a known Idle Worker. Running or Paused targets reject Submit; use WorkerNotify for advisory information during work already in progress." + } + Self::Notify => { + "Send an advisory notification to a known Runtime Worker so information can be incorporated into work already in progress without creating a queued Submit." + } Self::Cancel => "Cancel the current turn of a known Runtime Worker when allowed.", Self::Stop => "Stop a known Runtime Worker when allowed.", Self::Restore => { @@ -1079,7 +1083,9 @@ fn format_elapsed(elapsed_ms: u64) -> String { fn sub_worker_tool_declaration(operation: WorkerOperation) -> ToolDeclaration { let description = match operation { WorkerOperation::List => "List this Worker's direct SubWorkers.", - WorkerOperation::SendInput => "Send a new user turn to a direct SubWorker.", + WorkerOperation::SendInput => { + "Start a fresh user turn for an Idle direct SubWorker. Running or Paused SubWorkers reject Submit." + } WorkerOperation::Stop => { "Stop a direct SubWorker and release all authority delegated to its child session." } diff --git a/crates/worker/src/prompt/catalog.rs b/crates/worker/src/prompt/catalog.rs index df1261f6b..1a73bc21e 100644 --- a/crates/worker/src/prompt/catalog.rs +++ b/crates/worker/src/prompt/catalog.rs @@ -913,6 +913,9 @@ mod tests { "missing canonical tool {name}" ); } + assert!(orchestration.contains("Idle-only Submit")); + assert!(orchestration.contains("use `WorkerNotify` instead")); + assert!(orchestration.contains("do not create a queued Submit")); for alias in ["SubWorkerList", "SubWorkerSend", "SubWorkerStop"] { assert!( !orchestration.contains(alias), diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 087e55e73..c020c0782 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -1434,7 +1434,60 @@ where ) } - pub(crate) fn accept_from_source( + pub(crate) fn accept_idle_from_source( + &self, + submission_request_id: String, + input: Vec, + source_namespace: String, + provenance: WorkerHistoryProvenance, + ) -> Result { + self.accept_from_source( + submission_request_id, + input, + source_namespace, + provenance, + true, + ) + } + + pub(crate) fn replay_acceptance_from_source( + &self, + submission_request_id: String, + input: Vec, + source_namespace: String, + ) -> Result, PendingSubmissionError> { + if submission_request_id.trim().is_empty() { + return Err(PendingSubmissionError::EmptyRequestId); + } + if submission_request_id.len() > MAX_ACTIVATION_REQUEST_ID_BYTES { + return Err(PendingSubmissionError::RequestIdLimit); + } + if input.is_empty() { + return Err(PendingSubmissionError::EmptyInput); + } + let payload_digest = submission_payload_digest(&input); + let current = self + .state + .lock() + .expect("pending activation state poisoned"); + let Some(receipt) = current.receipts.iter().find(|receipt| { + receipt.submission_request_id == submission_request_id + && receipt.source_namespace == source_namespace + }) else { + return Ok(None); + }; + if receipt.payload_digest != payload_digest { + return Err(PendingSubmissionError::IdempotencyConflict); + } + Ok(Some(SubmissionAcceptance { + submission_request_id, + submission_id: receipt.submission_id.clone(), + disposition: receipt.disposition, + activation: None, + })) + } + + fn accept_from_source( &self, submission_request_id: String, input: Vec, diff --git a/crates/worker/tests/controller_test.rs b/crates/worker/tests/controller_test.rs index 0f996122b..326325dff 100644 --- a/crates/worker/tests/controller_test.rs +++ b/crates/worker/tests/controller_test.rs @@ -1376,8 +1376,48 @@ async fn events_are_broadcast() { } #[tokio::test] -async fn submit_while_running_is_durably_queued() { - // Keep the first turn in-flight until the second Submit is accepted. +async fn submit_requires_an_idle_worker() { + async fn wait_for_rejection( + rx: &mut tokio::sync::broadcast::Receiver, + expected_request_id: &str, + ) -> String { + tokio::time::timeout(std::time::Duration::from_secs(1), async { + loop { + if let Ok(Event::SubmissionRejected { + submission_request_id, + message, + }) = rx.recv().await + && submission_request_id == expected_request_id + { + break message; + } + } + }) + .await + .expect("Submit rejection") + } + + async fn wait_for_acceptance( + rx: &mut tokio::sync::broadcast::Receiver, + expected_request_id: &str, + ) -> protocol::SubmissionDisposition { + tokio::time::timeout(std::time::Duration::from_secs(1), async { + loop { + if let Ok(Event::SubmissionAccepted { + submission_request_id, + disposition, + .. + }) = rx.recv().await + && submission_request_id == expected_request_id + { + break disposition; + } + } + }) + .await + .expect("Submit acceptance") + } + let events = vec![ LlmEvent::text_block_start(0), LlmEvent::text_delta(0, "slow..."), @@ -1391,40 +1431,38 @@ async fn submit_while_running_is_durably_queued() { .send(Method::submit_text("request-first", "first")) .await .unwrap(); + let first = wait_for_acceptance(&mut rx, "request-first").await; + assert_eq!(first, protocol::SubmissionDisposition::Started); wait_for_status(&handle, WorkerStatus::Running).await; + handle - .send(Method::submit_text("request-second", "second")) + .send(Method::submit_text("request-first", "first")) + .await + .unwrap(); + let replay = wait_for_acceptance(&mut rx, "request-first").await; + assert_eq!(replay, protocol::SubmissionDisposition::Started); + + handle + .send(Method::submit_text("request-running", "second")) .await .unwrap(); - let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(2); - let mut accepted = None; - let mut pending_snapshot = None; - while tokio::time::Instant::now() < deadline { - match tokio::time::timeout(std::time::Duration::from_millis(100), rx.recv()).await { - Ok(Ok(Event::SubmissionAccepted { - submission_request_id, - disposition, - .. - })) if submission_request_id == "request-second" => accepted = Some(disposition), - Ok(Ok(Event::PendingSubmissionsChanged { pending })) - if pending.submissions.len() == 1 => - { - pending_snapshot = Some(pending) + let running_rejection = wait_for_rejection(&mut rx, "request-running").await; + assert!(running_rejection.contains("requires an idle Worker")); + assert!(running_rejection.contains("use Notify")); + + handle.send(Method::ListPendingSubmissions).await.unwrap(); + let pending = tokio::time::timeout(std::time::Duration::from_secs(1), async { + loop { + if let Ok(Event::PendingSubmissionsChanged { pending }) = rx.recv().await { + break pending; } - Ok(Ok(Event::Error { code, message })) if code == worker::ErrorCode::AlreadyRunning => { - panic!("Submit was busy-rejected: {message}") - } - _ => {} - } - if accepted.is_some() && pending_snapshot.is_some() { - break; } - } + }) + .await + .expect("pending snapshot"); + assert!(pending.submissions.is_empty()); - assert_eq!(accepted, Some(protocol::SubmissionDisposition::Queued)); - let pending_snapshot = pending_snapshot.expect("pending snapshot"); - assert_eq!(pending_snapshot.submissions.len(), 1); handle .send(Method::Pause { command: worker_command(&handle), @@ -1433,25 +1471,13 @@ async fn submit_while_running_is_durably_queued() { .unwrap(); wait_for_status(&handle, WorkerStatus::Paused).await; handle - .send(Method::ContinuePending { - expected_revision: pending_snapshot.revision, - expected_head_id: pending_snapshot.head_id.expect("pending head"), - }) + .send(Method::submit_text("request-paused", "third")) .await .unwrap(); - let rejection = tokio::time::timeout(std::time::Duration::from_secs(1), async { - loop { - if let Ok(Event::Error { code, message }) = rx.recv().await - && code == worker::ErrorCode::InvalidRequest - && message.contains("requires an idle Worker") - { - break message; - } - } - }) - .await - .expect("paused ContinuePending rejection"); - assert!(rejection.contains("Resume or Cancel")); + + let paused_rejection = wait_for_rejection(&mut rx, "request-paused").await; + assert!(paused_rejection.contains("requires an idle Worker")); + assert!(paused_rejection.contains("use Notify")); assert_eq!(handle.shared_state.catalog_status(), WorkerStatus::Paused); } @@ -2783,12 +2809,12 @@ async fn pause_then_resume_preserves_notifications_and_history_consistency() { assert!(!has_tool_call, "no orphan tool_call in history"); } -/// Paused with an orphan `tool_use` in history + a fresh `Method::Submit` -/// must produce a wire-valid next LLM request: the orphan is closed -/// with a synthetic `tool_result`, a system note is inserted, and the -/// new user input is appended. +/// Paused with an orphan `tool_use` in history must reject a fresh Submit. +/// After explicit Cancel returns the Worker to Idle, a new Submit must produce +/// a wire-valid next LLM request: the orphan is closed with a synthetic +/// `tool_result`, a system note is inserted, and the new user input is appended. #[tokio::test] -async fn paused_then_run_closes_orphan_tool_use_for_next_request() { +async fn paused_submit_is_rejected_before_cancel_and_fresh_run() { // Response 1: emit a tool_use block (complete with stop) targeting // our hanging tool. The Engine commits the ToolCall to history, // then parks inside `execute_tools` waiting on the tool — which is @@ -2858,10 +2884,44 @@ async fn paused_then_run_closes_orphan_tool_use_for_next_request() { tokio::time::sleep(std::time::Duration::from_millis(50)).await; assert_eq!(handle.shared_state.catalog_status(), WorkerStatus::Paused); - // New user input while Paused → `Worker::run` observes - // `last_run_interrupted` and runs its interrupt-prep step, which - // closes the orphan + injects a system note before the fresh user - // message. + let paused_request_id = protocol::new_submission_request_id(); + handle + .send(Method::submit_text( + paused_request_id.clone(), + "new request", + )) + .await + .unwrap(); + assert!( + drain_until(&mut rx, std::time::Duration::from_secs(2), |event| { + matches!( + event, + Event::SubmissionRejected { + submission_request_id, + message, + } if submission_request_id == &paused_request_id + && message.contains("requires an idle Worker") + ) + }) + .await, + "Paused Submit must be rejected" + ); + assert_eq!( + client_for_assert.captured_requests().len(), + 1, + "rejected Paused Submit must not start another LLM request" + ); + + handle + .send(Method::Cancel { + command: worker_command(&handle), + }) + .await + .unwrap(); + wait_for_status(&handle, WorkerStatus::Idle).await; + + // Once explicitly returned to Idle, fresh user input runs interrupt prep, + // which closes the orphan and inserts a system note before the user message. handle .send(Method::submit_text( protocol::new_submission_request_id(), @@ -2877,7 +2937,7 @@ async fn paused_then_run_closes_orphan_tool_use_for_next_request() { } )) .await, - "expected RunEnd::Finished after Paused→Run" + "expected RunEnd::Finished after Cancel→Submit" ); // The second LLM request carries the closure chain. Walk its items diff --git a/resources/prompts/common/worker-orchestration.md b/resources/prompts/common/worker-orchestration.md index c1f815d10..4505e85cb 100644 --- a/resources/prompts/common/worker-orchestration.md +++ b/resources/prompts/common/worker-orchestration.md @@ -4,6 +4,8 @@ When SubWorker-management tools are available, create direct children with `SubWorkerSpawn`, discover them with `WorkerList`, continue them with `WorkerSendInput`, and release their delegated authority with `WorkerStop`. Pass the exact `{ kind: "sub_worker", name }` subject returned by `WorkerList`; do not invent direct-only aliases. SubWorker notifications are background signals for the parent Worker to handle at a natural stopping point. Do not ignore routine follow-up, but do not interrupt the current user request unnecessarily. +`WorkerSendInput` is an Idle-only Submit that starts a fresh turn. Do not use it for a Worker that is Running or Paused. When `WorkerNotify` is available and information should be incorporated into work already in progress, use `WorkerNotify` instead; notifications are advisory and do not create a queued Submit. + The parent Worker does not need to keep a turn open or call tools solely to wait for a notification. Do not use `sleep` or polling loops just to wait for SubWorker output; if there is no useful immediate work, return control and handle the SubWorker when notified or when the user next asks. Before treating delegated SubWorker work as complete, inspect its committed session through worker-observation and verify concrete evidence such as worktree state, diff, and test results. Notifications are hints, not proof of completion. diff --git a/web/workspace/src/lib/workspace/console/composer-delivery.test.ts b/web/workspace/src/lib/workspace/console/composer-delivery.test.ts index 4d0522156..b2225e65e 100644 --- a/web/workspace/src/lib/workspace/console/composer-delivery.test.ts +++ b/web/workspace/src/lib/workspace/console/composer-delivery.test.ts @@ -20,15 +20,7 @@ const base = { hasAttachments: false, }; -Deno.test("running Composer enables Queue Submit and Notify but not immediate Submit", () => { - assertEquals( - canDeliverComposerDraft({ - ...base, - delivery: "queue", - workerState: "running", - }), - true, - ); +Deno.test("running Composer enables Notify but not Submit", () => { assertEquals( canDeliverComposerDraft({ ...base, @@ -47,16 +39,8 @@ Deno.test("running Composer enables Queue Submit and Notify but not immediate Su ); }); -Deno.test("running Queue Submit and Notify dispatch their protocol methods", () => { +Deno.test("running Notify dispatches its protocol method", () => { const sent: string[] = []; - assertEquals( - sendComposerDelivery( - { ...base, delivery: "queue", workerState: "running" }, - "submit", - (method) => sent.push(method), - ), - true, - ); assertEquals( sendComposerDelivery( { ...base, delivery: "notify", workerState: "running" }, @@ -65,10 +49,10 @@ Deno.test("running Queue Submit and Notify dispatch their protocol methods", () ), true, ); - assertEquals(sent.join(","), "submit,notify"); + assertEquals(sent.join(","), "notify"); }); -Deno.test("idle Composer enables only immediate Submit", () => { +Deno.test("idle Composer enables only Submit", () => { assertEquals( canDeliverComposerDraft({ ...base, @@ -77,14 +61,6 @@ Deno.test("idle Composer enables only immediate Submit", () => { }), true, ); - assertEquals( - canDeliverComposerDraft({ - ...base, - delivery: "queue", - workerState: "idle", - }), - false, - ); assertEquals( canDeliverComposerDraft({ ...base, @@ -95,11 +71,30 @@ Deno.test("idle Composer enables only immediate Submit", () => { ); }); -Deno.test("running delivery remains fenced by protocol, send state, and payload kind", () => { +Deno.test("paused Composer enables neither Submit nor Notify", () => { assertEquals( canDeliverComposerDraft({ ...base, - delivery: "queue", + delivery: "submit", + workerState: "paused", + }), + false, + ); + assertEquals( + canDeliverComposerDraft({ + ...base, + delivery: "notify", + workerState: "paused", + }), + false, + ); +}); + +Deno.test("running Notify remains fenced by protocol, send state, and payload kind", () => { + assertEquals( + canDeliverComposerDraft({ + ...base, + delivery: "notify", workerState: "running", protocolOpen: false, }), @@ -123,14 +118,4 @@ Deno.test("running delivery remains fenced by protocol, send state, and payload }), false, ); - assertEquals( - canDeliverComposerDraft({ - ...base, - delivery: "queue", - workerState: "running", - hasText: false, - hasAttachments: true, - }), - true, - ); }); diff --git a/web/workspace/src/lib/workspace/console/composer-delivery.ts b/web/workspace/src/lib/workspace/console/composer-delivery.ts index adf178839..7d908da02 100644 --- a/web/workspace/src/lib/workspace/console/composer-delivery.ts +++ b/web/workspace/src/lib/workspace/console/composer-delivery.ts @@ -1,4 +1,4 @@ -export type ComposerDelivery = "submit" | "queue" | "notify"; +export type ComposerDelivery = "submit" | "notify"; export type ComposerDeliveryState = { delivery: ComposerDelivery; @@ -11,7 +11,7 @@ export type ComposerDeliveryState = { /** * Resolve whether the current Composer draft can use one delivery action. - * Immediate Submit is idle-only; Queue and Notify are running-only. + * Immediate Submit is idle-only; Notify is running-only. */ export function canDeliverComposerDraft(state: ComposerDeliveryState): boolean { if (!state.protocolOpen || state.sending) return false; @@ -20,8 +20,6 @@ export function canDeliverComposerDraft(state: ComposerDeliveryState): boolean { switch (state.delivery) { case "submit": return state.workerState === "idle" && hasInput; - case "queue": - return state.workerState === "running" && hasInput; case "notify": return state.workerState === "running" && state.hasText && !state.hasAttachments; diff --git a/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts b/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts index 037adf4a2..427bb744f 100644 --- a/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts +++ b/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts @@ -1239,7 +1239,7 @@ Deno.test("Web Console switches main and direct SubWorker views from the Tasks r ); }); -Deno.test("Web Console uses Notify while running and exposes durable pending controls", async () => { +Deno.test("Web Console uses Notify while running and keeps recovery pending controls", async () => { const consolePage = await Deno.readTextFile( new URL( "./../../../routes/w/[workspaceId]/workers/[workerRef]/console/+page.svelte", @@ -1257,12 +1257,8 @@ Deno.test("Web Console uses Notify while running and exposes durable pending con 'method: "cancel_pending_submission"', 'method: "clear_pending_submissions"', 'method: "continue_pending"', - "handleQueueSubmit", "handleNotifySubmit", - 'submitDraft(composerInputElement?.snapshot() ?? draft, "queue")', - "disabled={!canQueueDraft}", "disabled={!canNotifyDraft}", - ">Queue Submit", ">Notify", ] ) { @@ -1272,6 +1268,13 @@ Deno.test("Web Console uses Notify while running and exposes durable pending con ); } + assert( + !consolePage.includes("handleQueueSubmit") && + !consolePage.includes(">Queue Submit") && + !consolePage.includes('delivery: "queue"'), + "running Console must not offer Submit queueing; callers should use Notify", + ); + const userCase = consolePage.slice( consolePage.indexOf('case "user":'), consolePage.indexOf('case "compact":'), diff --git a/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/+page.svelte b/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/+page.svelte index bf842bc3b..8f826cc0e 100644 --- a/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/+page.svelte +++ b/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/+page.svelte @@ -269,16 +269,6 @@ hasAttachments: draftHasAttachments, }), ); - const canQueueDraft = $derived( - canDeliverComposerDraft({ - delivery: "queue", - workerState, - protocolOpen: protocolState === "open", - sending, - hasText: draftHasText, - hasAttachments: draftHasAttachments, - }), - ); const canNotifyDraft = $derived( canDeliverComposerDraft({ delivery: "notify", @@ -729,10 +719,6 @@ void submitDraft(composerInputElement?.snapshot() ?? draft); } - function handleQueueSubmit() { - void submitDraft(composerInputElement?.snapshot() ?? draft, "queue"); - } - function handleNotifySubmit() { void submitDraft(composerInputElement?.snapshot() ?? draft, "notify"); } @@ -842,7 +828,7 @@ ) { if (delivery === "notify" && attachments.length > 0) { composerNotice = null; - sendError = "Notify accepts text only; remove attachments or queue a Submit."; + sendError = "Notify accepts text only; remove attachments or wait until the Worker is idle to Submit."; return; } const incompleteAttachment = attachments.find((attachment) => @@ -2151,11 +2137,6 @@
{#if workerRunning} -