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

refactor: compose Objective and SubWorker contributions through Feature installation

Browse Source
This commit is contained in:
Hare
2026-10-28 17:59:19 +09:00
parent a1fbe1e754
commit 14101d45dc
11 changed files with 890 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker/src/controller.rs
+23 -67
View File
@@ -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<C, St>(
worker: &mut Worker<C, St>,
@@ -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<dyn crate::feature::builtin::worker_observation::WorkerObservationProvider>,
> = 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,
crates/worker/src/feature.rs
+122 -1
View File
@@ -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};
@@ -620,6 +627,9 @@ pub struct FeatureDescriptor {
pub tools: Vec<ToolDeclaration>,
pub hooks: Vec<HookDeclaration>,
pub instructions: Vec<FeatureInstructionDeclaration>,
/// Named native WIP contributions staged with the rest of the install scope.
#[serde(default)]
pub wip_mounts: Vec<String>,
pub background_tasks: Vec<BackgroundTaskDeclaration>,
pub provides_services: Vec<ServiceDeclaration>,
pub requires_services: Vec<ServiceRequirement>,
@@ -640,6 +650,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(),
@@ -663,6 +674,11 @@ impl FeatureDescriptor {
self
}
pub fn with_wip_mount(mut self, name: impl Into<String>) -> Self {
self.wip_mounts.push(name.into());
self
}
pub fn with_instruction(mut self, instruction: FeatureInstructionDeclaration) -> Self {
self.instructions.push(instruction);
self
@@ -746,6 +762,7 @@ pub enum FeatureContributionKind {
Tool,
Hook,
Instruction,
WipMount,
BackgroundTask,
Service,
ProtocolProvider,
@@ -770,6 +787,8 @@ pub struct FeatureInstallReport {
pub runtime: FeatureRuntimeKind,
pub installed: bool,
pub installed_tools: Vec<String>,
#[serde(default)]
pub installed_wip_mounts: Vec<String>,
pub installed_hooks: Vec<HookDeclaration>,
pub installed_instructions: Vec<FeatureInstructionDeclaration>,
pub declared_background_tasks: Vec<BackgroundTaskDeclaration>,
@@ -788,6 +807,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(),
@@ -803,6 +823,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();
@@ -829,6 +850,7 @@ struct FeatureContributionDeclarations {
tools: HashSet<String>,
hooks: HashSet<(String, FeatureHookPoint)>,
instructions: HashSet<FeatureInstructionId>,
wip_mounts: HashSet<String>,
background_tasks: HashSet<String>,
provided_services: HashSet<(ServiceId, String)>,
required_services: HashSet<ServiceId>,
@@ -874,6 +896,7 @@ impl FeatureContributionDeclarations {
.iter()
.map(|provider| provider.id.clone())
.collect(),
wip_mounts: descriptor.wip_mounts.iter().cloned().collect(),
chat_invocations: descriptor.chat_invocations.clone(),
}
}
@@ -1304,6 +1327,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,
@@ -1548,6 +1618,7 @@ pub struct FeatureInstallContext<'a> {
feature_id: &'a FeatureId,
declarations: &'a FeatureContributionDeclarations,
pending_tools: &'a mut Vec<ToolDefinition>,
wip_mounts: Option<&'a mut crate::wip::WipMountRegistry>,
installed_tool_names: &'a mut HashMap<String, FeatureId>,
hook_builder: &'a mut HookRegistryBuilder,
background_task_builder: &'a mut FeatureBackgroundTaskRegistryBuilder,
@@ -1561,6 +1632,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,
@@ -2034,10 +2114,20 @@ impl FeatureRegistryBuilder {
}
/// Install modules into the existing Engine tool path and hook builder.
#[cfg(test)]
pub(crate) fn install_into_engine<C: LlmClient, A: Send + Sync>(
self,
worker: &mut Engine<C, Mutable, A>,
hook_builder: &mut HookRegistryBuilder,
) -> FeatureRegistryInstallReport {
self.install_into_engine_with_wip(worker, hook_builder, None)
}
pub(crate) fn install_into_engine_with_wip<C: LlmClient, A: Send + Sync>(
self,
worker: &mut Engine<C, Mutable, A>,
hook_builder: &mut HookRegistryBuilder,
wip_mounts: Option<&mut crate::wip::WipMountRegistry>,
) -> FeatureRegistryInstallReport {
let mut pending_tools = Vec::new();
worker.tool_server_handle().flush_pending();
@@ -2051,6 +2141,7 @@ impl FeatureRegistryBuilder {
&mut pending_tools,
hook_builder,
registered_tool_names,
wip_mounts,
);
if !report.has_errors() {
worker.register_tools(pending_tools);
@@ -2064,7 +2155,7 @@ impl FeatureRegistryBuilder {
pending_tools: &mut Vec<ToolDefinition>,
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(
@@ -2072,6 +2163,7 @@ impl FeatureRegistryBuilder {
pending_tools: &mut Vec<ToolDefinition>,
hook_builder: &mut HookRegistryBuilder,
mut installed_tool_names: HashMap<String, FeatureId>,
mut wip_mounts: Option<&mut crate::wip::WipMountRegistry>,
) -> FeatureRegistryInstallReport {
if self.installation == FeatureInstallation::Activate
&& let Some(error) = self
@@ -2149,6 +2241,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::<Vec<_>>();
let ordered_modules = plan
.ordered_indices
@@ -2171,6 +2264,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;
@@ -2213,6 +2307,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,
@@ -2231,6 +2326,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
@@ -2239,6 +2339,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!(
@@ -2273,6 +2380,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();
@@ -2290,6 +2402,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();
@@ -3779,4 +3896,8 @@ mod tests {
);
assert!(report.reports[0].installed);
}
mod composition_tests {
include!("feature/composition_tests.rs");
}
}
crates/worker/src/feature/builtin.rs
+1
View File
@@ -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;
crates/worker/src/feature/builtin/objective.rs
+2
View File
@@ -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)]
crates/worker/src/feature/builtin/objective/composition.rs
+107
View File
@@ -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<WorkspaceHttpObjectiveBackend>);
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<ToolPermissionConfig>)>,
}
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::<WorkspaceHttpObjectiveBackend>(&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<dyn WorkspaceClient>,
native: Option<(WipNamespaceRoute, Option<ToolPermissionConfig>)>,
) -> 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(())
}
crates/worker/src/feature/builtin/sub_worker.rs
+117
View File
@@ -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 super::manage_worker::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<HostSpawner>);
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::<HostSpawner>(&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<PathBuf>,
broker: Option<workdir::WorkdirToolBrokerRouter>,
children: Arc<SpawnedWorkerRegistry>,
prompts: Arc<ArcSwap<PromptCatalog>>,
) -> Option<Arc<dyn super::worker_observation::WorkerObservationProvider>> {
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),
))
}
crates/worker/src/feature/composition_tests.rs
+324
View File
@@ -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<WorkspaceResponse, WorkspaceClientError> {
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<ToolDefinition>,
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::<Vec<_>>();
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::<Vec<_>>(),
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)
);
}
}
}
crates/worker/src/spawn/tool.rs
+9 -7
View File
@@ -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();
crates/worker/src/wip.rs
+5 -1
View File
@@ -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<Vec<String>, ProtocolError>;
}
#[derive(Clone)]
pub struct WipSubtreeMount {
pub root: String,
pub provider: Arc<dyn WipSubtreeProvider>,
@@ -395,6 +397,7 @@ pub struct WipDynamicOperationContribution {
pub resolver: Arc<dyn WipDynamicOperationResolver>,
}
#[derive(Clone)]
struct MountedDynamicOperationContribution {
contributor: String,
operations: Vec<String>,
@@ -412,6 +415,7 @@ impl WipNamespaceRoute {
}
}
#[derive(Clone)]
struct MountedProjection {
projection: WipProjection,
operation_handlers: BTreeMap<String, (String, Arc<dyn WipOperationHandler>)>,
@@ -497,7 +501,7 @@ impl WipOperationHandler for OperationDispatchHandler {
}
}
#[derive(Default)]
#[derive(Clone, Default)]
pub struct WipMountRegistry {
mounts: BTreeMap<String, MountedProjection>,
additional_interfaces: BTreeMap<String, Vec<WipProjection>>,
crates/worker/src/worker.rs
+16 -17
View File
@@ -60,8 +60,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,
@@ -77,8 +77,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;
@@ -516,14 +514,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;
@@ -4898,10 +4888,6 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
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
@@ -5369,6 +5355,14 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
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();
@@ -5380,7 +5374,8 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
.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;
}
@@ -21004,6 +20999,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.
crates/worker/src/worker_composition_tests.rs
+164
View File
@@ -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<Arc<FeatureAdmissionClient>>,
) -> crate::WorkerBootstrap<BodyPublicationStore> {
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"
);
}
}
}
}
}