diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index 485ff09d2..7bfe7213c 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -240,6 +240,16 @@ pub enum Method { #[serde(skip_deserializing, default)] source: AuthenticatedInputSource, }, + /// Agent-controlled input: accept only at an Idle controller boundary, never + /// enqueue behind a run. Unlike human Submit, non-Idle admission is rejected. + #[cfg_attr(feature = "typescript", ts(skip))] + #[cfg_attr(feature = "json-schema", schemars(skip))] + SubmitIfIdle { + submission_request_id: String, + input: Vec, + #[serde(skip_deserializing, default)] + source: AuthenticatedInputSource, + }, /// Human-readable text injected into the target Worker's LLM context as a /// non-blocking system message. Accepted notifications are delivered at the /// next between-turn boundary, or start a coalesced notification run while diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index 8b564cb7a..1935f023e 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -1556,6 +1556,19 @@ fn authorize_runtime_protocol_method( submission_request_id, input, }, + protocol::Method::SubmitIfIdle { + submission_request_id, + input, + .. + } => protocol::Method::SubmitIfIdle { + source: transport_source.cloned().unwrap_or_else(|| { + protocol::AuthenticatedInputSource::Backend { + operation_id: submission_request_id.clone(), + } + }), + submission_request_id, + input, + }, protocol::Method::NotifyTracked { notification_request_id, message, diff --git a/crates/worker-runtime/src/interaction.rs b/crates/worker-runtime/src/interaction.rs index e3324e998..5fd3371f1 100644 --- a/crates/worker-runtime/src/interaction.rs +++ b/crates/worker-runtime/src/interaction.rs @@ -8,6 +8,8 @@ use serde::{Deserialize, Serialize}; #[serde(rename_all = "snake_case")] pub enum WorkerInputKind { User, + /// Agent input that must never enter the human Submit queue. + UserIfIdle, Notify, Compact, ListRewindTargets, diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index d39c0cbc9..f16d5f603 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -2090,7 +2090,10 @@ impl Runtime { mut input: WorkerInput, ) -> Result { validate_worker_input(&input)?; - if matches!(input.kind, WorkerInputKind::User | WorkerInputKind::Notify) { + if matches!( + input.kind, + WorkerInputKind::User | WorkerInputKind::UserIfIdle | WorkerInputKind::Notify + ) { let request_id = input .submission_request_id .clone() @@ -2098,7 +2101,11 @@ impl Runtime { .unwrap_or_else(|| Uuid::now_v7().to_string()); input.submission_request_id = Some(request_id); } - let expected_submission_id = (input.kind == WorkerInputKind::User).then(|| { + let expected_submission_id = matches!( + input.kind, + WorkerInputKind::User | WorkerInputKind::UserIfIdle + ) + .then(|| { input .submission_request_id .clone() @@ -2110,6 +2117,17 @@ impl Runtime { .clone() .expect("Notify request id must be assigned before dispatch") }); + if input.kind == WorkerInputKind::UserIfIdle { + let state = self.lock()?; + let worker = state.worker(worker_ref)?; + // Stopped/restoring Workers have no controller to enforce admission. + // Live busy state is checked atomically by the controller, not here. + if worker.status == WorkerStatus::Stopped || worker.has_pending_lifecycle_operation() { + return Err(RuntimeError::InvalidRequest( + "WorkerSendInput requires an Idle Worker and never queues input. Use WorkerNotify for information about work already in progress.".to_string(), + )); + } + } self.ensure_worker_execution(worker_ref)?; let (backend, handle) = { let state = self.lock()?; @@ -5319,7 +5337,7 @@ fn input_protocol_event(input: &WorkerInput) -> Option { match input.kind { // Submit is projected only after the Worker commits UserInput. Queued // payloads must never become model- or client-visible history early. - WorkerInputKind::User | WorkerInputKind::Notify => None, + WorkerInputKind::User | WorkerInputKind::UserIfIdle | WorkerInputKind::Notify => None, WorkerInputKind::Compact | WorkerInputKind::ListRewindTargets | WorkerInputKind::RegisterPeer => Some(protocol::Event::SystemItem { diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index 381a972a2..8b98d0f92 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -2982,7 +2982,7 @@ where } let (method, submission_request_id) = match input.kind { - WorkerInputKind::User => { + WorkerInputKind::User | WorkerInputKind::UserIfIdle => { let Some(submission_id) = input .submission_request_id .filter(|submission_id| !submission_id.trim().is_empty()) @@ -2992,18 +2992,26 @@ where "Runtime user input is missing its internal submission id", ); }; - ( + let segments = input + .segments + .unwrap_or_else(|| vec![Segment::text(input.content.trim().to_string())]); + let source = protocol::AuthenticatedInputSource::Backend { + operation_id: submission_id.clone(), + }; + let method = if input.kind == WorkerInputKind::UserIfIdle { + Method::SubmitIfIdle { + submission_request_id: submission_id.clone(), + input: segments, + source, + } + } else { Method::SubmitTracked { submission_request_id: submission_id.clone(), - input: input.segments.unwrap_or_else(|| { - vec![Segment::text(input.content.trim().to_string())] - }), - source: protocol::AuthenticatedInputSource::Backend { - operation_id: submission_id.clone(), - }, - }, - Some(submission_id), - ) + input: segments, + source, + } + }; + (method, Some(submission_id)) } WorkerInputKind::Notify => { unreachable!("Notify input is dispatched before ordinary input mapping") @@ -5581,6 +5589,124 @@ mod tests { assert_eq!(call_count.load(Ordering::SeqCst), 3); } + #[test] + #[cfg(feature = "ws-server")] + #[serial_test::serial(worker_allocation)] + fn idle_only_input_acknowledges_idle_and_rejects_running_paused_and_stopped() { + let client = MockClient::sequential(vec![MockResponse::Hang( + simple_text_events().into_iter().take(2).collect(), + )]); + 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("idle-only")).unwrap(); + let input = |id: &str| { + let mut input = WorkerInput::user(id); + input.kind = WorkerInputKind::UserIfIdle; + input.submission_request_id = Some(id.to_string()); + input + }; + let ack = runtime + .send_input(&detail.worker_ref, input("idle-input")) + .unwrap(); + assert_eq!(ack.submission.unwrap().submission_request_id, "idle-input"); + wait_for_adapter_state(&backend, &detail.worker_ref, WorkerStatus::Running); + let error = runtime + .send_input(&detail.worker_ref, input("running-input")) + .unwrap_err(); + assert!(error.to_string().contains("WorkerNotify"), "{error}"); + runtime + .send_protocol_method( + &detail.worker_ref, + Method::Pause { + command: adapter_command(&backend, &detail.worker_ref), + }, + ) + .unwrap(); + wait_for_adapter_state(&backend, &detail.worker_ref, WorkerStatus::Paused); + let error = runtime + .send_input(&detail.worker_ref, input("paused-input")) + .unwrap_err(); + assert!(error.to_string().contains("WorkerNotify"), "{error}"); + let handle = backend + .workers + .lock() + .unwrap() + .get(&detail.worker_ref) + .unwrap() + .handle + .clone(); + let pending = backend + .run_on_adapter_runtime(async move { + let mut events = handle.subscribe(); + handle + .send(Method::ListPendingSubmissions) + .await + .map_err(|e| e.to_string())?; + tokio::time::timeout(Duration::from_secs(2), async { + loop { + if let Ok(Event::PendingSubmissionsChanged { pending }) = + events.recv().await + { + break pending; + } + } + }) + .await + .map_err(|e| e.to_string()) + }) + .unwrap(); + assert!(pending.submissions.is_empty()); + // Human Submit still queues while paused, and Notify remains advisory. + let ack = runtime + .send_input(&detail.worker_ref, WorkerInput::user("human-input")) + .unwrap(); + assert_eq!( + ack.submission.unwrap().disposition, + protocol::SubmissionDisposition::Queued + ); + let mut notify = WorkerInput::user("advisory"); + notify.kind = WorkerInputKind::Notify; + assert!( + runtime + .send_input(&detail.worker_ref, notify) + .unwrap() + .notification + .is_some() + ); + wait_for_adapter_state(&backend, &detail.worker_ref, WorkerStatus::Paused); + let handle = backend + .workers + .lock() + .unwrap() + .get(&detail.worker_ref) + .unwrap() + .handle + .clone(); + let entries = serde_json::to_string(&handle.committed_entries()).unwrap(); + assert!(!entries.contains("running-input")); + assert!(!entries.contains("paused-input")); + runtime.stop_worker(&detail.worker_ref, None).unwrap(); + let error = runtime + .send_input(&detail.worker_ref, input("stopped-input")) + .unwrap_err(); + assert!(error.to_string().contains("WorkerNotify"), "{error}"); + } + #[test] fn stopped_runtime_worker_can_restore_and_accept_input() { let client = MockClient::new(simple_text_events()); diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index f925b7e60..18d72a2d8 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -65,6 +65,14 @@ pub struct WorkerHandle { impl WorkerHandle { pub async fn send(&self, method: Method) -> Result<(), mpsc::error::SendError> { + // Reject known-busy sends before channel dispatch, including maintenance + // that cannot receive methods until it finishes. The controller checks + // again at admission to cover an Idle -> busy race after this snapshot. + if self.shared_state.snapshot().state != WorkerState::Idle + && reject_idle_only_submission(&method, &self.working_event_tx) + { + return Ok(()); + } self.method_tx.send(method).await } @@ -1523,6 +1531,21 @@ where } } +fn reject_idle_only_submission(method: &Method, events: &broadcast::Sender) -> bool { + let Method::SubmitIfIdle { + submission_request_id, + .. + } = method + else { + return false; + }; + let _ = events.send(Event::SubmissionRejected { + submission_request_id: submission_request_id.clone(), + message: "WorkerSendInput requires an Idle Worker and never queues input. Use WorkerNotify for information about work already in progress.".to_string(), + }); + true +} + fn durably_accept_method_while_busy( method: Method, pending_submissions: &PendingSubmissionHandle, @@ -1531,6 +1554,9 @@ fn durably_accept_method_while_busy( where St: session_store::Store + Clone, { + if reject_idle_only_submission(&method, working_event_tx) { + return None; + } match method { Method::Submit { submission_request_id, @@ -1640,6 +1666,10 @@ fn reject_method_while_attention_locked( | Method::SubmitTracked { submission_request_id, .. + } + | Method::SubmitIfIdle { + submission_request_id, + .. } => { let _ = working_event_tx.send(Event::SubmissionRejected { submission_request_id: submission_request_id.clone(), @@ -1946,6 +1976,12 @@ async fn controller_loop( } }; + if shared_state.snapshot().state != WorkerState::Idle + && reject_idle_only_submission(&method, &working_event_tx) + { + continue; + } + if worker.has_pending_compaction_cleanup() { match worker.retry_pending_compaction_cleanup().await { Ok(()) => { @@ -2032,6 +2068,11 @@ async fn controller_loop( submission_request_id, input, source, + } + | Method::SubmitIfIdle { + submission_request_id, + input, + source, } => { let request_id = submission_request_id.clone(); let (source_namespace, provenance) = @@ -3017,6 +3058,9 @@ where } } } + Some(method @ Method::SubmitIfIdle { .. }) => { + reject_idle_only_submission(&method, working_event_tx); + } Some(Method::SubmitTracked { submission_request_id, input, @@ -3558,6 +3602,36 @@ mod tests { tokio::time::timeout(timeout, accept).await.ok().flatten() } + #[tokio::test] + async fn idle_only_input_is_rejected_in_busy_maintenance_admission() { + let env = make_env().await; + let mut events = env.working_event_tx.subscribe(); + let method = Method::SubmitIfIdle { + submission_request_id: "maintenance-input".to_string(), + input: vec![protocol::Segment::text("must not queue")], + source: protocol::AuthenticatedInputSource::Backend { + operation_id: "maintenance-input".to_string(), + }, + }; + assert!( + durably_accept_method_while_busy( + method, + &env.pending_submissions, + &env.working_event_tx + ) + .is_none() + ); + let event = events.try_recv().unwrap(); + assert!( + matches!(event, Event::SubmissionRejected { submission_request_id, message } + if submission_request_id == "maintenance-input" && message.contains("WorkerNotify")) + ); + assert!( + events.try_recv().is_err(), + "must not emit durable acceptance or queue changes" + ); + } + #[tokio::test] async fn parent_originated_finished_fires_turn_ended() { let mut env = make_env().await; diff --git a/crates/worker/src/feature/builtin/manage_worker.rs b/crates/worker/src/feature/builtin/manage_worker.rs index 1e2b4d964..4155db86d 100644 --- a/crates/worker/src/feature/builtin/manage_worker.rs +++ b/crates/worker/src/feature/builtin/manage_worker.rs @@ -245,11 +245,11 @@ impl WorkerControlService for WorkspaceWorkerControlService { "unknown Worker or permission not granted".to_string(), ) })?; - record - .session - .send(content) - .await - .map_err(|error| WorkspaceClientError::Request(error.to_string()))?; + record.session.send(content).await.map_err(|error| { + WorkspaceClientError::Request(format!( + "WorkerSendInput requires an Idle SubWorker and never queues input: {error}" + )) + })?; Ok(WorkspaceResponse { status: 200, body: serde_json::json!({ @@ -855,7 +855,7 @@ impl WorkerOperation { "Spawn a Backend/Runtime Worker session in one existing Workspace Workdir. The model-facing input remains singular and the shared lifecycle wraps it as the canonical attachment collection under the stable `workdir` alias. 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. Idle Workers can start a fresh turn; Running or Paused Workers durably queue it for a later turn. Use WorkerNotify for advisory information during work already in progress. Direct SubWorkers require Idle." + "Start a fresh turn for an Idle Worker. Non-Idle Workers reject WorkerSendInput without queueing. Use WorkerNotify for advisory information during work already in progress. Direct SubWorkers also require Idle." } 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." @@ -1536,6 +1536,17 @@ mod tests { assert_eq!(client.requests.lock().unwrap().len(), 1); } + #[test] + fn send_input_rejection_is_a_tool_error_with_notify_guidance() { + let error = tool_output(WorkerOperation::SendInput, WorkspaceResponse { + status: 400, + body: "WorkerSendInput requires an Idle Worker. Use WorkerNotify for work already in progress.".to_string(), + }).unwrap_err(); + assert!(matches!(error, ToolError::ExecutionFailed(_))); + assert!(error.to_string().contains("WorkerNotify")); + assert!(!error.to_string().contains("completed")); + } + #[tokio::test] async fn control_service_routes_by_subject_without_cross_provider_fallback() { let client = Arc::new(RecordingWorkspaceClient::default()); diff --git a/crates/worker/src/prompt/catalog.rs b/crates/worker/src/prompt/catalog.rs index 296872011..bba265bdd 100644 --- a/crates/worker/src/prompt/catalog.rs +++ b/crates/worker/src/prompt/catalog.rs @@ -937,8 +937,8 @@ mod tests { "missing canonical tool {name}" ); } - assert!(orchestration.contains("durably queue it for a later turn")); - assert!(orchestration.contains("Direct SubWorkers still require Idle")); + assert!(orchestration.contains("requires an Idle Worker and never queues input")); + assert!(orchestration.contains("Direct SubWorkers also require Idle")); assert!(orchestration.contains("use `WorkerNotify` instead")); assert!(orchestration.contains("do not create a queued Submit")); for alias in ["SubWorkerList", "SubWorkerSend", "SubWorkerStop"] { diff --git a/crates/worker/tests/controller_test.rs b/crates/worker/tests/controller_test.rs index 7802cb4c5..75d6e8ee4 100644 --- a/crates/worker/tests/controller_test.rs +++ b/crates/worker/tests/controller_test.rs @@ -2895,6 +2895,97 @@ async fn drain_until bool>( } } +fn idle_only_input(request_id: &str) -> Method { + Method::SubmitIfIdle { + submission_request_id: request_id.to_string(), + input: vec![protocol::Segment::text(request_id)], + source: protocol::AuthenticatedInputSource::Backend { + operation_id: request_id.to_string(), + }, + } +} + +#[tokio::test] +async fn idle_only_submit_rejects_admission_race_and_paused_without_persisting() { + let client = MockClient::sequential(vec![MockResponse::Hang(vec![ + LlmEvent::text_block_start(0), + LlmEvent::text_delta(0, "partial"), + ])]); + let handle = spawn_controller(make_worker(client.clone()).await).await; + let mut rx = handle.subscribe(); + assert_eq!(handle.shared_state.catalog_status(), WorkerStatus::Idle); + // Both callers can observe Idle. Admission of the first request must make + // the second fail rather than silently scheduling a later turn. + handle.send(idle_only_input("first-idle")).await.unwrap(); + handle.send(idle_only_input("racing-input")).await.unwrap(); + let mut accepted_first = false; + assert!( + drain_until(&mut rx, std::time::Duration::from_secs(2), |event| { + match event { + Event::SubmissionAccepted { + submission_request_id, + .. + } => { + assert_eq!(submission_request_id, "first-idle"); + accepted_first = true; + false + } + Event::SubmissionRejected { + submission_request_id, + message, + } => { + assert_eq!(submission_request_id, "racing-input"); + assert!(message.contains("WorkerNotify")); + true + } + _ => false, + } + }) + .await + ); + assert!(accepted_first); + wait_for_status(&handle, WorkerStatus::Running).await; + handle + .send(Method::Pause { + command: worker_command(&handle), + }) + .await + .unwrap(); + wait_for_status(&handle, WorkerStatus::Paused).await; + handle.send(idle_only_input("paused-input")).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-input" && message.contains("WorkerNotify")) + }) + .await + ); + assert_eq!(handle.shared_state.catalog_status(), WorkerStatus::Paused); + handle.send(Method::ListPendingSubmissions).await.unwrap(); + assert!( + drain_until(&mut rx, std::time::Duration::from_secs(2), |event| { + if let Event::PendingSubmissionsChanged { pending } = event { + assert!(pending.submissions.is_empty()); + true + } else { + false + } + }) + .await + ); + assert_eq!(client.captured_requests().len(), 1); + let entries = serde_json::to_string(&handle.committed_entries()).unwrap(); + assert!(!entries.contains("racing-input")); + assert!(!entries.contains("paused-input")); + handle + .send(Method::Cancel { + command: worker_command(&handle), + }) + .await + .unwrap(); + wait_for_status(&handle, WorkerStatus::Idle).await; +} + /// Paused → Running → Idle. A notification accepted while Paused must not /// auto-resume the Worker, and explicit Resume must inject it while preserving /// the interrupted turn's history consistency. diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 839f945d7..40ab2b09b 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -696,6 +696,8 @@ pub struct WorkerLifecycleResult { #[serde(rename_all = "snake_case")] pub enum WorkerInputKind { User, + /// Agent input that must never enter the human Submit queue. + UserIfIdle, Notify, Compact, ListRewindTargets, @@ -3037,6 +3039,7 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { let input = EmbeddedWorkerInput { kind: match request.kind { WorkerInputKind::User => EmbeddedWorkerInputKind::User, + WorkerInputKind::UserIfIdle => EmbeddedWorkerInputKind::UserIfIdle, WorkerInputKind::Notify => EmbeddedWorkerInputKind::Notify, WorkerInputKind::Compact => EmbeddedWorkerInputKind::Compact, WorkerInputKind::ListRewindTargets => EmbeddedWorkerInputKind::ListRewindTargets, @@ -5028,6 +5031,7 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { let input = EmbeddedWorkerInput { kind: match request.kind { WorkerInputKind::User => EmbeddedWorkerInputKind::User, + WorkerInputKind::UserIfIdle => EmbeddedWorkerInputKind::UserIfIdle, WorkerInputKind::Notify => EmbeddedWorkerInputKind::Notify, WorkerInputKind::Compact => EmbeddedWorkerInputKind::Compact, WorkerInputKind::ListRewindTargets => EmbeddedWorkerInputKind::ListRewindTargets, diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 08e09b32d..6a715f6cc 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -17659,7 +17659,30 @@ async fn send_known_worker_input( &path.worker, permission, )?; - scoped_send_runtime_worker_input(State(api), AxumPath(path), Json(request)).await + let mut request = worker_input_request_from_api(request)?; + let idle_only = request.kind == WorkerInputKind::User; + if idle_only { + request.kind = WorkerInputKind::UserIfIdle; + } + let Json(result) = + dispatch_runtime_worker_input(api, path.worker.runtime_id, path.worker.worker_id, request)?; + if result.state != server_api::WorkerOperationState::Accepted { + let detail = result + .diagnostics + .iter() + .map(|diagnostic| diagnostic.message.as_str()) + .collect::>() + .join("; "); + let message = if idle_only { + format!( + "WorkerSendInput requires an Idle Worker and never queues input. Use WorkerNotify for information about work already in progress. {detail}" + ) + } else { + detail + }; + return Err(Error::InvalidInput(message).into()); + } + Ok(Json(result)) } async fn cancel_known_worker( @@ -24426,6 +24449,15 @@ async fn send_runtime_worker_input( Json(request): Json, ) -> ApiResult> { let request = worker_input_request_from_api(request)?; + dispatch_runtime_worker_input(api, runtime_id, worker_id, request) +} + +fn dispatch_runtime_worker_input( + api: WorkspaceApi, + runtime_id: String, + worker_id: String, + request: WorkerInputRequest, +) -> ApiResult> { let worker = resolve_workspace_worker_reference(&api, &runtime_id, &worker_id)?; if api .store @@ -24535,7 +24567,9 @@ pub(crate) fn authorize_browser_worker_method( message, source: source.clone(), }), - protocol::Method::SubmitTracked { .. } | protocol::Method::NotifyTracked { .. } => { + protocol::Method::SubmitTracked { .. } + | protocol::Method::SubmitIfIdle { .. } + | protocol::Method::NotifyTracked { .. } => { Err("authenticated Worker input source is server-owned") } other => Ok(other), @@ -29862,6 +29896,7 @@ mod tests { #[derive(Clone, Default)] struct WorkdirlessFixtureRuntime { + input_requests: Arc>>, workers: Arc>>, spawn_requests: Arc>>, reject_next_spawn: Arc>, @@ -29988,6 +30023,30 @@ mod tests { } impl crate::hosts::WorkspaceWorkerRuntime for WorkdirlessFixtureRuntime { + fn send_input( + &self, + worker_id: &str, + request: WorkerInputRequest, + ) -> crate::hosts::WorkerInputResult { + let idle_only = request.kind == WorkerInputKind::UserIfIdle; + self.input_requests.lock().unwrap().push(request); + crate::hosts::WorkerInputResult { + state: if idle_only { + InternalWorkerOperationState::Rejected + } else { + InternalWorkerOperationState::Accepted + }, + disposition: if idle_only { + crate::hosts::WorkerInputDisposition::Rejected + } else { + crate::hosts::WorkerInputDisposition::Accepted + }, + worker: RuntimeWorkerRef::new(Self::RUNTIME_ID, worker_id), + runtime_run_id: None, + diagnostics: Vec::new(), + } + } + fn runtime_id(&self) -> &str { Self::RUNTIME_ID } @@ -33327,6 +33386,76 @@ mod tests { assert_eq!(current.assignment_id, "existing-coder-assignment"); } + #[tokio::test] + async fn worker_control_input_is_idle_only_and_returns_rejection_as_http_error() { + let fixture = guarded_spawn_fixture().await; + let response = post_guarded_spawn( + &fixture, + guarded_spawn_payload(&fixture, "idle-only-control"), + ) + .await; + assert_eq!(response.status(), StatusCode::OK); + let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let created: BrowserCreateWorkerResponse = serde_json::from_slice(&bytes).unwrap(); + let mut headers = HeaderMap::new(); + headers.insert( + "x-yoi-runtime-id", + fixture.controller.runtime_id.parse().unwrap(), + ); + headers.insert( + "x-yoi-worker-id", + fixture.controller.worker_id.parse().unwrap(), + ); + let path = || ScopedRuntimeWorkerPath { + workspace_id: TEST_WORKSPACE_ID.to_string(), + worker: RuntimeWorkerRef::new(&created.runtime_id, &created.worker_id), + }; + let request = |kind: &str| server_api::RuntimeWorkerInputRequest { + kind: Some(kind.to_string()), + content: "follow-up".to_string(), + segments: None, + }; + let error = send_known_worker_input( + State(fixture.api.clone()), + AxumPath(path()), + headers.clone(), + Json(request("user")), + ) + .await + .unwrap_err(); + let response = error.into_response(); + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + assert!(String::from_utf8_lossy(&bytes).contains("WorkerNotify")); + // Machine Notify is unchanged; the human endpoint must still use User. + let Json(notify) = send_known_worker_input( + State(fixture.api.clone()), + AxumPath(path()), + headers, + Json(request("notify")), + ) + .await + .unwrap(); + assert_eq!(notify.state, server_api::WorkerOperationState::Accepted); + let Json(human) = scoped_send_runtime_worker_input( + State(fixture.api.clone()), + AxumPath(path()), + Json(request("user")), + ) + .await + .unwrap(); + assert_eq!(human.state, server_api::WorkerOperationState::Accepted); + let requests = fixture.runtime.input_requests.lock().unwrap(); + assert_eq!( + requests.iter().map(|r| r.kind.clone()).collect::>(), + vec![ + WorkerInputKind::UserIfIdle, + WorkerInputKind::Notify, + WorkerInputKind::User + ] + ); + } + #[tokio::test] async fn worker_control_spawn_retry_converges_on_one_worker_and_one_grant() { let workspace = tempfile::tempdir().unwrap(); diff --git a/resources/prompts/common/worker-orchestration.md b/resources/prompts/common/worker-orchestration.md index 50a47793f..82258d87a 100644 --- a/resources/prompts/common/worker-orchestration.md +++ b/resources/prompts/common/worker-orchestration.md @@ -4,7 +4,7 @@ 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. -For Runtime Workers, `WorkerSendInput` submits user input: Idle Workers can start a fresh turn, while Running or Paused Workers durably queue it for a later turn. Queueing does not implicitly resume a paused run; Cancel/error retains pending input until explicitly continued. Direct SubWorkers still require Idle for `WorkerSendInput`. 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. +`WorkerSendInput` requires an Idle Worker and never queues input. Non-Idle Workers reject it; when information should be incorporated into work already in progress, use `WorkerNotify` instead. Notifications are advisory and do not create a queued Submit. Direct SubWorkers also require Idle for `WorkerSendInput`; do not send follow-up input until they are Idle. The Submit queue is for human-controlled Web/TUI interaction, not Orchestrator scheduling. Do not retry SendInput in a polling loop or cancel/pause work to bypass this restriction. 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.