Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

fix: require idle workers for submit

Browse Source
This commit is contained in:
Hare
2026-09-23 00:36:54 +09:00
parent d8444e49bd
commit 9857bf04f7
16 changed files with 451 additions and 264 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker/src/controller.rs
+118 -90
View File
@@ -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<Event>,
) {
let _ = working_event_tx.send(Event::SubmissionRejected {
submission_request_id,
message: SUBMIT_REQUIRES_IDLE_MESSAGE.into(),
});
}
fn replay_or_reject_submit_while_not_idle<St>(
pending_submissions: &PendingSubmissionHandle<St>,
submission_request_id: String,
input: Vec<protocol::Segment>,
source_namespace: String,
working_event_tx: &broadcast::Sender<Event>,
) 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<St>(
method: Method,
pending_submissions: &PendingSubmissionHandle<St>,
@@ -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<C, St>(
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<C, St>(
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();
crates/worker/src/feature/builtin/manage_worker.rs
+9 -3
View File
@@ -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."
}
crates/worker/src/prompt/catalog.rs
+3
View File
@@ -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),
crates/worker/src/worker.rs
+54 -1
View File
@@ -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<Segment>,
source_namespace: String,
provenance: WorkerHistoryProvenance,
) -> Result<SubmissionAcceptance, PendingSubmissionError> {
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<Segment>,
source_namespace: String,
) -> Result<Option<SubmissionAcceptance>, 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<Segment>,
crates/worker/tests/controller_test.rs
+115 -55
View File
@@ -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<Event>,
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<Event>,
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