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

worker: separate subworkers from workspace workers

Browse Source
This commit is contained in:
Hare
2026-08-05 03:36:59 +09:00
parent 50726e4cf3
commit ba009c0a20
35 changed files with 747 additions and 261 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker/src/controller.rs
+30 -25
View File
@@ -8,9 +8,7 @@ use session_store::WorkerMetadataStore;
use session_store::{LogEntry, Store};
use tokio::sync::{broadcast, mpsc, oneshot};
use crate::discovery::{
WorkerDiscovery, list_workers_tool, restore_worker_tool, send_to_peer_worker_tool,
};
use crate::discovery::WorkerDiscovery;
use crate::feature::FeatureRegistryBuilder;
use crate::in_flight::{InFlightEvents, snapshot_from_guard};
use crate::ipc::alerter::Alerter;
@@ -23,9 +21,11 @@ use crate::shutdown_after_idle::{
ShutdownAfterIdleRequest, TicketIntakeReadyShutdownHook, is_ticket_intake_role,
take_shutdown_request_after_status,
};
use crate::spawn::comm_tools::{read_worker_output_tool, send_to_worker_tool, stop_worker_tool};
use crate::spawn::comm_tools::{
sub_worker_list_tool, sub_worker_read_output_tool, sub_worker_send_tool, sub_worker_stop_tool,
};
use crate::spawn::registry::SpawnedWorkerRegistry;
use crate::spawn::tool::spawn_worker_tool;
use crate::spawn::tool::sub_worker_spawn_tool;
use crate::worker::{SystemItemCommitter, Worker, WorkerError, WorkerRunResult};
use protocol::{
AlertLevel, AlertSource, ErrorCode, Event, Method, RewindTargetId, RunResult, Segment,
@@ -582,7 +582,7 @@ fn wire_event_bridges_on_engine<C, St>(
}
/// Register the builtin file-manipulation tools, optional memory tools,
/// and the Worker-orchestration tools (SpawnWorker + comm) on the Worker's
/// and the Worker-orchestration tools (SubWorkerSpawn + comm) on the Worker's
/// Engine. Returns the WorkdirSession handle used to attach a `WorkerFsView` to
/// the shared state.
async fn register_worker_tools<C, St>(
@@ -610,7 +610,6 @@ where
let spawner_name = worker.manifest().worker.name.clone();
let spawner_manifest = worker.manifest().clone();
let prompts = worker.prompts().clone();
let worker_metadata_store = worker.store().clone();
let self_parent_socket = worker.callback_socket().cloned();
// Resolve the existing Worker–Workdir binding into the domain provider.
@@ -684,6 +683,21 @@ where
crate::feature::builtin::manage_workdir::manage_workdir_feature(workspace_client),
);
}
if feature_config.worker.enabled {
let workspace_client = worker.workspace_client_handle();
let has_workspace_identity = workspace_client.workspace_id().is_some_and(|workspace_id| {
!workspace_id.is_empty() && !workspace_id.chars().any(char::is_control)
});
if !workspace_client.is_available() || !has_workspace_identity {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidInput,
"Worker tools require Backend Workspace API authority",
));
}
feature_registry.add_module(
crate::feature::builtin::manage_worker::manage_worker_feature(workspace_client),
);
}
for module in crate::feature::plugin::plugin_tool_features_if_enabled(
feature_config.plugins.enabled,
&worker.manifest().plugins,
@@ -698,7 +712,7 @@ where
}
}
if feature_config.workers.enabled {
if feature_config.sub_worker.enabled {
worker.register_worker_orchestration_instruction();
}
@@ -756,16 +770,16 @@ where
}
}
// Worker-orchestration tools (SpawnWorker + the four comm tools) share
// Worker-orchestration tools (SubWorkerSpawn + the four comm tools) share
// the Worker-scoped `SpawnedWorkerRegistry` (also consumed by the main
// loop's `WorkerEvent` handler). Expose them only behind the explicit
// profile feature and require delegation authority up front so enabling
// the surface cannot imply broad child scope by accident.
if feature_config.workers.enabled {
if feature_config.sub_worker.enabled {
if spawner_manifest.delegation_scope.allow.is_empty() {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidInput,
"[feature.workers].enabled = true requires non-empty [[delegation_scope.allow]]",
"[feature.sub_worker].enabled = true requires non-empty [[delegation_scope.allow]]",
));
}
let spawner_cwd = local_filesystem
@@ -783,7 +797,7 @@ where
"worker spawn tools require local Worker filesystem authority",
)
})?;
engine.register_tool(spawn_worker_tool(
engine.register_tool(sub_worker_spawn_tool(
spawner_name.clone(),
spawner_socket,
runtime_base.clone(),
@@ -795,19 +809,10 @@ where
scope_handle,
prompts,
));
engine.register_tool(send_to_worker_tool(spawned_registry.clone()));
engine.register_tool(read_worker_output_tool(spawned_registry.clone()));
engine.register_tool(stop_worker_tool(spawned_registry.clone()));
let discovery = WorkerDiscovery::new(
worker_metadata_store,
spawner_name,
runtime_base,
Some(spawner_cwd),
spawned_registry,
);
engine.register_tool(list_workers_tool(discovery.clone()));
engine.register_tool(restore_worker_tool(discovery.clone()));
engine.register_tool(send_to_peer_worker_tool(discovery));
engine.register_tool(sub_worker_list_tool(spawned_registry.clone()));
engine.register_tool(sub_worker_send_tool(spawned_registry.clone()));
engine.register_tool(sub_worker_read_output_tool(spawned_registry.clone()));
engine.register_tool(sub_worker_stop_tool(spawned_registry));
}
}
let _feature_install_report = worker.install_features(feature_registry);
crates/worker/src/entrypoint.rs
+1 -1
View File
@@ -56,7 +56,7 @@ struct Cli {
/// Claim a scope allocation pre-registered by a spawning Worker, rather
/// than installing a new top-level allocation. Used only when this
/// process is launched by `SpawnWorker`; end users should never pass it.
/// process is launched by `SubWorkerSpawn`; end users should never pass it.
#[arg(long)]
adopt: bool,
crates/worker/src/feature/builtin.rs
+1
View File
@@ -5,6 +5,7 @@
//! an external plugin-loading surface.
pub mod manage_workdir;
pub mod manage_worker;
pub mod memory;
pub mod objective;
pub mod session_explore;
crates/worker/src/feature/builtin/manage_worker.rs
+361
View File
@@ -0,0 +1,361 @@
//! Workspace-authority-backed Worker session management tools.
use std::sync::Arc;
use async_trait::async_trait;
use llm_engine::tool::{
Tool, ToolDefinition, ToolError, ToolExecutionContext, ToolMeta, ToolOutput,
};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use crate::feature::{
FeatureDescriptor, FeatureInstallContext, FeatureInstallError, FeatureModule, ToolContribution,
ToolDeclaration,
};
use crate::worker::{WorkspaceClient, WorkspaceRequest, WorkspaceRequestMethod};
const FEATURE_ID: &str = "worker";
const FEATURE_NAME: &str = "Worker";
const FEATURE_DESCRIPTION: &str =
"Workspace-authority tools for managing Workdir-bound Backend/Runtime Worker sessions.";
#[derive(Clone, Debug)]
pub struct ManageWorkerFeature {
client: Arc<dyn WorkspaceClient>,
}
pub fn manage_worker_feature(client: Arc<dyn WorkspaceClient>) -> ManageWorkerFeature {
ManageWorkerFeature { client }
}
impl FeatureModule for ManageWorkerFeature {
fn descriptor(&self) -> FeatureDescriptor {
let mut descriptor = FeatureDescriptor::builtin(FEATURE_ID, FEATURE_NAME)
.with_description(FEATURE_DESCRIPTION);
for operation in WorkerOperation::ALL {
descriptor = descriptor.with_tool(ToolDeclaration::new(
operation.tool_name(),
operation.description(),
));
}
descriptor
}
fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> {
let workspace_id = self
.client
.workspace_id()
.filter(|value| !value.trim().is_empty())
.ok_or_else(|| {
FeatureInstallError::InvalidDescriptor(
"worker feature requires a Workspace id".to_string(),
)
})?
.to_string();
for operation in WorkerOperation::ALL {
let definition = match operation {
WorkerOperation::List => definition::<WorkerListInput>(
operation,
self.client.clone(),
workspace_id.clone(),
),
WorkerOperation::Spawn => definition::<WorkerSpawnInput>(
operation,
self.client.clone(),
workspace_id.clone(),
),
WorkerOperation::Stop => definition::<WorkerStopInput>(
operation,
self.client.clone(),
workspace_id.clone(),
),
WorkerOperation::Restore => definition::<WorkerTargetInput>(
operation,
self.client.clone(),
workspace_id.clone(),
),
};
context
.tools()
.register(ToolContribution::new(operation.tool_name(), definition))?;
}
Ok(())
}
}
#[derive(Debug, Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
struct WorkerListInput {}
#[derive(Debug, Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
struct WorkerSpawnInput {
runtime_id: String,
working_directory_id: String,
profile: String,
#[serde(default)]
display_name: Option<String>,
#[serde(default)]
initial_text: Option<String>,
#[serde(default)]
relative_cwd: Option<String>,
}
#[derive(Debug, Serialize)]
struct WorkerSpawnRequest {
runtime_id: String,
display_name: String,
profile: String,
initial_text: String,
working_directory: WorkerWorkingDirectorySelection,
}
#[derive(Debug, Serialize)]
struct WorkerWorkingDirectorySelection {
working_directory_id: String,
#[serde(skip_serializing_if = "Option::is_none")]
relative_cwd: Option<String>,
}
#[derive(Debug, Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
struct WorkerTargetInput {
runtime_id: String,
worker_id: String,
}
#[derive(Debug, Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
struct WorkerStopInput {
runtime_id: String,
worker_id: String,
#[serde(default)]
reason: Option<String>,
}
struct WorkspaceWorkerTool {
operation: WorkerOperation,
client: Arc<dyn WorkspaceClient>,
workspace_id: String,
}
#[derive(Debug, Clone, Copy)]
enum WorkerOperation {
List,
Spawn,
Stop,
Restore,
}
impl WorkerOperation {
const ALL: [Self; 4] = [Self::List, Self::Spawn, Self::Stop, Self::Restore];
fn tool_name(self) -> &'static str {
match self {
Self::List => "WorkerList",
Self::Spawn => "WorkerSpawn",
Self::Stop => "WorkerStop",
Self::Restore => "WorkerRestore",
}
}
fn description(self) -> &'static str {
match self {
Self::List => {
"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."
}
Self::Stop => "Stop a Backend/Runtime Worker session in the current Workspace.",
Self::Restore => {
"Restore a stopped Backend/Runtime Worker session in the current Workspace."
}
}
}
}
#[async_trait]
impl Tool for WorkspaceWorkerTool {
async fn execute(
&self,
input_json: &str,
_ctx: ToolExecutionContext,
) -> Result<ToolOutput, ToolError> {
let request = match self.operation {
WorkerOperation::List => {
parse::<WorkerListInput>(input_json, "WorkerList")?;
WorkspaceRequest::get(format!("/api/w/{}/workers", self.workspace_id))
}
WorkerOperation::Spawn => {
let input = parse::<WorkerSpawnInput>(input_json, "WorkerSpawn")?;
let request = WorkerSpawnRequest {
runtime_id: authority_id(&input.runtime_id, "runtime_id")?,
display_name: input
.display_name
.filter(|value| !value.trim().is_empty())
.unwrap_or_else(|| "Workspace Worker".to_string()),
profile: non_empty(input.profile, "profile")?,
initial_text: input.initial_text.unwrap_or_default(),
working_directory: WorkerWorkingDirectorySelection {
working_directory_id: authority_id(
&input.working_directory_id,
"working_directory_id",
)?,
relative_cwd: input
.relative_cwd
.map(|value| validate_relative_cwd(&value))
.transpose()?,
},
};
WorkspaceRequest::json(
WorkspaceRequestMethod::Post,
format!("/api/w/{}/workers", self.workspace_id),
serde_json::to_string(&request)
.map_err(|error| ToolError::ExecutionFailed(error.to_string()))?,
)
}
WorkerOperation::Stop => {
let input = parse::<WorkerStopInput>(input_json, "WorkerStop")?;
let runtime_id = authority_id(&input.runtime_id, "runtime_id")?;
let worker_id = authority_id(&input.worker_id, "worker_id")?;
WorkspaceRequest::json(
WorkspaceRequestMethod::Post,
format!(
"/api/w/{}/runtimes/{runtime_id}/workers/{worker_id}/stop",
self.workspace_id
),
serde_json::json!({ "reason": input.reason }).to_string(),
)
}
WorkerOperation::Restore => {
let input = parse::<WorkerTargetInput>(input_json, "WorkerRestore")?;
let runtime_id = authority_id(&input.runtime_id, "runtime_id")?;
let worker_id = authority_id(&input.worker_id, "worker_id")?;
WorkspaceRequest::json(
WorkspaceRequestMethod::Post,
format!(
"/api/w/{}/runtimes/{runtime_id}/workers/{worker_id}/restore",
self.workspace_id
),
"{}",
)
}
};
let response = self
.client
.execute(request)
.map_err(|error| ToolError::ExecutionFailed(error.to_string()))?;
if !response.is_success() {
return Err(ToolError::ExecutionFailed(format!(
"Workspace Worker operation returned HTTP {}: {}",
response.status, response.body
)));
}
Ok(ToolOutput {
summary: format!("{} completed", self.operation.tool_name()),
content: Some(response.body),
})
}
}
fn definition<I: JsonSchema + 'static>(
operation: WorkerOperation,
client: Arc<dyn WorkspaceClient>,
workspace_id: String,
) -> ToolDefinition {
Arc::new(move || {
let schema = schemars::schema_for!(I);
let schema_value = serde_json::to_value(schema).unwrap_or(serde_json::json!({}));
let meta = ToolMeta::new(operation.tool_name())
.description(operation.description())
.input_schema(schema_value);
let tool: Arc<dyn Tool> = Arc::new(WorkspaceWorkerTool {
operation,
client: client.clone(),
workspace_id: workspace_id.clone(),
});
(meta, tool)
})
}
fn parse<T: for<'de> Deserialize<'de>>(input: &str, tool: &str) -> Result<T, ToolError> {
serde_json::from_str(input)
.map_err(|error| ToolError::InvalidArgument(format!("invalid {tool} input: {error}")))
}
fn authority_id(value: &str, field: &str) -> Result<String, ToolError> {
let value = non_empty(value.to_string(), field)?;
if value.contains('/') || value.contains('?') || value.contains('#') {
return Err(ToolError::InvalidArgument(format!(
"{field} must be an authority id, not a path or URL"
)));
}
Ok(value)
}
fn non_empty(value: String, field: &str) -> Result<String, ToolError> {
let value = value.trim().to_string();
if value.is_empty() {
return Err(ToolError::InvalidArgument(format!(
"{field} must not be empty"
)));
}
Ok(value)
}
fn validate_relative_cwd(value: &str) -> Result<String, ToolError> {
let value = value.trim();
if value.is_empty()
|| value.starts_with('/')
|| value.split('/').any(|part| matches!(part, "" | "." | ".."))
{
return Err(ToolError::InvalidArgument(
"relative_cwd must be a normalized relative path inside the Workdir".to_string(),
));
}
Ok(value.to_string())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn worker_tool_family_is_distinct_from_sub_worker_tools() {
assert_eq!(
WorkerOperation::ALL.map(WorkerOperation::tool_name),
["WorkerList", "WorkerSpawn", "WorkerStop", "WorkerRestore"]
);
}
#[test]
fn worker_spawn_request_uses_authority_ids_without_runtime_paths() {
let request = WorkerSpawnRequest {
runtime_id: "runtime-1".to_string(),
display_name: "Coder".to_string(),
profile: "builtin:coder".to_string(),
initial_text: "Implement the Ticket".to_string(),
working_directory: WorkerWorkingDirectorySelection {
working_directory_id: "wd-1".to_string(),
relative_cwd: Some("repo".to_string()),
},
};
let value = serde_json::to_value(request).unwrap();
assert_eq!(value["runtime_id"], "runtime-1");
assert_eq!(value["working_directory"]["working_directory_id"], "wd-1");
assert!(value.get("cwd").is_none());
assert!(value.get("runtime_url").is_none());
assert!(value["working_directory"].get("mode").is_none());
}
#[test]
fn worker_inputs_reject_paths_and_parent_traversal() {
assert!(authority_id("https://runtime.example", "runtime_id").is_err());
assert!(authority_id("runtime/id", "runtime_id").is_err());
assert!(validate_relative_cwd("../repo").is_err());
assert!(validate_relative_cwd("/repo").is_err());
assert_eq!(validate_relative_cwd("repo/src").unwrap(), "repo/src");
}
}
crates/worker/src/ipc/event.rs
+1 -1
View File
@@ -59,7 +59,7 @@ pub fn fire_and_forget(socket: Option<PathBuf>, event: WorkerEvent) {
/// Only events classified by `WorkerEvent::should_notify_agent` are injected
/// into the parent's LLM context as system messages; control-plane-only events
/// keep this renderer for diagnostics/tests. Agent-visible summaries are kept
/// deliberately short — the LLM can always call `ReadWorkerOutput` to fetch more
/// deliberately short — the LLM can always call `SubWorkerReadOutput` to fetch more
/// detail if the event summary is not enough.
pub fn render_event(event: &WorkerEvent) -> String {
match event {
crates/worker/src/prompt/catalog.rs
+12 -12
View File
@@ -88,9 +88,9 @@ pub enum WorkerPrompt {
WorkerOrchestrationGuidanceSection,
/// Weak Companion Notify payload for explicit Orchestrator Ticket events.
TicketEventCompanionNotice,
/// LLM-facing description for the SpawnWorker tool, including discovered
/// LLM-facing description for the SubWorkerSpawn tool, including discovered
/// profile selectors.
SpawnWorkerToolDescription,
SubWorkerSpawnToolDescription,
}
impl WorkerPrompt {
@@ -107,7 +107,7 @@ impl WorkerPrompt {
Self::ResidentMemorySummarySection => "resident_memory_summary_section",
Self::WorkerOrchestrationGuidanceSection => "worker_orchestration_guidance_section",
Self::TicketEventCompanionNotice => "ticket_event_companion_notice",
Self::SpawnWorkerToolDescription => "spawn_worker_tool_description",
Self::SubWorkerSpawnToolDescription => "sub_worker_spawn_tool_description",
}
}
@@ -126,7 +126,7 @@ impl WorkerPrompt {
WorkerPrompt::ResidentMemorySummarySection,
WorkerPrompt::WorkerOrchestrationGuidanceSection,
WorkerPrompt::TicketEventCompanionNotice,
WorkerPrompt::SpawnWorkerToolDescription,
WorkerPrompt::SubWorkerSpawnToolDescription,
];
pub const KEYS: &'static [&'static str] = &[
@@ -141,7 +141,7 @@ impl WorkerPrompt {
"resident_memory_summary_section",
"worker_orchestration_guidance_section",
"ticket_event_companion_notice",
"spawn_worker_tool_description",
"sub_worker_spawn_tool_description",
];
}
@@ -384,8 +384,8 @@ impl PromptCatalog {
)
}
/// Render `WorkerPrompt::SpawnWorkerToolDescription`.
pub fn spawn_worker_tool_description(
/// Render `WorkerPrompt::SubWorkerSpawnToolDescription`.
pub fn sub_worker_spawn_tool_description(
&self,
available_profiles: &str,
default_profile: &str,
@@ -396,7 +396,7 @@ impl PromptCatalog {
m.insert("available_profiles", Value::from(available_profiles));
m.insert("default_profile", Value::from(default_profile));
m.insert("profile_diagnostic", Value::from(profile_diagnostic));
self.render(WorkerPrompt::SpawnWorkerToolDescription, Value::from(m))
self.render(WorkerPrompt::SubWorkerSpawnToolDescription, Value::from(m))
}
}
@@ -722,8 +722,8 @@ compact_system = "PREFIX\n{% include \"$yoi/internal/compact_system\" %}"
fn worker_orchestration_guidance_section_renders_resource_body() {
let cat = PromptCatalog::builtins_only().unwrap();
let rendered = cat.worker_orchestration_guidance_section().unwrap();
assert!(rendered.contains("## Worker orchestration"));
assert!(rendered.contains("spawned Worker notifications are background signals"));
assert!(rendered.contains("## SubWorker orchestration"));
assert!(rendered.contains("SubWorker notifications are background signals"));
assert!(rendered.contains("does not need to keep a turn open"));
assert!(rendered.contains("Do not use `sleep` or polling loops"));
assert!(rendered.contains("worktree state, diff, and test results"));
@@ -732,10 +732,10 @@ compact_system = "PREFIX\n{% include \"$yoi/internal/compact_system\" %}"
}
#[test]
fn spawn_worker_tool_description_renders_profile_block() {
fn sub_worker_spawn_tool_description_renders_profile_block() {
let cat = PromptCatalog::builtins_only().unwrap();
let rendered = cat
.spawn_worker_tool_description(
.sub_worker_spawn_tool_description(
"- `project:coder` — Coder\n- `project:reviewer` — Reviewer",
"project:coder",
"",
crates/worker/src/prompt/system.rs
+27 -25
View File
@@ -206,12 +206,12 @@ struct ToolCapabilities {
memory_query: bool,
memory_read_document: bool,
memory_update_document: bool,
worker_spawn: bool,
worker_send: bool,
worker_read_output: bool,
worker_stop: bool,
worker_list: bool,
worker_restore: bool,
sub_worker_spawn: bool,
sub_worker_send: bool,
sub_worker_read_output: bool,
sub_worker_stop: bool,
sub_worker_list: bool,
sub_worker_restore: bool,
}
impl ToolCapabilities {
@@ -222,12 +222,11 @@ impl ToolCapabilities {
"MemoryQuery" => capabilities.memory_query = true,
"MemoryReadDocument" => capabilities.memory_read_document = true,
"MemoryUpdateDocument" => capabilities.memory_update_document = true,
"SpawnWorker" => capabilities.worker_spawn = true,
"SendToWorker" => capabilities.worker_send = true,
"ReadWorkerOutput" => capabilities.worker_read_output = true,
"StopWorker" => capabilities.worker_stop = true,
"ListWorkers" => capabilities.worker_list = true,
"RestoreWorker" => capabilities.worker_restore = true,
"SubWorkerSpawn" => capabilities.sub_worker_spawn = true,
"SubWorkerSend" => capabilities.sub_worker_send = true,
"SubWorkerReadOutput" => capabilities.sub_worker_read_output = true,
"SubWorkerStop" => capabilities.sub_worker_stop = true,
"SubWorkerList" => capabilities.sub_worker_list = true,
_ => {}
}
}
@@ -246,13 +245,13 @@ impl ToolCapabilities {
self.memory_update_document
}
fn worker_management(self) -> bool {
self.worker_spawn
|| self.worker_send
|| self.worker_read_output
|| self.worker_stop
|| self.worker_list
|| self.worker_restore
fn sub_worker_management(self) -> bool {
self.sub_worker_spawn
|| self.sub_worker_send
|| self.sub_worker_read_output
|| self.sub_worker_stop
|| self.sub_worker_list
|| self.sub_worker_restore
}
fn to_minijinja_value(self) -> Value {
@@ -269,7 +268,10 @@ impl ToolCapabilities {
Value::from(self.memory_update_document),
);
map.insert("memory_mutation", Value::from(self.memory_mutation()));
map.insert("worker_management", Value::from(self.worker_management()));
map.insert(
"sub_worker_management",
Value::from(self.sub_worker_management()),
);
Value::from(map)
}
}
@@ -419,7 +421,7 @@ mod tests {
.unwrap()
}
fn worker_orchestration_instruction() -> FeatureInstructionDeclaration {
fn sub_worker_orchestration_instruction() -> FeatureInstructionDeclaration {
FeatureInstructionDeclaration::new(
crate::feature::FeatureInstructionId::builtin("worker.orchestration"),
"$yoi/common/worker-orchestration",
@@ -595,13 +597,13 @@ mod tests {
let tmpl = SystemPromptTemplate::parse("$yoi/default", loader).unwrap();
let dir = TempDir::new().unwrap();
let scope = build_scope(dir.path());
let instructions = [worker_orchestration_instruction()];
let instructions = [sub_worker_orchestration_instruction()];
let mut ctx = ctx(dir.path(), &scope, vec!["Read".into()], None);
ctx.feature_instructions = &instructions;
let rendered = tmpl.render(&ctx).unwrap();
assert!(rendered.contains("## Worker orchestration"));
assert!(rendered.contains("spawned Worker notifications are background signals"));
assert!(rendered.contains("## SubWorker orchestration"));
assert!(rendered.contains("SubWorker notifications are background signals"));
assert!(rendered.contains("does not need to keep a turn open"));
assert!(rendered.contains("Do not use `sleep` or polling loops"));
assert!(rendered.contains("worktree state, diff, and test results"));
@@ -610,7 +612,7 @@ mod tests {
}
#[test]
fn worker_orchestration_guidance_is_omitted_without_worker_management_tools() {
fn worker_orchestration_guidance_is_omitted_without_sub_worker_management_tools() {
let loader = PromptLoader::builtins_only();
let tmpl = SystemPromptTemplate::parse("$yoi/default", loader).unwrap();
let dir = TempDir::new().unwrap();
crates/worker/src/spawn/comm_tools.rs
+103 -47
View File
@@ -1,6 +1,6 @@
//! Worker-to-Worker communication tools.
//!
//! Three tools in one module: `SendToWorker`, `ReadWorkerOutput`, `StopWorker`,
//! Three tools in one module: `SubWorkerSend`, `SubWorkerReadOutput`, `SubWorkerStop`,
//! all built on the same `SpawnedWorkerRegistry` handed in by
//! the controller. Each operation is request-response: connect to the
//! target's Unix socket, perform one method exchange, disconnect.
@@ -18,7 +18,7 @@ use llm_engine::llm_client::types::{ContentPart, Item, Role};
use llm_engine::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput};
use protocol::stream::{JsonLineReader, JsonLineWriter};
use protocol::{ErrorCode, Event, InvokeKind, Method};
use serde::Deserialize;
use serde::{Deserialize, Serialize};
use session_store::LogEntry;
use tokio::net::UnixStream;
@@ -35,40 +35,96 @@ const SOCKET_OP_TIMEOUT: Duration = Duration::from_secs(5);
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct NameInput {
/// Name of a previously spawned Worker.
/// Name of a previously spawned SubWorker.
name: String,
}
// ---------------------------------------------------------------------------
// SendToWorker
// ---------------------------------------------------------------------------
const SEND_TO_POD_DESCRIPTION: &str = "Send a text message to a previously spawned Worker. The spawned Worker \
processes it as a user turn. Fails if the Worker is already executing a \
turn — retry after it finishes. Does not wait for the turn to complete; \
use `ReadWorkerOutput` to fetch results afterwards.";
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct SendToWorkerInput {
/// Target Worker name.
#[serde(deny_unknown_fields)]
struct SubWorkerListInput {}
#[derive(Debug, Serialize)]
struct SubWorkerListItem {
name: String,
/// Text delivered to the Worker as the next user message.
message: String,
}
struct SendToWorkerTool {
struct SubWorkerListTool {
registry: Arc<SpawnedWorkerRegistry>,
}
#[async_trait]
impl Tool for SendToWorkerTool {
impl Tool for SubWorkerListTool {
async fn execute(
&self,
input_json: &str,
_ctx: llm_engine::tool::ToolExecutionContext,
) -> Result<ToolOutput, ToolError> {
let input: SendToWorkerInput = serde_json::from_str(input_json)
.map_err(|e| ToolError::InvalidArgument(format!("invalid SendToWorker input: {e}")))?;
let _input: SubWorkerListInput = serde_json::from_str(input_json).map_err(|error| {
ToolError::InvalidArgument(format!("invalid SubWorkerList input: {error}"))
})?;
let items = self
.registry
.list()
.await
.into_iter()
.map(|record| SubWorkerListItem {
name: record.worker_name,
})
.collect::<Vec<_>>();
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),
})
}
}
pub fn sub_worker_list_tool(registry: Arc<SpawnedWorkerRegistry>) -> 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<dyn Tool> = 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 `SubWorkerReadOutput` to fetch results afterwards.";
#[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<SpawnedWorkerRegistry>,
}
#[async_trait]
impl Tool for SubWorkerSendTool {
async fn execute(
&self,
input_json: &str,
_ctx: llm_engine::tool::ToolExecutionContext,
) -> Result<ToolOutput, ToolError> {
let input: SubWorkerSendInput = serde_json::from_str(input_json)
.map_err(|e| ToolError::InvalidArgument(format!("invalid SubWorkerSend input: {e}")))?;
let record = self
.registry
.get(&input.name)
@@ -98,14 +154,14 @@ impl Tool for SendToWorkerTool {
}
}
pub fn send_to_worker_tool(registry: Arc<SpawnedWorkerRegistry>) -> ToolDefinition {
pub fn sub_worker_send_tool(registry: Arc<SpawnedWorkerRegistry>) -> ToolDefinition {
Arc::new(move || {
let schema = schemars::schema_for!(SendToWorkerInput);
let schema = schemars::schema_for!(SubWorkerSendInput);
let schema_value = serde_json::to_value(schema).unwrap_or(serde_json::json!({}));
let meta = ToolMeta::new("SendToWorker")
let meta = ToolMeta::new("SubWorkerSend")
.description(SEND_TO_POD_DESCRIPTION)
.input_schema(schema_value);
let tool: Arc<dyn Tool> = Arc::new(SendToWorkerTool {
let tool: Arc<dyn Tool> = Arc::new(SubWorkerSendTool {
registry: registry.clone(),
});
(meta, tool)
@@ -113,27 +169,27 @@ pub fn send_to_worker_tool(registry: Arc<SpawnedWorkerRegistry>) -> ToolDefiniti
}
// ---------------------------------------------------------------------------
// ReadWorkerOutput
// SubWorkerReadOutput
// ---------------------------------------------------------------------------
const READ_POD_OUTPUT_DESCRIPTION: &str = "Fetch new assistant text from a spawned Worker since the last read. \
Uses an internal cursor per-Worker so consecutive calls return only \
newly-produced output. Returns the Worker's current status and the new \
text, or reports `stopped` if the Worker can no longer be reached.";
const READ_POD_OUTPUT_DESCRIPTION: &str = "Fetch new assistant text from a SubWorker since the last read. \
Uses an internal cursor per-SubWorker so consecutive calls return only \
newly-produced output. Returns the SubWorker's current status and the new \
text, or reports `stopped` if the SubWorker can no longer be reached.";
struct ReadWorkerOutputTool {
struct SubWorkerReadOutputTool {
registry: Arc<SpawnedWorkerRegistry>,
}
#[async_trait]
impl Tool for ReadWorkerOutputTool {
impl Tool for SubWorkerReadOutputTool {
async fn execute(
&self,
input_json: &str,
_ctx: llm_engine::tool::ToolExecutionContext,
) -> Result<ToolOutput, ToolError> {
let input: NameInput = serde_json::from_str(input_json).map_err(|e| {
ToolError::InvalidArgument(format!("invalid ReadWorkerOutput input: {e}"))
ToolError::InvalidArgument(format!("invalid SubWorkerReadOutput input: {e}"))
})?;
let record = self
.registry
@@ -178,14 +234,14 @@ impl Tool for ReadWorkerOutputTool {
}
}
pub fn read_worker_output_tool(registry: Arc<SpawnedWorkerRegistry>) -> ToolDefinition {
pub fn sub_worker_read_output_tool(registry: Arc<SpawnedWorkerRegistry>) -> 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("ReadWorkerOutput")
let meta = ToolMeta::new("SubWorkerReadOutput")
.description(READ_POD_OUTPUT_DESCRIPTION)
.input_schema(schema_value);
let tool: Arc<dyn Tool> = Arc::new(ReadWorkerOutputTool {
let tool: Arc<dyn Tool> = Arc::new(SubWorkerReadOutputTool {
registry: registry.clone(),
});
(meta, tool)
@@ -193,26 +249,26 @@ pub fn read_worker_output_tool(registry: Arc<SpawnedWorkerRegistry>) -> ToolDefi
}
// ---------------------------------------------------------------------------
// StopWorker
// SubWorkerStop
// ---------------------------------------------------------------------------
const STOP_POD_DESCRIPTION: &str = "Terminate a spawned Worker and reclaim the delegated scope. The Worker \
const STOP_POD_DESCRIPTION: &str = "Terminate a spawned SubWorker and reclaim the delegated scope. The SubWorker \
receives `Shutdown`; its scope entry is released in the machine-wide \
registry so the spawner can spawn a new Worker over the same paths.";
registry so the parent Worker can spawn a new SubWorker over the same paths.";
struct StopWorkerTool {
struct SubWorkerStopTool {
registry: Arc<SpawnedWorkerRegistry>,
}
#[async_trait]
impl Tool for StopWorkerTool {
impl Tool for SubWorkerStopTool {
async fn execute(
&self,
input_json: &str,
_ctx: llm_engine::tool::ToolExecutionContext,
) -> Result<ToolOutput, ToolError> {
let input: NameInput = serde_json::from_str(input_json)
.map_err(|e| ToolError::InvalidArgument(format!("invalid StopWorker input: {e}")))?;
.map_err(|e| ToolError::InvalidArgument(format!("invalid SubWorkerStop input: {e}")))?;
let record = self
.registry
.get(&input.name)
@@ -244,14 +300,14 @@ impl Tool for StopWorkerTool {
}
}
pub fn stop_worker_tool(registry: Arc<SpawnedWorkerRegistry>) -> ToolDefinition {
pub fn sub_worker_stop_tool(registry: Arc<SpawnedWorkerRegistry>) -> 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("StopWorker")
let meta = ToolMeta::new("SubWorkerStop")
.description(STOP_POD_DESCRIPTION)
.input_schema(schema_value);
let tool: Arc<dyn Tool> = Arc::new(StopWorkerTool {
let tool: Arc<dyn Tool> = Arc::new(SubWorkerStopTool {
registry: registry.clone(),
});
(meta, tool)
@@ -335,13 +391,13 @@ where
}
}
/// Failure modes distinguished by `SendToWorker`.
/// Failure modes distinguished by `SubWorkerSend`.
#[derive(Debug)]
pub(crate) enum SendRunError {
/// Target Worker responded with `Error { AlreadyRunning }` — the
/// Target SubWorker responded with `Error { AlreadyRunning }` — the
/// caller can retry once the current turn ends.
AlreadyRunning,
/// Target Worker explicitly rejected the run after delivery reached the
/// Target SubWorker explicitly rejected the run after delivery reached the
/// controller.
Rejected { code: ErrorCode, message: String },
/// Transport, protocol, timeout, or unexpected EOF before acceptance
crates/worker/src/spawn/registry.rs
+3 -3
View File
@@ -1,12 +1,12 @@
//! Shared registry of Workers spawned by this Worker.
//!
//! `SpawnWorker` writes here; the worker-comm tools (`SendToWorker`,
//! `ReadWorkerOutput`, `StopWorker`) read and mutate the same instance. Discovery
//! `SubWorkerSpawn` writes here; the worker-comm tools (`SubWorkerSend`,
//! `SubWorkerReadOutput`, `SubWorkerStop`) read and mutate the same instance. Discovery
//! tools consult this registry together with durable Worker state. Runtime
//! write-through still materialises `spawned_workers.json`, but durable state lives
//! in the spawner's Worker metadata.
//!
//! `ReadWorkerOutput` additionally owns a per-spawned-worker cursor here so
//! `SubWorkerReadOutput` additionally owns a per-spawned-worker cursor here so
//! two consecutive reads yield only new assistant text. The cursor is
//! an item-index into the child's history; push-only history makes
//! index stable across reads.
crates/worker/src/spawn/tool.rs
+53 -47
View File
@@ -1,8 +1,8 @@
//! `SpawnWorker` tool — launch a new Worker process as a child of this one.
//! `SubWorkerSpawn` tool — launch a new SubWorker process as a child of this one.
//!
//! Wires worker-allocation delegation, child manifest-config construction, subprocess
//! launch, and socket handoff into a single `Tool` implementation. When
//! the LLM calls `SpawnWorker`, a fresh Worker runtime command is exec'd in its own
//! the LLM calls `SubWorkerSpawn`, a fresh SubWorker runtime command is exec'd in its own
//! process group, the worker-allocation is updated atomically, and the child's
//! first turn is kicked off by handing its socket a `Method::Run`.
@@ -34,13 +34,13 @@ use crate::spawn::comm_tools::{SendRunError, send_run_and_confirm};
use crate::spawn::registry::SpawnedWorkerRegistry;
use protocol::WorkerEvent;
/// How long we will wait for the spawned Worker's socket to become
/// How long we will wait for the spawned SubWorker's socket to become
/// connectable before treating the spawn as failed.
const SOCKET_WAIT_TIMEOUT: Duration = Duration::from_secs(10);
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct SpawnWorkerInput {
/// Identifier for the spawned Worker. Must be unique machine-wide.
struct SubWorkerSpawnInput {
/// Identifier for the spawned SubWorker. Must be unique machine-wide.
name: String,
/// Profile selector for child role configuration. Omit or use `default`
/// for the effective child default profile, use `inherit` to derive
@@ -53,13 +53,13 @@ struct SpawnWorkerInput {
#[serde(default)]
instruction: Option<String>,
/// Child process/tool working directory. This is not the runtime workspace
/// root and grants no filesystem authority. When omitted, the spawned Worker
/// root and grants no filesystem authority. When omitted, the spawned SubWorker
/// starts in the spawner's current working directory.
#[serde(default)]
cwd: Option<PathBuf>,
/// First message sent to the spawned Worker via `Method::Run`.
/// First message sent to the spawned SubWorker via `Method::Run`.
task: String,
/// Allow rules delegated to the spawned Worker. Must be a subset of the
/// Allow rules delegated to the spawned SubWorker. Must be a subset of the
/// spawner's explicit delegation authority; direct tool scope alone is not
/// sufficient. Omit `recursive` for normal workspace/worktree delegation; it defaults to true.
scope: Vec<ScopeRuleInput>,
@@ -189,7 +189,7 @@ fn parse_spawn_profile_selector(raw: Option<&str>) -> Result<SpawnProfileSelecto
|| raw.ends_with(".nix")
{
return Err(format!(
"SpawnWorker.profile accepts `default`, `inherit`, or registry selectors only; path-like selector `{raw}` is not allowed"
"SubWorkerSpawn.profile accepts `default`, `inherit`, or registry selectors only; path-like selector `{raw}` is not allowed"
));
}
if let Some((prefix, name)) = raw.split_once(':') {
@@ -199,12 +199,14 @@ fn parse_spawn_profile_selector(raw: Option<&str>) -> Result<SpawnProfileSelecto
"project" => ProfileRegistrySource::Project,
_ => {
return Err(format!(
"unsupported SpawnWorker.profile selector prefix `{prefix}`; use builtin:, user:, project:, default, or inherit"
"unsupported SubWorkerSpawn.profile selector prefix `{prefix}`; use builtin:, user:, project:, default, or inherit"
));
}
};
if name.is_empty() {
return Err("SpawnWorker.profile registry selector has an empty profile name".into());
return Err(
"SubWorkerSpawn.profile registry selector has an empty profile name".into(),
);
}
return Ok(SpawnProfileSelector::Registry(
ProfileSelector::source_named(source, name),
@@ -213,30 +215,30 @@ fn parse_spawn_profile_selector(raw: Option<&str>) -> Result<SpawnProfileSelecto
Ok(SpawnProfileSelector::Registry(ProfileSelector::named(raw)))
}
/// Runtime dependencies the `SpawnWorker` tool needs in order to launch a
/// child Worker and record the handoff locally. Constructed by the Worker
/// 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.
pub struct SpawnWorkerTool {
/// Spawner's own worker name — becomes the spawned Worker's
pub struct SubWorkerSpawnTool {
/// Spawner's own worker name — becomes the spawned SubWorker's
/// `delegated_from` in the worker-allocation.
spawner_name: String,
/// Path to the spawner's Unix socket. Handed to the child via
/// `--callback` so its `WorkerEvent` callbacks have somewhere to land.
callback_socket: PathBuf,
/// Root of the `$XDG_RUNTIME_DIR/yoi/` tree, used to predict
/// the spawned Worker's socket path before the child has bound it.
/// the spawned SubWorker's socket path before the child has bound it.
runtime_base: PathBuf,
/// Inherited runtime workspace root for Profile/project/Ticket/workflow/
/// memory context. SpawnWorker `cwd` must not affect this value.
/// memory context. SubWorkerSpawn `cwd` must not affect this value.
workspace_root: PathBuf,
/// Directory the spawned Worker's tools should use when the LLM did not
/// Directory the spawned SubWorker's tools should use when the LLM did not
/// override it. Defaults to the spawner's cwd.
spawner_cwd: PathBuf,
/// Optional typed runtime command injected by tests. Production resolves
/// the runtime command from `std::env::current_exe()` at launch time.
runtime_command: Option<WorkerRuntimeCommand>,
/// Shared registry of spawned children, also used by the
/// worker-comm tools (`SendToWorker` / `ReadWorkerOutput` / `StopWorker`) and by
/// worker-comm tools (`SubWorkerSend` / `SubWorkerReadOutput` / `SubWorkerStop`) and by
/// Worker discovery. Writes the list to runtime and durable Worker state on
/// each add.
registry: Arc<SpawnedWorkerRegistry>,
@@ -266,7 +268,7 @@ pub struct SpawnWorkerTool {
delegation_scope: DelegationScope,
}
impl SpawnWorkerTool {
impl SubWorkerSpawnTool {
fn new(
spawner_name: String,
callback_socket: PathBuf,
@@ -299,14 +301,15 @@ impl SpawnWorkerTool {
}
#[async_trait]
impl Tool for SpawnWorkerTool {
impl Tool for SubWorkerSpawnTool {
async fn execute(
&self,
input_json: &str,
_ctx: llm_engine::tool::ToolExecutionContext,
) -> Result<ToolOutput, ToolError> {
let input: SpawnWorkerInput = serde_json::from_str(input_json)
.map_err(|e| ToolError::InvalidArgument(format!("invalid SpawnWorker input: {e}")))?;
let input: SubWorkerSpawnInput = serde_json::from_str(input_json).map_err(|e| {
ToolError::InvalidArgument(format!("invalid SubWorkerSpawn input: {e}"))
})?;
// `delegate_scope` catches this too (as `DuplicateWorkerName`), but
// the dedicated message is kinder to the LLM — which gets the
@@ -438,7 +441,7 @@ impl Tool for SpawnWorkerTool {
}
}
impl SpawnWorkerTool {
impl SubWorkerSpawnTool {
async fn exec_child(
&self,
worker_name: &str,
@@ -508,7 +511,7 @@ impl SpawnWorkerTool {
fn validate_delegation_scope(&self, scope_allow: &[ScopeRule]) -> Result<(), ToolError> {
if self.delegation_scope.is_empty() && !scope_allow.is_empty() {
return Err(ToolError::InvalidArgument(
"SpawnWorker requires delegation authority, but this Worker has no delegation scope grant; direct filesystem scope only authorizes this Worker's own tools".into(),
"SubWorkerSpawn requires delegation authority, but this Worker has no delegation scope grant; direct filesystem scope only authorizes this Worker's own tools".into(),
));
}
for rule in scope_allow {
@@ -566,29 +569,32 @@ fn validate_spawn_cwd(
};
if !cwd.is_absolute() {
return Err(ToolError::InvalidArgument(format!(
"SpawnWorker.cwd must be absolute: {}",
"SubWorkerSpawn.cwd must be absolute: {}",
cwd.display()
)));
}
let metadata = std::fs::metadata(cwd).map_err(|e| {
if e.kind() == std::io::ErrorKind::NotFound {
ToolError::InvalidArgument(format!("SpawnWorker.cwd does not exist: {}", cwd.display()))
ToolError::InvalidArgument(format!(
"SubWorkerSpawn.cwd does not exist: {}",
cwd.display()
))
} else {
ToolError::InvalidArgument(format!(
"SpawnWorker.cwd is not usable: {}: {e}",
"SubWorkerSpawn.cwd is not usable: {}: {e}",
cwd.display()
))
}
})?;
if !metadata.is_dir() {
return Err(ToolError::InvalidArgument(format!(
"SpawnWorker.cwd must be a directory: {}",
"SubWorkerSpawn.cwd must be a directory: {}",
cwd.display()
)));
}
let canonical = std::fs::canonicalize(cwd).map_err(|e| {
ToolError::InvalidArgument(format!(
"SpawnWorker.cwd is not usable: {}: {e}",
"SubWorkerSpawn.cwd is not usable: {}: {e}",
cwd.display()
))
})?;
@@ -598,12 +604,12 @@ fn validate_spawn_cwd(
})
.map_err(|e| {
ToolError::InvalidArgument(format!(
"requested child scope cannot validate SpawnWorker.cwd: {e}"
"requested child scope cannot validate SubWorkerSpawn.cwd: {e}"
))
})?;
if !child_scope.is_readable(&canonical) {
return Err(ToolError::InvalidArgument(format!(
"SpawnWorker.cwd {} is outside the child's delegated readable scope; cwd grants no authority, so add an explicit read or write scope rule covering it",
"SubWorkerSpawn.cwd {} is outside the child's delegated readable scope; cwd grants no authority, so add an explicit read or write scope rule covering it",
cwd.display()
)));
}
@@ -617,7 +623,7 @@ fn validate_spawn_cwd(
///
/// The child's tool working directory is carried separately through
/// the child runtime entrypoint; it is not part of the manifest.
impl SpawnWorkerTool {
impl SubWorkerSpawnTool {
fn build_spawn_config_json(
&self,
name: &str,
@@ -651,7 +657,7 @@ fn build_spawn_config_json_for_profile(
SpawnProfileSelector::Default | SpawnProfileSelector::Registry(_) => {
let registry = available_profiles.registry.as_ref().ok_or_else(|| {
format!(
"profile discovery failed for SpawnWorker: {}{}",
"profile discovery failed for SubWorkerSpawn: {}{}",
available_profiles.diagnostic().if_empty("unknown error"),
available_profiles.error_suffix()
)
@@ -728,7 +734,7 @@ impl IfEmpty for str {
fn profile_error_with_available(error: ProfileError, available: &AvailableProfiles) -> String {
format!(
"invalid SpawnWorker.profile: {error}{}",
"invalid SubWorkerSpawn.profile: {error}{}",
available.error_suffix()
)
}
@@ -878,8 +884,8 @@ fn worker_allocation_err_to_tool(e: ScopeLockError) -> ToolError {
}
}
/// Factory for the `SpawnWorker` tool.
pub fn spawn_worker_tool(
/// Factory for the `SubWorkerSpawn` tool.
pub fn sub_worker_spawn_tool(
spawner_name: String,
callback_socket: PathBuf,
runtime_base: PathBuf,
@@ -891,7 +897,7 @@ pub fn spawn_worker_tool(
spawner_scope: SharedScope,
prompts: Arc<PromptCatalog>,
) -> ToolDefinition {
spawn_worker_tool_impl(
sub_worker_spawn_tool_impl(
spawner_name,
callback_socket,
runtime_base,
@@ -907,7 +913,7 @@ pub fn spawn_worker_tool(
}
#[doc(hidden)]
pub fn spawn_worker_tool_with_runtime_command(
pub fn sub_worker_spawn_tool_with_runtime_command(
spawner_name: String,
callback_socket: PathBuf,
runtime_base: PathBuf,
@@ -920,7 +926,7 @@ pub fn spawn_worker_tool_with_runtime_command(
prompts: Arc<PromptCatalog>,
runtime_command: WorkerRuntimeCommand,
) -> ToolDefinition {
spawn_worker_tool_impl(
sub_worker_spawn_tool_impl(
spawner_name,
callback_socket,
runtime_base,
@@ -935,7 +941,7 @@ pub fn spawn_worker_tool_with_runtime_command(
)
}
fn spawn_worker_tool_impl(
fn sub_worker_spawn_tool_impl(
spawner_name: String,
callback_socket: PathBuf,
runtime_base: PathBuf,
@@ -949,25 +955,25 @@ fn spawn_worker_tool_impl(
runtime_command: Option<WorkerRuntimeCommand>,
) -> ToolDefinition {
Arc::new(move || {
let schema = schemars::schema_for!(SpawnWorkerInput);
let schema = schemars::schema_for!(SubWorkerSpawnInput);
let schema_value = serde_json::to_value(schema).unwrap_or(serde_json::json!({}));
let available_profiles = AvailableProfiles::discover(&workspace_root);
let description = prompts
.spawn_worker_tool_description(
.sub_worker_spawn_tool_description(
&available_profiles.compact_list(),
&available_profiles.default_label(),
available_profiles.diagnostic(),
)
.unwrap_or_else(|e| {
format!(
"Spawn a new Worker process to work on a delegated task. Profile description rendering failed: {e}. Available profiles:\n{}",
"Spawn a new SubWorker process to split context for a delegated task. Profile description rendering failed: {e}. Available profiles:\n{}",
available_profiles.compact_list()
)
});
let meta = ToolMeta::new("SpawnWorker")
let meta = ToolMeta::new("SubWorkerSpawn")
.description(description)
.input_schema(schema_value);
let tool: Arc<dyn Tool> = Arc::new(SpawnWorkerTool::new(
let tool: Arc<dyn Tool> = Arc::new(SubWorkerSpawnTool::new(
spawner_name.clone(),
callback_socket.clone(),
runtime_base.clone(),
@@ -1002,7 +1008,7 @@ mod tests {
#[test]
fn spawn_worker_input_schema_includes_optional_cwd() {
let schema = serde_json::to_value(schemars::schema_for!(SpawnWorkerInput)).unwrap();
let schema = serde_json::to_value(schemars::schema_for!(SubWorkerSpawnInput)).unwrap();
let properties = schema
.get("properties")
.and_then(serde_json::Value::as_object)
crates/worker/src/worker.rs
+2 -2
View File
@@ -654,7 +654,7 @@ pub struct Worker<C: LlmClient, St: Store> {
/// and compaction so updates propagate at the next permission check.
scope: SharedScope,
/// Filesystem authority this Worker may pass to spawned children. Direct tools
/// continue to use `scope`; SpawnWorker validates requested child scope here.
/// continue to use `scope`; SubWorkerSpawn validates requested child scope here.
delegation_scope: DelegationScope,
hook_builder: HookRegistryBuilder,
interceptor_installed: bool,
@@ -3794,7 +3794,7 @@ where
/// The Worker's working directory is captured once here from the
/// process's `std::env::current_dir()` — callers that want a
/// different cwd must `cd` before constructing the Worker (e.g. the
/// `SpawnWorker` tool sets `Command::current_dir` on the child). The
/// `SubWorkerSpawn` tool sets `Command::current_dir` on the child). The
/// captured cwd is canonicalised and validated against
/// `manifest.scope`.
///
crates/worker/tests/controller_test.rs
+18 -18
View File
@@ -342,7 +342,7 @@ async fn feature_flags_default_to_core_tool_surface_only() {
assert_eq!(names, vec!["Bash", "Edit", "Glob", "Grep", "Read", "Write"]);
assert!(!names.iter().any(|name| name == "TaskCreate"));
assert!(!names.iter().any(|name| name == "WebSearch"));
assert!(!names.iter().any(|name| name == "SpawnWorker"));
assert!(!names.iter().any(|name| name == "SubWorkerSpawn"));
}
#[tokio::test]
@@ -386,7 +386,7 @@ permission = "write"
assert!(names.iter().any(|name| name == "TaskUpdate"));
assert!(names.iter().any(|name| name == "WebSearch"));
assert!(names.iter().any(|name| name == "WebFetch"));
assert!(!names.iter().any(|name| name == "SpawnWorker"));
assert!(!names.iter().any(|name| name == "SubWorkerSpawn"));
assert!(!names.iter().any(|name| name == "MemoryRead"));
}
@@ -394,34 +394,34 @@ permission = "write"
async fn project_role_tool_surfaces_keep_task_disabled_and_workers_role_scoped() {
struct Case {
role: &'static str,
workers_enabled: bool,
sub_worker_enabled: bool,
}
let cases = [
Case {
role: "orchestrator",
workers_enabled: true,
sub_worker_enabled: true,
},
Case {
role: "coder",
workers_enabled: false,
sub_worker_enabled: false,
},
Case {
role: "intake",
workers_enabled: false,
sub_worker_enabled: false,
},
Case {
role: "reviewer",
workers_enabled: false,
sub_worker_enabled: false,
},
Case {
role: "companion",
workers_enabled: false,
sub_worker_enabled: false,
},
];
for case in cases {
let delegation = if case.workers_enabled {
let delegation = if case.sub_worker_enabled {
r#"
[[delegation_scope.allow]]
target = "/tmp"
@@ -446,8 +446,8 @@ max_tokens = 100
[feature.task]
enabled = false
[feature.workers]
enabled = {workers_enabled}
[feature.sub_worker]
enabled = {sub_worker_enabled}
[[scope.allow]]
target = "./"
@@ -455,7 +455,7 @@ permission = "write"
{delegation}
"#,
role = case.role,
workers_enabled = case.workers_enabled,
sub_worker_enabled = case.sub_worker_enabled,
delegation = delegation,
);
let client = MockClient::new(simple_text_events());
@@ -474,16 +474,16 @@ permission = "write"
case.role
);
assert_eq!(
names.iter().any(|name| name == "SpawnWorker"),
case.workers_enabled,
"{} role Worker tool exposure mismatch: {names:?}",
names.iter().any(|name| name == "SubWorkerSpawn"),
case.sub_worker_enabled,
"{} role SubWorker tool exposure mismatch: {names:?}",
case.role
);
}
}
#[tokio::test]
async fn workers_feature_requires_delegation_scope() {
async fn sub_worker_feature_requires_delegation_scope() {
let manifest = r#"
[worker]
name = "worker-management-feature-test"
@@ -496,7 +496,7 @@ model_id = "test-model"
[engine]
max_tokens = 100
[feature.workers]
[feature.sub_worker]
enabled = true
[[scope.allow]]
@@ -510,7 +510,7 @@ permission = "write"
assert!(result.is_err());
let message = result.err().unwrap().to_string();
assert!(
message.contains("[feature.workers].enabled = true requires non-empty"),
message.contains("[feature.sub_worker].enabled = true requires non-empty"),
"unexpected error: {message}"
);
}
crates/worker/tests/spawn_worker_test.rs
+10 -10
View File
@@ -1,4 +1,4 @@
//! Integration tests for the `SpawnWorker` tool.
//! Integration tests for the `SubWorkerSpawn` tool.
//!
//! These tests exercise the tool's worker-allocation delegation, subprocess
//! launch, socket handoff, and `spawned_workers.json` write through an injected
@@ -24,7 +24,7 @@ use tokio::net::UnixListener;
use worker::runtime::dir::{RuntimeDir, SpawnedWorkerRecord};
use worker::runtime::worker_allocation::{self, LockFileGuard};
use worker::spawn::registry::SpawnedWorkerRegistry;
use worker::spawn::tool::spawn_worker_tool_with_runtime_command;
use worker::spawn::tool::sub_worker_spawn_tool_with_runtime_command;
/// Serialises tests that mutate `YOI_RUNTIME_DIR` across the
/// thread-pooled test harness.
@@ -203,7 +203,7 @@ fn which_sh() -> String {
}
/// Tests don't exercise the model — they intercept the spawned
/// child via a mock socket — but `spawn_worker_tool` needs a value to
/// child via a mock socket — but `sub_worker_spawn_tool` needs a value to
/// embed in the overlay TOML. Any well-formed `ModelManifest` works.
fn dummy_model() -> ModelManifest {
ModelManifest {
@@ -289,7 +289,7 @@ async fn spawn_worker_launches_runtime_in_workspace_and_process_cwd() {
let received = accept_one_method(listener);
let registry = SpawnedWorkerRegistry::new(spawner_rd);
let def = spawn_worker_tool_with_runtime_command(
let def = sub_worker_spawn_tool_with_runtime_command(
"root".into(),
spawner_socket,
runtime_base,
@@ -349,7 +349,7 @@ async fn spawn_worker_omitted_cwd_preserves_spawner_cwd() {
let received = accept_one_method(listener);
let registry = SpawnedWorkerRegistry::new(spawner_rd);
let def = spawn_worker_tool_with_runtime_command(
let def = sub_worker_spawn_tool_with_runtime_command(
"root".into(),
spawner_socket,
runtime_base,
@@ -400,7 +400,7 @@ async fn spawn_worker_delegates_scope_and_sends_run() {
let registry = SpawnedWorkerRegistry::new(spawner_rd.clone());
let spawner_scope = shared_scope_for(allow_root.path());
let def = spawn_worker_tool_with_runtime_command(
let def = sub_worker_spawn_tool_with_runtime_command(
"root".into(),
spawner_socket.clone(),
runtime_base.clone(),
@@ -493,7 +493,7 @@ async fn spawn_worker_requires_explicit_delegation_even_with_direct_scope() {
assert!(direct.is_writable(&allow_root.path().join("direct.txt")));
let registry = SpawnedWorkerRegistry::new(spawner_rd.clone());
let def = spawn_worker_tool_with_runtime_command(
let def = sub_worker_spawn_tool_with_runtime_command(
"root".into(),
spawner_socket,
runtime_base,
@@ -560,7 +560,7 @@ async fn spawn_worker_rejects_child_non_recursive_scope_under_parent_non_recursi
let manifest = dummy_manifest_with_scopes(direct_scope, delegation_scope);
let registry = SpawnedWorkerRegistry::new(spawner_rd.clone());
let def = spawn_worker_tool_with_runtime_command(
let def = sub_worker_spawn_tool_with_runtime_command(
"root".into(),
spawner_socket,
runtime_base,
@@ -612,7 +612,7 @@ async fn spawn_worker_rejects_scope_outside_spawner() {
let registry = SpawnedWorkerRegistry::new(spawner_rd);
let spawner_scope = shared_scope_for(allow_root.path());
let def = spawn_worker_tool_with_runtime_command(
let def = sub_worker_spawn_tool_with_runtime_command(
"root".into(),
spawner_socket,
runtime_base,
@@ -686,7 +686,7 @@ async fn spawn_worker_rolls_back_reservation_when_socket_never_appears() {
let registry = SpawnedWorkerRegistry::new(spawner_rd);
let spawner_scope = shared_scope_for(allow_root.path());
let def = spawn_worker_tool_with_runtime_command(
let def = sub_worker_spawn_tool_with_runtime_command(
"root".into(),
spawner_socket,
runtime_base,
crates/worker/tests/worker_comm_tools_test.rs
+22 -20
View File
@@ -1,5 +1,5 @@
//! Integration tests for the worker-comm tools (`SendToWorker`,
//! `ReadWorkerOutput`, `StopWorker`).
//! Integration tests for the worker-comm tools (`SubWorkerSend`,
//! `SubWorkerReadOutput`, `SubWorkerStop`).
//!
//! The real child Worker binary is not started. Instead each test stands
//! up a mock `UnixListener` that speaks the socket protocol directly:
@@ -25,7 +25,9 @@ use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use worker::runtime::dir::{RuntimeDir, SpawnedWorkerRecord};
use worker::runtime::worker_allocation::{self, LockFileGuard};
use worker::spawn::comm_tools::{read_worker_output_tool, send_to_worker_tool, stop_worker_tool};
use worker::spawn::comm_tools::{
sub_worker_read_output_tool, sub_worker_send_tool, sub_worker_stop_tool,
};
use worker::spawn::registry::SpawnedWorkerRegistry;
/// Serialises env-mutating tests. The test harness runs tasks across
@@ -148,7 +150,7 @@ fn accept_one_method(listener: UnixListener) -> JoinHandle<Option<Method>> {
}
/// Accept one connection, send the protocol's connect-time snapshot,
/// read one `Method`, then write `response` back. Used by `SendToWorker`
/// read one `Method`, then write `response` back. Used by `SubWorkerSend`
/// tests to mock the real controller's `TurnStart` acknowledgement (or
/// its `AlreadyRunning` rejection).
fn accept_method_and_respond(
@@ -171,7 +173,7 @@ fn accept_method_and_respond(
/// Pretend to be a spawned Worker whose connect-time snapshot carries a
/// fixed set of assistant items. Sends `Event::Snapshot` immediately on
/// every accept — the real Worker does the same, so `ReadWorkerOutput`'s
/// every accept — the real Worker does the same, so `SubWorkerReadOutput`'s
/// `fetch_history` just consumes the first non-Alert event.
fn serve_history(listener: UnixListener, items: Vec<Item>) -> JoinHandle<()> {
tokio::spawn(async move {
@@ -249,7 +251,7 @@ fn assistant(text: &str) -> Item {
}
// ---------------------------------------------------------------------------
// SendToWorker
// SubWorkerSend
// ---------------------------------------------------------------------------
#[tokio::test]
@@ -257,11 +259,11 @@ async fn send_to_worker_delivers_run_method() {
let (tmp, registry, _rd) = setup_registry().await;
let (socket, listener) = bind_mock_socket(tmp.path(), "child").await;
// Mock the controller's accept path: after reading the method,
// ack with `TurnStart` so `SendToWorker`'s confirmation loop succeeds.
// ack with `TurnStart` so `SubWorkerSend`'s confirmation loop succeeds.
let received = accept_method_and_respond(listener, Event::TurnStart { turn: 1 });
register_child(&registry, "child", &socket, tmp.path()).await;
let def = send_to_worker_tool(registry);
let def = sub_worker_send_tool(registry);
let (_meta, tool) = def();
let input = json!({ "name": "child", "message": "hello there" }).to_string();
let output: ToolOutput = tool.execute(&input, Default::default()).await.unwrap();
@@ -284,7 +286,7 @@ async fn send_to_worker_delivers_run_method() {
#[tokio::test]
async fn send_to_worker_errors_on_unknown_worker() {
let (_tmp, registry, _rd) = setup_registry().await;
let def = send_to_worker_tool(registry);
let def = sub_worker_send_tool(registry);
let (_meta, tool) = def();
let input = json!({ "name": "nope", "message": "hi" }).to_string();
let err = tool.execute(&input, Default::default()).await.unwrap_err();
@@ -306,7 +308,7 @@ async fn send_to_worker_errors_when_worker_already_running() {
);
register_child(&registry, "child", &socket, tmp.path()).await;
let def = send_to_worker_tool(registry);
let def = sub_worker_send_tool(registry);
let (_meta, tool) = def();
let input = json!({ "name": "child", "message": "hi" }).to_string();
let err = tool.execute(&input, Default::default()).await.unwrap_err();
@@ -323,7 +325,7 @@ async fn send_to_worker_errors_when_worker_already_running() {
}
// ---------------------------------------------------------------------------
// ReadWorkerOutput
// SubWorkerReadOutput
// ---------------------------------------------------------------------------
#[tokio::test]
@@ -339,7 +341,7 @@ async fn read_worker_output_returns_new_assistant_text_then_empty_on_second_call
];
let _server = serve_history(listener, items);
let def = read_worker_output_tool(registry);
let def = sub_worker_read_output_tool(registry);
let (_meta, tool) = def();
let input = json!({ "name": "child" }).to_string();
@@ -370,7 +372,7 @@ async fn read_worker_output_reports_stopped_on_dead_socket() {
let dead_socket = tmp.path().join("dead.sock");
register_child(&registry, "child", &dead_socket, tmp.path()).await;
let def = read_worker_output_tool(registry);
let def = sub_worker_read_output_tool(registry);
let (_meta, tool) = def();
let input = json!({ "name": "child" }).to_string();
let output: ToolOutput = tool.execute(&input, Default::default()).await.unwrap();
@@ -378,7 +380,7 @@ async fn read_worker_output_reports_stopped_on_dead_socket() {
}
// ---------------------------------------------------------------------------
// StopWorker
// SubWorkerStop
// ---------------------------------------------------------------------------
#[tokio::test]
@@ -408,7 +410,7 @@ async fn stop_worker_sends_shutdown_and_releases_scope() {
// Seed workers.json with a restored top-level `spawner` allocation whose
// scope_deny contains the delegated child path plus the live child
// allocation — mimics a parent resumed after SpawnWorker.
// allocation — mimics a parent resumed after SubWorkerSpawn.
{
let mut g = LockFileGuard::open(&lock_path).unwrap();
let rule = ScopeRule {
@@ -451,7 +453,7 @@ async fn stop_worker_sends_shutdown_and_releases_scope() {
let received = accept_one_method(listener);
register_child(&registry, "child", &socket, tmp.path()).await;
let def = stop_worker_tool(registry.clone());
let def = sub_worker_stop_tool(registry.clone());
let (_meta, tool) = def();
let input = json!({ "name": "child" }).to_string();
let output: ToolOutput = tool.execute(&input, Default::default()).await.unwrap();
@@ -492,11 +494,11 @@ async fn stop_worker_succeeds_even_when_child_unreachable() {
}
// No live listener — socket never bound. Registered record points
// at a dead path. StopWorker should still clean up local bookkeeping.
// at a dead path. SubWorkerStop should still clean up local bookkeeping.
let dead_socket = tmp.path().join("dead.sock");
register_child(&registry, "child", &dead_socket, tmp.path()).await;
let def = stop_worker_tool(registry.clone());
let def = sub_worker_stop_tool(registry.clone());
let (_meta, tool) = def();
let input = json!({ "name": "child" }).to_string();
let output: ToolOutput = tool.execute(&input, Default::default()).await.unwrap();
@@ -550,7 +552,7 @@ async fn restored_registry_uses_worker_state_without_runtime_file() {
.await
.unwrap();
let def = send_to_worker_tool(restored.clone());
let def = sub_worker_send_tool(restored.clone());
let (_meta, tool) = def();
let input = json!({ "name": "child", "message": "after restart" }).to_string();
tool.execute(&input, Default::default()).await.unwrap();
@@ -562,7 +564,7 @@ async fn restored_registry_uses_worker_state_without_runtime_file() {
other => panic!("expected Run, got {other:?}"),
}
let def = stop_worker_tool(restored.clone());
let def = sub_worker_stop_tool(restored.clone());
let (_meta, tool) = def();
tool.execute(&json!({ "name": "child" }).to_string(), Default::default())
.await