diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index cdae53d00..3bc7c180d 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -22,7 +22,6 @@ use crate::shutdown_after_idle::{ take_shutdown_request_after_status, }; use crate::spawn::registry::{PreparedSpawnedWorkerRegistryLoad, SpawnedWorkerRegistry}; -use crate::spawn::tool::sub_worker_spawn_tool; use crate::worker::{ PendingSubmissionHandle, SystemItemCommitter, Worker, WorkerError, WorkerRunResult, }; @@ -1427,10 +1426,9 @@ fn validate_memory_lifecycle_targets( Ok(()) } -/// Register the builtin file-manipulation tools, optional memory tools, -/// 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. +/// Compose Worker Features with Host-owned live bindings and install their +/// staged contributions. Remaining direct registrations are Worker-lifetime +/// infrastructure (paste/Workdir) and the image/web migration follow-ups. pub(crate) async fn register_worker_tools( worker: &mut Worker, @@ -1463,7 +1461,6 @@ where let task_feature = worker.task_feature(); let web_config = worker.manifest().web.clone(); let mcp_config = worker.manifest().mcp.clone(); - let spawner_name = worker.manifest().worker.name.clone(); let spawner_manifest = worker.manifest().clone(); let spawner_workspace_context = worker.workspace_context_handle(); let pending_submissions = worker.pending_submission_handle(); @@ -1900,10 +1897,6 @@ where } } - if feature_config.sub_worker.enabled { - worker.register_worker_orchestration_instruction(); - } - let host_worker_observation_provider = worker.worker_observation_provider(); let wip_permissions = worker.manifest().permissions.clone(); let mut wip_mount_registry = crate::wip::WipMountRegistry::new(); @@ -1973,7 +1966,6 @@ where } { let workspace_client = worker.workspace_client_handle(); - let engine = worker.engine_mut(); // Allocate cross-domain input reference coordinates before mounting either provider. // Allocation is not publication or authorization; current handlers still check both. @@ -2050,66 +2042,29 @@ where })?; } - // Objective tools expose scoped project Objective reads and mutations through - // the Backend Workspace API. Workers must not guess local `.yoi/objectives` - // paths or read/edit Objective files directly. - if feature_config.objective.enabled { - if workspace_client.is_available() && workspace_client.workspace_id().is_some() { - for definition in crate::feature::builtin::objective::workspace_http_objective_tools( - workspace_client.clone(), - ) { - engine.register_tool(definition); - } - if let Some(namespace_route) = &objective_namespace { - crate::feature::builtin::objective::mount_workspace_http_objective_wip( - &mut wip_mount_registry, - workspace_client.clone(), - wip_permissions.clone(), - &namespace_route, - ) - .map_err(|error| { - std::io::Error::new( - std::io::ErrorKind::InvalidInput, - format!("mount Objective WIP projection: {error}"), - ) - })?; - } - } else { - return Err(std::io::Error::new( - std::io::ErrorKind::InvalidInput, - "objective tools require Backend Workspace API authority", - )); - } - } + crate::feature::builtin::objective::composition::compose( + &mut feature_registry, + feature_config.objective.enabled, + workspace_client, + objective_namespace.map(|route| (route, wip_permissions.clone())), + )?; let mut observation_providers: Vec< Arc, > = Vec::new(); - // Worker-orchestration tools derive child filesystem authority from the - // active provider-backed Workdir session. The tool remains registered - // without one so invocation fails deterministically until the parent - // attaches a Workdir. - if feature_config.sub_worker.enabled { - let spawner_workspace_root = local_workspace_root - .clone() - .unwrap_or_else(|| PathBuf::from("/")); - engine.register_tool(sub_worker_spawn_tool( - spawner_name.clone(), - spawner_workspace_context, - parent_notifications, - runtime_base.clone(), - spawner_workspace_root, - workdir_tool_broker, - spawned_registry.clone(), - spawner_manifest, - prompts, - )); - observation_providers.push(Arc::new( - crate::feature::builtin::worker_observation::SpawnedSubWorkerObservationProvider::new( - spawned_registry, - ), - )); + if let Some(provider) = crate::feature::builtin::sub_worker::compose( + &mut feature_registry, + spawner_manifest, + spawner_workspace_context, + parent_notifications, + runtime_base, + local_workspace_root, + workdir_tool_broker, + spawned_registry, + prompts, + ) { + observation_providers.push(provider); } if let Some(provider) = host_worker_observation_provider { observation_providers.push(provider); @@ -2126,7 +2081,8 @@ where ); } } - let feature_install_report = worker.install_features(feature_registry); + let feature_install_report = + worker.install_features_with_wip(feature_registry, Some(&mut wip_mount_registry)); if feature_install_report.has_errors() { return Err(std::io::Error::new( std::io::ErrorKind::InvalidInput, diff --git a/crates/worker/src/feature.rs b/crates/worker/src/feature.rs index a59075e87..0f03a3c42 100644 --- a/crates/worker/src/feature.rs +++ b/crates/worker/src/feature.rs @@ -10,6 +10,13 @@ //! the normal Engine tool path, hooks are installed through //! [`crate::hook::HookRegistryBuilder`], and provider output is represented as //! ordinary feature reports/diagnostics instead of a separate authority layer. +//! +//! Native WIP contributions use a Host-supplied staging registry. Descriptor +//! preflight and the existing module/scope rollback cover their publication +//! metadata together with Tools, instructions, hooks, and services. Snapshotting +//! mount metadata clones live handles, not permission decisions or Client +//! observations. This is install rollback, not runtime lease/revocation; final +//! WIP aggregation and namespace allocation remain Host-owned. use std::any::{Any, type_name}; use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; @@ -625,6 +632,9 @@ pub struct FeatureDescriptor { pub tools: Vec, pub hooks: Vec, pub instructions: Vec, + /// Named native WIP contributions staged with the rest of the install scope. + #[serde(default)] + pub wip_mounts: Vec, pub background_tasks: Vec, pub provides_services: Vec, pub requires_services: Vec, @@ -645,6 +655,7 @@ impl FeatureDescriptor { tools: Vec::new(), hooks: Vec::new(), instructions: Vec::new(), + wip_mounts: Vec::new(), background_tasks: Vec::new(), provides_services: Vec::new(), requires_services: Vec::new(), @@ -668,6 +679,11 @@ impl FeatureDescriptor { self } + pub fn with_wip_mount(mut self, name: impl Into) -> Self { + self.wip_mounts.push(name.into()); + self + } + pub fn with_instruction(mut self, instruction: FeatureInstructionDeclaration) -> Self { self.instructions.push(instruction); self @@ -751,6 +767,7 @@ pub enum FeatureContributionKind { Tool, Hook, Instruction, + WipMount, BackgroundTask, Service, ProtocolProvider, @@ -775,6 +792,8 @@ pub struct FeatureInstallReport { pub runtime: FeatureRuntimeKind, pub installed: bool, pub installed_tools: Vec, + #[serde(default)] + pub installed_wip_mounts: Vec, pub installed_hooks: Vec, pub installed_instructions: Vec, pub declared_background_tasks: Vec, @@ -793,6 +812,7 @@ impl FeatureInstallReport { runtime: descriptor.runtime.clone(), installed: false, installed_tools: Vec::new(), + installed_wip_mounts: Vec::new(), installed_hooks: Vec::new(), installed_instructions: Vec::new(), declared_background_tasks: Vec::new(), @@ -808,6 +828,7 @@ impl FeatureInstallReport { fn clear_installed_contributions(&mut self) { self.installed = false; self.installed_tools.clear(); + self.installed_wip_mounts.clear(); self.installed_hooks.clear(); self.installed_instructions.clear(); self.declared_background_tasks.clear(); @@ -834,6 +855,7 @@ struct FeatureContributionDeclarations { tools: HashSet, hooks: HashSet<(String, FeatureHookPoint)>, instructions: HashSet, + wip_mounts: HashSet, background_tasks: HashSet, provided_services: HashSet<(ServiceId, String)>, required_services: HashSet, @@ -879,6 +901,7 @@ impl FeatureContributionDeclarations { .iter() .map(|provider| provider.id.clone()) .collect(), + wip_mounts: descriptor.wip_mounts.iter().cloned().collect(), chat_invocations: descriptor.chat_invocations.clone(), } } @@ -1309,6 +1332,53 @@ impl FeatureInstructionRegistrar<'_> { } } +/// Native projection registrar. Mounts mutate only a Host-supplied staging +/// registry; module or scope failure restores its previous publication exactly. +/// Namespace allocation stays with trusted composition, not the consumer. +pub struct FeatureWipRegistrar<'a> { + feature_id: &'a FeatureId, + declarations: &'a FeatureContributionDeclarations, + registry: Option<&'a mut crate::wip::WipMountRegistry>, + report: &'a mut FeatureInstallReport, +} + +impl FeatureWipRegistrar<'_> { + pub fn register( + &mut self, + name: &str, + mount: impl FnOnce(&mut crate::wip::WipMountRegistry) -> Result<(), crate::wip::WipMountError>, + ) -> Result<(), FeatureInstallError> { + if !self.declarations.wip_mounts.contains(name) { + return Err(reject_undeclared_contribution( + self.feature_id, + self.report, + FeatureContributionKind::WipMount, + name, + )); + } + if self + .report + .installed_wip_mounts + .iter() + .any(|installed| installed == name) + { + return Err(FeatureInstallError::InvalidDescriptor(format!( + "duplicate WIP contribution: {name}" + ))); + } + let registry = self.registry.as_deref_mut().ok_or_else(|| { + FeatureInstallError::InvalidDescriptor(format!( + "WIP contribution {name} requires a Host mount registry" + )) + })?; + mount(registry).map_err(|error| { + FeatureInstallError::InvalidDescriptor(format!("mount {name}: {error}")) + })?; + self.report.installed_wip_mounts.push(name.to_owned()); + Ok(()) + } +} + /// Registrar for executable, Worker-managed background task contributions. pub struct BackgroundTaskRegistrar<'a> { feature_id: &'a FeatureId, @@ -1553,6 +1623,7 @@ pub struct FeatureInstallContext<'a> { feature_id: &'a FeatureId, declarations: &'a FeatureContributionDeclarations, pending_tools: &'a mut Vec, + wip_mounts: Option<&'a mut crate::wip::WipMountRegistry>, installed_tool_names: &'a mut HashMap, hook_builder: &'a mut HookRegistryBuilder, background_task_builder: &'a mut FeatureBackgroundTaskRegistryBuilder, @@ -1566,6 +1637,15 @@ impl FeatureInstallContext<'_> { self.feature_id } + pub fn wip(&mut self) -> FeatureWipRegistrar<'_> { + FeatureWipRegistrar { + feature_id: self.feature_id, + declarations: self.declarations, + registry: self.wip_mounts.as_deref_mut(), + report: self.report, + } + } + pub fn tools(&mut self) -> ToolContributionRegistrar<'_> { ToolContributionRegistrar { feature_id: self.feature_id, @@ -2039,10 +2119,20 @@ impl FeatureRegistryBuilder { } /// Install modules into the existing Engine tool path and hook builder. + #[cfg(test)] pub(crate) fn install_into_engine( self, worker: &mut Engine, hook_builder: &mut HookRegistryBuilder, + ) -> FeatureRegistryInstallReport { + self.install_into_engine_with_wip(worker, hook_builder, None) + } + + pub(crate) fn install_into_engine_with_wip( + self, + worker: &mut Engine, + hook_builder: &mut HookRegistryBuilder, + wip_mounts: Option<&mut crate::wip::WipMountRegistry>, ) -> FeatureRegistryInstallReport { let mut pending_tools = Vec::new(); worker.tool_server_handle().flush_pending(); @@ -2056,6 +2146,7 @@ impl FeatureRegistryBuilder { &mut pending_tools, hook_builder, registered_tool_names, + wip_mounts, ); if !report.has_errors() { worker.register_tools(pending_tools); @@ -2069,7 +2160,7 @@ impl FeatureRegistryBuilder { pending_tools: &mut Vec, hook_builder: &mut HookRegistryBuilder, ) -> FeatureRegistryInstallReport { - self.install_into_pending_with_registered(pending_tools, hook_builder, HashMap::new()) + self.install_into_pending_with_registered(pending_tools, hook_builder, HashMap::new(), None) } fn install_into_pending_with_registered( @@ -2077,6 +2168,7 @@ impl FeatureRegistryBuilder { pending_tools: &mut Vec, hook_builder: &mut HookRegistryBuilder, mut installed_tool_names: HashMap, + mut wip_mounts: Option<&mut crate::wip::WipMountRegistry>, ) -> FeatureRegistryInstallReport { if self.installation == FeatureInstallation::Activate && let Some(error) = self @@ -2154,6 +2246,7 @@ impl FeatureRegistryBuilder { let mut reports = Vec::with_capacity(plan.ordered_indices.len()); let install_hook_checkpoint = hook_builder.checkpoint(); let install_tool_checkpoint = pending_tools.len(); + let install_wip_checkpoint = wip_mounts.as_deref().cloned(); let mut modules = self.modules.into_iter().map(Some).collect::>(); let ordered_modules = plan .ordered_indices @@ -2176,6 +2269,7 @@ impl FeatureRegistryBuilder { let service_checkpoint = service_registry.clone(); let invocation_checkpoint = invocation_registry.clone(); let tool_checkpoint = pending_tools.len(); + let wip_checkpoint = wip_mounts.as_deref().cloned(); let installed_tool_checkpoint = installed_tool_names.clone(); let mut required_service_failed = false; @@ -2218,6 +2312,7 @@ impl FeatureRegistryBuilder { feature_id: &descriptor.id, declarations: &declarations, pending_tools, + wip_mounts: wip_mounts.as_deref_mut(), installed_tool_names: &mut installed_tool_names, hook_builder, background_task_builder: &mut background_task_builder, @@ -2236,6 +2331,11 @@ impl FeatureRegistryBuilder { service_registry = service_checkpoint.clone(); invocation_registry = invocation_checkpoint.clone(); pending_tools.truncate(tool_checkpoint); + if let (Some(registry), Some(checkpoint)) = + (wip_mounts.as_deref_mut(), &wip_checkpoint) + { + *registry = checkpoint.clone(); + } installed_tool_names = installed_tool_checkpoint.clone(); report.clear_installed_contributions(); report @@ -2244,6 +2344,13 @@ impl FeatureRegistryBuilder { } } if report.installed { + for name in &descriptor.wip_mounts { + if !report.installed_wip_mounts.contains(name) { + report.diagnostics.push(FeatureDiagnostic::error(format!( + "declared WIP contribution {name} was not installed" + ))); + } + } for hook in &descriptor.hooks { if !report.installed_hooks.contains(hook) { report.diagnostics.push(FeatureDiagnostic::error(format!( @@ -2278,6 +2385,11 @@ impl FeatureRegistryBuilder { service_registry = service_checkpoint.clone(); invocation_registry = invocation_checkpoint.clone(); pending_tools.truncate(tool_checkpoint); + if let (Some(registry), Some(checkpoint)) = + (wip_mounts.as_deref_mut(), &wip_checkpoint) + { + *registry = checkpoint.clone(); + } installed_tool_names = installed_tool_checkpoint.clone(); report.clear_installed_contributions(); report.clear_installed_contributions(); @@ -2295,6 +2407,11 @@ impl FeatureRegistryBuilder { if failed { hook_builder.rollback_to(install_hook_checkpoint); pending_tools.truncate(install_tool_checkpoint); + if let (Some(registry), Some(checkpoint)) = + (wip_mounts.as_deref_mut(), install_wip_checkpoint) + { + *registry = checkpoint; + } for report in &mut reports { if report.installed { report.clear_installed_contributions(); @@ -3785,4 +3902,8 @@ mod tests { ); assert!(report.reports[0].installed); } + + mod composition_tests { + include!("feature/composition_tests.rs"); + } } diff --git a/crates/worker/src/feature/builtin.rs b/crates/worker/src/feature/builtin.rs index 817151949..c83039ce9 100644 --- a/crates/worker/src/feature/builtin.rs +++ b/crates/worker/src/feature/builtin.rs @@ -35,6 +35,7 @@ pub mod objective; pub mod orchestration; mod resource_projection; pub mod session_explore; +pub(crate) mod sub_worker; pub(crate) mod subjektiv_memory; pub(crate) mod subjektiv_session; pub mod task; diff --git a/crates/worker/src/feature/builtin/objective.rs b/crates/worker/src/feature/builtin/objective.rs index 321dcfcc6..0b303bc2a 100644 --- a/crates/worker/src/feature/builtin/objective.rs +++ b/crates/worker/src/feature/builtin/objective.rs @@ -17,6 +17,8 @@ //! logic. Query/detail uses canonical references and bounded bodies/events, omitting internal //! storage IDs from output. +pub(crate) mod composition; + use std::collections::BTreeMap; use std::sync::Arc; #[cfg(test)] diff --git a/crates/worker/src/feature/builtin/objective/composition.rs b/crates/worker/src/feature/builtin/objective/composition.rs new file mode 100644 index 000000000..7001fcbec --- /dev/null +++ b/crates/worker/src/feature/builtin/objective/composition.rs @@ -0,0 +1,107 @@ +//! Objective registration belongs to the Feature install scope, not the Engine +//! bootstrap. The Host supplies the live API; resolved enablement never grants +//! operation authority. The existing backend revalidates every access/effect. + +use super::*; +use crate::feature::{ + FeatureDescriptor, FeatureInstallContext, FeatureInstallError, FeatureModule, + FeatureRegistryBuilder, ServiceDeclaration, ServiceId, ServiceRequirement, ToolContribution, + ToolDeclaration, +}; + +const BACKEND_SERVICE: &str = "objective.workspace-api"; +const WIP_MOUNT: &str = "objective.native"; + +fn backend_declaration() -> ServiceDeclaration { + ServiceDeclaration::new( + ServiceId::builtin(BACKEND_SERVICE), + "1", + "Host-bound Objective Workspace API; registration does not authorize operations", + ) +} + +struct ObjectiveBackendFeature(Arc); + +impl FeatureModule for ObjectiveBackendFeature { + fn descriptor(&self) -> FeatureDescriptor { + FeatureDescriptor::builtin("objective-backend", "Objective Backend") + .with_provided_service(backend_declaration()) + } + + fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> { + context + .services() + .provide(backend_declaration(), self.0.clone()) + } +} + +pub(crate) struct ObjectiveFeature { + native: Option<(WipNamespaceRoute, Option)>, +} + +impl FeatureModule for ObjectiveFeature { + fn descriptor(&self) -> FeatureDescriptor { + let mut descriptor = FeatureDescriptor::builtin("objective", "Objective") + .with_description("Resolved Objective enablement publishes Tools and, in WIP mode, native projection; no separate instruction contribution") + .with_service_requirement(ServiceRequirement::required( + ServiceId::builtin(BACKEND_SERVICE), + "Objective Tools and native WIP share the Host-bound backend", + )); + for name in OBJECTIVE_TOOL_NAMES { + descriptor = + descriptor.with_tool(ToolDeclaration::new(name, "Workspace Objective operation")); + } + if self.native.is_some() { + descriptor = descriptor.with_wip_mount(WIP_MOUNT); + } + descriptor + } + + fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> { + let backend = context + .services() + .require::(&ServiceId::builtin(BACKEND_SERVICE))?; + for definition in workspace_http_objective_tools(backend.client.clone()) { + let name = definition().0.name; + context + .tools() + .register(ToolContribution::new(name, definition))?; + } + if let Some((route, permissions)) = &self.native { + context.wip().register(WIP_MOUNT, |registry| { + mount_workspace_http_objective_wip( + registry, + backend.client.clone(), + permissions.clone(), + route, + ) + })?; + } + Ok(()) + } +} + +/// Trusted composition selects both provider and consumer. Ticket cross-domain +/// coordinates are allocated by the Host even when Objective publication is off. +/// A consumer neither enables another Feature nor constructs authority. +pub(crate) fn compose( + registry: &mut FeatureRegistryBuilder, + enabled: bool, + client: Arc, + native: Option<(WipNamespaceRoute, Option)>, +) -> std::io::Result<()> { + if !enabled { + return Ok(()); + } + if !client.is_available() || client.workspace_id().is_none() { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "objective tools require Backend Workspace API authority", + )); + } + registry.add_module(ObjectiveBackendFeature(Arc::new( + WorkspaceHttpObjectiveBackend::new(client), + ))); + registry.add_module(ObjectiveFeature { native }); + Ok(()) +} diff --git a/crates/worker/src/feature/builtin/sub_worker.rs b/crates/worker/src/feature/builtin/sub_worker.rs new file mode 100644 index 000000000..0b7958fbe --- /dev/null +++ b/crates/worker/src/feature/builtin/sub_worker.rs @@ -0,0 +1,117 @@ +//! SubWorker publication is Worker-lifetime composition. Host-bound spawn +//! inputs retain delegation, command grants, notifications, and child ownership +//! in the existing spawn implementation. No scope or authority is inferred from +//! enablement, Profile identity, a service handle, or attachment presence. + +use std::path::PathBuf; +use std::sync::Arc; + +use agen::tool::ToolDefinition; +use arc_swap::ArcSwap; +use manifest::WorkerManifest; + +use crate::feature::contracts::WORKER_CONTROL_SERVICE_ID; +use crate::feature::{ + FeatureDescriptor, FeatureInstallContext, FeatureInstallError, FeatureInstructionContribution, + FeatureInstructionDeclaration, FeatureInstructionId, FeatureModule, FeatureRegistryBuilder, + ServiceDeclaration, ServiceId, ServiceRequirement, ToolContribution, ToolDeclaration, +}; +use crate::prompt::catalog::PromptCatalog; +use crate::spawn::registry::SpawnedWorkerRegistry; +use crate::spawn::tool::{ParentNotificationTarget, sub_worker_spawn_tool}; +use crate::worker::WorkerWorkspaceContext; + +const SPAWNER_SERVICE: &str = "sub-worker.spawner"; + +fn spawner_declaration() -> ServiceDeclaration { + ServiceDeclaration::new( + ServiceId::builtin(SPAWNER_SERVICE), + "1", + "Host-bound parent-owned spawner; live delegation and command checks remain per operation", + ) +} + +fn instruction() -> FeatureInstructionDeclaration { + FeatureInstructionDeclaration::new( + FeatureInstructionId::builtin("worker.orchestration"), + "common.worker_orchestration", + "Worker orchestration guidance", + ) + .expect("static Worker orchestration instruction declaration is valid") +} + +struct HostSpawner(ToolDefinition); +struct SubWorkerHostFeature(Arc); + +impl FeatureModule for SubWorkerHostFeature { + fn descriptor(&self) -> FeatureDescriptor { + FeatureDescriptor::builtin("sub-worker-host", "SubWorker Host") + .with_provided_service(spawner_declaration()) + } + fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> { + context + .services() + .provide(spawner_declaration(), self.0.clone()) + } +} + +struct SubWorkerFeature; + +impl FeatureModule for SubWorkerFeature { + fn descriptor(&self) -> FeatureDescriptor { + FeatureDescriptor::builtin("sub-worker", "SubWorker") + .with_description("Resolved SubWorker enablement publishes spawn and orchestration guidance together; no native WIP projection (normal Tool compatibility applies)") + .with_tool(ToolDeclaration::new("SubWorkerSpawn", "Start a parent-owned Internal SubWorker")) + .with_instruction(instruction()) + .with_service_requirement(ServiceRequirement::required(ServiceId::builtin(SPAWNER_SERVICE), + "Only the Host supplies live spawn inputs")) + .with_service_requirement(ServiceRequirement::required(ServiceId::builtin(WORKER_CONTROL_SERVICE_ID), + "Spawned children belong to the explicitly composed Worker control backend")) + } + fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> { + let spawner = context + .services() + .require::(&ServiceId::builtin(SPAWNER_SERVICE))?; + context + .tools() + .register(ToolContribution::new("SubWorkerSpawn", spawner.0.clone()))?; + context + .instructions() + .register(FeatureInstructionContribution::new(instruction())) + } +} + +/// Live handles are supplied by the Host. This product composition function +/// consumes the resolved saved manifest identically for create and restore. +#[allow(clippy::too_many_arguments)] +pub(crate) fn compose( + registry: &mut FeatureRegistryBuilder, + manifest: WorkerManifest, + workspace_context: WorkerWorkspaceContext, + parent_notifications: ParentNotificationTarget, + runtime_base: PathBuf, + workspace_root: Option, + broker: Option, + children: Arc, + prompts: Arc>, +) -> Option> { + if !manifest.feature.sub_worker.enabled { + return None; + } + let tool = sub_worker_spawn_tool( + manifest.worker.name.clone(), + workspace_context, + parent_notifications, + runtime_base, + workspace_root.unwrap_or_else(|| PathBuf::from("/")), + broker, + children.clone(), + manifest, + prompts, + ); + registry.add_module(SubWorkerHostFeature(Arc::new(HostSpawner(tool)))); + registry.add_module(SubWorkerFeature); + Some(Arc::new( + super::worker_observation::SpawnedSubWorkerObservationProvider::new(children), + )) +} diff --git a/crates/worker/src/feature/composition_tests.rs b/crates/worker/src/feature/composition_tests.rs new file mode 100644 index 000000000..6283e04f1 --- /dev/null +++ b/crates/worker/src/feature/composition_tests.rs @@ -0,0 +1,324 @@ +// Production Feature composition and installer; only the Workspace transport is +// replaced. Registration must not use it. Spawn execution/Host policy and saved +// bootstrap are covered by their owning suites, not simulated by this fixture. +use super::*; +use crate::feature::builtin::{manage_worker, objective, sub_worker}; +use crate::spawn::{registry::SpawnedWorkerRegistry, tool::ParentNotificationTarget}; +use crate::wip::WipMountRegistry; +use crate::worker::{ + WorkerWorkspaceContext, WorkspaceClient, WorkspaceClientError, WorkspaceRequest, + WorkspaceResponse, +}; +use manifest::{Scope, SharedScope, WorkerManifest}; + +#[derive(Debug)] +struct RegistrationClient { + available: bool, + identity: bool, +} +impl WorkspaceClient for RegistrationClient { + fn workspace_id(&self) -> Option<&str> { + self.identity.then_some("workspace") + } + fn kind(&self) -> &str { + "composition-fixture" + } + fn is_available(&self) -> bool { + self.available + } + fn execute(&self, _: WorkspaceRequest) -> Result { + panic!("registration must not access protected resources or ask for Feature activation") + } +} + +fn manifest() -> WorkerManifest { + WorkerManifest::from_toml("[worker]\nname='composition-parent'\n[model]\nscheme='anthropic'\nmodel_id='fixture'\n[engine]\n[scope]\nallow=[]").unwrap() +} + +fn compose_sub_worker(registry: &mut FeatureRegistryBuilder, enabled: bool, control: bool) { + let mut manifest = manifest(); + manifest.feature.sub_worker.enabled = enabled; + let children = SpawnedWorkerRegistry::new_internal( + manifest.worker.name.clone(), + SharedScope::new(Scope::empty()), + ); + if control { + registry.add_module(manage_worker::sub_worker_control_feature( + Arc::new(RegistrationClient { + available: false, + identity: false, + }), + children.clone(), + )); + } + let provider = sub_worker::compose( + registry, + manifest, + WorkerWorkspaceContext::no_workspace(), + ParentNotificationTarget::Durable(Arc::new(|_| {})), + std::path::PathBuf::from("unused-runtime"), + None, + None, + children, + Arc::new(arc_swap::ArcSwap::from( + crate::prompt::catalog::PromptCatalog::builtins_only().unwrap(), + )), + ); + assert_eq!(provider.is_some(), enabled); +} + +fn install( + registry: FeatureRegistryBuilder, + pending: &mut Vec, + mounts: &mut WipMountRegistry, +) -> FeatureRegistryInstallReport { + registry.install_into_pending_with_registered( + pending, + &mut HookRegistryBuilder::default(), + HashMap::new(), + Some(mounts), + ) +} + +#[test] +fn objective_enablement_requires_live_workspace_binding_not_operation_grants() { + for enabled in [false, true] { + for available in [false, true] { + for identity in [false, true] { + for native in [false, true] { + let mut registry = FeatureRegistryBuilder::new(); + let mut mounts = WipMountRegistry::new(); + let route = mounts + .allocate_namespace("objective", "objectives") + .unwrap(); + let result = objective::composition::compose( + &mut registry, + enabled, + Arc::new(RegistrationClient { + available, + identity, + }), + native.then_some((route, None)), + ); + assert_eq!(result.is_ok(), !enabled || (available && identity)); + if result.is_err() { + assert!(registry.is_empty()); + continue; + } + let mut pending = Vec::new(); + let report = install(registry, &mut pending, &mut mounts); + assert!(!report.has_errors(), "{}", report.error_message()); + assert_eq!(pending.len(), if enabled { 7 } else { 0 }); + assert_eq!( + mounts.routes().any(|route| route == "/objectives"), + enabled && native + ); + if enabled { + let objective = report + .reports + .iter() + .find(|r| r.feature_id == FeatureId::builtin("objective")) + .unwrap(); + assert_eq!( + objective.resolved_service_requirements[0].id, + ServiceId::builtin("objective.workspace-api") + ); + assert_eq!(objective.installed_wip_mounts.len(), usize::from(native)); + assert!(objective.installed_instructions.is_empty()); + } + } + } + } + } +} + +#[test] +fn sub_worker_enablement_publishes_spawn_and_instruction_without_delegation_or_workdir() { + for enabled in [false, true] { + let mut registry = FeatureRegistryBuilder::new(); + compose_sub_worker(&mut registry, enabled, enabled); + let mut pending = Vec::new(); + let report = install(registry, &mut pending, &mut WipMountRegistry::new()); + assert!(!report.has_errors(), "{}", report.error_message()); + assert_eq!( + pending + .iter() + .filter(|tool| tool().0.name == "SubWorkerSpawn") + .count(), + usize::from(enabled) + ); + assert_eq!( + report + .installed_instruction_contributions() + .iter() + .filter(|instruction| instruction.id + == FeatureInstructionId::builtin("worker.orchestration")) + .count(), + usize::from(enabled) + ); + if enabled { + let feature = report + .reports + .iter() + .find(|r| r.feature_id == FeatureId::builtin("sub-worker")) + .unwrap(); + assert_eq!(feature.resolved_service_requirements.len(), 2); + assert!(feature.installed_wip_mounts.is_empty()); + } + } +} + +#[test] +fn missing_control_dependency_prevents_objective_and_spawn_publication() { + let mut registry = FeatureRegistryBuilder::new(); + let mut mounts = WipMountRegistry::new(); + let route = mounts + .allocate_namespace("objective", "objectives") + .unwrap(); + objective::composition::compose( + &mut registry, + true, + Arc::new(RegistrationClient { + available: true, + identity: true, + }), + Some((route, None)), + ) + .unwrap(); + compose_sub_worker(&mut registry, true, false); + let mut pending = vec![dummy_tool("Existing")]; + let report = install(registry, &mut pending, &mut mounts); + assert!(matches!( + report.plan_error, + Some(FeaturePlanError::MissingServiceProvider { .. }) + )); + assert_eq!(pending.len(), 1); + assert_eq!(pending[0]().0.name, "Existing"); + assert!(mounts.routes().next().is_none()); + assert!(report.installed_instruction_contributions().is_empty()); +} + +#[test] +fn late_install_failure_rolls_back_migrated_tools_instructions_services_and_wip_only() { + let client = Arc::new(RegistrationClient { + available: true, + identity: true, + }); + let mut mounts = WipMountRegistry::new(); + let existing = mounts + .allocate_namespace("objective", "existing-objectives") + .unwrap(); + objective::mount_workspace_http_objective_wip(&mut mounts, client.clone(), None, &existing) + .unwrap(); + let before = mounts.routes().map(str::to_owned).collect::>(); + let mut registry = FeatureRegistryBuilder::new(); + let route = mounts + .allocate_namespace("objective", "objectives") + .unwrap(); + objective::composition::compose(&mut registry, true, client, Some((route, None))).unwrap(); + compose_sub_worker(&mut registry, true, true); + registry.add_module(ToolFeature { + descriptor: FeatureDescriptor::builtin("duplicate", "Duplicate") + .with_tool(ToolDeclaration::new("QueryObjective", "late duplicate")), + contribution_name: "QueryObjective", + model_visible_name: "QueryObjective", + }); + let mut pending = vec![dummy_tool("Existing")]; + let report = install(registry, &mut pending, &mut mounts); + assert!(report.has_errors()); + assert_eq!(pending.len(), 1); + assert_eq!(pending[0]().0.name, "Existing"); + assert_eq!( + mounts.routes().map(str::to_owned).collect::>(), + before + ); + assert!(report.installed_instruction_contributions().is_empty()); + assert!( + !report + .services + .provides(&ServiceId::builtin("objective.workspace-api")) + ); + assert!( + !report + .services + .provides(&ServiceId::builtin("sub-worker.spawner")) + ); + for report in report.reports { + assert!(!report.installed); + assert!(report.installed_tools.is_empty()); + assert!(report.installed_wip_mounts.is_empty()); + } +} + +struct PartialWipFailure { + route: crate::wip::WipNamespaceRoute, + declare: bool, +} +impl FeatureModule for PartialWipFailure { + fn descriptor(&self) -> FeatureDescriptor { + let descriptor = FeatureDescriptor::builtin("partial-wip", "Partial WIP"); + if self.declare { + descriptor.with_wip_mount("partial") + } else { + descriptor + } + } + fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> { + context.wip().register("partial", |mounts| { + let client = Arc::new(RegistrationClient { + available: true, + identity: true, + }); + objective::mount_workspace_http_objective_wip( + mounts, + client.clone(), + None, + &self.route, + )?; + // Deterministic failure after the first mount changed publication metadata. + objective::mount_workspace_http_objective_wip(mounts, client, None, &self.route) + }) + } +} + +#[test] +fn partial_or_undeclared_wip_failure_restores_host_mounts_and_discards_successful_siblings() { + for declare in [false, true] { + let mut mounts = WipMountRegistry::new(); + let route = mounts + .allocate_namespace("objective", "partial-objectives") + .unwrap(); + let mut registry = FeatureRegistryBuilder::new(); + compose_sub_worker(&mut registry, true, true); + registry.add_module(PartialWipFailure { route, declare }); + let mut pending = vec![dummy_tool("Existing")]; + let report = install(registry, &mut pending, &mut mounts); + assert!(report.has_errors()); + assert!(mounts.routes().next().is_none()); + assert_eq!( + mounts.namespace_root("objective"), + Some("/partial-objectives") + ); + assert_eq!(pending.len(), 1); + assert!(report.installed_instruction_contributions().is_empty()); + assert!( + report + .reports + .iter() + .all(|report| !report.installed && report.installed_wip_mounts.is_empty()) + ); + let failure = report + .reports + .iter() + .find(|report| report.feature_id == FeatureId::builtin("partial-wip")) + .unwrap(); + if !declare { + assert!( + failure + .skipped + .iter() + .any(|skip| skip.kind == FeatureContributionKind::WipMount) + ); + } + } +} diff --git a/crates/worker/src/spawn/tool.rs b/crates/worker/src/spawn/tool.rs index 1b8f357ef..8fde928ae 100644 --- a/crates/worker/src/spawn/tool.rs +++ b/crates/worker/src/spawn/tool.rs @@ -571,13 +571,15 @@ impl Tool for SubWorkerSpawnTool { })?; let child_change_tracker = child.tracker().cloned(); #[cfg(test)] - let installed_tools = child - .engine() - .tool_server_handle() - .tool_definitions_sorted() - .into_iter() - .map(|definition| definition.name) - .collect(); + let installed_tools = { + let server = child.engine().tool_server_handle(); + server.flush_pending(); + server + .tool_definitions_sorted() + .into_iter() + .map(|definition| definition.name) + .collect() + }; let child_name = input.name.clone(); let registry = Arc::downgrade(&self.registry); let parent_notifications = self.parent_notifications.clone(); diff --git a/crates/worker/src/wip.rs b/crates/worker/src/wip.rs index f1e989446..76fa98e94 100644 --- a/crates/worker/src/wip.rs +++ b/crates/worker/src/wip.rs @@ -313,6 +313,7 @@ pub trait WipDynamicItemResolver: Send + Sync { } } +#[derive(Clone)] pub struct WipDynamicMount { pub collection_route: String, pub capability: String, @@ -378,6 +379,7 @@ pub trait WipSubtreeProvider: Send + Sync { async fn children(&self, path: &str) -> Result, ProtocolError>; } +#[derive(Clone)] pub struct WipSubtreeMount { pub root: String, pub provider: Arc, @@ -395,6 +397,7 @@ pub struct WipDynamicOperationContribution { pub resolver: Arc, } +#[derive(Clone)] struct MountedDynamicOperationContribution { contributor: String, operations: Vec, @@ -412,6 +415,7 @@ impl WipNamespaceRoute { } } +#[derive(Clone)] struct MountedProjection { projection: WipProjection, operation_handlers: BTreeMap)>, @@ -497,7 +501,7 @@ impl WipOperationHandler for OperationDispatchHandler { } } -#[derive(Default)] +#[derive(Clone, Default)] pub struct WipMountRegistry { mounts: BTreeMap, additional_interfaces: BTreeMap>, diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index b2073b8aa..799b67f80 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -61,8 +61,8 @@ use crate::feature::session::{ SessionExtensionHandle, }; use crate::feature::{ - FeatureInstructionDeclaration, FeatureInstructionId, FeatureInvocationRegistry, - FeatureRegistryBuilder, FeatureRegistryInstallReport, dedupe_instruction_contributions, + FeatureInstructionDeclaration, FeatureInvocationRegistry, FeatureRegistryBuilder, + FeatureRegistryInstallReport, dedupe_instruction_contributions, }; use crate::hook::{ BeforeSessionRewriteAction, BeforeSessionRewriteContext, Hook, HookHistoryRange, @@ -78,8 +78,6 @@ use crate::internal_worker::{ }; const LARGE_PASTE_INLINE_MAX_BYTES: usize = 32 * 1024; -const WORKER_ORCHESTRATION_INSTRUCTION_ID: &str = "worker.orchestration"; -const WORKER_ORCHESTRATION_PROMPT_REF: &str = "common.worker_orchestration"; const FEATURE_HOOK_CHAIN_TIMEOUT: Duration = Duration::from_secs(30); const SESSION_PENDING_ACTIVATIONS_EXTENSION_DOMAIN: &str = "worker.pending_activations.v1"; const MAX_PENDING_SUBMISSIONS: usize = 32; @@ -517,14 +515,6 @@ fn hook_run_exit(exit: &EngineRunExit) -> RunCommittedExit { } } -fn worker_orchestration_instruction() -> FeatureInstructionDeclaration { - FeatureInstructionDeclaration::new( - FeatureInstructionId::builtin(WORKER_ORCHESTRATION_INSTRUCTION_ID), - WORKER_ORCHESTRATION_PROMPT_REF, - "Worker orchestration guidance", - ) - .expect("static Worker orchestration instruction declaration is valid") -} use crate::ipc::alerter::Alerter; use crate::ipc::interceptor::WorkerInterceptor; use crate::ipc::notify_buffer::NotifyBuffer; @@ -4899,10 +4889,6 @@ impl Worker { self.feature_instructions = dedupe_instruction_contributions(instructions); } - pub fn register_worker_orchestration_instruction(&mut self) { - self.register_feature_instruction(worker_orchestration_instruction()); - } - /// Toggle all resident sections in the system prompt. /// /// Default `true`: normal Workers may expose each resident section according @@ -5370,6 +5356,14 @@ impl Worker { pub fn install_features( &mut self, registry: FeatureRegistryBuilder, + ) -> FeatureRegistryInstallReport { + self.install_features_with_wip(registry, None) + } + + pub(crate) fn install_features_with_wip( + &mut self, + registry: FeatureRegistryBuilder, + wip_mounts: Option<&mut crate::wip::WipMountRegistry>, ) -> FeatureRegistryInstallReport { self.backend_job.get_or_init(|| None); let client = self.workspace_client_handle(); @@ -5381,7 +5375,8 @@ impl Worker { .map_err(|error| error.to_string()) }); let worker = self.engine.as_mut().expect("worker taken during run"); - let report = registry.install_into_engine(worker, &mut self.hook_builder); + let report = + registry.install_into_engine_with_wip(worker, &mut self.hook_builder, wip_mounts); if report.has_errors() { return report; } @@ -21005,6 +21000,10 @@ permission = "write" WorkerManifest::from_toml(toml_str).unwrap() } + mod composition_tests { + include!("worker_composition_tests.rs"); + } + // Real initial-seed/replay/fork/compaction and Memory source wiring. Only the // Host HTTP boundary and provider transport are fixtures; Host policy evaluation // itself is covered by workspace-server authorization tests, not asserted here. diff --git a/crates/worker/src/worker_composition_tests.rs b/crates/worker/src/worker_composition_tests.rs new file mode 100644 index 000000000..7a7baa55d --- /dev/null +++ b/crates/worker/src/worker_composition_tests.rs @@ -0,0 +1,164 @@ +// Production bootstrap, Controller registration, immutable manifest storage and +// restore. Model and Workspace transport are fixtures; no live external policy +// outcome is claimed. All cases have no Workdir attachment or delegation grant. +use super::*; + +fn bootstrap( + temp: &tempfile::TempDir, + manifest: WorkerManifest, + store: BodyPublicationStore, + client: Option>, +) -> crate::WorkerBootstrap { + let context = client.map_or_else(WorkerWorkspaceContext::no_workspace, |client| { + WorkerWorkspaceContext::with_client( + Some(WorkspaceId::new("feature-restore-workspace").unwrap()), + client, + ) + }); + crate::WorkerBootstrap::new( + manifest, + store, + PromptCatalogSource::builtins_only(), + context, + WorkerFilesystemAuthority::None, + crate::WorkerBootstrapLayout::RuntimeManagedRun { + run_dir: temp + .path() + .join("runtime/runs") + .join(uuid::Uuid::now_v7().to_string()), + }, + ) + .with_model_client(NoopClient) +} + +fn assert_publication( + controller: &crate::BootstrappedWorker, + objective: bool, + sub_worker: bool, + mode: manifest::WorkerMode, +) { + let names = controller.handle.shared_state.greeting().tools; + if mode == manifest::WorkerMode::Tools { + for tool in ["Read", "Write", "Edit", "Glob", "Grep", "Bash"] { + assert!( + names.iter().any(|name| name == tool), + "empty attachment must retain {tool}: {names:?}" + ); + } + assert_eq!( + names + .iter() + .filter(|name| name.as_str() == "QueryObjective") + .count(), + usize::from(objective) + ); + assert_eq!( + names + .iter() + .filter(|name| name.as_str() == "SubWorkerSpawn") + .count(), + usize::from(sub_worker) + ); + } else { + assert_eq!(names, ["Inspect", "Invoke", "Tree"]); + } + let state = session_store::collect_state(&controller.handle.committed_entries()); + let prompt = state.system_prompt.unwrap(); + assert_eq!(prompt.contains("## SubWorker orchestration"), sub_worker); +} + +#[tokio::test] +async fn objective_and_sub_worker_saved_composition_survives_create_and_restore_matrix() { + for mode in [manifest::WorkerMode::Tools, manifest::WorkerMode::Wip] { + for workspace in [false, true] { + for objective in [false, true] { + for sub_worker in [false, true] { + let temp = tempfile::tempdir().unwrap(); + let _sandbox = + worker_allocation::test_util::RuntimeDirSandbox::new(temp.path()); + let (mut manifest, store) = feature_restore_fixture(&temp); + manifest.worker.mode = mode; + manifest.feature.ticket.enabled = false; + manifest.feature.workdir_catalog.enabled = false; + manifest.feature.objective.enabled = objective; + manifest.feature.sub_worker.enabled = sub_worker; + let client = Arc::new(FeatureAdmissionClient::default()); + // Denied operation permissions must not prevent local publication. + *client.response_status.lock().unwrap() = Some(403); + let prepared = bootstrap( + &temp, + manifest.clone(), + store.clone(), + workspace.then(|| client.clone()), + ) + .prepare() + .await + .unwrap(); + assert!( + prepared + .worker() + .workdir_sessions() + .only_session() + .is_none() + ); + assert!( + prepared + .worker() + .manifest() + .delegation_scope + .allow + .is_empty() + ); + let created = prepared.start().await; + let rejected = objective && !workspace; + if rejected { + assert!( + format!("{:?}", created.err().unwrap()).contains( + "objective tools require Backend Workspace API authority" + ) + ); + } else { + let controller = created.unwrap(); + assert_publication(&controller, objective, sub_worker, mode); + stop_body_publication_controller(controller).await; + } + let mut fallback = manifest.clone(); + fallback.feature.objective.enabled = !objective; + fallback.feature.sub_worker.enabled = !sub_worker; + fallback.worker.mode = if mode == manifest::WorkerMode::Tools { + manifest::WorkerMode::Wip + } else { + manifest::WorkerMode::Tools + }; + let restored = bootstrap( + &temp, + fallback, + store.clone(), + workspace.then(|| client.clone()), + ) + .prepare_restored(&manifest.worker.name) + .await + .unwrap(); + assert_eq!(restored.worker().manifest().feature, manifest.feature); + assert_eq!(restored.worker().manifest().worker.mode, mode); + let restored = restored.start().await; + if rejected { + assert!( + format!("{:?}", restored.err().unwrap()).contains( + "objective tools require Backend Workspace API authority" + ) + ); + } else { + let controller = restored.unwrap(); + assert_publication(&controller, objective, sub_worker, mode); + stop_body_publication_controller(controller).await; + } + assert!( + client.requests.lock().unwrap().is_empty(), + "composition never requests Feature activation or protected data" + ); + } + } + } + } +} diff --git a/crates/workspace-server/src/server_ticket_tools_backend_tests.rs b/crates/workspace-server/src/server_ticket_tools_backend_tests.rs index e59da167c..3574f540b 100644 --- a/crates/workspace-server/src/server_ticket_tools_backend_tests.rs +++ b/crates/workspace-server/src/server_ticket_tools_backend_tests.rs @@ -196,6 +196,26 @@ impl TicketSurface { let reference_field = if name == "ShowTicket" { "id" } else { "ticket" }; let path = format!("/tickets/{}", input[reference_field].as_str().unwrap()); input.as_object_mut().unwrap().remove(reference_field); + match name { + "ShowTicket" => { + let mut page = serde_json::Map::new(); + for (legacy, native) in [("event_limit", "limit"), ("event_cursor", "cursor")] { + if let Some(value) = input.as_object_mut().unwrap().remove(legacy) { + page.insert(native.into(), value); + } + } + if !page.is_empty() { + input["page"] = page.into(); + } + } + "TicketAssignmentAnswer" => { + let fields = input.as_object_mut().unwrap(); + let assignment_id = fields.remove("assignment_id").unwrap(); + let expected_last_event_id = fields.remove("expected_last_event_id").unwrap(); + input["observed"] = serde_json::json!({"assignment_id":assignment_id,"expected_last_event_id":expected_last_event_id}); + } + _ => {} + } // Observe the real item representation before invoking a mutation. Tracking // reads alone do not establish the Ticket WIP object validator. Explicit // refresh also reconciles stale observations after an unknown result;