From 64ced7dbadff0bcebae2cf7b99ffd3e52e61e269 Mon Sep 17 00:00:00 2001 From: Hare Date: Mon, 10 Aug 2026 02:01:08 +0900 Subject: [PATCH] worker: make coder review routing durable --- Cargo.lock | 1 + crates/protocol/src/lib.rs | 24 ++ crates/worker-runtime/Cargo.toml | 1 + crates/worker-runtime/src/execution.rs | 54 ++++ crates/worker-runtime/src/http_server.rs | 36 ++- crates/worker-runtime/src/interaction.rs | 6 + crates/worker-runtime/src/runtime.rs | 170 ++++++++++- crates/worker-runtime/src/worker_backend.rs | 279 ++++++++++++++++-- crates/worker/src/controller.rs | 53 +++- .../src/feature/builtin/manage_worker.rs | 39 ++- crates/worker/src/lib.rs | 8 +- crates/worker/src/spawn/tool.rs | 99 +++++-- crates/worker/src/worker.rs | 31 +- crates/workspace-server/src/hosts.rs | 20 +- .../src/runtime_subscription_tests.rs | 18 +- crates/workspace-server/src/server.rs | 186 ++++++++++-- docs/design/flow-state-graph.md | 4 +- ...8-06-subworker-completion-idle-auto-run.md | 32 ++ 18 files changed, 925 insertions(+), 136 deletions(-) create mode 100644 docs/report/2026-08-06-subworker-completion-idle-auto-run.md diff --git a/Cargo.lock b/Cargo.lock index e2a882bc..50900db7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6053,6 +6053,7 @@ dependencies = [ "tokio-tungstenite 0.29.0", "toml", "tower", + "uuid", "workdir", "worker", ] diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index 33e7f216..d33bd9d0 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -31,6 +31,15 @@ pub enum Method { Run { input: Vec, }, + /// Runtime-internal Run carrying an opaque correlation id that is committed + /// with the resulting UserInput entry. This variant is not serializable on + /// the public Client → Worker protocol. + #[serde(skip)] + #[cfg_attr(feature = "typescript", ts(skip))] + RunTracked { + input: Vec, + submission_id: String, + }, /// Human-readable text injected into the target Worker's LLM context /// as a non-blocking system message. `auto_run` controls whether an /// idle target is kicked into `RunForNotification`; weak notifications @@ -938,6 +947,21 @@ mod tests { )); } + #[test] + fn runtime_tracked_run_is_not_public_protocol_json() { + let method = Method::RunTracked { + input: vec![Segment::text("private")], + submission_id: "submission-1".to_string(), + }; + assert!(serde_json::to_string(&method).is_err()); + assert!( + serde_json::from_str::( + r#"{"method":"run_tracked","input":[],"submission_id":"forged"}"#, + ) + .is_err() + ); + } + #[test] fn segment_unknown_variant_decodes_as_unknown() { // A future client sends a segment kind this Worker has never heard of. diff --git a/crates/worker-runtime/Cargo.toml b/crates/worker-runtime/Cargo.toml index 0af34f4a..d4ca9d64 100644 --- a/crates/worker-runtime/Cargo.toml +++ b/crates/worker-runtime/Cargo.toml @@ -41,6 +41,7 @@ tar.workspace = true thiserror = { workspace = true } tokio = { workspace = true, features = ["net", "rt", "sync", "time"] } toml.workspace = true +uuid = { workspace = true, features = ["v7"] } tower = { workspace = true, features = ["util"], optional = true } worker.workspace = true workdir.workspace = true diff --git a/crates/worker-runtime/src/execution.rs b/crates/worker-runtime/src/execution.rs index b75315da..c4a618cf 100644 --- a/crates/worker-runtime/src/execution.rs +++ b/crates/worker-runtime/src/execution.rs @@ -36,6 +36,16 @@ pub enum WorkerExecutionOperation { Cancel, } +/// Evidence that a user input reached the durable Worker session boundary. +/// +/// This is intentionally distinct from accepting a method on the Worker's +/// in-memory channel. For Flow submissions, the committed UserInput entry also +/// carries the initial Flow runtime-state extension. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct WorkerInputCommitAck { + pub submission_id: String, +} + /// Typed execution result class. Results are transient operation outcomes and /// are not persisted as Worker lifecycle authority. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] @@ -45,6 +55,8 @@ pub struct WorkerExecutionResult { pub run_state: WorkerExecutionRunState, #[serde(default, skip_serializing_if = "Option::is_none")] pub message: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub input_commit: Option, } /// Backend result class for a Worker execution operation. @@ -68,6 +80,23 @@ impl WorkerExecutionResult { outcome: WorkerExecutionOutcome::Accepted, run_state, message: None, + input_commit: None, + } + } + + pub fn accepted_input_committed( + operation: WorkerExecutionOperation, + run_state: WorkerExecutionRunState, + submission_id: impl Into, + ) -> Self { + Self { + operation, + outcome: WorkerExecutionOutcome::Accepted, + run_state, + message: None, + input_commit: Some(WorkerInputCommitAck { + submission_id: submission_id.into(), + }), } } @@ -77,6 +106,7 @@ impl WorkerExecutionResult { outcome: WorkerExecutionOutcome::Busy, run_state: WorkerExecutionRunState::Busy, message: Some(message.into()), + input_commit: None, } } @@ -86,6 +116,7 @@ impl WorkerExecutionResult { outcome: WorkerExecutionOutcome::Rejected, run_state: WorkerExecutionRunState::Stopped, message: Some(message.into()), + input_commit: None, } } @@ -95,6 +126,7 @@ impl WorkerExecutionResult { outcome: WorkerExecutionOutcome::Errored, run_state: WorkerExecutionRunState::Errored, message: Some(message.into()), + input_commit: None, } } @@ -104,6 +136,7 @@ impl WorkerExecutionResult { outcome: WorkerExecutionOutcome::Unsupported, run_state: WorkerExecutionRunState::Stopped, message: Some(message.into()), + input_commit: None, } } @@ -472,6 +505,27 @@ impl WorkerExecutionBackendRef { } } +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn input_commit_ack_survives_json_round_trip() { + let result = WorkerExecutionResult::accepted_input_committed( + WorkerExecutionOperation::Input, + WorkerExecutionRunState::Busy, + "submission-1", + ); + + let json = serde_json::to_string(&result).unwrap(); + assert!(json.contains("\"submission_id\":\"submission-1\"")); + assert_eq!( + serde_json::from_str::(&json).unwrap(), + result + ); + } +} + impl fmt::Debug for WorkerExecutionBackendRef { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("WorkerExecutionBackendRef") diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index ad586010..f08f245b 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -2208,12 +2208,20 @@ mod tests { fn dispatch_input( &self, _handle: &WorkerExecutionHandle, - _input: WorkerInput, + input: WorkerInput, ) -> WorkerExecutionResult { - WorkerExecutionResult::accepted( - WorkerExecutionOperation::Input, - WorkerExecutionRunState::Idle, - ) + if let Some(submission_id) = input.submission_id { + WorkerExecutionResult::accepted_input_committed( + WorkerExecutionOperation::Input, + WorkerExecutionRunState::Idle, + submission_id, + ) + } else { + WorkerExecutionResult::accepted( + WorkerExecutionOperation::Input, + WorkerExecutionRunState::Idle, + ) + } } fn stop_worker(&self, _handle: &WorkerExecutionHandle) -> WorkerExecutionResult { @@ -2515,12 +2523,20 @@ mod ws_tests { fn dispatch_input( &self, _handle: &WorkerExecutionHandle, - _input: WorkerInput, + input: WorkerInput, ) -> WorkerExecutionResult { - WorkerExecutionResult::accepted( - WorkerExecutionOperation::Input, - WorkerExecutionRunState::Idle, - ) + if let Some(submission_id) = input.submission_id { + WorkerExecutionResult::accepted_input_committed( + WorkerExecutionOperation::Input, + WorkerExecutionRunState::Idle, + submission_id, + ) + } else { + WorkerExecutionResult::accepted( + WorkerExecutionOperation::Input, + WorkerExecutionRunState::Idle, + ) + } } fn dispatch_method( diff --git a/crates/worker-runtime/src/interaction.rs b/crates/worker-runtime/src/interaction.rs index 225f051e..d76c675d 100644 --- a/crates/worker-runtime/src/interaction.rs +++ b/crates/worker-runtime/src/interaction.rs @@ -25,6 +25,10 @@ impl WorkerInputKind { pub struct WorkerInput { pub kind: WorkerInputKind, pub content: String, + /// Runtime-generated correlation id. This is never accepted from public + /// JSON input and is consumed only by the execution backend. + #[serde(skip)] + pub submission_id: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub segments: Option>, } @@ -34,6 +38,7 @@ impl WorkerInput { Self { kind: WorkerInputKind::User, content: content.into(), + submission_id: None, segments: None, } } @@ -42,6 +47,7 @@ impl WorkerInput { Self { kind: WorkerInputKind::Notify, content: content.into(), + submission_id: None, segments: None, } } diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 48475fbe..e4c8b3fa 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -40,6 +40,7 @@ use std::sync::{Arc, Mutex, MutexGuard, Weak}; #[cfg(feature = "ws-server")] use tokio::sync::broadcast; use tokio::sync::mpsc; +use uuid::Uuid; /// Workspace-scoped Runtime authorization context supplied by a trusted backend. #[derive(Clone, Debug, Eq, PartialEq)] @@ -565,10 +566,12 @@ impl Runtime { } }; - if let Some(initial_input) = { + if let Some(mut initial_input) = { let state = self.lock()?; state.worker(&worker_ref)?.request.initial_input.clone() } { + let expected_submission_id = Uuid::now_v7().to_string(); + initial_input.submission_id = Some(expected_submission_id.clone()); let dispatch_result = backend.dispatch_input(&handle, initial_input.clone()); if !dispatch_result.is_accepted() { let _ = backend.stop_worker(&handle); @@ -581,15 +584,32 @@ impl Runtime { result: dispatch_result, }); } + let has_commit_ack = dispatch_result + .input_commit + .as_ref() + .is_some_and(|ack| ack.submission_id == expected_submission_id); + if !has_commit_ack { + let _ = backend.stop_worker(&handle); + self.rollback_failed_create(&worker_ref)?; + let result = WorkerExecutionResult::rejected( + WorkerExecutionOperation::Input, + "execution backend accepted initial input without a durable session commit acknowledgement", + ); + return Err(RuntimeError::WorkerExecutionRejected { + worker_id: worker_ref.worker_id.clone(), + operation: result.operation, + outcome: result.outcome, + message: result.message_or_default(), + result, + }); + } + let initial_run_state = dispatch_result.run_state; let detail = self.commit_created_worker( &worker_ref, handle, - WorkerExecutionRunState::Busy, + initial_run_state, working_directory, - WorkerExecutionResult::accepted( - WorkerExecutionOperation::Input, - WorkerExecutionRunState::Busy, - ), + dispatch_result, )?; self.record_input_observation(&worker_ref, initial_input)?; Ok(detail) @@ -936,9 +956,16 @@ impl Runtime { pub fn send_input( &self, worker_ref: &WorkerRef, - input: WorkerInput, + mut input: WorkerInput, ) -> Result { validate_worker_input(&input)?; + let expected_submission_id = if input.kind == WorkerInputKind::User { + let submission_id = Uuid::now_v7().to_string(); + input.submission_id = Some(submission_id.clone()); + Some(submission_id) + } else { + None + }; self.ensure_worker_execution(worker_ref)?; let (backend, handle) = { let state = self.lock()?; @@ -975,6 +1002,25 @@ impl Runtime { result: dispatch_result, }); } + if let Some(expected_submission_id) = expected_submission_id + && dispatch_result + .input_commit + .as_ref() + .is_none_or(|ack| ack.submission_id != expected_submission_id) + { + let result = WorkerExecutionResult::rejected( + WorkerExecutionOperation::Input, + "execution backend did not acknowledge the committed Runtime submission id", + ); + self.record_execution_result(worker_ref, result.clone())?; + return Err(RuntimeError::WorkerExecutionRejected { + worker_id: worker_ref.worker_id.clone(), + operation: result.operation, + outcome: result.outcome, + message: result.message_or_default(), + result, + }); + } let mut state = self.lock()?; state.ensure_running()?; @@ -2460,6 +2506,7 @@ mod tests { let input = WorkerInput { kind: WorkerInputKind::User, content: String::new(), + submission_id: None, segments: Some(vec![protocol::Segment::Flow { selector: "builtin:coder-review".to_string(), }]), @@ -2472,6 +2519,7 @@ mod tests { let input = WorkerInput { kind: WorkerInputKind::User, content: String::new(), + submission_id: None, segments: Some(Vec::new()), }; assert!(matches!( @@ -2486,6 +2534,7 @@ mod tests { request.initial_input = Some(WorkerInput { kind: WorkerInputKind::User, content: String::new(), + submission_id: None, segments: Some(vec![protocol::Segment::Flow { selector: "builtin:coder-review".to_string(), }]), @@ -2580,6 +2629,7 @@ mod tests { restore_count: Mutex, contexts: Mutex>, dispatched_inputs: Mutex>, + preserve_commit_ack_submission_id: AtomicBool, #[cfg(feature = "ws-server")] snapshots: Mutex>, } @@ -2589,6 +2639,11 @@ mod tests { *self.dispatch_result.lock().unwrap() = Some(result); } + fn preserve_commit_ack_submission_id(&self) { + self.preserve_commit_ack_submission_id + .store(true, Ordering::SeqCst); + } + #[cfg(feature = "ws-server")] fn set_worker_snapshot(&self, worker_ref: &WorkerRef, snapshot: protocol::Event) { self.snapshots @@ -2656,17 +2711,29 @@ mod tests { _handle: &WorkerExecutionHandle, input: WorkerInput, ) -> WorkerExecutionResult { + let submission_id = input.submission_id.clone(); self.dispatched_inputs.lock().unwrap().push(input); - self.dispatch_result + let mut result = self + .dispatch_result .lock() .unwrap() .clone() .unwrap_or_else(|| { - WorkerExecutionResult::accepted( + WorkerExecutionResult::accepted_input_committed( WorkerExecutionOperation::Input, WorkerExecutionRunState::Idle, + "test-submission", ) - }) + }); + if !self + .preserve_commit_ack_submission_id + .load(Ordering::SeqCst) + && let (Some(ack), Some(submission_id)) = + (result.input_commit.as_mut(), submission_id) + { + ack.submission_id = submission_id; + } + result } fn stop_worker(&self, _handle: &WorkerExecutionHandle) -> WorkerExecutionResult { @@ -3212,6 +3279,68 @@ mod tests { assert!(runtime.list_workers().unwrap().is_empty()); } + #[test] + fn create_worker_uses_committed_input_ack_run_state() { + let (runtime, backend) = runtime_and_backend(); + backend.set_dispatch_result(WorkerExecutionResult::accepted_input_committed( + WorkerExecutionOperation::Input, + WorkerExecutionRunState::Idle, + "test-submission", + )); + let mut request = task_request("committed initial input is already idle"); + request.initial_input = Some(WorkerInput::user("start the ticket")); + + let detail = runtime.create_worker(request).unwrap(); + + assert_eq!(detail.status, WorkerStatus::Idle); + } + + #[test] + fn create_worker_rejects_mismatched_input_commit_acknowledgement() { + let (runtime, backend) = runtime_and_backend(); + backend.preserve_commit_ack_submission_id(); + backend.set_dispatch_result(WorkerExecutionResult::accepted_input_committed( + WorkerExecutionOperation::Input, + WorkerExecutionRunState::Busy, + "forged-submission", + )); + let mut request = task_request("mismatched initial input commit ack"); + request.initial_input = Some(WorkerInput::user("start the ticket")); + + let error = runtime.create_worker(request).unwrap_err(); + + assert!(matches!( + error, + RuntimeError::WorkerExecutionRejected { + outcome: crate::execution::WorkerExecutionOutcome::Rejected, + .. + } + )); + assert!(runtime.list_workers().unwrap().is_empty()); + } + + #[test] + fn create_worker_rejects_initial_input_without_commit_acknowledgement() { + let (runtime, backend) = runtime_and_backend(); + backend.set_dispatch_result(WorkerExecutionResult::accepted( + WorkerExecutionOperation::Input, + WorkerExecutionRunState::Busy, + )); + let mut request = task_request("missing initial input commit ack"); + request.initial_input = Some(WorkerInput::user("start the ticket")); + + let error = runtime.create_worker(request).unwrap_err(); + + assert!(matches!( + error, + RuntimeError::WorkerExecutionRejected { + outcome: crate::execution::WorkerExecutionOutcome::Rejected, + .. + } + )); + assert!(runtime.list_workers().unwrap().is_empty()); + } + #[test] fn create_worker_without_execution_backend_is_rejected_and_not_persisted() { let runtime = Runtime::new_memory(); @@ -3356,11 +3485,12 @@ mod tests { fn dispatch_input( &self, _handle: &WorkerExecutionHandle, - _input: WorkerInput, + input: WorkerInput, ) -> WorkerExecutionResult { - WorkerExecutionResult::accepted( + WorkerExecutionResult::accepted_input_committed( WorkerExecutionOperation::Input, WorkerExecutionRunState::Idle, + input.submission_id.expect("Runtime submission id"), ) } } @@ -3442,6 +3572,7 @@ mod tests { request.initial_input = Some(WorkerInput { kind: WorkerInputKind::User, content: String::new(), + submission_id: None, segments: Some(vec![ protocol::Segment::Flow { selector: "builtin:coder-review".to_string(), @@ -3479,6 +3610,7 @@ mod tests { let input = WorkerInput { kind: WorkerInputKind::User, content: String::new(), + submission_id: None, segments: Some(vec![protocol::Segment::Flow { selector: "builtin:coder-review".to_string(), }]), @@ -3488,10 +3620,16 @@ mod tests { .send_input(&detail.worker_ref, input.clone()) .unwrap(); - assert_eq!( - backend.dispatched_inputs.lock().unwrap().as_slice(), - &[input] - ); + let dispatched = backend.dispatched_inputs.lock().unwrap(); + assert_eq!(dispatched.len(), 1); + assert_eq!(dispatched[0].kind, input.kind); + assert_eq!(dispatched[0].content, input.content); + assert_eq!(dispatched[0].segments, input.segments); + let submission_id = dispatched[0] + .submission_id + .as_deref() + .expect("Runtime submission id"); + Uuid::parse_str(submission_id).expect("submission id UUID"); } #[cfg(feature = "ws-server")] diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index 7cf021d0..0e58922c 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -31,8 +31,8 @@ use crate::working_directory::{ }; use async_trait::async_trait; use manifest::paths; -use protocol::{Method, Segment, WorkerStatus}; -use session_store::{CombinedStore, FsStore, FsWorkerStore, collect_state}; +use protocol::{Event, Method, Segment, WorkerStatus}; +use session_store::{CombinedStore, FsStore, FsWorkerStore, LogEntry, collect_state}; use tokio::runtime::Runtime; #[cfg(feature = "ws-server")] use tokio::sync::broadcast; @@ -46,15 +46,29 @@ use worker::feature::builtin::{ #[cfg(feature = "ws-server")] use worker::ipc::protocol_session::{live_log_entry_event, subscribe_worker_protocol_session}; use worker::{ - PromptLoader, RuntimeWorkspaceHttpClient, SegmentLogSink, Worker, WorkerController, - WorkerError, WorkerFilesystemAuthority, WorkerHandle, WorkerSharedState, - WorkerWorkspaceContext, WorkspaceClient, WorkspaceId, + PromptLoader, RuntimeWorkspaceHttpClient, SegmentLogSink, + WORKER_INPUT_SUBMISSION_EXTENSION_DOMAIN, Worker, WorkerController, WorkerError, + WorkerFilesystemAuthority, WorkerHandle, WorkerSharedState, WorkerWorkspaceContext, + WorkspaceClient, WorkspaceId, }; const DEFAULT_BACKEND_ID: &str = "worker-crate"; const RUNTIME_TASK_TIMEOUT: Duration = Duration::from_secs(10); +// Keep this below the adapter task timeout so a failed acknowledgement task +// returns a typed execution error instead of leaving the outer waiter to time out. +const USER_INPUT_COMMIT_TIMEOUT: Duration = Duration::from_secs(9); static NEXT_RUNTIME_ARTIFACT_ROOT: AtomicU64 = AtomicU64::new(1); +fn user_input_has_submission(entry: &LogEntry, submission_id: &str) -> bool { + let LogEntry::UserInput { extensions, .. } = entry else { + return false; + }; + extensions.iter().any(|extension| { + extension.domain == WORKER_INPUT_SUBMISSION_EXTENSION_DOMAIN + && extension.payload["submission_id"].as_str() == Some(submission_id) + }) +} + #[derive(Clone)] enum RuntimeArtifactRoot { Owned(Arc), @@ -949,6 +963,131 @@ where .unwrap_or_else(|message| WorkerExecutionResult::errored(operation, message)) } + fn send_user_input_and_wait_for_commit( + &self, + operation: WorkerExecutionOperation, + worker: WorkerHandle, + method: Method, + submission_id: String, + accepted_run_state: WorkerExecutionRunState, + ) -> WorkerExecutionResult { + let acknowledged_submission_id = submission_id.clone(); + self.run_on_adapter_runtime(async move { + // Subscribe before enqueueing the input so the acknowledgement cannot + // race with a fast Worker commit. The opaque submission id is stored in + // the same UserInput entry as the transformed Flow input and its state. + let (_, mut committed_entries) = worker.sink.subscribe_with_snapshot(); + let committed_probe = worker.clone(); + let mut events = worker.subscribe(); + worker + .send(method) + .await + .map_err(|err| format!("failed to send Worker method: {err}"))?; + + let timeout_probe = committed_probe.clone(); + let timeout_submission_id = submission_id.clone(); + let acknowledgement = tokio::time::timeout(USER_INPUT_COMMIT_TIMEOUT, async move { + let input_was_committed = || { + committed_probe + .committed_entries() + .iter() + .any(|entry| user_input_has_submission(entry, &submission_id)) + }; + loop { + tokio::select! { + entry = committed_entries.recv() => { + match entry { + Ok(entry) if user_input_has_submission(&entry, &submission_id) => { + return Ok(()); + } + Ok(_) => {} + Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => { + if input_was_committed() { + return Ok(()); + } + return Err(format!( + "worker input commit acknowledgement lagged by {skipped} entry event(s)" + )); + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => { + if input_was_committed() { + return Ok(()); + } + return Err( + "worker entry stream closed before user input was committed" + .to_string(), + ); + } + } + } + event = events.recv() => { + match event { + Ok(Event::Error { message, .. }) => { + if input_was_committed() { + return Ok(()); + } + return Err(format!( + "worker rejected user input before session commit: {message}" + )); + } + Ok(Event::Shutdown) => { + if input_was_committed() { + return Ok(()); + } + return Err( + "worker shut down before user input was committed".to_string() + ); + } + Ok(_) => {} + Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => { + if input_was_committed() { + return Ok(()); + } + return Err(format!( + "worker input commit acknowledgement lagged by {skipped} protocol event(s)" + )); + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => { + if input_was_committed() { + return Ok(()); + } + return Err( + "worker event stream closed before user input was committed" + .to_string(), + ); + } + } + } + } + } + }) + .await; + + match acknowledgement { + Ok(result) => result, + Err(_) => { + if timeout_probe + .committed_entries() + .iter() + .any(|entry| user_input_has_submission(entry, &timeout_submission_id)) + { + Ok(()) + } else { + Err("timed out waiting for worker user input commit".to_string()) + } + } + } + }) + .map(|_| { + WorkerExecutionResult::accepted_input_committed( + operation, + accepted_run_state, + acknowledged_submission_id, + ) + }) + .unwrap_or_else(|message| WorkerExecutionResult::errored(operation, message)) + } + fn connect_handle( &self, operation: WorkerExecutionOperation, @@ -1045,6 +1184,7 @@ fn method_starts_turn(method: &Method) -> bool { matches!( method, Method::Run { .. } + | Method::RunTracked { .. } | Method::Notify { auto_run: true, .. } | Method::Resume | Method::Compact @@ -1062,6 +1202,7 @@ fn accepted_notify_run_state(status: WorkerStatus, auto_run: bool) -> WorkerExec fn accepted_run_state_for_method(method: &Method) -> WorkerExecutionRunState { match method { Method::Run { .. } + | Method::RunTracked { .. } | Method::Notify { auto_run: true, .. } | Method::Resume | Method::Compact => WorkerExecutionRunState::Busy, @@ -1400,35 +1541,66 @@ where ); } - let method = match input.kind { - WorkerInputKind::User => Method::Run { - input: input - .segments - .unwrap_or_else(|| vec![Segment::text(input.content.trim().to_string())]), - }, + let (method, submission_id) = match input.kind { + WorkerInputKind::User => { + let Some(submission_id) = input + .submission_id + .filter(|submission_id| !submission_id.trim().is_empty()) + else { + busy.store(false, Ordering::SeqCst); + return WorkerExecutionResult::rejected( + WorkerExecutionOperation::Input, + "Runtime user input is missing its internal submission id", + ); + }; + ( + Method::RunTracked { + input: input.segments.unwrap_or_else(|| { + vec![Segment::text(input.content.trim().to_string())] + }), + submission_id: submission_id.clone(), + }, + Some(submission_id), + ) + } WorkerInputKind::Notify => { unreachable!("Notify input is dispatched before the turn-start busy guard") } - WorkerInputKind::Compact => Method::Compact, - WorkerInputKind::ListRewindTargets => Method::ListRewindTargets, - WorkerInputKind::RegisterPeer => Method::RegisterPeer { - name: input.content.trim().to_string(), - }, + WorkerInputKind::Compact => (Method::Compact, None), + WorkerInputKind::ListRewindTargets => (Method::ListRewindTargets, None), + WorkerInputKind::RegisterPeer => ( + Method::RegisterPeer { + name: input.content.trim().to_string(), + }, + None, + ), }; let accepted_run_state = match method { - Method::Run { .. } | Method::Notify { .. } | Method::Compact => { - WorkerExecutionRunState::Busy - } + Method::Run { .. } + | Method::RunTracked { .. } + | Method::Notify { .. } + | Method::Compact => WorkerExecutionRunState::Busy, _ => WorkerExecutionRunState::Idle, }; let accepted_is_idle = accepted_run_state == WorkerExecutionRunState::Idle; + let waits_for_user_input_commit = submission_id.is_some(); - let result = self.send_method( - WorkerExecutionOperation::Input, - worker, - method, - accepted_run_state, - ); + let result = if waits_for_user_input_commit { + self.send_user_input_and_wait_for_commit( + WorkerExecutionOperation::Input, + worker, + method, + submission_id.expect("tracked Run has submission id"), + accepted_run_state, + ) + } else { + self.send_method( + WorkerExecutionOperation::Input, + worker, + method, + accepted_run_state, + ) + }; if accepted_is_idle || result.outcome != crate::execution::WorkerExecutionOutcome::Accepted { busy.store(false, Ordering::SeqCst); @@ -1608,7 +1780,7 @@ mod tests { use llm_engine::llm_client::event::{Event as LlmEvent, ResponseStatus, StatusEvent}; use llm_engine::llm_client::{ClientError, LlmClient, Request}; use manifest::{Scope, WorkerManifest}; - use session_store::WorkerMetadataStore; + use session_store::{LogEntry, WorkerMetadataStore}; #[test] fn notify_run_state_allows_running_worker_inbox_delivery() { @@ -2175,6 +2347,61 @@ mod tests { ); } + #[test] + fn create_with_initial_input_returns_after_session_commit() { + let client = MockClient::new(simple_text_events()); + 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 mut request = create_request("initial-commit"); + request.initial_input = Some(WorkerInput::user("start the ticket")); + + let detail = runtime.create_worker(request).unwrap(); + + let entries = backend + .workers + .lock() + .unwrap() + .get(&detail.worker_ref) + .expect("live Worker execution") + .handle + .committed_entries(); + assert!(entries.iter().any(|entry| { + matches!( + entry, + LogEntry::UserInput { segments, .. } + if segments == &vec![Segment::text("start the ticket")] + ) + })); + let submission_id = entries + .iter() + .find_map(|entry| { + let LogEntry::UserInput { extensions, .. } = entry else { + return None; + }; + extensions + .iter() + .find(|extension| extension.domain == WORKER_INPUT_SUBMISSION_EXTENSION_DOMAIN) + .and_then(|extension| extension.payload["submission_id"].as_str()) + }) + .expect("committed input submission id"); + uuid::Uuid::parse_str(submission_id).expect("opaque submission id is a UUID"); + } + #[test] fn adapter_dispatches_user_input_through_worker_run_lifecycle() { let client = MockClient::new(simple_text_events()); diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 2e4d6d4f..e5dc274c 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -5,7 +5,7 @@ use std::sync::atomic::Ordering; use llm_engine::EngineError; use llm_engine::llm_client::client::LlmClient; use session_store::WorkerMetadataStore; -use session_store::{LogEntry, Store}; +use session_store::{LogEntry, SessionExtension, Store}; use tokio::sync::{broadcast, mpsc, oneshot}; use crate::discovery::WorkerDiscovery; @@ -24,7 +24,10 @@ use crate::shutdown_after_idle::{ use crate::spawn::comm_tools::{sub_worker_list_tool, sub_worker_send_tool, sub_worker_stop_tool}; use crate::spawn::registry::SpawnedWorkerRegistry; use crate::spawn::tool::sub_worker_spawn_tool; -use crate::worker::{SystemItemCommitter, Worker, WorkerError, WorkerRunResult}; +use crate::worker::{ + SystemItemCommitter, WORKER_INPUT_SUBMISSION_EXTENSION_DOMAIN, Worker, WorkerError, + WorkerRunResult, +}; use protocol::{ AlertLevel, AlertSource, ErrorCode, Event, Method, RewindTargetId, RunResult, Segment, TurnResult, WorkerStatus, @@ -160,6 +163,10 @@ async fn finish_controller_run( /// `worker.run_for_notification()` drains the NotifyBuffer on its own. enum PendingRun { Run(Vec), + RunTracked { + input: Vec, + extension: SessionExtension, + }, /// Self-initiated turn kicked from the notify buffer. The carried /// `InvokeKind` is the trigger that flipped the Worker from IDLE /// (Notify or WorkerEvent) and is recorded by the Invoke marker @@ -177,7 +184,7 @@ impl PendingRun { /// notify buffer (Notify / inbound WorkerEvent) and stays silent. fn is_parent_originated(&self) -> bool { match self { - PendingRun::Run(_) | PendingRun::Resume => true, + PendingRun::Run(_) | PendingRun::RunTracked { .. } | PendingRun::Resume => true, PendingRun::RunForNotification(_) => false, } } @@ -340,6 +347,7 @@ impl WorkerController { bash_output_dir, runtime_base.to_path_buf(), spawned_registry.clone(), + Some(method_tx.downgrade()), ) .await?; @@ -591,6 +599,7 @@ pub(crate) async fn register_worker_tools( bash_output_dir: PathBuf, runtime_base: PathBuf, spawned_registry: Arc, + parent_method_tx: Option>, ) -> std::io::Result> where C: LlmClient + Clone + 'static, @@ -621,7 +630,11 @@ where let spawner_name = worker.manifest().worker.name.clone(); let spawner_manifest = worker.manifest().clone(); let spawner_workspace_context = worker.workspace_context_handle(); - let parent_notifies = worker.notify_buffer_handle(); + let parent_notifications = parent_method_tx + .map(crate::spawn::tool::ParentNotificationTarget::Controller) + .unwrap_or_else(|| { + crate::spawn::tool::ParentNotificationTarget::Buffer(worker.notify_buffer_handle()) + }); let prompts = worker.prompts().clone(); // Resolve the existing Worker–Workdir binding into the domain provider. // Tools only consume the provider handle; they do not own its root, cwd, @@ -810,7 +823,7 @@ where engine.register_tool(sub_worker_spawn_tool( spawner_name.clone(), spawner_workspace_context, - parent_notifies, + parent_notifications, runtime_base.clone(), spawner_workspace_root, spawner_cwd.clone(), @@ -933,6 +946,21 @@ async fn controller_loop( ) .await } + PendingRun::RunTracked { input, extension } => { + drive_turn( + worker.run_with_input_extensions(input, vec![extension]), + &mut method_rx, + &event_tx, + &cancel_tx, + &shared_state, + ¬ify_buffer, + self_parent_socket.as_ref(), + &spawner_name, + &spawned_registry, + parent_originated, + ) + .await + } PendingRun::RunForNotification(kind) => { drive_turn( worker.run_for_notification(kind), @@ -1018,6 +1046,19 @@ async fn controller_loop( pending = Some(PendingRun::Run(input)); } + Method::RunTracked { + input, + submission_id, + } => { + // Runtime-correlated submissions retain their opaque id in the + // same durable UserInput record used for Flow state. + let extension = SessionExtension::new( + WORKER_INPUT_SUBMISSION_EXTENSION_DOMAIN, + serde_json::json!({ "submission_id": submission_id }), + ); + pending = Some(PendingRun::RunTracked { input, extension }); + } + Method::Notify { message, auto_run } => { // Client-side live echo is delivered as `Event::SystemItem` // once the interceptor commits the corresponding @@ -1411,7 +1452,7 @@ where shutdown_requested = true; let _ = cancel_tx.try_send(()); } - Some(Method::Run { .. } | Method::Resume) => { + Some(Method::Run { .. } | Method::RunTracked { .. } | Method::Resume) => { let _ = event_tx.send(Event::Error { code: ErrorCode::AlreadyRunning, message: "Worker is already executing a turn".into(), diff --git a/crates/worker/src/feature/builtin/manage_worker.rs b/crates/worker/src/feature/builtin/manage_worker.rs index f34723f9..3a6ea594 100644 --- a/crates/worker/src/feature/builtin/manage_worker.rs +++ b/crates/worker/src/feature/builtin/manage_worker.rs @@ -96,6 +96,9 @@ struct WorkerSpawnInput { runtime_id: String, working_directory_id: String, profile: String, + /// Optional queued Ticket to assign atomically to the new Coder Worker. + #[serde(default)] + ticket_id: Option, #[serde(default)] display_name: Option, /// Normal typed initial user submission delivered after spawn. An empty @@ -106,11 +109,19 @@ struct WorkerSpawnInput { relative_cwd: Option, } +#[derive(Debug, Serialize)] +struct WorkerSpawnTicketAssignmentRequest { + ticket_id: String, + operation_id: String, +} + #[derive(Debug, Serialize)] struct WorkerSpawnRequest { runtime_id: String, display_name: String, profile: String, + #[serde(skip_serializing_if = "Option::is_none")] + ticket_assignment: Option, initial_submit: Vec, working_directory: WorkerWorkingDirectorySelection, } @@ -170,7 +181,7 @@ impl WorkerOperation { "List Backend/Runtime Worker sessions in the current Workspace. SubWorkers are excluded." } 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." + "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. Set `ticket_id` with a Flow segment in `initial_submit` to atomically assign a queued Ticket to the new Coder Worker; the operation id is derived from the durable tool call rather than model input." } Self::Stop => "Stop a Backend/Runtime Worker session in the current Workspace.", Self::Restore => { @@ -185,7 +196,7 @@ impl Tool for WorkspaceWorkerTool { async fn execute( &self, input_json: &str, - _ctx: ToolExecutionContext, + ctx: ToolExecutionContext, ) -> Result { let request = match self.operation { WorkerOperation::List => { @@ -194,6 +205,17 @@ impl Tool for WorkspaceWorkerTool { } WorkerOperation::Spawn => { let input = parse::(input_json, "WorkerSpawn")?; + let ticket_assignment = input + .ticket_id + .map(|ticket_id| { + let ticket_id = authority_id(&ticket_id, "ticket_id")?; + let call_id = non_empty(ctx.call_id.clone(), "tool call_id")?; + Ok::<_, ToolError>(WorkerSpawnTicketAssignmentRequest { + operation_id: format!("worker-spawn:{ticket_id}:{call_id}"), + ticket_id, + }) + }) + .transpose()?; let request = WorkerSpawnRequest { runtime_id: authority_id(&input.runtime_id, "runtime_id")?, display_name: input @@ -201,6 +223,7 @@ impl Tool for WorkspaceWorkerTool { .filter(|value| !value.trim().is_empty()) .unwrap_or_else(|| "Workspace Worker".to_string()), profile: non_empty(input.profile, "profile")?, + ticket_assignment, initial_submit: input.initial_submit, working_directory: WorkerWorkingDirectorySelection { working_directory_id: authority_id( @@ -372,13 +395,14 @@ mod tests { "runtime_id": "runtime-1", "working_directory_id": "workdir-1", "profile": "builtin:coder", + "ticket_id": "00001KZ9E0DBS", "initial_submit": [ { "kind": "flow", "selector": "builtin:coder-review" }, { "kind": "text", "content": "Implement Ticket 00001" } ] }) .to_string(), - ToolExecutionContext::direct(), + ToolExecutionContext::new("call-1", "batch-1", 0), ) .await .unwrap(); @@ -394,6 +418,13 @@ mod tests { "builtin:coder-review" ); assert_eq!(body["initial_submit"][1]["kind"], "text"); + assert_eq!( + body["ticket_assignment"], + serde_json::json!({ + "ticket_id": "00001KZ9E0DBS", + "operation_id": "worker-spawn:00001KZ9E0DBS:call-1" + }) + ); assert!(body.get("initial_text").is_none()); } @@ -410,6 +441,7 @@ mod tests { let schema = serde_json::to_value(schemars::schema_for!(WorkerSpawnInput)).unwrap(); let text = serde_json::to_string(&schema).unwrap(); assert!(text.contains("initial_submit")); + assert!(text.contains("ticket_id")); assert!(text.contains("selector")); assert!(text.contains("flow")); assert!(!text.contains("initial_text")); @@ -421,6 +453,7 @@ mod tests { runtime_id: "runtime-1".to_string(), display_name: "Coder".to_string(), profile: "builtin:coder".to_string(), + ticket_assignment: None, initial_submit: vec![ Segment::Flow { selector: "builtin:coder-review".to_string(), diff --git a/crates/worker/src/lib.rs b/crates/worker/src/lib.rs index 185d1cad..3c0d1e12 100644 --- a/crates/worker/src/lib.rs +++ b/crates/worker/src/lib.rs @@ -40,9 +40,9 @@ pub use runtime::dir::RuntimeDir; pub use segment_log_sink::SegmentLogSink; pub use shared_state::WorkerSharedState; pub use worker::{ - LocalWorkingDirectory, RuntimeWorkspaceHttpClient, Worker, WorkerError, - WorkerFilesystemAuthority, WorkerRunResult, WorkerWorkspaceContext, WorkspaceClient, - WorkspaceClientError, WorkspaceId, WorkspaceIdError, WorkspaceRequest, WorkspaceRequestMethod, - WorkspaceResponse, apply_worker_manifest, marker_workspace_client, + LocalWorkingDirectory, RuntimeWorkspaceHttpClient, WORKER_INPUT_SUBMISSION_EXTENSION_DOMAIN, + Worker, WorkerError, WorkerFilesystemAuthority, WorkerRunResult, WorkerWorkspaceContext, + WorkspaceClient, WorkspaceClientError, WorkspaceId, WorkspaceIdError, WorkspaceRequest, + WorkspaceRequestMethod, WorkspaceResponse, apply_worker_manifest, marker_workspace_client, unavailable_workspace_client, }; diff --git a/crates/worker/src/spawn/tool.rs b/crates/worker/src/spawn/tool.rs index df25c8b5..6e64423d 100644 --- a/crates/worker/src/spawn/tool.rs +++ b/crates/worker/src/spawn/tool.rs @@ -18,6 +18,7 @@ use manifest::{ WorkerManifest, WorkerManifestConfig, WorkerMetaConfig, }; use serde::Deserialize; +use tokio::sync::mpsc; use crate::PromptLoader; use crate::controller::register_worker_tools; @@ -27,6 +28,7 @@ use crate::internal_worker::{ use crate::prompt::catalog::PromptCatalog; use crate::spawn::registry::SpawnedWorkerRegistry; use crate::worker::{Worker, WorkerFilesystemAuthority}; +use protocol::Method; #[derive(Debug, Deserialize, schemars::JsonSchema)] struct SubWorkerSpawnInput { @@ -205,6 +207,39 @@ fn parse_spawn_profile_selector(raw: Option<&str>) -> Result), + Buffer(crate::ipc::notify_buffer::NotifyBuffer), +} + +impl ParentNotificationTarget { + fn notify(&self, message: String, auto_run: bool) { + match self { + Self::Controller(parent_method_tx) => { + let Some(parent_method_tx) = parent_method_tx.upgrade() else { + tracing::warn!( + "parent Worker controller closed before Internal SubWorker completion notification" + ); + return; + }; + tokio::spawn(async move { + if let Err(error) = parent_method_tx + .send(Method::Notify { message, auto_run }) + .await + { + tracing::warn!( + %error, + "failed to notify parent Worker about Internal SubWorker completion" + ); + } + }); + } + Self::Buffer(parent_notifies) => parent_notifies.push_notify(message, auto_run), + } + } +} + /// Runtime dependencies the `SubWorkerSpawn` tool needs in order to launch a /// child SubWorker and record the handoff locally. Constructed by the Worker /// controller once per Worker lifetime. @@ -212,7 +247,7 @@ pub struct SubWorkerSpawnTool { /// Spawner's own Worker name, used for direct-child identity collision checks. spawner_name: String, workspace_context: crate::worker::WorkerWorkspaceContext, - parent_notifies: crate::ipc::notify_buffer::NotifyBuffer, + parent_notifications: ParentNotificationTarget, /// Runtime-owned root used only for bounded Internal Worker tool artifacts such as Bash spill /// output. It is not an Internal Worker identity or catalog location. runtime_base: PathBuf, @@ -256,7 +291,7 @@ impl SubWorkerSpawnTool { fn new( spawner_name: String, workspace_context: crate::worker::WorkerWorkspaceContext, - parent_notifies: crate::ipc::notify_buffer::NotifyBuffer, + parent_notifications: ParentNotificationTarget, runtime_base: PathBuf, workspace_root: PathBuf, spawner_cwd: PathBuf, @@ -270,7 +305,7 @@ impl SubWorkerSpawnTool { Self { spawner_name, workspace_context, - parent_notifies, + parent_notifications, runtime_base, workspace_root, spawner_cwd, @@ -368,6 +403,7 @@ impl Tool for SubWorkerSpawnTool { .join("bash-output"), self.runtime_base.clone(), child_registry, + None, ) .await .map_err(|error| { @@ -392,7 +428,7 @@ impl Tool for SubWorkerSpawnTool { let child_name = input.name.clone(); let registry = Arc::downgrade(&self.registry); - let parent_notifies = self.parent_notifies.clone(); + let parent_notifications = self.parent_notifications.clone(); let session_result = prepare_internal_worker_session( child, store, @@ -408,10 +444,10 @@ impl Tool for SubWorkerSpawnTool { } } } - parent_notifies.push_notify( - format!("SubWorker `{child_name}` turn ended with status {status:?}. Inspect its committed session with worker-observation tools before making completion decisions."), - true, + let message = format!( + "SubWorker `{child_name}` turn ended with status {status:?}. Inspect its committed session with worker-observation tools before making completion decisions." ); + parent_notifications.notify(message, true); })), ) .await; @@ -764,10 +800,10 @@ fn manifest_to_reusable_config(manifest: &WorkerManifest) -> WorkerManifestConfi /// tool-result budget — debugging beyond this should read the file /// directly. /// Factory for the `SubWorkerSpawn` tool. -pub fn sub_worker_spawn_tool( +pub(crate) fn sub_worker_spawn_tool( spawner_name: String, workspace_context: crate::worker::WorkerWorkspaceContext, - parent_notifies: crate::ipc::notify_buffer::NotifyBuffer, + parent_notifications: ParentNotificationTarget, runtime_base: PathBuf, workspace_root: PathBuf, spawner_cwd: PathBuf, @@ -779,7 +815,7 @@ pub fn sub_worker_spawn_tool( sub_worker_spawn_tool_impl( spawner_name, workspace_context, - parent_notifies, + parent_notifications, runtime_base, workspace_root, spawner_cwd, @@ -793,7 +829,7 @@ pub fn sub_worker_spawn_tool( fn sub_worker_spawn_tool_impl( spawner_name: String, workspace_context: crate::worker::WorkerWorkspaceContext, - parent_notifies: crate::ipc::notify_buffer::NotifyBuffer, + parent_notifications: ParentNotificationTarget, runtime_base: PathBuf, workspace_root: PathBuf, spawner_cwd: PathBuf, @@ -824,7 +860,7 @@ fn sub_worker_spawn_tool_impl( let tool: Arc = Arc::new(SubWorkerSpawnTool::new( spawner_name.clone(), workspace_context.clone(), - parent_notifies.clone(), + parent_notifications.clone(), runtime_base.clone(), workspace_root.clone(), spawner_cwd.clone(), @@ -845,6 +881,7 @@ mod tests { use super::*; use std::pin::Pin; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; + use std::time::Duration; use crate::WorkspaceId; use async_trait::async_trait; @@ -896,7 +933,18 @@ extract_threshold = 4000 "#; #[tokio::test] - async fn reviewer_profile_spawns_as_workspace_aware_internal_session() { + async fn parent_controller_notification_target_does_not_keep_channel_open() { + let (parent_method_tx, mut parent_method_rx) = mpsc::channel(1); + let target = ParentNotificationTarget::Controller(parent_method_tx.downgrade()); + + drop(parent_method_tx); + + assert!(parent_method_rx.recv().await.is_none()); + target.notify("late completion".to_string(), true); + } + + #[tokio::test] + async fn reviewer_profile_spawns_and_notifies_parent_controller() { let runtime = TempDir::new().unwrap(); let workspace_root = runtime.path().join("project"); let available_profiles = write_project_profile_registry( @@ -927,11 +975,11 @@ extract_threshold = 4000 ) .unwrap(); let prompt_loader = PromptLoader::new(None, Some(workspace_prompts)); - let parent_notifies = crate::ipc::notify_buffer::NotifyBuffer::new(); + let (parent_method_tx, mut parent_method_rx) = mpsc::channel(8); let tool = SubWorkerSpawnTool::new( "parent".into(), workspace_context, - parent_notifies.clone(), + ParentNotificationTarget::Controller(parent_method_tx.downgrade()), runtime.path().to_path_buf(), workspace_root.clone(), workspace_root.clone(), @@ -995,11 +1043,17 @@ extract_threshold = 4000 assert_eq!(calls.load(Ordering::SeqCst), 1); assert!(observed_parent_write_revoked.load(Ordering::SeqCst)); assert!(observed_instruction_override.load(Ordering::SeqCst)); - assert_eq!(parent_notifies.len(), 1); - assert!( - parent_notifies.has_auto_run_pending(), - "SubWorker completion must auto-invoke the parent" - ); + let completion = tokio::time::timeout(Duration::from_secs(1), parent_method_rx.recv()) + .await + .expect("SubWorker completion must wake the parent method channel") + .expect("parent method channel remains open"); + assert!(matches!( + completion, + Method::Notify { + message, + auto_run: true, + } if message.contains("SubWorker `reviewer-child` turn ended with status Idle") + )); assert!(!runtime.path().join("reviewer-child/sock").exists()); let duplicate_error = tool @@ -1018,7 +1072,10 @@ extract_threshold = 4000 1, "duplicate rejection must not invoke the child provider" ); - assert_eq!(parent_notifies.len(), 1); + assert!(matches!( + parent_method_rx.try_recv(), + Err(tokio::sync::mpsc::error::TryRecvError::Empty) + )); let context = llm_engine::tool::ToolExecutionContext::direct(); let list = (crate::spawn::comm_tools::sub_worker_list_tool(registry.clone()))().1; diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 8839dc05..11ca450d 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -650,6 +650,8 @@ where } } +pub const WORKER_INPUT_SUBMISSION_EXTENSION_DOMAIN: &str = "worker.input-submission.v1"; + /// An independent agent execution unit. /// /// Holds a [`Engine`] directly and persists session state via @@ -2182,19 +2184,24 @@ impl Worker { /// the Engine is aborted, history is compacted, and execution resumes /// automatically. pub async fn run(&mut self, input: Vec) -> Result { + self.run_with_input_extensions(input, Vec::new()).await + } + + pub(crate) async fn run_with_input_extensions( + &mut self, + input: Vec, + mut input_extensions: Vec, + ) -> Result { let (input, pending_flow_state) = self.prepare_flow_input(input)?; - let input_extensions = pending_flow_state - .as_ref() - .map(|state| { - serde_json::to_value(state) - .map(|payload| SessionExtension::new(FLOW_RUNTIME_EXTENSION_DOMAIN, payload)) - .map_err(|error| { - WorkerError::FlowInput(format!("serialize Flow runtime state: {error}")) - }) - }) - .transpose()? - .into_iter() - .collect(); + if let Some(state) = pending_flow_state.as_ref() { + let payload = serde_json::to_value(state).map_err(|error| { + WorkerError::FlowInput(format!("serialize Flow runtime state: {error}")) + })?; + input_extensions.push(SessionExtension::new( + FLOW_RUNTIME_EXTENSION_DOMAIN, + payload, + )); + } // Paused→Run transition: if the previous turn was cut short, // any `Item::ToolCall` whose tool never produced a matching diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 44ff693d..d86ad12b 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -418,6 +418,7 @@ fn initial_worker_input(segments: &[Segment]) -> Option { Some(EmbeddedWorkerInput { kind: EmbeddedWorkerInputKind::User, content: Segment::flatten_to_text(segments), + submission_id: None, segments: Some(segments.to_vec()), }) } @@ -2181,6 +2182,7 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { WorkerInputKind::RegisterPeer => EmbeddedWorkerInputKind::RegisterPeer, }, content: request.content, + submission_id: None, segments: request.segments, }; match self.runtime.send_input(&worker_ref, input) { @@ -3148,6 +3150,7 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { WorkerInputKind::RegisterPeer => EmbeddedWorkerInputKind::RegisterPeer, }, content: request.content, + submission_id: None, segments: request.segments, }; match self.post_json::<_, RuntimeHttpWorkerInputResponse>( @@ -4326,6 +4329,7 @@ mod tests { "missing test context", ); }; + let submission_id = input.submission_id.clone(); let content = input.content; std::thread::spawn(move || { std::thread::sleep(std::time::Duration::from_millis(10)); @@ -4342,10 +4346,18 @@ mod tests { status: protocol::WorkerStatus::Idle, }); }); - worker_runtime::execution::WorkerExecutionResult::accepted( - worker_runtime::execution::WorkerExecutionOperation::Input, - WorkerExecutionRunState::Busy, - ) + if let Some(submission_id) = submission_id { + worker_runtime::execution::WorkerExecutionResult::accepted_input_committed( + worker_runtime::execution::WorkerExecutionOperation::Input, + WorkerExecutionRunState::Busy, + submission_id, + ) + } else { + worker_runtime::execution::WorkerExecutionResult::accepted( + worker_runtime::execution::WorkerExecutionOperation::Input, + WorkerExecutionRunState::Busy, + ) + } } } diff --git a/crates/workspace-server/src/runtime_subscription_tests.rs b/crates/workspace-server/src/runtime_subscription_tests.rs index 26b37d16..08a93fd7 100644 --- a/crates/workspace-server/src/runtime_subscription_tests.rs +++ b/crates/workspace-server/src/runtime_subscription_tests.rs @@ -29,12 +29,20 @@ impl WorkerExecutionBackend for TestExecutionBackend { fn dispatch_input( &self, _handle: &WorkerExecutionHandle, - _input: worker_runtime::interaction::WorkerInput, + input: worker_runtime::interaction::WorkerInput, ) -> WorkerExecutionResult { - WorkerExecutionResult::accepted( - WorkerExecutionOperation::Input, - WorkerExecutionRunState::Busy, - ) + if let Some(submission_id) = input.submission_id { + WorkerExecutionResult::accepted_input_committed( + WorkerExecutionOperation::Input, + WorkerExecutionRunState::Busy, + submission_id, + ) + } else { + WorkerExecutionResult::accepted( + WorkerExecutionOperation::Input, + WorkerExecutionRunState::Busy, + ) + } } fn stop_worker(&self, _handle: &WorkerExecutionHandle) -> WorkerExecutionResult { diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 844141a1..3661cf86 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -65,12 +65,13 @@ use crate::config::{BackendRuntimesConfigFile, RemoteRuntimeConfigFile, resolve_ use crate::hosts::{ ConfigBundleCheckResult, ConfigBundleSyncResult, DiagnosticSeverity, EmbeddedWorkerRuntime, HostSummary, RemoteRuntimeConfig, RemoteWorkerRuntime, RuntimeDiagnostic, RuntimeRegistry, - RuntimeRegistryError, RuntimeRegistryUnregisterResult, RuntimeSummary, WorkerCapabilitySummary, - WorkerCompletionsRequest, WorkerCompletionsResult, WorkerImplementationSummary, - WorkerInputKind, WorkerInputRequest, WorkerInputResult, WorkerLifecycleRequest, - WorkerLifecycleResult, WorkerOperationState, WorkerRestoreResult, + RuntimeRegistryError, RuntimeRegistryUnregisterResult, RuntimeSummary, TicketWorkerRole, + WorkerCapabilitySummary, WorkerCompletionsRequest, WorkerCompletionsResult, + WorkerImplementationSummary, WorkerInputKind, WorkerInputRequest, WorkerInputResult, + WorkerLifecycleRequest, WorkerLifecycleResult, WorkerOperationState, WorkerRestoreResult, WorkerSpawnAcceptanceRequirement, WorkerSpawnIntent, WorkerSpawnRequest, WorkerSpawnResult, - WorkerSpawnWorkingDirectoryRequest, WorkerSummary, WorkerWorkspaceSummary, + WorkerSpawnWorkingDirectoryRequest, WorkerSummary, WorkerTicketAssignmentRequest, + WorkerWorkspaceSummary, }; use crate::identity::WorkspaceIdentity; use crate::memory_backend::execute_memory_backend_operation_with_authority; @@ -1554,6 +1555,13 @@ pub struct BrowserWorkspaceOrchestratorResponse { pub diagnostics: Vec, } +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct CreateWorkspaceWorkerTicketAssignmentRequest { + pub ticket_id: String, + pub operation_id: String, +} + #[derive(Debug, Deserialize)] #[serde(deny_unknown_fields)] pub struct CreateWorkspaceWorkerRequest { @@ -1562,6 +1570,8 @@ pub struct CreateWorkspaceWorkerRequest { #[serde(default)] pub profile: Option, #[serde(default)] + pub ticket_assignment: Option, + #[serde(default)] pub initial_submit: Vec, #[serde(default)] pub working_directory: Option, @@ -6856,12 +6866,71 @@ fn validate_worker_initial_submit(segments: &[Segment]) -> Result<()> { Ok(()) } +fn browser_worker_spawn_policy( + ticket_assignment: Option, + initial_submit: &[Segment], +) -> Result<( + WorkerSpawnIntent, + WorkerSpawnAcceptanceRequirement, + Option, +)> { + let expected_segments = initial_submit.len(); + match ticket_assignment { + Some(assignment) => { + if !initial_submit + .iter() + .any(|segment| matches!(segment, Segment::Flow { .. })) + { + return Err(Error::InvalidInput( + "Ticket-assigned Coder spawn requires one Flow segment in initial_submit" + .to_string(), + )); + } + let ticket_id = assignment.ticket_id.trim().to_string(); + if ticket_id.is_empty() { + return Err(Error::InvalidInput( + "ticket_id must not be empty".to_string(), + )); + } + let operation_id = assignment.operation_id.trim().to_string(); + if operation_id.is_empty() { + return Err(Error::InvalidInput( + "assignment operation_id must not be empty".to_string(), + )); + } + Ok(( + WorkerSpawnIntent::TicketRole { + ticket_id: ticket_id.clone(), + role: TicketWorkerRole::Coder, + }, + WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments }, + Some(WorkerTicketAssignmentRequest { + ticket_id, + operation_id, + }), + )) + } + None => Ok(( + WorkerSpawnIntent::WorkspaceCoding, + WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments }, + None, + )), + } +} + async fn create_workspace_worker( State(api): State, Json(request): Json, ) -> ApiResult> { - let profile = request - .profile + let CreateWorkspaceWorkerRequest { + runtime_id, + display_name, + profile, + ticket_assignment, + initial_submit, + working_directory, + } = request; + let profile = profile .as_deref() .map(str::trim) .filter(|profile| !profile.is_empty()) @@ -6898,7 +6967,7 @@ async fn create_workspace_worker( } else { None }; - let display_name = sanitize_worker_display_name(&request.display_name).ok_or_else(|| { + let display_name = sanitize_worker_display_name(&display_name).ok_or_else(|| { settings_bad_request( "invalid_worker_display_name", "display_name must contain at least one non-control character", @@ -6907,33 +6976,28 @@ async fn create_workspace_worker( if display_name == crate::hosts::WORKSPACE_ORCHESTRATOR_SINGLETON_KEY { return Err(Error::ReservedWorkerName(display_name).into()); } - let initial_submit = request.initial_submit; validate_worker_initial_submit(&initial_submit)?; - let expected_segments = initial_submit.len(); - let selected_working_directory_id = request - .working_directory + let selected_working_directory_id = working_directory .as_ref() .map(|selection| selection.working_directory_id.clone()); - let resolved_working_directory = - request - .working_directory - .map(|selection| WorkingDirectoryClaim { - working_directory_id: selection.working_directory_id, - relative_cwd: selection.relative_cwd, - }); + let resolved_working_directory = working_directory.map(|selection| WorkingDirectoryClaim { + working_directory_id: selection.working_directory_id, + relative_cwd: selection.relative_cwd, + }); validate_working_directory_claim_for_browser(resolved_working_directory.as_ref())?; if resolved_working_directory.is_none() { - reject_no_workdir_for_non_embedded_runtime(&request.runtime_id)?; + reject_no_workdir_for_non_embedded_runtime(&runtime_id)?; } - let runtime_id = request.runtime_id.clone(); + let (intent, acceptance, ticket_assignment) = + browser_worker_spawn_policy(ticket_assignment, &initial_submit)?; let result = api.spawn_workspace_worker( &runtime_id, WorkerSpawnRequest { requested_worker_name: Some(display_name.clone()), - intent: WorkerSpawnIntent::WorkspaceCoding, - acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments }, + intent, + acceptance, profile: profile_selector, - ticket_assignment: None, + ticket_assignment, initial_submit, working_directory_request: None, resolved_working_directory_request: None, @@ -10271,6 +10335,60 @@ mod tests { ); } + #[test] + fn worker_ticket_assignment_projects_coder_intent_and_run_acceptance() { + let initial_submit = vec![ + Segment::Flow { + selector: "builtin:coder-review".to_string(), + }, + Segment::text("Implement Ticket 00001KZ9E0DBS"), + ]; + let assignment_request = || CreateWorkspaceWorkerTicketAssignmentRequest { + ticket_id: "00001KZ9E0DBS".to_string(), + operation_id: "worker-spawn:00001KZ9E0DBS:call-1".to_string(), + }; + let (intent, acceptance, assignment) = + browser_worker_spawn_policy(Some(assignment_request()), &initial_submit).unwrap(); + + assert_eq!( + intent, + WorkerSpawnIntent::TicketRole { + ticket_id: "00001KZ9E0DBS".to_string(), + role: TicketWorkerRole::Coder, + } + ); + assert_eq!( + acceptance, + WorkerSpawnAcceptanceRequirement::RunAccepted { + expected_segments: 2, + } + ); + assert_eq!( + assignment, + Some(WorkerTicketAssignmentRequest { + ticket_id: "00001KZ9E0DBS".to_string(), + operation_id: "worker-spawn:00001KZ9E0DBS:call-1".to_string(), + }) + ); + assert!( + browser_worker_spawn_policy( + Some(CreateWorkspaceWorkerTicketAssignmentRequest { + ticket_id: " ".to_string(), + operation_id: "operation".to_string(), + }), + &initial_submit, + ) + .is_err() + ); + assert!( + browser_worker_spawn_policy( + Some(assignment_request()), + &[Segment::text("Ticket text without Flow")], + ) + .is_err() + ); + } + #[tokio::test] async fn worker_source_auth_rejects_cross_workspace_mutation() { let workspace = tempfile::tempdir().unwrap(); @@ -10282,6 +10400,7 @@ mod tests { runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), display_name: "Scoped Worker".to_string(), profile: Some("builtin:coder".to_string()), + ticket_assignment: None, initial_submit: Vec::new(), working_directory: None, }), @@ -10315,6 +10434,7 @@ mod tests { runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), display_name: "Generic Orchestrator Profile Worker".to_string(), profile: Some("builtin:orchestrator".to_string()), + ticket_assignment: None, initial_submit: Vec::new(), working_directory: None, }), @@ -10329,6 +10449,7 @@ mod tests { runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), display_name: crate::hosts::WORKSPACE_ORCHESTRATOR_SINGLETON_KEY.to_string(), profile: Some("builtin:orchestrator".to_string()), + ticket_assignment: None, initial_submit: Vec::new(), working_directory: None, }), @@ -11046,6 +11167,7 @@ mod tests { .get(handle.worker_ref()) .cloned() .expect("execution context"); + let submission_id = input.submission_id.clone(); let content = input.content.clone(); std::thread::spawn(move || { std::thread::sleep(std::time::Duration::from_millis(25)); @@ -11053,10 +11175,18 @@ mod tests { text: format!("server companion echoed: {content}"), }); }); - worker_runtime::execution::WorkerExecutionResult::accepted( - worker_runtime::execution::WorkerExecutionOperation::Input, - worker_runtime::execution::WorkerExecutionRunState::Idle, - ) + if let Some(submission_id) = submission_id { + worker_runtime::execution::WorkerExecutionResult::accepted_input_committed( + worker_runtime::execution::WorkerExecutionOperation::Input, + worker_runtime::execution::WorkerExecutionRunState::Idle, + submission_id, + ) + } else { + worker_runtime::execution::WorkerExecutionResult::accepted( + worker_runtime::execution::WorkerExecutionOperation::Input, + worker_runtime::execution::WorkerExecutionRunState::Idle, + ) + } } } diff --git a/docs/design/flow-state-graph.md b/docs/design/flow-state-graph.md index 43d84b7b..3fc49249 100644 --- a/docs/design/flow-state-graph.md +++ b/docs/design/flow-state-graph.md @@ -69,7 +69,9 @@ Flow invocation uses the normal Submit/Run segment vector rather than a Worker-c Runtime accepts exactly one Flow segment only when the resolved Profile enables `feature.flow` and a Workspace client is available. The Worker asks Workspace authority only for an immutable source snapshot, creates the instance locally, replaces the Flow segment with the entered state's instructions, and commits that runtime state atomically with the remaining Submit segments before LLM execution. A Worker with an active Flow rejects the duplicate input without changing its local state or events. -The generic model-facing `WorkerSpawn` accepts `initial_submit: Vec` and routes them unchanged through the shared Workspace spawn request into Runtime `CreateWorkerRequest.initial_input`. It does not have a parallel `initial_text` or a role-specific `SpawnCoder` wrapper. Backend derives the flat content projection from the canonical segment vector, validates Flow shape before spawn, and includes the segment vector in lifecycle idempotency fingerprints. Restoring the same Worker never replays spawn initial segments. +The generic model-facing `WorkerSpawn` accepts `initial_submit: Vec` and routes them unchanged through the shared Workspace spawn request into Runtime `CreateWorkerRequest.initial_input`. It does not have a parallel `initial_text` or a role-specific `SpawnCoder` wrapper. Backend derives the flat content projection from the canonical segment vector, validates Flow shape before spawn, and includes the segment vector in lifecycle idempotency fingerprints. Runtime does not commit Worker creation or report spawn success merely because the initial Run method entered the Worker's in-memory channel: Runtime assigns the Submit an opaque id, the Worker commits that id as an extension on the same `UserInput` entry as any initial `FlowRuntimeState`, and the execution backend must return a matching typed input-commit acknowledgement. Restoring the same Worker never replays spawn initial segments. + +When an Orchestrator supplies `ticket_id` to generic `WorkerSpawn`, the Worker tool derives the assignment operation id from the durable tool-call id rather than accepting lifecycle authority from model input. The shared Workspace worker-create route projects that request into a Coder Ticket-role intent and atomically applies the existing queued-Ticket assignment operation only after Runtime has returned the input-commit acknowledgement. A spawn or pre-commit input failure therefore leaves the Ticket queued and unassigned. `RequestFlowTransition` accepts only: diff --git a/docs/report/2026-08-06-subworker-completion-idle-auto-run.md b/docs/report/2026-08-06-subworker-completion-idle-auto-run.md new file mode 100644 index 00000000..01053ec9 --- /dev/null +++ b/docs/report/2026-08-06-subworker-completion-idle-auto-run.md @@ -0,0 +1,32 @@ +# SubWorker completion notification did not wake an idle parent + +Date: 2026-08-06 +Ticket: `00001KZKNWP5X` + +## Observed behavior + +A Reviewer SubWorker finished after its parent Worker had returned to idle. Although the completion notification was marked `auto_run: true`, the parent did not run until the user submitted another message. The notification appeared only in that later turn. + +## Root cause + +The completion callback called `NotifyBuffer::push_notify(..., true)` directly. A running parent checks that buffer at turn end and can stage a follow-up, but an idle controller waits on its method channel. Writing the buffer alone therefore could not wake an already-idle parent. + +## Fix + +Normal controller-owned Workers now give the SubWorker tool a weak sender for the parent method channel. Completion is delivered through the existing: + +```rust +Method::Notify { + message, + auto_run: true, +} +``` + +path, which commits the notification through the normal inbox and wakes an idle controller. A `WeakSender` avoids a controller/tool/channel reference cycle that would otherwise keep the controller alive after external handles are dropped. Internal Worker sessions without a controller retain the direct buffer target. + +## Regression coverage + +- SubWorker completion sends exactly one `Method::Notify { auto_run: true }` to the parent controller channel. +- The controller notification target does not keep the method channel alive after the strong sender is dropped. +- Existing running-parent notification follow-up tests remain green. +- `cargo test -p worker --lib`: 514 passed.