diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 41b33f44..ec4437db 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -884,8 +884,10 @@ where .register_tools(tools::web_builtin_tools(web_config)); } + let worker_enabled = feature_config.worker.enabled; + let sub_worker_enabled = feature_config.sub_worker.enabled; let mut feature_registry = FeatureRegistryBuilder::new(); - if feature_config.sub_worker.enabled { + if sub_worker_enabled && !worker_enabled { feature_registry.add_module( crate::feature::builtin::manage_worker::sub_worker_control_feature( worker.workspace_client_handle(), @@ -970,7 +972,7 @@ where feature_registry.add_module( crate::feature::builtin::manage_worker::manage_worker_feature( workspace_client, - Some(spawned_registry.clone()), + sub_worker_enabled.then(|| spawned_registry.clone()), feature_config.worker.direct_spawn, ), ); diff --git a/crates/worker/src/feature/builtin/manage_worker.rs b/crates/worker/src/feature/builtin/manage_worker.rs index 12d481b4..1cd54dfa 100644 --- a/crates/worker/src/feature/builtin/manage_worker.rs +++ b/crates/worker/src/feature/builtin/manage_worker.rs @@ -63,6 +63,7 @@ struct WorkspaceWorkerControlService { client: Arc, workspace_id: String, registry: Option>, + runtime_worker_control: bool, } impl std::fmt::Debug for WorkspaceWorkerControlService { @@ -169,6 +170,11 @@ impl WorkerControlService for WorkspaceWorkerControlService { worker_id: &str, reason: &str, ) -> Result { + if !self.runtime_worker_control { + return Err(WorkspaceClientError::Unavailable( + "Runtime Worker control is not enabled for this Worker".to_string(), + )); + } self.client .execute_worker_remove(runtime_id, worker_id, reason) } @@ -177,6 +183,11 @@ impl WorkerControlService for WorkspaceWorkerControlService { &self, request: WorkspaceRequest, ) -> Result { + if !self.runtime_worker_control { + return Err(WorkspaceClientError::Request( + "Runtime Worker control is not enabled for this Worker".to_string(), + )); + } self.client.execute(request) } @@ -204,6 +215,11 @@ impl WorkerControlService for WorkspaceWorkerControlService { runtime_id, worker_id, } => { + if !self.runtime_worker_control { + return Err(WorkspaceClientError::Unavailable( + "Runtime Worker control is not enabled for this Worker".to_string(), + )); + } let response = self.client.execute(WorkspaceRequest::get(format!( "/api/w/{}/worker-control/workers", self.workspace_id @@ -348,6 +364,7 @@ pub fn manage_worker_feature( client: client.clone(), workspace_id, registry, + runtime_worker_control: true, }); ManageWorkerFeature { client, @@ -356,6 +373,12 @@ pub fn manage_worker_feature( } } +const SUB_WORKER_CONTROL_OPERATIONS: &[WorkerOperation] = &[ + WorkerOperation::List, + WorkerOperation::SendInput, + WorkerOperation::Stop, +]; + pub struct SubWorkerControlFeature { client: Arc, registry: Arc, @@ -369,13 +392,19 @@ impl SubWorkerControlFeature { impl FeatureModule for SubWorkerControlFeature { fn descriptor(&self) -> FeatureDescriptor { - FeatureDescriptor::builtin("sub-worker-control", "SubWorker Control") - .with_description("Parent-owned SubWorker control service provider") + let mut descriptor = FeatureDescriptor::builtin("sub-worker-control", "SubWorker Control") + .with_description( + "Parent-owned SubWorker lifecycle through the canonical Worker control surface", + ) .with_provided_service(ServiceDeclaration::new( ServiceId::builtin(WORKER_CONTROL_SERVICE_ID), WORKER_LIFECYCLE_SERVICE_VERSION, "Parent-owned SubWorker discovery and control operations", - )) + )); + for operation in SUB_WORKER_CONTROL_OPERATIONS { + descriptor = descriptor.with_tool(sub_worker_tool_declaration(*operation)); + } + descriptor } fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> { @@ -383,6 +412,7 @@ impl FeatureModule for SubWorkerControlFeature { workspace_id: self.client.workspace_id().unwrap_or_default().to_string(), client: self.client.clone(), registry: Some(self.registry.clone()), + runtime_worker_control: false, }); context.services().provide( ServiceDeclaration::new( @@ -390,8 +420,16 @@ impl FeatureModule for SubWorkerControlFeature { WORKER_LIFECYCLE_SERVICE_VERSION, "Parent-owned SubWorker discovery and control operations", ), - control, - ) + control.clone(), + )?; + for operation in SUB_WORKER_CONTROL_OPERATIONS { + context.tools().register(worker_tool_contribution( + *operation, + control.clone(), + false, + ))?; + } + Ok(()) } } @@ -418,10 +456,7 @@ impl FeatureModule for ManageWorkerFeature { )); for operation in WorkerOperation::ALL { if operation != WorkerOperation::Spawn || self.direct_spawn { - descriptor = descriptor.with_tool(ToolDeclaration::new( - operation.tool_name(), - operation.description(), - )); + descriptor = descriptor.with_tool(worker_tool_declaration(operation)); } } descriptor @@ -463,29 +498,11 @@ impl FeatureModule for ManageWorkerFeature { if operation == WorkerOperation::Spawn && !self.direct_spawn { continue; } - let definition = match operation { - WorkerOperation::List => { - definition::(operation, self.control.clone()) - } - WorkerOperation::Spawn => { - definition::(operation, self.control.clone()) - } - WorkerOperation::SendInput | WorkerOperation::Notify => { - definition::(operation, self.control.clone()) - } - WorkerOperation::Cancel | WorkerOperation::Stop => { - definition::(operation, self.control.clone()) - } - WorkerOperation::Restore => { - definition::(operation, self.control.clone()) - } - WorkerOperation::Remove => { - definition::(operation, self.control.clone()) - } - }; - context - .tools() - .register(ToolContribution::new(operation.tool_name(), definition))?; + context.tools().register(worker_tool_contribution( + operation, + self.control.clone(), + true, + ))?; } Ok(()) } @@ -558,6 +575,34 @@ struct WorkerTargetInput { subject: WorkerSubjectInput, } +#[derive(Debug, Deserialize, JsonSchema)] +#[serde(tag = "kind", rename_all = "snake_case")] +enum DirectSubWorkerSubjectInput { + SubWorker { name: String }, +} + +impl From for WorkerSubjectInput { + fn from(subject: DirectSubWorkerSubjectInput) -> Self { + match subject { + DirectSubWorkerSubjectInput::SubWorker { name } => Self::SubWorker { name }, + } + } +} + +#[derive(Debug, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +struct DirectSubWorkerMessageInput { + subject: DirectSubWorkerSubjectInput, + content: String, +} + +#[derive(Debug, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +struct DirectSubWorkerStopInput { + subject: DirectSubWorkerSubjectInput, + reason: Option, +} + #[derive(Debug, Deserialize, JsonSchema)] #[serde(deny_unknown_fields)] struct WorkerMessageInput { @@ -583,6 +628,7 @@ struct WorkerRemoveInput { struct WorkspaceWorkerTool { operation: WorkerOperation, control: Arc, + runtime_worker_control: bool, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -654,14 +700,20 @@ impl Tool for WorkspaceWorkerTool { let response = match self.operation { WorkerOperation::List => { parse::(input_json, "WorkerList")?; - let response = self - .control - .execute_runtime(WorkspaceRequest::get(format!( - "/api/w/{}/worker-control/workers", - self.control.workspace_id() - ))) - .await - .map_err(control_tool_error)?; + let response = if self.runtime_worker_control { + self.control + .execute_runtime(WorkspaceRequest::get(format!( + "/api/w/{}/worker-control/workers", + self.control.workspace_id() + ))) + .await + .map_err(control_tool_error)? + } else { + WorkspaceResponse { + status: 200, + body: r#"{"items":[]}"#.to_string(), + } + }; self.with_subworkers(response)? } WorkerOperation::Spawn => { @@ -701,7 +753,18 @@ impl Tool for WorkspaceWorkerTool { .map_err(control_tool_error)? } WorkerOperation::SendInput | WorkerOperation::Notify => { - let input = parse::(input_json, self.operation.tool_name())?; + let input = if self.runtime_worker_control { + parse::(input_json, self.operation.tool_name())? + } else { + let input = parse::( + input_json, + self.operation.tool_name(), + )?; + WorkerMessageInput { + subject: input.subject.into(), + content: input.content, + } + }; let content = non_empty(input.content, "content")?; if content.len() > 16 * 1024 { return Err(ToolError::ExecutionFailed( @@ -738,7 +801,16 @@ impl Tool for WorkspaceWorkerTool { } } WorkerOperation::Cancel | WorkerOperation::Stop => { - let input = parse::(input_json, self.operation.tool_name())?; + let input = if self.runtime_worker_control { + parse::(input_json, self.operation.tool_name())? + } else { + let input = + parse::(input_json, self.operation.tool_name())?; + WorkerStopInput { + subject: input.subject.into(), + reason: input.reason, + } + }; match input.subject { WorkerSubjectInput::SubWorker { name } => { if self.operation != WorkerOperation::Stop { @@ -911,7 +983,7 @@ fn render_subworker_stop_summary(summary: &SubWorkerStopSummary) -> String { .as_ref() .map(|stat| format!("+{}/-{} Changes · ", stat.added, stat.deleted)) .unwrap_or_default(); - format!("SubWorkerStop - done\n {tools}\n {changes}{elapsed}",) + format!("WorkerStop - done\n {tools}\n {changes}{elapsed}",) } fn format_elapsed(elapsed_ms: u64) -> String { @@ -925,9 +997,62 @@ 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::Stop => { + "Stop a direct SubWorker and release all authority delegated to its child session." + } + _ => unreachable!("unsupported direct SubWorker operation"), + }; + ToolDeclaration::new(operation.tool_name(), description) +} + +fn worker_tool_declaration(operation: WorkerOperation) -> ToolDeclaration { + ToolDeclaration::new(operation.tool_name(), operation.description()) +} + +fn worker_tool_contribution( + operation: WorkerOperation, + control: Arc, + runtime_worker_control: bool, +) -> ToolContribution { + let definition = match operation { + WorkerOperation::List => { + definition::(operation, control, runtime_worker_control) + } + WorkerOperation::Spawn => { + definition::(operation, control, runtime_worker_control) + } + WorkerOperation::SendInput | WorkerOperation::Notify => { + if runtime_worker_control { + definition::(operation, control, true) + } else { + definition::(operation, control, false) + } + } + WorkerOperation::Cancel | WorkerOperation::Stop => { + if runtime_worker_control { + definition::(operation, control, true) + } else { + definition::(operation, control, false) + } + } + WorkerOperation::Restore => { + definition::(operation, control, runtime_worker_control) + } + WorkerOperation::Remove => { + definition::(operation, control, runtime_worker_control) + } + }; + ToolContribution::new(operation.tool_name(), definition) +} + fn definition( operation: WorkerOperation, control: Arc, + runtime_worker_control: bool, ) -> ToolDefinition { Arc::new(move || { let schema = schemars::schema_for!(I); @@ -938,6 +1063,7 @@ fn definition( let tool: Arc = Arc::new(WorkspaceWorkerTool { operation, control: control.clone(), + runtime_worker_control, }); (meta, tool) }) @@ -1036,11 +1162,101 @@ mod tests { } } + #[derive(Debug, Default)] + struct RecordingSubWorkerControl { + sent: Mutex>, + stopped: Mutex>, + } + + #[async_trait] + impl WorkerLifecycleService for RecordingSubWorkerControl { + async fn spawn( + &self, + _request: WorkerLifecycleSpawnRequest, + ) -> Result { + Err(WorkspaceClientError::Unavailable( + "Runtime Worker control is disabled in this test".to_string(), + )) + } + } + + #[async_trait] + impl WorkerControlService for RecordingSubWorkerControl { + fn workspace_id(&self) -> &str { + "workspace-test" + } + + fn known_subworkers(&self) -> Vec { + Vec::new() + } + + async fn spawn_worker( + &self, + _request: WorkerLifecycleSpawnRequest, + ) -> Result { + Err(WorkspaceClientError::Unavailable( + "Runtime Worker control is disabled in this test".to_string(), + )) + } + + fn remove_runtime_worker( + &self, + _runtime_id: &str, + _worker_id: &str, + _reason: &str, + ) -> Result { + Err(WorkspaceClientError::Unavailable( + "Runtime Worker control is disabled in this test".to_string(), + )) + } + + async fn execute_runtime( + &self, + _request: WorkspaceRequest, + ) -> Result { + Err(WorkspaceClientError::Unavailable( + "Runtime Worker control is disabled in this test".to_string(), + )) + } + + async fn ensure_permission( + &self, + _subject: &crate::feature::builtin::WorkerObservationSubjectRef, + _permission: &str, + ) -> Result<(), WorkspaceClientError> { + Ok(()) + } + + async fn send_subworker( + &self, + name: &str, + content: String, + ) -> Result { + self.sent.lock().unwrap().push((name.to_string(), content)); + Ok(WorkspaceResponse { + status: 200, + body: serde_json::json!({ "status": "accepted" }).to_string(), + }) + } + + async fn stop_subworker( + &self, + name: &str, + ) -> Result { + self.stopped.lock().unwrap().push(name.to_string()); + Ok(WorkspaceResponse { + status: 200, + body: serde_json::json!({ "status": "stopped" }).to_string(), + }) + } + } + fn test_control(client: Arc) -> Arc { Arc::new(WorkspaceWorkerControlService { client, workspace_id: "workspace%2Ftest".to_string(), registry: None, + runtime_worker_control: true, }) } @@ -1050,6 +1266,7 @@ mod tests { let tool = WorkspaceWorkerTool { operation: WorkerOperation::Spawn, control: test_control(client.clone()), + runtime_worker_control: true, }; tool.execute( &serde_json::json!({ @@ -1128,6 +1345,123 @@ mod tests { assert!(report.services.providers().is_empty()); } + #[tokio::test] + async fn sub_worker_control_surface_lists_registry_without_workspace_authority() { + let client = Arc::new(RecordingWorkspaceClient::default()); + let runtime_base = tempfile::tempdir().unwrap(); + let runtime_dir = Arc::new( + crate::runtime::dir::RuntimeDir::create(runtime_base.path(), "sub-worker-control") + .await + .unwrap(), + ); + let registry = SpawnedWorkerRegistry::new(runtime_dir); + let control: Arc = Arc::new(WorkspaceWorkerControlService { + client: client.clone(), + workspace_id: "workspace%2Ftest".to_string(), + registry: Some(registry), + runtime_worker_control: false, + }); + let tool = WorkspaceWorkerTool { + operation: WorkerOperation::List, + control, + runtime_worker_control: false, + }; + + let output = tool + .execute( + "{}", + ToolExecutionContext::new("call-list", "batch-list", 0), + ) + .await + .unwrap(); + + assert!(client.requests.lock().unwrap().is_empty()); + let value: serde_json::Value = + serde_json::from_str(output.content.as_deref().unwrap()).unwrap(); + assert_eq!(value["items"], serde_json::json!([])); + } + + #[tokio::test] + async fn canonical_send_and_stop_tools_route_to_direct_subworker_control() { + let control = Arc::new(RecordingSubWorkerControl::default()); + let send_tool = WorkspaceWorkerTool { + operation: WorkerOperation::SendInput, + control: control.clone(), + runtime_worker_control: false, + }; + let stop_tool = WorkspaceWorkerTool { + operation: WorkerOperation::Stop, + control: control.clone(), + runtime_worker_control: false, + }; + + send_tool + .execute( + r#"{"subject":{"kind":"sub_worker","name":"reviewer"},"content":"continue"}"#, + ToolExecutionContext::new("call-send", "batch-control", 0), + ) + .await + .unwrap(); + stop_tool + .execute( + r#"{"subject":{"kind":"sub_worker","name":"reviewer"},"reason":"review completed"}"#, + ToolExecutionContext::new("call-stop", "batch-control", 1), + ) + .await + .unwrap(); + + assert_eq!( + *control.sent.lock().unwrap(), + vec![("reviewer".to_string(), "continue".to_string())] + ); + assert_eq!( + *control.stopped.lock().unwrap(), + vec!["reviewer".to_string()] + ); + } + + #[tokio::test] + async fn sub_worker_control_surface_rejects_runtime_subject_without_workspace_call() { + let client = Arc::new(RecordingWorkspaceClient::default()); + let control: Arc = Arc::new(WorkspaceWorkerControlService { + client: client.clone(), + workspace_id: "workspace%2Ftest".to_string(), + registry: None, + runtime_worker_control: false, + }); + let permission_error = control + .ensure_permission( + &crate::feature::builtin::WorkerObservationSubjectRef::RuntimeWorker { + runtime_id: "runtime-1".to_string(), + worker_id: "worker-1".to_string(), + }, + "stop", + ) + .await + .unwrap_err(); + assert!(matches!( + permission_error, + WorkspaceClientError::Unavailable(_) + )); + + let tool = WorkspaceWorkerTool { + operation: WorkerOperation::Stop, + control, + runtime_worker_control: false, + }; + + let error = tool + .execute( + r#"{"subject":{"kind":"runtime_worker","runtime_id":"runtime-1","worker_id":"worker-1"},"reason":"not authorized"}"#, + ToolExecutionContext::new("call-stop", "batch-stop", 0), + ) + .await + .unwrap_err(); + + assert!(matches!(error, ToolError::InvalidArgument(_))); + assert!(client.requests.lock().unwrap().is_empty()); + } + #[test] fn worker_service_can_remain_enabled_without_direct_spawn_surface() { let client = Arc::new(RecordingWorkspaceClient::default()); @@ -1146,7 +1480,7 @@ mod tests { } #[test] - fn worker_tool_family_is_distinct_from_sub_worker_tools() { + fn canonical_worker_tool_family_uses_one_subject_based_namespace() { assert_eq!( WorkerOperation::ALL.map(WorkerOperation::tool_name), [ @@ -1254,6 +1588,7 @@ mod tests { WorkspaceWorkerTool { operation, control: test_control(client.clone()), + runtime_worker_control: true, } .execute( &args.to_string(), @@ -1278,6 +1613,7 @@ mod tests { let tool = WorkspaceWorkerTool { operation: WorkerOperation::Remove, control: test_control(client.clone()), + runtime_worker_control: true, }; tool.execute( &serde_json::json!({ @@ -1327,6 +1663,7 @@ mod tests { let tool = WorkspaceWorkerTool { operation: WorkerOperation::Remove, control: test_control(client.clone()), + runtime_worker_control: true, }; for reason in [" ".to_string(), "x".repeat(513)] { let _error = tool @@ -1397,7 +1734,7 @@ mod tests { assert_eq!( output.summary, - "SubWorkerStop - done\n 26 Read, 5 Grep\n +215/-148 Changes · 1m 18s" + "WorkerStop - done\n 26 Read, 5 Grep\n +215/-148 Changes · 1m 18s" ); assert_eq!( serde_json::from_str::(output.content.as_deref().unwrap()) diff --git a/crates/worker/src/prompt/catalog.rs b/crates/worker/src/prompt/catalog.rs index 87986cff..fc3c9dc1 100644 --- a/crates/worker/src/prompt/catalog.rs +++ b/crates/worker/src/prompt/catalog.rs @@ -901,6 +901,23 @@ mod tests { .unwrap() .contains("BOUNDARY_MARKER") ); - catalog.worker_orchestration_guidance_section().unwrap(); + let orchestration = catalog.worker_orchestration_guidance_section().unwrap(); + for name in [ + "SubWorkerSpawn", + "WorkerList", + "WorkerSendInput", + "WorkerStop", + ] { + assert!( + orchestration.contains(name), + "missing canonical tool {name}" + ); + } + for alias in ["SubWorkerList", "SubWorkerSend", "SubWorkerStop"] { + assert!( + !orchestration.contains(alias), + "guidance referenced stale alias {alias}" + ); + } } } diff --git a/crates/worker/src/prompt/system.rs b/crates/worker/src/prompt/system.rs index 8dd78771..ffd9da5a 100644 --- a/crates/worker/src/prompt/system.rs +++ b/crates/worker/src/prompt/system.rs @@ -181,12 +181,19 @@ impl ToolCapabilities { "MemoryReadDocument" => capabilities.memory_read_document = true, "MemoryUpdateDocument" => capabilities.memory_update_document = true, "SubWorkerSpawn" => capabilities.sub_worker_spawn = true, - "SubWorkerSend" => capabilities.sub_worker_send = true, - "SubWorkerStop" => capabilities.sub_worker_stop = true, - "SubWorkerList" => capabilities.sub_worker_list = true, _ => {} } } + if capabilities.sub_worker_spawn { + for name in names { + match name.as_str() { + "WorkerSendInput" => capabilities.sub_worker_send = true, + "WorkerStop" => capabilities.sub_worker_stop = true, + "WorkerList" => capabilities.sub_worker_list = true, + _ => {} + } + } + } capabilities } @@ -316,6 +323,35 @@ fn append_trailing_section( mod tests { use super::*; + #[test] + fn sub_worker_capabilities_follow_the_registered_canonical_control_tools() { + let names = [ + "SubWorkerSpawn", + "WorkerList", + "WorkerSendInput", + "WorkerStop", + ] + .map(str::to_string); + let capabilities = ToolCapabilities::from_tool_names(&names); + assert!(capabilities.sub_worker_management()); + assert!(capabilities.sub_worker_list); + assert!(capabilities.sub_worker_send); + assert!(capabilities.sub_worker_stop); + + let stale_aliases = [ + "SubWorkerSpawn", + "SubWorkerList", + "SubWorkerSend", + "SubWorkerStop", + ] + .map(str::to_string); + let capabilities = ToolCapabilities::from_tool_names(&stale_aliases); + assert!(capabilities.sub_worker_spawn); + assert!(!capabilities.sub_worker_list); + assert!(!capabilities.sub_worker_send); + assert!(!capabilities.sub_worker_stop); + } + #[test] fn rejects_legacy_prefix_relative_and_missing_names() { for reference in ["legacy/custom", "custom.md", "../custom", "missing"] { diff --git a/crates/worker/src/spawn/comm_tools.rs b/crates/worker/src/spawn/comm_tools.rs index 49f9ab78..27e6845f 100644 --- a/crates/worker/src/spawn/comm_tools.rs +++ b/crates/worker/src/spawn/comm_tools.rs @@ -1,226 +1,19 @@ -#![cfg_attr(not(test), allow(dead_code, unused_imports))] - -//! Parent-facing tools for in-process Internal SubWorker sessions. +//! Socket communication retained for the legacy top-level Worker callback protocol. //! -//! Legacy direct-child tool constructors are test-only; production exposes the -//! registry through the unified `worker.control` service and Worker tools. -//! There is no Runtime catalog lookup or child socket transport, so a Worker can operate only on -//! its direct Internal children. The socket helper at the bottom remains solely for the legacy -//! top-level Worker callback protocol and is not part of SubWorker communication. +//! Direct Internal SubWorker lifecycle is exposed through `worker.control` and the +//! canonical Worker tools, not a second SubWorker-specific tool family. use std::path::Path; -use std::sync::Arc; use std::time::Duration; -use agen::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput}; -use async_trait::async_trait; use protocol::stream::{JsonLineReader, JsonLineWriter}; use protocol::{Event, Method}; -use serde::{Deserialize, Serialize}; use tokio::net::UnixStream; -use crate::spawn::registry::SpawnedWorkerRegistry; - /// Timeout applied to each socket-level operation — connect, write, /// read. Kept short so a stuck child doesn't block the spawner's turn. const SOCKET_OP_TIMEOUT: Duration = Duration::from_secs(5); -// --------------------------------------------------------------------------- -// Shared input types -// --------------------------------------------------------------------------- - -#[derive(Debug, Deserialize, schemars::JsonSchema)] -struct NameInput { - /// Name of a previously spawned SubWorker. - name: String, -} - -#[derive(Debug, Deserialize, schemars::JsonSchema)] -#[serde(deny_unknown_fields)] -struct SubWorkerListInput {} - -#[derive(Debug, Serialize)] -struct SubWorkerListItem { - name: String, -} - -struct SubWorkerListTool { - registry: Arc, -} - -#[async_trait] -impl Tool for SubWorkerListTool { - async fn execute( - &self, - input_json: &str, - _ctx: agen::tool::ToolExecutionContext, - ) -> Result { - let _input: SubWorkerListInput = serde_json::from_str(input_json).map_err(|error| { - ToolError::InvalidArgument(format!("invalid SubWorkerList input: {error}")) - })?; - let items = self - .registry - .list_internal() - .into_iter() - .map(|record| SubWorkerListItem { - name: record.worker_name, - }) - .collect::>(); - let count = items.len(); - let content = serde_json::to_string_pretty(&serde_json::json!({ "sub_workers": items })) - .map_err(|error| ToolError::ExecutionFailed(error.to_string()))?; - Ok(ToolOutput { - summary: format!("listed {count} child SubWorker(s)"), - content: Some(content), - attachments: Vec::new(), - }) - } -} - -#[cfg(test)] -pub fn sub_worker_list_tool(registry: Arc) -> ToolDefinition { - Arc::new(move || { - let schema = schemars::schema_for!(SubWorkerListInput); - let schema_value = serde_json::to_value(schema).unwrap_or(serde_json::json!({})); - let meta = ToolMeta::new("SubWorkerList") - .description("List child SubWorkers owned by this Worker. Peer Workers and general Runtime Workers are excluded.") - .input_schema(schema_value); - let tool: Arc = Arc::new(SubWorkerListTool { - registry: registry.clone(), - }); - (meta, tool) - }) -} - -// --------------------------------------------------------------------------- -// SubWorkerSend -// --------------------------------------------------------------------------- - -const SEND_TO_POD_DESCRIPTION: &str = "Send a text message to a previously spawned SubWorker. The SubWorker \ -processes it as a user turn. Fails if the SubWorker is already executing a \ -turn — retry after it finishes. Does not wait for the turn to complete; \ -use worker-observation tools to inspect its committed session."; - -#[derive(Debug, Deserialize, schemars::JsonSchema)] -struct SubWorkerSendInput { - /// Target SubWorker name. - name: String, - /// Text delivered to the SubWorker as the next user message. - message: String, -} - -struct SubWorkerSendTool { - registry: Arc, -} - -#[async_trait] -impl Tool for SubWorkerSendTool { - async fn execute( - &self, - input_json: &str, - _ctx: agen::tool::ToolExecutionContext, - ) -> Result { - let input: SubWorkerSendInput = serde_json::from_str(input_json) - .map_err(|e| ToolError::InvalidArgument(format!("invalid SubWorkerSend input: {e}")))?; - if let Some(record) = self.registry.get_internal(&input.name) { - record.session.send(input.message).await.map_err(|error| { - ToolError::ExecutionFailed(format!("send to `{}`: {error}", input.name)) - })?; - return Ok(ToolOutput { - summary: format!("sent message to `{}`", input.name), - content: None, - attachments: Vec::new(), - }); - } - Err(unknown_worker_err(&input.name)) - } -} - -#[cfg(test)] -pub fn sub_worker_send_tool(registry: Arc) -> ToolDefinition { - Arc::new(move || { - let schema = schemars::schema_for!(SubWorkerSendInput); - let schema_value = serde_json::to_value(schema).unwrap_or(serde_json::json!({})); - let meta = ToolMeta::new("SubWorkerSend") - .description(SEND_TO_POD_DESCRIPTION) - .input_schema(schema_value); - let tool: Arc = Arc::new(SubWorkerSendTool { - registry: registry.clone(), - }); - (meta, tool) - }) -} - -// --------------------------------------------------------------------------- -// SubWorkerStop -// --------------------------------------------------------------------------- - -const STOP_POD_DESCRIPTION: &str = "Cancel and stop a spawned Internal SubWorker session, remove it from the parent's direct-child registry, and reclaim delegated Write scope."; - -struct SubWorkerStopTool { - registry: Arc, -} - -#[async_trait] -impl Tool for SubWorkerStopTool { - async fn execute( - &self, - input_json: &str, - _ctx: agen::tool::ToolExecutionContext, - ) -> Result { - let input: NameInput = serde_json::from_str(input_json) - .map_err(|e| ToolError::InvalidArgument(format!("invalid SubWorkerStop input: {e}")))?; - if let Some(summary) = self - .registry - .remove_internal(&input.name) - .await - .map_err(|error| ToolError::ExecutionFailed(error.to_string()))? - { - return Ok(ToolOutput { - summary: format!( - "SubWorkerStop - done\n {} tool kind{}\n {}ms", - summary.tool_counts.len(), - if summary.tool_counts.len() == 1 { - "" - } else { - "s" - }, - summary.elapsed_ms, - ), - content: Some( - serde_json::to_string(&summary) - .map_err(|error| ToolError::ExecutionFailed(error.to_string()))?, - ), - attachments: Vec::new(), - }); - } - Err(unknown_worker_err(&input.name)) - } -} - -#[cfg(test)] -pub fn sub_worker_stop_tool(registry: Arc) -> ToolDefinition { - Arc::new(move || { - let schema = schemars::schema_for!(NameInput); - let schema_value = serde_json::to_value(schema).unwrap_or(serde_json::json!({})); - let meta = ToolMeta::new("SubWorkerStop") - .description(STOP_POD_DESCRIPTION) - .input_schema(schema_value); - let tool: Arc = Arc::new(SubWorkerStopTool { - registry: registry.clone(), - }); - (meta, tool) - }) -} - -// --------------------------------------------------------------------------- -// Helpers -// --------------------------------------------------------------------------- - -fn unknown_worker_err(name: &str) -> ToolError { - ToolError::InvalidArgument(format!("no spawned worker named `{name}`")) -} - /// Connect with a timeout, drain the server's connect-time snapshot, /// write one `Method` line, flush, and close. /// diff --git a/crates/worker/src/spawn/tool.rs b/crates/worker/src/spawn/tool.rs index 8e4843d7..9cd314dd 100644 --- a/crates/worker/src/spawn/tool.rs +++ b/crates/worker/src/spawn/tool.rs @@ -1226,14 +1226,11 @@ extract_threshold = 4000 Err(tokio::sync::mpsc::error::TryRecvError::Empty) )); - let context = agen::tool::ToolExecutionContext::direct(); - let list = (crate::spawn::comm_tools::sub_worker_list_tool(registry.clone()))().1; - let listed = list.execute("{}", context.clone()).await.unwrap(); assert!( - listed - .content - .unwrap_or_default() - .contains("reviewer-child") + registry + .list_internal() + .iter() + .any(|record| record.worker_name == "reviewer-child") ); let observation = @@ -1256,13 +1253,11 @@ extract_threshold = 4000 .contains("reviewed") ); - let send = (crate::spawn::comm_tools::sub_worker_send_tool(registry.clone()))().1; - send.execute( - r#"{"name":"reviewer-child","message":"review follow-up"}"#, - context.clone(), - ) - .await - .unwrap(); + record + .session + .send("review follow-up".to_string()) + .await + .unwrap(); assert_eq!( record.session.wait_until_idle().await, crate::internal_worker::InternalWorkerSessionStatus::Idle @@ -1277,12 +1272,11 @@ extract_threshold = 4000 assert!(latest_capture.session.entries.len() > first_capture.session.entries.len()); fail_requests.store(true, Ordering::SeqCst); - send.execute( - r#"{"name":"reviewer-child","message":"trigger terminal failure"}"#, - context.clone(), - ) - .await - .unwrap(); + record + .session + .send("trigger terminal failure".to_string()) + .await + .unwrap(); assert_eq!( record.session.wait_until_idle().await, InternalWorkerSessionStatus::Stopped @@ -1298,10 +1292,13 @@ extract_threshold = 4000 ); assert!(registry.get_internal("reviewer-child").is_some()); - let stop = (crate::spawn::comm_tools::sub_worker_stop_tool(registry.clone()))().1; - stop.execute(r#"{"name":"reviewer-child"}"#, context) - .await - .unwrap(); + assert!( + registry + .remove_internal("reviewer-child") + .await + .unwrap() + .is_some() + ); assert!(registry.get_internal("reviewer-child").is_none()); assert!(spawner_scope.snapshot().is_writable(&workspace_root)); @@ -1315,9 +1312,6 @@ extract_threshold = 4000 .await .unwrap(); assert!(spawner_scope.snapshot().is_writable(&workspace_root)); - drop(list); - drop(send); - drop(stop); drop(observation); drop(tool); drop(registry); diff --git a/crates/worker/tests/controller_test.rs b/crates/worker/tests/controller_test.rs index 92191f3e..9a984657 100644 --- a/crates/worker/tests/controller_test.rs +++ b/crates/worker/tests/controller_test.rs @@ -18,7 +18,8 @@ use workdir::{ use worker::{ Event, Method, Worker, WorkerController, WorkerFilesystemAuthority, WorkerHandle, - WorkerManifest, WorkerStatus, WorkerWorkspaceContext, + WorkerManifest, WorkerStatus, WorkerWorkspaceContext, WorkspaceClient, WorkspaceClientError, + WorkspaceRequest, WorkspaceResponse, }; type TestStore = CombinedStore; @@ -179,9 +180,48 @@ async fn make_worker_with_pwd( make_worker_with_pwd_and_manifest(client, MANIFEST_TOML).await } +#[derive(Debug)] +struct NoopWorkspaceClient; + +impl WorkspaceClient for NoopWorkspaceClient { + fn workspace_id(&self) -> Option<&str> { + Some("workspace-test") + } + + fn kind(&self) -> &str { + "test-noop" + } + + fn is_available(&self) -> bool { + true + } + + fn execute( + &self, + _request: WorkspaceRequest, + ) -> Result { + Err(WorkspaceClientError::Unavailable( + "test client does not execute requests".to_string(), + )) + } +} + async fn make_worker_with_pwd_and_manifest( client: MockClient, manifest_toml: &str, +) -> (Worker, std::path::PathBuf) { + make_worker_with_pwd_manifest_and_workspace_context( + client, + manifest_toml, + WorkerWorkspaceContext::local_filesystem(None), + ) + .await +} + +async fn make_worker_with_pwd_manifest_and_workspace_context( + client: MockClient, + manifest_toml: &str, + workspace_context: WorkerWorkspaceContext, ) -> (Worker, std::path::PathBuf) { let manifest = WorkerManifest::from_toml(manifest_toml).unwrap(); let store_tmp = tempfile::tempdir().unwrap(); @@ -202,16 +242,9 @@ async fn make_worker_with_pwd_and_manifest( let worker = Engine::<_, agen::state::Mutable, worker::SessionHistoryMetadata>::new_annotated(client); let authority = WorkerFilesystemAuthority::local(pwd.clone(), pwd.clone()); - let worker = Worker::new( - manifest, - worker, - store, - WorkerWorkspaceContext::local_filesystem(None), - authority, - scope, - ) - .await - .unwrap(); + let worker = Worker::new(manifest, worker, store, workspace_context, authority, scope) + .await + .unwrap(); (worker, pwd) } @@ -663,6 +696,83 @@ permission = "write" "{} role SubWorker tool exposure mismatch: {names:?}", case.role ); + for control_tool in ["WorkerList", "WorkerSendInput", "WorkerStop"] { + assert_eq!( + names.iter().any(|name| name == control_tool), + case.sub_worker_enabled, + "{} role {control_tool} exposure mismatch: {names:?}", + case.role + ); + } + for stale_alias in ["SubWorkerList", "SubWorkerSend", "SubWorkerStop"] { + assert!( + !names.iter().any(|name| name == stale_alias), + "{} role exposed stale alias {stale_alias}: {names:?}", + case.role + ); + } + } +} + +#[tokio::test] +async fn worker_and_sub_worker_features_install_one_canonical_control_surface() { + let manifest = r#" +[worker] +name = "combined-worker-control-feature-test" +pwd = "./" + +[model] +scheme = "anthropic" +model_id = "test-model" + +[engine] +max_tokens = 100 + +[feature.worker] +enabled = true +direct_spawn = false + +[feature.sub_worker] +enabled = true + +[[scope.allow]] +target = "./" +permission = "write" + +[[delegation_scope.allow]] +target = "/tmp" +permission = "write" +"#; + let client = MockClient::new(simple_text_events()); + let client_for_assert = client.clone(); + let worker = make_worker_with_pwd_manifest_and_workspace_context( + client, + manifest, + WorkerWorkspaceContext::with_client(None, Arc::new(NoopWorkspaceClient)), + ) + .await + .0; + let handle = spawn_controller(worker).await; + + handle.send(Method::run_text("Hello")).await.unwrap(); + wait_for_status(&handle, WorkerStatus::Idle).await; + + let request = wait_for_captured_request(&client_for_assert).await; + let names = request_tool_names(&request); + assert!(names.iter().any(|name| name == "SubWorkerSpawn")); + assert!(!names.iter().any(|name| name == "WorkerSpawn")); + for control_tool in ["WorkerList", "WorkerSendInput", "WorkerStop"] { + assert_eq!( + names + .iter() + .filter(|name| name.as_str() == control_tool) + .count(), + 1, + "expected one {control_tool} contribution: {names:?}" + ); + } + for stale_alias in ["SubWorkerList", "SubWorkerSend", "SubWorkerStop"] { + assert!(!names.iter().any(|name| name == stale_alias)); } } diff --git a/docs/design/session-observation.md b/docs/design/session-observation.md index ae7eb9fa..24ce5dc0 100644 --- a/docs/design/session-observation.md +++ b/docs/design/session-observation.md @@ -44,4 +44,4 @@ Observation is read-only evidence access. It does not authorize Ticket, Memory, ## SubWorker output -SubWorkers no longer expose a separate output cursor tool. `SubWorkerList`, `SubWorkerSend`, and `SubWorkerStop` retain parent-owned lifecycle control, while committed child output is read through `worker-observation`. Turn-completion notifications carry no transcript and only tell the parent to inspect the authoritative committed session at a natural boundary. +SubWorkers no longer expose a separate output cursor or lifecycle tool family. `WorkerList`, `WorkerSendInput`, and `WorkerStop` retain parent-owned lifecycle control through a `{ kind: "sub_worker", name }` subject, while committed child output is read through `worker-observation`. Turn-completion notifications carry no transcript and only tell the parent to inspect the authoritative committed session at a natural boundary. diff --git a/resources/prompts/common/worker-orchestration.md b/resources/prompts/common/worker-orchestration.md index 6bf73286..c1f815d1 100644 --- a/resources/prompts/common/worker-orchestration.md +++ b/resources/prompts/common/worker-orchestration.md @@ -2,7 +2,7 @@ --- ## SubWorker orchestration -When SubWorker-management tools are available, SubWorker notifications are background signals for the parent Worker to handle at a natural stopping point. Do not ignore routine follow-up, but do not interrupt the current user request unnecessarily. +When SubWorker-management tools are available, create direct children with `SubWorkerSpawn`, discover them with `WorkerList`, continue them with `WorkerSendInput`, and release their delegated authority with `WorkerStop`. Pass the exact `{ kind: "sub_worker", name }` subject returned by `WorkerList`; do not invent direct-only aliases. SubWorker notifications are background signals for the parent Worker to handle at a natural stopping point. Do not ignore routine follow-up, but do not interrupt the current user request unnecessarily. The parent Worker does not need to keep a turn open or call tools solely to wait for a notification. Do not use `sleep` or polling loops just to wait for SubWorker output; if there is no useful immediate work, return control and handle the SubWorker when notified or when the user next asks.