diff --git a/crates/manifest/src/config.rs b/crates/manifest/src/config.rs index b71520ac..3dfbc816 100644 --- a/crates/manifest/src/config.rs +++ b/crates/manifest/src/config.rs @@ -83,7 +83,9 @@ pub struct FeatureConfigPartial { #[serde(default)] pub web: Option, #[serde(default)] - pub workers: Option, + pub sub_worker: Option, + #[serde(default)] + pub worker: Option, #[serde(default)] pub objective: Option, #[serde(default)] @@ -100,7 +102,12 @@ impl FeatureConfigPartial { task: merge_option(self.task, other.task, FeatureFlagConfigPartial::merge), memory: merge_option(self.memory, other.memory, MemoryFeatureConfigPartial::merge), web: merge_option(self.web, other.web, FeatureFlagConfigPartial::merge), - workers: merge_option(self.workers, other.workers, FeatureFlagConfigPartial::merge), + sub_worker: merge_option( + self.sub_worker, + other.sub_worker, + FeatureFlagConfigPartial::merge, + ), + worker: merge_option(self.worker, other.worker, FeatureFlagConfigPartial::merge), objective: merge_option( self.objective, other.objective, @@ -179,8 +186,12 @@ impl From for FeatureConfig { .map(MemoryFeatureConfig::from) .unwrap_or_default(), web: value.web.map(FeatureFlagConfig::from).unwrap_or_default(), - workers: value - .workers + sub_worker: value + .sub_worker + .map(FeatureFlagConfig::from) + .unwrap_or_default(), + worker: value + .worker .map(FeatureFlagConfig::from) .unwrap_or_default(), objective: value @@ -267,7 +278,8 @@ impl From for FeatureConfigPartial { task: Some(value.task.into()), memory: Some(value.memory.into()), web: Some(value.web.into()), - workers: Some(value.workers.into()), + sub_worker: Some(value.sub_worker.into()), + worker: Some(value.worker.into()), objective: Some(value.objective.into()), manage_workdir: Some(value.manage_workdir.into()), ticket: Some(value.ticket.into()), @@ -417,6 +429,15 @@ pub(crate) fn reject_removed_manifest_fields(s: &str) -> Result<(), toml::de::Er "unknown field in manifest: memory.extract_worker_max_input_tokens (removed)", )); } + if value + .get("feature") + .and_then(toml::Value::as_table) + .is_some_and(|table| table.contains_key("workers")) + { + return Err(toml::de::Error::custom( + "unknown field in manifest: feature.workers (removed; use feature.sub_worker)", + )); + } Ok(()) } @@ -424,8 +445,8 @@ impl WorkerManifestConfig { /// Parse a partial manifest from a TOML string. Unknown top-level or /// nested fields emit a `tracing::warn!` and are ignored; use /// `tracing_subscriber` with `WARN` enabled to surface them to the - /// operator. Removed fields that must not be silently ignored (currently - /// `compaction.prune_protected_turns`) are rejected before deserialization. + /// operator. Removed fields with an explicit replacement (including + /// `feature.workers`) are rejected before deserialization. pub fn from_toml(s: &str) -> Result { reject_removed_manifest_fields(s)?; let de = toml::Deserializer::parse(s)?; @@ -1814,7 +1835,7 @@ worker_max_turns = 7 assert!(!manifest.feature.task.enabled); assert!(!manifest.feature.memory.enabled); assert!(!manifest.feature.web.enabled); - assert!(!manifest.feature.workers.enabled); + assert!(!manifest.feature.sub_worker.enabled); assert!(!manifest.feature.objective.enabled); assert!(!manifest.feature.manage_workdir.enabled); assert!(!manifest.feature.ticket.enabled); @@ -1949,7 +1970,7 @@ enabled = true assert!(manifest.feature.ticket.orchestration_control); assert!(manifest.feature.objective.enabled); assert!(manifest.feature.web.enabled); - assert!(!manifest.feature.workers.enabled); + assert!(!manifest.feature.sub_worker.enabled); } #[test] diff --git a/crates/manifest/src/lib.rs b/crates/manifest/src/lib.rs index 54d16032..36266fff 100644 --- a/crates/manifest/src/lib.rs +++ b/crates/manifest/src/lib.rs @@ -111,7 +111,9 @@ pub struct FeatureConfig { #[serde(default)] pub web: FeatureFlagConfig, #[serde(default)] - pub workers: FeatureFlagConfig, + pub sub_worker: FeatureFlagConfig, + #[serde(default)] + pub worker: FeatureFlagConfig, #[serde(default)] pub objective: FeatureFlagConfig, #[serde(default)] @@ -128,7 +130,8 @@ impl Default for FeatureConfig { task: FeatureFlagConfig::disabled(), memory: MemoryFeatureConfig::disabled(), web: FeatureFlagConfig::disabled(), - workers: FeatureFlagConfig::disabled(), + sub_worker: FeatureFlagConfig::disabled(), + worker: FeatureFlagConfig::disabled(), objective: FeatureFlagConfig::disabled(), manage_workdir: FeatureFlagConfig::disabled(), ticket: TicketFeatureConfig::default(), @@ -405,7 +408,7 @@ pub struct MemoryConfig { /// system-prompt section. `None` ⇒ enabled. #[serde(default)] pub inject_summary: Option, - /// Language used by memory extraction / consolidation workers for durable + /// Language used by memory extraction / consolidation sub_worker for durable /// memory text. Free-form so workspaces can use names like /// `English`, `Japanese`, or locale tags. `None` ⇒ /// [`defaults::MEMORY_LANGUAGE`]. diff --git a/crates/manifest/src/profile.rs b/crates/manifest/src/profile.rs index 05df53f5..729a78b8 100644 --- a/crates/manifest/src/profile.rs +++ b/crates/manifest/src/profile.rs @@ -436,7 +436,7 @@ impl ProfileResolver { } } /// Resolve a registry/default selector against an already-discovered - /// registry. Callers such as SpawnWorker use this to bind discovery to the + /// registry. Callers such as SubWorkerSpawn use this to bind discovery to the /// Worker's cwd instead of the process current directory. pub fn resolve_from_registry( &self, @@ -936,7 +936,7 @@ fn builtin_profile_artifact(label: &str) -> Option { value["feature"]["task"] = serde_json::json!({ "enabled": false }); value["feature"]["memory"] = serde_json::json!({ "enabled": true, "staging": true }); value["feature"]["web"] = serde_json::json!({ "enabled": false }); - value["feature"]["workers"] = serde_json::json!({ "enabled": false }); + value["feature"]["sub_worker"] = serde_json::json!({ "enabled": false }); value["feature"]["objective"] = serde_json::json!({ "enabled": false }); value["feature"]["ticket"] = serde_json::json!({ "enabled": false, "thread": false }); Some(value) @@ -962,7 +962,8 @@ fn builtin_base_profile_artifact() -> serde_json::Value { "task": { "enabled": true }, "memory": { "enabled": true }, "web": { "enabled": true }, - "workers": { "enabled": true }, + "sub_worker": { "enabled": true }, + "worker": { "enabled": false }, "objective": { "enabled": true }, "ticket": { "enabled": true, "authoring": true, "thread": true } }, @@ -990,14 +991,15 @@ fn apply_role_profile( task: bool, memory: bool, web: bool, - workers: bool, + sub_worker: bool, ) { value["slug"] = serde_json::Value::String(slug.to_string()); value["description"] = serde_json::Value::String(description.to_string()); value["feature"]["task"] = serde_json::json!({ "enabled": task }); value["feature"]["memory"] = serde_json::json!({ "enabled": memory }); value["feature"]["web"] = serde_json::json!({ "enabled": web }); - value["feature"]["workers"] = serde_json::json!({ "enabled": workers }); + value["feature"]["sub_worker"] = serde_json::json!({ "enabled": sub_worker }); + value["feature"]["worker"] = serde_json::json!({ "enabled": slug == "orchestrator" }); value["feature"]["manage_workdir"] = serde_json::json!({ "enabled": slug == "orchestrator" }); let ticket = match slug { "companion" => serde_json::json!({ "enabled": true, "authoring": true, "thread": true }), @@ -1456,7 +1458,8 @@ mod tests { let companion = resolve("companion"); assert!(companion.feature.task.enabled); - assert!(companion.feature.workers.enabled); + assert!(companion.feature.sub_worker.enabled); + assert!(!companion.feature.worker.enabled); assert!(companion.scope.allow.is_empty()); assert!(companion.scope.deny.is_empty()); assert!(companion.delegation_scope.allow.is_empty()); @@ -1487,7 +1490,8 @@ mod tests { let intake = resolve("intake"); assert!(intake.feature.task.enabled); - assert!(!intake.feature.workers.enabled); + assert!(!intake.feature.sub_worker.enabled); + assert!(!intake.feature.worker.enabled); assert!(intake.feature.ticket.enabled); assert!(intake.feature.ticket.enabled); assert!(intake.feature.ticket.authoring); @@ -1504,7 +1508,8 @@ mod tests { let orchestrator = resolve("orchestrator"); assert!(orchestrator.feature.task.enabled); - assert!(!orchestrator.feature.workers.enabled); + assert!(!orchestrator.feature.sub_worker.enabled); + assert!(orchestrator.feature.worker.enabled); assert!(orchestrator.feature.ticket.enabled); assert!(orchestrator.feature.ticket.enabled); assert!(!orchestrator.feature.ticket.authoring); @@ -1524,7 +1529,8 @@ mod tests { let coder = resolve("coder"); assert!(coder.feature.task.enabled); - assert!(!coder.feature.workers.enabled); + assert!(!coder.feature.sub_worker.enabled); + assert!(!coder.feature.worker.enabled); assert!(coder.scope.allow.is_empty()); assert!(coder.delegation_scope.allow.is_empty()); assert_eq!(coder.model.ref_.as_deref(), Some("codex-oauth/gpt-5.5")); @@ -1540,7 +1546,8 @@ mod tests { assert!(!coder.feature.ticket.orchestration_control); let reviewer = resolve("reviewer"); assert!(reviewer.feature.task.enabled); - assert!(!reviewer.feature.workers.enabled); + assert!(!reviewer.feature.sub_worker.enabled); + assert!(!reviewer.feature.worker.enabled); assert!(reviewer.feature.ticket.enabled); assert!(reviewer.feature.ticket.enabled); assert!(!reviewer.feature.ticket.authoring); @@ -1692,7 +1699,7 @@ enabled = false [feature.web] enabled = true -[feature.workers] +[feature.sub_worker] enabled = true [feature.ticket] @@ -1716,7 +1723,7 @@ orchestration_control = false assert!(resolved.manifest.feature.task.enabled); assert!(!resolved.manifest.feature.memory.enabled); assert!(resolved.manifest.feature.web.enabled); - assert!(resolved.manifest.feature.workers.enabled); + assert!(resolved.manifest.feature.sub_worker.enabled); assert!(resolved.manifest.feature.ticket.enabled); assert!(!resolved.manifest.feature.ticket.authoring); assert!(!resolved.manifest.feature.ticket.thread); diff --git a/crates/manifest/src/scope.rs b/crates/manifest/src/scope.rs index fad12dea..67756c61 100644 --- a/crates/manifest/src/scope.rs +++ b/crates/manifest/src/scope.rs @@ -331,7 +331,7 @@ impl Scope { /// Build a new [`Scope`] equal to `self` with `extra_deny` appended /// to the deny set. Used by dynamic-scope shrink paths - /// (e.g. SpawnWorker-style delegation that strips Write from the + /// (e.g. SubWorkerSpawn-style delegation that strips Write from the /// spawner without touching its allow rules). pub fn with_added_deny_rules( &self, diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index fe33ca35..5a8e29a0 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -125,7 +125,7 @@ pub enum WorkerEvent { /// Child has stopped (controller loop is exiting). ShutDown { worker_name: String }, - /// Child sub-delegated scope to a grandchild Worker via `SpawnWorker`. + /// Child SubWorker sub-delegated scope to a grandchild SubWorker via `SubWorkerSpawn`. /// /// Control-plane only: receivers apply registry side effects and /// propagate upward, but do not expose this as an agent notification. diff --git a/crates/session-analytics/src/lib.rs b/crates/session-analytics/src/lib.rs index 88466ead..4b5bef3e 100644 --- a/crates/session-analytics/src/lib.rs +++ b/crates/session-analytics/src/lib.rs @@ -1598,8 +1598,22 @@ fn tool_kind(name: &str) -> &'static str { "Read" | "Write" | "Edit" | "Glob" | "Grep" => "filesystem", "Bash" => "shell", "WebFetch" | "WebSearch" => "web", - "SpawnWorker" | "SendToWorker" | "SendToPeerWorker" | "ReadWorkerOutput" - | "ListWorkers" | "StopWorker" | "RestoreWorker" => "worker", + "SubWorkerSpawn" + | "SubWorkerSend" + | "SubWorkerReadOutput" + | "SubWorkerList" + | "SubWorkerStop" + | "WorkerList" + | "WorkerSpawn" + | "WorkerStop" + | "WorkerRestore" + | "SpawnWorker" + | "SendToWorker" + | "SendToPeerWorker" + | "ReadWorkerOutput" + | "ListWorkers" + | "StopWorker" + | "RestoreWorker" => "worker", // Legacy session logs used the pre-rename peer tool name; keep analytics classification only. /* legacy session-log tool name only */ LEGACY_SEND_TO_PEER_POD_TOOL => "worker", diff --git a/crates/tui/src/dashboard/mod.rs b/crates/tui/src/dashboard/mod.rs index d72c422d..c6942f9e 100644 --- a/crates/tui/src/dashboard/mod.rs +++ b/crates/tui/src/dashboard/mod.rs @@ -4966,7 +4966,7 @@ fn orchestrator_queue_notification_message( ) -> String { let title = ticket.title.replace(['\r', '\n'], " "); format!( - "Workspace Dashboard Queue for Ticket `{}`, title `{}`: human authorized Orchestrator routing; this is not an unattended scheduler. Read the Ticket and inspect current Orchestrator workspace state. If unblocked, record routing and transition state queued -> inprogress before any worktree/SpawnWorker implementation side effects. After inprogress acceptance, create the delegated implementation worktree with tracked `.yoi` project records visible and generated/local/runtime/log/lock/secret-like `.yoi` paths excluded, then run sibling coder/reviewer Workers through typed Ticket role launch surfaces. After reviewer approval and blocker resolution, integrate the implementation branch into the orchestration branch automatically, validate in the Orchestrator worktree, record the outcome, and clean up only child implementation worktrees/branches. Do not read, write, validate, merge, clean up, or run git operations in the root/original workspace. If blocked, record a concise reason and leave the Ticket queued or return it to planning with the missing-information reason.", + "Workspace Dashboard Queue for Ticket `{}`, title `{}`: human authorized Orchestrator routing; this is not an unattended scheduler. Read the Ticket and inspect current Orchestrator workspace state. If unblocked, record routing and transition state queued -> inprogress before any worktree/WorkerSpawn implementation side effects. After inprogress acceptance, create the delegated implementation worktree with tracked `.yoi` project records visible and generated/local/runtime/log/lock/secret-like `.yoi` paths excluded, then run sibling coder/reviewer Workers through typed Ticket role launch surfaces. After reviewer approval and blocker resolution, integrate the implementation branch into the orchestration branch automatically, validate in the Orchestrator worktree, record the outcome, and clean up only child implementation worktrees/branches. Do not read, write, validate, merge, clean up, or run git operations in the root/original workspace. If blocked, record a concise reason and leave the Ticket queued or return it to planning with the missing-information reason.", ticket.id, title.trim() ) diff --git a/crates/tui/src/dashboard/tests.rs b/crates/tui/src/dashboard/tests.rs index c18ed9ec..b0e21082 100644 --- a/crates/tui/src/dashboard/tests.rs +++ b/crates/tui/src/dashboard/tests.rs @@ -823,7 +823,7 @@ fn ticket_queue_notification_message_carries_routing_contract() { assert!(message.contains("Read the Ticket")); assert!(message.contains("inspect current Orchestrator workspace state")); assert!(message.contains("transition state queued -> inprogress")); - assert!(message.contains("before any worktree/SpawnWorker implementation side effects")); + assert!(message.contains("before any worktree/WorkerSpawn implementation side effects")); assert!(message.contains("After inprogress acceptance")); assert!(message.contains("implementation worktree")); assert!(message.contains("tracked `.yoi` project records visible")); diff --git a/crates/tui/src/setup_model.rs b/crates/tui/src/setup_model.rs index b554dd0c..951b434b 100644 --- a/crates/tui/src/setup_model.rs +++ b/crates/tui/src/setup_model.rs @@ -233,7 +233,7 @@ enabled = true [feature.web] enabled = true -[feature.workers] +[feature.sub_worker] enabled = false [feature.ticket] diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 55e3fa8a..424bd110 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -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( } /// 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( @@ -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); diff --git a/crates/worker/src/entrypoint.rs b/crates/worker/src/entrypoint.rs index 7c6d3823..60bc4657 100644 --- a/crates/worker/src/entrypoint.rs +++ b/crates/worker/src/entrypoint.rs @@ -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, diff --git a/crates/worker/src/feature/builtin.rs b/crates/worker/src/feature/builtin.rs index 45057c0d..c92bf10d 100644 --- a/crates/worker/src/feature/builtin.rs +++ b/crates/worker/src/feature/builtin.rs @@ -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; diff --git a/crates/worker/src/feature/builtin/manage_worker.rs b/crates/worker/src/feature/builtin/manage_worker.rs new file mode 100644 index 00000000..7b93967e --- /dev/null +++ b/crates/worker/src/feature/builtin/manage_worker.rs @@ -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, +} + +pub fn manage_worker_feature(client: Arc) -> 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::( + operation, + self.client.clone(), + workspace_id.clone(), + ), + WorkerOperation::Spawn => definition::( + operation, + self.client.clone(), + workspace_id.clone(), + ), + WorkerOperation::Stop => definition::( + operation, + self.client.clone(), + workspace_id.clone(), + ), + WorkerOperation::Restore => definition::( + 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, + #[serde(default)] + initial_text: Option, + #[serde(default)] + relative_cwd: Option, +} + +#[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, +} + +#[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, +} + +struct WorkspaceWorkerTool { + operation: WorkerOperation, + client: Arc, + 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 { + let request = match self.operation { + WorkerOperation::List => { + parse::(input_json, "WorkerList")?; + WorkspaceRequest::get(format!("/api/w/{}/workers", self.workspace_id)) + } + WorkerOperation::Spawn => { + let input = parse::(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::(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::(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( + operation: WorkerOperation, + client: Arc, + 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 = Arc::new(WorkspaceWorkerTool { + operation, + client: client.clone(), + workspace_id: workspace_id.clone(), + }); + (meta, tool) + }) +} + +fn parse Deserialize<'de>>(input: &str, tool: &str) -> Result { + serde_json::from_str(input) + .map_err(|error| ToolError::InvalidArgument(format!("invalid {tool} input: {error}"))) +} + +fn authority_id(value: &str, field: &str) -> Result { + 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 { + 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 { + 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"); + } +} diff --git a/crates/worker/src/ipc/event.rs b/crates/worker/src/ipc/event.rs index 0c8349cf..d4155719 100644 --- a/crates/worker/src/ipc/event.rs +++ b/crates/worker/src/ipc/event.rs @@ -59,7 +59,7 @@ pub fn fire_and_forget(socket: Option, 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 { diff --git a/crates/worker/src/prompt/catalog.rs b/crates/worker/src/prompt/catalog.rs index 4d1be72d..3c780a0c 100644 --- a/crates/worker/src/prompt/catalog.rs +++ b/crates/worker/src/prompt/catalog.rs @@ -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", "", diff --git a/crates/worker/src/prompt/system.rs b/crates/worker/src/prompt/system.rs index f4fb8b09..5f4b1dd2 100644 --- a/crates/worker/src/prompt/system.rs +++ b/crates/worker/src/prompt/system.rs @@ -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(); diff --git a/crates/worker/src/spawn/comm_tools.rs b/crates/worker/src/spawn/comm_tools.rs index 7c726761..e86b105d 100644 --- a/crates/worker/src/spawn/comm_tools.rs +++ b/crates/worker/src/spawn/comm_tools.rs @@ -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, } #[async_trait] -impl Tool for SendToWorkerTool { +impl Tool for SubWorkerListTool { async fn execute( &self, input_json: &str, _ctx: llm_engine::tool::ToolExecutionContext, ) -> Result { - 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::>(); + 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) -> 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 `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, +} + +#[async_trait] +impl Tool for SubWorkerSendTool { + async fn execute( + &self, + input_json: &str, + _ctx: llm_engine::tool::ToolExecutionContext, + ) -> Result { + 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) -> ToolDefinition { +pub fn sub_worker_send_tool(registry: Arc) -> 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 = Arc::new(SendToWorkerTool { + let tool: Arc = Arc::new(SubWorkerSendTool { registry: registry.clone(), }); (meta, tool) @@ -113,27 +169,27 @@ pub fn send_to_worker_tool(registry: Arc) -> 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, } #[async_trait] -impl Tool for ReadWorkerOutputTool { +impl Tool for SubWorkerReadOutputTool { async fn execute( &self, input_json: &str, _ctx: llm_engine::tool::ToolExecutionContext, ) -> Result { 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) -> ToolDefinition { +pub fn sub_worker_read_output_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("ReadWorkerOutput") + let meta = ToolMeta::new("SubWorkerReadOutput") .description(READ_POD_OUTPUT_DESCRIPTION) .input_schema(schema_value); - let tool: Arc = Arc::new(ReadWorkerOutputTool { + let tool: Arc = Arc::new(SubWorkerReadOutputTool { registry: registry.clone(), }); (meta, tool) @@ -193,26 +249,26 @@ pub fn read_worker_output_tool(registry: Arc) -> 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, } #[async_trait] -impl Tool for StopWorkerTool { +impl Tool for SubWorkerStopTool { async fn execute( &self, input_json: &str, _ctx: llm_engine::tool::ToolExecutionContext, ) -> Result { 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) -> ToolDefinition { +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("StopWorker") + let meta = ToolMeta::new("SubWorkerStop") .description(STOP_POD_DESCRIPTION) .input_schema(schema_value); - let tool: Arc = Arc::new(StopWorkerTool { + let tool: Arc = 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 diff --git a/crates/worker/src/spawn/registry.rs b/crates/worker/src/spawn/registry.rs index 3081f6d4..900cb99f 100644 --- a/crates/worker/src/spawn/registry.rs +++ b/crates/worker/src/spawn/registry.rs @@ -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. diff --git a/crates/worker/src/spawn/tool.rs b/crates/worker/src/spawn/tool.rs index 6bd00454..56b75dce 100644 --- a/crates/worker/src/spawn/tool.rs +++ b/crates/worker/src/spawn/tool.rs @@ -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, /// 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, - /// 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, @@ -189,7 +189,7 @@ fn parse_spawn_profile_selector(raw: Option<&str>) -> Result) -> Result 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, /// 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, @@ -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 { - 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, ) -> 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, 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, ) -> 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 = Arc::new(SpawnWorkerTool::new( + let tool: Arc = 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) diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 1611359e..c051ec91 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -654,7 +654,7 @@ pub struct Worker { /// 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`. /// diff --git a/crates/worker/tests/controller_test.rs b/crates/worker/tests/controller_test.rs index 7d09ad0b..51735481 100644 --- a/crates/worker/tests/controller_test.rs +++ b/crates/worker/tests/controller_test.rs @@ -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}" ); } diff --git a/crates/worker/tests/spawn_worker_test.rs b/crates/worker/tests/spawn_worker_test.rs index 1f080f2d..c06d94b6 100644 --- a/crates/worker/tests/spawn_worker_test.rs +++ b/crates/worker/tests/spawn_worker_test.rs @@ -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, diff --git a/crates/worker/tests/worker_comm_tools_test.rs b/crates/worker/tests/worker_comm_tools_test.rs index 91a04544..7ff4b63e 100644 --- a/crates/worker/tests/worker_comm_tools_test.rs +++ b/crates/worker/tests/worker_comm_tools_test.rs @@ -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> { } /// 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) -> 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(®istry, "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(®istry, "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(®istry, "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(®istry, "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(®istry, "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 diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 74903e0a..4e10c56d 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -3930,7 +3930,7 @@ mod tests { } #[test] - fn embedded_orchestrator_profile_enables_manage_workdir() { + fn embedded_orchestrator_profile_enables_workdir_and_worker_authority() { let root = tempfile::tempdir().unwrap(); let broker = BackendResourceBroker::default(); let runtime_id = "runtime-test"; @@ -3962,7 +3962,8 @@ mod tests { .unwrap(); assert!(manifest.feature.manage_workdir.enabled); - assert!(!manifest.feature.workers.enabled); + assert!(!manifest.feature.sub_worker.enabled); + assert!(manifest.feature.worker.enabled); } #[test] diff --git a/docs/design/profiles-manifests-prompts.md b/docs/design/profiles-manifests-prompts.md index 6386cd7e..88773689 100644 --- a/docs/design/profiles-manifests-prompts.md +++ b/docs/design/profiles-manifests-prompts.md @@ -65,9 +65,9 @@ name = "MCP_UPSTREAM_TOKEN" Local stdio MCP servers are ordinary local executables running with the user's OS permissions. Yoi's feature flags, Plugin permissions, and MCP config validation are not an operating-system sandbox and cannot prevent filesystem/network/process side effects once a later lifecycle implementation chooses to spawn a configured server. -## Spawned Workers +## SubWorkers -`SpawnWorker.profile` is optional and resolves through defaults when omitted. The only concrete capability delegation in the tool call is `SpawnWorker.scope`, and it must be a subset of the parent's effective scope. +`SubWorkerSpawn.profile` is optional and resolves through defaults when omitted. The only concrete capability delegation in the tool call is `SubWorkerSpawn.scope`, and it must be a subset of the parent Worker's effective scope. `inherit` derives reusable settings from the parent's resolved Manifest while replacing child identity and delegated scope. It should not blindly reuse the parent's original Profile source or runtime state. diff --git a/docs/development/work-items.md b/docs/development/work-items.md index e5a24312..cb8b898c 100644 --- a/docs/development/work-items.md +++ b/docs/development/work-items.md @@ -253,7 +253,7 @@ Unless explicitly authorized otherwise, final merge, cleanup, design-boundary de Before closing, verify concrete evidence: -- child Worker output via `ReadWorkerOutput`; +- SubWorker output via `SubWorkerReadOutput`; - worktree state and diff; - validation command output; - review result; diff --git a/resources/profiles/base.dcdl b/resources/profiles/base.dcdl index d04d30dd..cd78e186 100644 --- a/resources/profiles/base.dcdl +++ b/resources/profiles/base.dcdl @@ -25,7 +25,8 @@ feature = { task = { enabled = true; }; memory = { enabled = true; }; web = { enabled = true; }; - workers = { enabled = true; }; + sub_worker = { enabled = true; }; + worker = { enabled = false; }; objective = { enabled = true; }; ticket = { enabled = true; authoring = true; thread = true; }; }; diff --git a/resources/profiles/coder.dcdl b/resources/profiles/coder.dcdl index 4fdcad5e..e010da1a 100644 --- a/resources/profiles/coder.dcdl +++ b/resources/profiles/coder.dcdl @@ -7,7 +7,8 @@ import "./base.dcdl" // { task = { enabled = true; }; memory = { enabled = true; }; web = { enabled = true; }; - workers = { enabled = false; }; + sub_worker = { enabled = false; }; + worker = { enabled = false; }; ticket = { enabled = true; thread = true; }; }; } diff --git a/resources/profiles/companion.dcdl b/resources/profiles/companion.dcdl index 50e70e1f..150db6a0 100644 --- a/resources/profiles/companion.dcdl +++ b/resources/profiles/companion.dcdl @@ -7,7 +7,8 @@ import "./base.dcdl" // { task = { enabled = true; }; memory = { enabled = true; }; web = { enabled = true; }; - workers = { enabled = true; }; + sub_worker = { enabled = true; }; + worker = { enabled = false; }; ticket = { enabled = true; authoring = true; thread = true; }; }; } diff --git a/resources/profiles/intake.dcdl b/resources/profiles/intake.dcdl index 5e88e598..ae69be4b 100644 --- a/resources/profiles/intake.dcdl +++ b/resources/profiles/intake.dcdl @@ -7,7 +7,8 @@ import "./base.dcdl" // { task = { enabled = true; }; memory = { enabled = true; }; web = { enabled = true; }; - workers = { enabled = false; }; + sub_worker = { enabled = false; }; + worker = { enabled = false; }; ticket = { enabled = true; authoring = true; thread = true; intake = true; }; }; } diff --git a/resources/profiles/memory-consolidation.dcdl b/resources/profiles/memory-consolidation.dcdl index 34a030eb..689fb934 100644 --- a/resources/profiles/memory-consolidation.dcdl +++ b/resources/profiles/memory-consolidation.dcdl @@ -7,7 +7,8 @@ import "./base.dcdl" // { task = { enabled = false; }; memory = { enabled = true; staging = true; }; web = { enabled = false; }; - workers = { enabled = false; }; + sub_worker = { enabled = false; }; + worker = { enabled = false; }; objective = { enabled = false; }; ticket = { enabled = false; thread = false; }; }; diff --git a/resources/profiles/orchestrator.dcdl b/resources/profiles/orchestrator.dcdl index 76c263cc..5c82075a 100644 --- a/resources/profiles/orchestrator.dcdl +++ b/resources/profiles/orchestrator.dcdl @@ -7,7 +7,8 @@ import "./base.dcdl" // { task = { enabled = true; }; memory = { enabled = true; }; web = { enabled = true; }; - workers = { enabled = false; }; + sub_worker = { enabled = false; }; + worker = { enabled = true; }; manage_workdir = { enabled = true; }; ticket = { enabled = true; thread = true; orchestration_control = true; }; }; diff --git a/resources/profiles/reviewer.dcdl b/resources/profiles/reviewer.dcdl index 012a729f..d3938983 100644 --- a/resources/profiles/reviewer.dcdl +++ b/resources/profiles/reviewer.dcdl @@ -7,7 +7,8 @@ import "./base.dcdl" // { task = { enabled = true; }; memory = { enabled = true; }; web = { enabled = true; }; - workers = { enabled = false; }; + sub_worker = { enabled = false; }; + worker = { enabled = false; }; ticket = { enabled = true; thread = true; }; }; } diff --git a/resources/prompts/common/worker-orchestration.md b/resources/prompts/common/worker-orchestration.md index bcab4a7b..9c2a0104 100644 --- a/resources/prompts/common/worker-orchestration.md +++ b/resources/prompts/common/worker-orchestration.md @@ -1,11 +1,11 @@ --- -## Worker orchestration +## SubWorker orchestration -When Worker-management tools are available, spawned Worker notifications are background signals for the parent 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, 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 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 Worker output; if there is no useful immediate work, return control and handle the child when notified or when the user next asks. +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. -Before treating delegated work as complete, read the child output and inspect concrete evidence such as worktree state, diff, and test results. Notifications are hints, not proof of completion. +Before treating delegated SubWorker work as complete, read the SubWorker output and inspect concrete evidence such as worktree state, diff, and test results. Notifications are hints, not proof of completion. Peer Workers made visible by reciprocal metadata registration are not spawned children. Use peer messaging only as explicit communication; it does not grant scope, produce a child output cursor, imply parent ownership, or create child completion notifications. Peer sends require a live peer and do not auto-restore stopped peers. diff --git a/resources/prompts/internal.toml b/resources/prompts/internal.toml index 5b3bd2a5..01bd6f41 100644 --- a/resources/prompts/internal.toml +++ b/resources/prompts/internal.toml @@ -53,12 +53,12 @@ worker_orchestration_guidance_section = "{% include \"$yoi/common/worker-orchest ticket_event_companion_notice = "{% include \"$yoi/worker/ticket_event_companion_notice\" %}" -spawn_worker_tool_description = """\ -Spawn a new Worker process to work on a delegated task. The spawner's write scope is reduced by the scope passed here; the spawned Worker receives its own socket and starts running `task` immediately. The spawned Worker outlives the spawner's current turn and can be contacted again through its socket path. +sub_worker_spawn_tool_description = """\ +Spawn a new SubWorker process to split context for a delegated task. The parent Worker's write scope is reduced by the scope passed here; the SubWorker receives its own socket and starts running `task` immediately. The SubWorker outlives the parent Worker's current turn and can be contacted again through its socket path. Optional `cwd`: when provided, it is the child process/tool default working directory only. It must be an absolute existing directory covered by the child's delegated readable scope, and it does not change workspace/Profile/memory/Ticket roots or grant authority. -Profile selection: `profile` may be omitted or set to `default` to use the effective child default profile, set to `inherit` to derive reusable child configuration from this Worker, or set to one of the registry selectors below. Raw/path profile selectors are not accepted by SpawnWorker. `scope` is always the only delegated filesystem capability; profile scope is replaced by the explicit SpawnWorker scope. +Profile selection: `profile` may be omitted or set to `default` to use the effective child default profile, set to `inherit` to derive reusable child configuration from this Worker, or set to one of the registry selectors below. Raw/path profile selectors are not accepted by SubWorkerSpawn. `scope` is always the only delegated filesystem capability; profile scope is replaced by the explicit SubWorkerSpawn scope. Default profile: {{ default_profile }} Special selector: inherit — derive reusable model/worker/tool policy from the spawner while replacing worker.name and scope.