fix: reconcile Feature composition with neutral Worker contracts

This commit is contained in:
2026-10-29 05:04:39 +09:00
12 changed files with 910 additions and 93 deletions
+19 -63
View File
@@ -22,7 +22,6 @@ use crate::shutdown_after_idle::{
take_shutdown_request_after_status, take_shutdown_request_after_status,
}; };
use crate::spawn::registry::{PreparedSpawnedWorkerRegistryLoad, SpawnedWorkerRegistry}; use crate::spawn::registry::{PreparedSpawnedWorkerRegistryLoad, SpawnedWorkerRegistry};
use crate::spawn::tool::sub_worker_spawn_tool;
use crate::worker::{ use crate::worker::{
PendingSubmissionHandle, SystemItemCommitter, Worker, WorkerError, WorkerRunResult, PendingSubmissionHandle, SystemItemCommitter, Worker, WorkerError, WorkerRunResult,
}; };
@@ -1427,10 +1426,9 @@ fn validate_memory_lifecycle_targets(
Ok(()) Ok(())
} }
/// Register the builtin file-manipulation tools, optional memory tools, /// Compose Worker Features with Host-owned live bindings and install their
/// and the Worker-orchestration tools (SubWorkerSpawn + comm) on the Worker's /// staged contributions. Remaining direct registrations are Worker-lifetime
/// Engine. Returns the WorkdirSession handle used to attach a `WorkerFsView` to /// infrastructure (paste/Workdir) and the image/web migration follow-ups.
/// the shared state.
pub(crate) async fn register_worker_tools<C, St>( pub(crate) async fn register_worker_tools<C, St>(
worker: &mut Worker<C, St>, worker: &mut Worker<C, St>,
@@ -1463,7 +1461,6 @@ where
let task_feature = worker.task_feature(); let task_feature = worker.task_feature();
let web_config = worker.manifest().web.clone(); let web_config = worker.manifest().web.clone();
let mcp_config = worker.manifest().mcp.clone(); let mcp_config = worker.manifest().mcp.clone();
let spawner_name = worker.manifest().worker.name.clone();
let spawner_manifest = worker.manifest().clone(); let spawner_manifest = worker.manifest().clone();
let spawner_workspace_context = worker.workspace_context_handle(); let spawner_workspace_context = worker.workspace_context_handle();
let pending_submissions = worker.pending_submission_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 host_worker_observation_provider = worker.worker_observation_provider();
let wip_permissions = worker.manifest().permissions.clone(); let wip_permissions = worker.manifest().permissions.clone();
let mut wip_mount_registry = crate::wip::WipMountRegistry::new(); let mut wip_mount_registry = crate::wip::WipMountRegistry::new();
@@ -1973,7 +1966,6 @@ where
} }
{ {
let workspace_client = worker.workspace_client_handle(); let workspace_client = worker.workspace_client_handle();
let engine = worker.engine_mut();
// Allocate cross-domain input reference coordinates before mounting either provider. // Allocate cross-domain input reference coordinates before mounting either provider.
// Allocation is not publication or authorization; current handlers still check both. // 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 crate::feature::builtin::objective::composition::compose(
// the Backend Workspace API. Workers must not guess local `.yoi/objectives` &mut feature_registry,
// paths or read/edit Objective files directly. feature_config.objective.enabled,
if feature_config.objective.enabled { workspace_client,
if workspace_client.is_available() && workspace_client.workspace_id().is_some() { objective_namespace.map(|route| (route, wip_permissions.clone())),
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",
));
}
}
let mut observation_providers: Vec< let mut observation_providers: Vec<
Arc<dyn crate::feature::contracts::WorkerObservationProvider>, Arc<dyn crate::feature::contracts::WorkerObservationProvider>,
> = Vec::new(); > = Vec::new();
// Worker-orchestration tools derive child filesystem authority from the if let Some(provider) = crate::feature::builtin::sub_worker::compose(
// active provider-backed Workdir session. The tool remains registered &mut feature_registry,
// without one so invocation fails deterministically until the parent spawner_manifest,
// 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, spawner_workspace_context,
parent_notifications, parent_notifications,
runtime_base.clone(), runtime_base,
spawner_workspace_root, local_workspace_root,
workdir_tool_broker, workdir_tool_broker,
spawned_registry.clone(),
spawner_manifest,
prompts,
));
observation_providers.push(Arc::new(
crate::feature::builtin::worker_observation::SpawnedSubWorkerObservationProvider::new(
spawned_registry, spawned_registry,
), prompts,
)); ) {
observation_providers.push(provider);
} }
if let Some(provider) = host_worker_observation_provider { if let Some(provider) = host_worker_observation_provider {
observation_providers.push(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() { if feature_install_report.has_errors() {
return Err(std::io::Error::new( return Err(std::io::Error::new(
std::io::ErrorKind::InvalidInput, std::io::ErrorKind::InvalidInput,
+122 -1
View File
@@ -10,6 +10,13 @@
//! the normal Engine tool path, hooks are installed through //! the normal Engine tool path, hooks are installed through
//! [`crate::hook::HookRegistryBuilder`], and provider output is represented as //! [`crate::hook::HookRegistryBuilder`], and provider output is represented as
//! ordinary feature reports/diagnostics instead of a separate authority layer. //! 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::any::{Any, type_name};
use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
@@ -625,6 +632,9 @@ pub struct FeatureDescriptor {
pub tools: Vec<ToolDeclaration>, pub tools: Vec<ToolDeclaration>,
pub hooks: Vec<HookDeclaration>, pub hooks: Vec<HookDeclaration>,
pub instructions: Vec<FeatureInstructionDeclaration>, 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 background_tasks: Vec<BackgroundTaskDeclaration>,
pub provides_services: Vec<ServiceDeclaration>, pub provides_services: Vec<ServiceDeclaration>,
pub requires_services: Vec<ServiceRequirement>, pub requires_services: Vec<ServiceRequirement>,
@@ -645,6 +655,7 @@ impl FeatureDescriptor {
tools: Vec::new(), tools: Vec::new(),
hooks: Vec::new(), hooks: Vec::new(),
instructions: Vec::new(), instructions: Vec::new(),
wip_mounts: Vec::new(),
background_tasks: Vec::new(), background_tasks: Vec::new(),
provides_services: Vec::new(), provides_services: Vec::new(),
requires_services: Vec::new(), requires_services: Vec::new(),
@@ -668,6 +679,11 @@ impl FeatureDescriptor {
self 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 { pub fn with_instruction(mut self, instruction: FeatureInstructionDeclaration) -> Self {
self.instructions.push(instruction); self.instructions.push(instruction);
self self
@@ -751,6 +767,7 @@ pub enum FeatureContributionKind {
Tool, Tool,
Hook, Hook,
Instruction, Instruction,
WipMount,
BackgroundTask, BackgroundTask,
Service, Service,
ProtocolProvider, ProtocolProvider,
@@ -775,6 +792,8 @@ pub struct FeatureInstallReport {
pub runtime: FeatureRuntimeKind, pub runtime: FeatureRuntimeKind,
pub installed: bool, pub installed: bool,
pub installed_tools: Vec<String>, pub installed_tools: Vec<String>,
#[serde(default)]
pub installed_wip_mounts: Vec<String>,
pub installed_hooks: Vec<HookDeclaration>, pub installed_hooks: Vec<HookDeclaration>,
pub installed_instructions: Vec<FeatureInstructionDeclaration>, pub installed_instructions: Vec<FeatureInstructionDeclaration>,
pub declared_background_tasks: Vec<BackgroundTaskDeclaration>, pub declared_background_tasks: Vec<BackgroundTaskDeclaration>,
@@ -793,6 +812,7 @@ impl FeatureInstallReport {
runtime: descriptor.runtime.clone(), runtime: descriptor.runtime.clone(),
installed: false, installed: false,
installed_tools: Vec::new(), installed_tools: Vec::new(),
installed_wip_mounts: Vec::new(),
installed_hooks: Vec::new(), installed_hooks: Vec::new(),
installed_instructions: Vec::new(), installed_instructions: Vec::new(),
declared_background_tasks: Vec::new(), declared_background_tasks: Vec::new(),
@@ -808,6 +828,7 @@ impl FeatureInstallReport {
fn clear_installed_contributions(&mut self) { fn clear_installed_contributions(&mut self) {
self.installed = false; self.installed = false;
self.installed_tools.clear(); self.installed_tools.clear();
self.installed_wip_mounts.clear();
self.installed_hooks.clear(); self.installed_hooks.clear();
self.installed_instructions.clear(); self.installed_instructions.clear();
self.declared_background_tasks.clear(); self.declared_background_tasks.clear();
@@ -834,6 +855,7 @@ struct FeatureContributionDeclarations {
tools: HashSet<String>, tools: HashSet<String>,
hooks: HashSet<(String, FeatureHookPoint)>, hooks: HashSet<(String, FeatureHookPoint)>,
instructions: HashSet<FeatureInstructionId>, instructions: HashSet<FeatureInstructionId>,
wip_mounts: HashSet<String>,
background_tasks: HashSet<String>, background_tasks: HashSet<String>,
provided_services: HashSet<(ServiceId, String)>, provided_services: HashSet<(ServiceId, String)>,
required_services: HashSet<ServiceId>, required_services: HashSet<ServiceId>,
@@ -879,6 +901,7 @@ impl FeatureContributionDeclarations {
.iter() .iter()
.map(|provider| provider.id.clone()) .map(|provider| provider.id.clone())
.collect(), .collect(),
wip_mounts: descriptor.wip_mounts.iter().cloned().collect(),
chat_invocations: descriptor.chat_invocations.clone(), 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. /// Registrar for executable, Worker-managed background task contributions.
pub struct BackgroundTaskRegistrar<'a> { pub struct BackgroundTaskRegistrar<'a> {
feature_id: &'a FeatureId, feature_id: &'a FeatureId,
@@ -1553,6 +1623,7 @@ pub struct FeatureInstallContext<'a> {
feature_id: &'a FeatureId, feature_id: &'a FeatureId,
declarations: &'a FeatureContributionDeclarations, declarations: &'a FeatureContributionDeclarations,
pending_tools: &'a mut Vec<ToolDefinition>, pending_tools: &'a mut Vec<ToolDefinition>,
wip_mounts: Option<&'a mut crate::wip::WipMountRegistry>,
installed_tool_names: &'a mut HashMap<String, FeatureId>, installed_tool_names: &'a mut HashMap<String, FeatureId>,
hook_builder: &'a mut HookRegistryBuilder, hook_builder: &'a mut HookRegistryBuilder,
background_task_builder: &'a mut FeatureBackgroundTaskRegistryBuilder, background_task_builder: &'a mut FeatureBackgroundTaskRegistryBuilder,
@@ -1566,6 +1637,15 @@ impl FeatureInstallContext<'_> {
self.feature_id 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<'_> { pub fn tools(&mut self) -> ToolContributionRegistrar<'_> {
ToolContributionRegistrar { ToolContributionRegistrar {
feature_id: self.feature_id, feature_id: self.feature_id,
@@ -2039,10 +2119,20 @@ impl FeatureRegistryBuilder {
} }
/// Install modules into the existing Engine tool path and hook builder. /// Install modules into the existing Engine tool path and hook builder.
#[cfg(test)]
pub(crate) fn install_into_engine<C: LlmClient, A: Send + Sync>( pub(crate) fn install_into_engine<C: LlmClient, A: Send + Sync>(
self, self,
worker: &mut Engine<C, Mutable, A>, worker: &mut Engine<C, Mutable, A>,
hook_builder: &mut HookRegistryBuilder, 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 { ) -> FeatureRegistryInstallReport {
let mut pending_tools = Vec::new(); let mut pending_tools = Vec::new();
worker.tool_server_handle().flush_pending(); worker.tool_server_handle().flush_pending();
@@ -2056,6 +2146,7 @@ impl FeatureRegistryBuilder {
&mut pending_tools, &mut pending_tools,
hook_builder, hook_builder,
registered_tool_names, registered_tool_names,
wip_mounts,
); );
if !report.has_errors() { if !report.has_errors() {
worker.register_tools(pending_tools); worker.register_tools(pending_tools);
@@ -2069,7 +2160,7 @@ impl FeatureRegistryBuilder {
pending_tools: &mut Vec<ToolDefinition>, pending_tools: &mut Vec<ToolDefinition>,
hook_builder: &mut HookRegistryBuilder, hook_builder: &mut HookRegistryBuilder,
) -> FeatureRegistryInstallReport { ) -> 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( fn install_into_pending_with_registered(
@@ -2077,6 +2168,7 @@ impl FeatureRegistryBuilder {
pending_tools: &mut Vec<ToolDefinition>, pending_tools: &mut Vec<ToolDefinition>,
hook_builder: &mut HookRegistryBuilder, hook_builder: &mut HookRegistryBuilder,
mut installed_tool_names: HashMap<String, FeatureId>, mut installed_tool_names: HashMap<String, FeatureId>,
mut wip_mounts: Option<&mut crate::wip::WipMountRegistry>,
) -> FeatureRegistryInstallReport { ) -> FeatureRegistryInstallReport {
if self.installation == FeatureInstallation::Activate if self.installation == FeatureInstallation::Activate
&& let Some(error) = self && let Some(error) = self
@@ -2154,6 +2246,7 @@ impl FeatureRegistryBuilder {
let mut reports = Vec::with_capacity(plan.ordered_indices.len()); let mut reports = Vec::with_capacity(plan.ordered_indices.len());
let install_hook_checkpoint = hook_builder.checkpoint(); let install_hook_checkpoint = hook_builder.checkpoint();
let install_tool_checkpoint = pending_tools.len(); 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 mut modules = self.modules.into_iter().map(Some).collect::<Vec<_>>();
let ordered_modules = plan let ordered_modules = plan
.ordered_indices .ordered_indices
@@ -2176,6 +2269,7 @@ impl FeatureRegistryBuilder {
let service_checkpoint = service_registry.clone(); let service_checkpoint = service_registry.clone();
let invocation_checkpoint = invocation_registry.clone(); let invocation_checkpoint = invocation_registry.clone();
let tool_checkpoint = pending_tools.len(); let tool_checkpoint = pending_tools.len();
let wip_checkpoint = wip_mounts.as_deref().cloned();
let installed_tool_checkpoint = installed_tool_names.clone(); let installed_tool_checkpoint = installed_tool_names.clone();
let mut required_service_failed = false; let mut required_service_failed = false;
@@ -2218,6 +2312,7 @@ impl FeatureRegistryBuilder {
feature_id: &descriptor.id, feature_id: &descriptor.id,
declarations: &declarations, declarations: &declarations,
pending_tools, pending_tools,
wip_mounts: wip_mounts.as_deref_mut(),
installed_tool_names: &mut installed_tool_names, installed_tool_names: &mut installed_tool_names,
hook_builder, hook_builder,
background_task_builder: &mut background_task_builder, background_task_builder: &mut background_task_builder,
@@ -2236,6 +2331,11 @@ impl FeatureRegistryBuilder {
service_registry = service_checkpoint.clone(); service_registry = service_checkpoint.clone();
invocation_registry = invocation_checkpoint.clone(); invocation_registry = invocation_checkpoint.clone();
pending_tools.truncate(tool_checkpoint); 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(); installed_tool_names = installed_tool_checkpoint.clone();
report.clear_installed_contributions(); report.clear_installed_contributions();
report report
@@ -2244,6 +2344,13 @@ impl FeatureRegistryBuilder {
} }
} }
if report.installed { 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 { for hook in &descriptor.hooks {
if !report.installed_hooks.contains(hook) { if !report.installed_hooks.contains(hook) {
report.diagnostics.push(FeatureDiagnostic::error(format!( report.diagnostics.push(FeatureDiagnostic::error(format!(
@@ -2278,6 +2385,11 @@ impl FeatureRegistryBuilder {
service_registry = service_checkpoint.clone(); service_registry = service_checkpoint.clone();
invocation_registry = invocation_checkpoint.clone(); invocation_registry = invocation_checkpoint.clone();
pending_tools.truncate(tool_checkpoint); 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(); installed_tool_names = installed_tool_checkpoint.clone();
report.clear_installed_contributions(); report.clear_installed_contributions();
report.clear_installed_contributions(); report.clear_installed_contributions();
@@ -2295,6 +2407,11 @@ impl FeatureRegistryBuilder {
if failed { if failed {
hook_builder.rollback_to(install_hook_checkpoint); hook_builder.rollback_to(install_hook_checkpoint);
pending_tools.truncate(install_tool_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 { for report in &mut reports {
if report.installed { if report.installed {
report.clear_installed_contributions(); report.clear_installed_contributions();
@@ -3785,4 +3902,8 @@ mod tests {
); );
assert!(report.reports[0].installed); assert!(report.reports[0].installed);
} }
mod composition_tests {
include!("feature/composition_tests.rs");
}
} }
+1
View File
@@ -35,6 +35,7 @@ pub mod objective;
pub mod orchestration; pub mod orchestration;
mod resource_projection; mod resource_projection;
pub mod session_explore; pub mod session_explore;
pub(crate) mod sub_worker;
pub(crate) mod subjektiv_memory; pub(crate) mod subjektiv_memory;
pub(crate) mod subjektiv_session; pub(crate) mod subjektiv_session;
pub mod task; pub mod task;
@@ -17,6 +17,8 @@
//! logic. Query/detail uses canonical references and bounded bodies/events, omitting internal //! logic. Query/detail uses canonical references and bounded bodies/events, omitting internal
//! storage IDs from output. //! storage IDs from output.
pub(crate) mod composition;
use std::collections::BTreeMap; use std::collections::BTreeMap;
use std::sync::Arc; use std::sync::Arc;
#[cfg(test)] #[cfg(test)]
@@ -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(())
}
@@ -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<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 crate::feature::contracts::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),
))
}
@@ -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)
);
}
}
}
+6 -4
View File
@@ -571,13 +571,15 @@ impl Tool for SubWorkerSpawnTool {
})?; })?;
let child_change_tracker = child.tracker().cloned(); let child_change_tracker = child.tracker().cloned();
#[cfg(test)] #[cfg(test)]
let installed_tools = child let installed_tools = {
.engine() let server = child.engine().tool_server_handle();
.tool_server_handle() server.flush_pending();
server
.tool_definitions_sorted() .tool_definitions_sorted()
.into_iter() .into_iter()
.map(|definition| definition.name) .map(|definition| definition.name)
.collect(); .collect()
};
let child_name = input.name.clone(); let child_name = input.name.clone();
let registry = Arc::downgrade(&self.registry); let registry = Arc::downgrade(&self.registry);
let parent_notifications = self.parent_notifications.clone(); let parent_notifications = self.parent_notifications.clone();
+5 -1
View File
@@ -313,6 +313,7 @@ pub trait WipDynamicItemResolver: Send + Sync {
} }
} }
#[derive(Clone)]
pub struct WipDynamicMount { pub struct WipDynamicMount {
pub collection_route: String, pub collection_route: String,
pub capability: String, pub capability: String,
@@ -378,6 +379,7 @@ pub trait WipSubtreeProvider: Send + Sync {
async fn children(&self, path: &str) -> Result<Vec<String>, ProtocolError>; async fn children(&self, path: &str) -> Result<Vec<String>, ProtocolError>;
} }
#[derive(Clone)]
pub struct WipSubtreeMount { pub struct WipSubtreeMount {
pub root: String, pub root: String,
pub provider: Arc<dyn WipSubtreeProvider>, pub provider: Arc<dyn WipSubtreeProvider>,
@@ -395,6 +397,7 @@ pub struct WipDynamicOperationContribution {
pub resolver: Arc<dyn WipDynamicOperationResolver>, pub resolver: Arc<dyn WipDynamicOperationResolver>,
} }
#[derive(Clone)]
struct MountedDynamicOperationContribution { struct MountedDynamicOperationContribution {
contributor: String, contributor: String,
operations: Vec<String>, operations: Vec<String>,
@@ -412,6 +415,7 @@ impl WipNamespaceRoute {
} }
} }
#[derive(Clone)]
struct MountedProjection { struct MountedProjection {
projection: WipProjection, projection: WipProjection,
operation_handlers: BTreeMap<String, (String, Arc<dyn WipOperationHandler>)>, operation_handlers: BTreeMap<String, (String, Arc<dyn WipOperationHandler>)>,
@@ -497,7 +501,7 @@ impl WipOperationHandler for OperationDispatchHandler {
} }
} }
#[derive(Default)] #[derive(Clone, Default)]
pub struct WipMountRegistry { pub struct WipMountRegistry {
mounts: BTreeMap<String, MountedProjection>, mounts: BTreeMap<String, MountedProjection>,
additional_interfaces: BTreeMap<String, Vec<WipProjection>>, additional_interfaces: BTreeMap<String, Vec<WipProjection>>,
+16 -17
View File
@@ -61,8 +61,8 @@ use crate::feature::session::{
SessionExtensionHandle, SessionExtensionHandle,
}; };
use crate::feature::{ use crate::feature::{
FeatureInstructionDeclaration, FeatureInstructionId, FeatureInvocationRegistry, FeatureInstructionDeclaration, FeatureInvocationRegistry, FeatureRegistryBuilder,
FeatureRegistryBuilder, FeatureRegistryInstallReport, dedupe_instruction_contributions, FeatureRegistryInstallReport, dedupe_instruction_contributions,
}; };
use crate::hook::{ use crate::hook::{
BeforeSessionRewriteAction, BeforeSessionRewriteContext, Hook, HookHistoryRange, BeforeSessionRewriteAction, BeforeSessionRewriteContext, Hook, HookHistoryRange,
@@ -78,8 +78,6 @@ use crate::internal_worker::{
}; };
const LARGE_PASTE_INLINE_MAX_BYTES: usize = 32 * 1024; 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 FEATURE_HOOK_CHAIN_TIMEOUT: Duration = Duration::from_secs(30);
const SESSION_PENDING_ACTIVATIONS_EXTENSION_DOMAIN: &str = "worker.pending_activations.v1"; const SESSION_PENDING_ACTIVATIONS_EXTENSION_DOMAIN: &str = "worker.pending_activations.v1";
const MAX_PENDING_SUBMISSIONS: usize = 32; 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::alerter::Alerter;
use crate::ipc::interceptor::WorkerInterceptor; use crate::ipc::interceptor::WorkerInterceptor;
use crate::ipc::notify_buffer::NotifyBuffer; use crate::ipc::notify_buffer::NotifyBuffer;
@@ -4899,10 +4889,6 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
self.feature_instructions = dedupe_instruction_contributions(instructions); 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. /// Toggle all resident sections in the system prompt.
/// ///
/// Default `true`: normal Workers may expose each resident section according /// Default `true`: normal Workers may expose each resident section according
@@ -5370,6 +5356,14 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
pub fn install_features( pub fn install_features(
&mut self, &mut self,
registry: FeatureRegistryBuilder, 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 { ) -> FeatureRegistryInstallReport {
self.backend_job.get_or_init(|| None); self.backend_job.get_or_init(|| None);
let client = self.workspace_client_handle(); let client = self.workspace_client_handle();
@@ -5381,7 +5375,8 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
.map_err(|error| error.to_string()) .map_err(|error| error.to_string())
}); });
let worker = self.engine.as_mut().expect("worker taken during run"); 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() { if report.has_errors() {
return report; return report;
} }
@@ -21005,6 +21000,10 @@ permission = "write"
WorkerManifest::from_toml(toml_str).unwrap() 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 // Real initial-seed/replay/fork/compaction and Memory source wiring. Only the
// Host HTTP boundary and provider transport are fixtures; Host policy evaluation // Host HTTP boundary and provider transport are fixtures; Host policy evaluation
// itself is covered by workspace-server authorization tests, not asserted here. // itself is covered by workspace-server authorization tests, not asserted here.
@@ -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"
);
}
}
}
}
}
@@ -196,6 +196,26 @@ impl TicketSurface {
let reference_field = if name == "ShowTicket" { "id" } else { "ticket" }; let reference_field = if name == "ShowTicket" { "id" } else { "ticket" };
let path = format!("/tickets/{}", input[reference_field].as_str().unwrap()); let path = format!("/tickets/{}", input[reference_field].as_str().unwrap());
input.as_object_mut().unwrap().remove(reference_field); 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 // Observe the real item representation before invoking a mutation. Tracking
// reads alone do not establish the Ticket WIP object validator. Explicit // reads alone do not establish the Ticket WIP object validator. Explicit
// refresh also reconciles stale observations after an unknown result; // refresh also reconciles stale observations after an unknown result;