fix: restrict worker send input to idle admission

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<Method>> {
+        // 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<Event>) -> 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<St>(
     method: Method,
     pending_submissions: &PendingSubmissionHandle<St>,
@@ -1531,6 +1554,9 @@ fn durably_accept_method_while_busy<St>(
 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<C, St>(
             }
         };
 
+        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<C, St>(
                 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;

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

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"] {

crates/worker/tests/controller_test.rs

@@ -2895,6 +2895,97 @@ async fn drain_until<F: FnMut(&Event) -> 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.