diff --git a/CONTEXT.md b/CONTEXT.md new file mode 100644 index 0000000000..14f8740543 --- /dev/null +++ b/CONTEXT.md @@ -0,0 +1,161 @@ +# DimOS Runtime Deployment Language + +This context defines the language for discussing how DimOS modules retain stable identity while their implementations run in different environments or on different machines. + +## Language + +**In-Environment Python Module**: +A normal Python `Module` whose implementation is instantiated inside a local `PythonWorker` and shares that worker's Python environment. +_Avoid_: ExternalModule, remote normal Python module + +**PythonWorker Preservation Boundary**: +The rule that deployment work does not replace the existing local `PythonWorker` path. In-environment Python modules continue using its lightweight object and RPC protocol. +_Avoid_: Universal external worker path, PythonWorker replacement + +**ExternalModule**: +A declarative `Module` subclass whose implementation runs outside `PythonWorker`. It declares the DimOS-facing streams, config, RPC surface, and an implementation reference, but contains no build, process, watchdog, or transport behavior. +_Avoid_: NativeModule compatibility wrapper, process supervisor + +**External Implementation Reference**: +The `ExternalModule` declaration that identifies what a runtime handle launches. It may be written as a string or `pathlib.Path`; the convention-discovered implementation layout, not the Python value type, determines whether it denotes a Python class or native executable. +_Avoid_: Deployment-level implementation override, value-type runtime discriminator + +**NativeModule Compatibility Path**: +The existing `NativeModule` design in which a Python wrapper runs in `PythonWorker` and directly builds, launches, logs, and supervises a native subprocess. It remains available during migration to `ExternalModule`. +_Avoid_: Final external-module architecture, flag-day migration + +**Convention Preset**: +A built-in preparation strategy selected automatically from a standard implementation layout. V1 heuristics recognize `python/pyproject.toml`, `rust/Cargo.toml`, and `cpp/CMakeLists.txt`; Deployment Specs override the inferred preset only for exceptional deployments. +_Avoid_: Hand-written prepare commands for every module, required manifest for standard layouts + +**Unambiguous Convention Resolution**: +The rule that automatic preparation succeeds only when exactly one supported implementation layout is detected. No match or multiple matches fail planning unless the Module Deployment provides explicit Preparation and Runtime Environment behavior that selects and validates one implementation folder within the discovered package root. +_Avoid_: Convention precedence, silently selecting one of several implementations + +**Deployment Spec**: +A first-class runnable deployment description that references a Blueprint, defines reusable named targets, and groups each module's execution target, build target, Preparation, and Runtime Environment choices in a Module Deployment. It does not redefine module identity or Blueprint wiring. +_Avoid_: Module package declaration, Blueprint replacement, deployment registry entry + +**Execution Target**: +A named machine or execution substrate on which modules may be prepared and run. In v1, each target name identifies one distinct machine. +_Avoid_: Module implementation, Module Deployment + +**Resolved Target Platform**: +The single detected and optionally asserted platform identity of an Execution Target. Worker bootstrap, Preparation, and artifact validation consume this identity rather than declaring independent destination platforms. +_Avoid_: Preparation-specific target platform, competing compiler-target declarations + +**Deployment Root**: +The configured directory on an Execution Target containing DimOS-managed control environments, source snapshots, artifacts, runtime environments, run state, logs, caches, and locks. External system stores and provisioned devices may remain outside it. +_Avoid_: Generic source workspace, scattered DimOS state directories + +**Hermetic Worker Bootstrap**: +The target bootstrap model in which DimOS transfers a pinned environment tool and provisions Python plus the version-matched ExternalWorker control environment inside the Deployment Root rather than relying on target-global installations. +_Avoid_: Module Runtime Environment, unmanaged system Python dependency + +**Module Deployment**: +The grouped deployment policy for one Module: where it builds, where it executes, how deployable material is prepared, and how its runtime environment is materialized. Omitted choices use local placement or convention-derived behavior. +_Avoid_: Parallel assignment, build-location, and override maps + +**Implicit Local Target**: +The execution target that exists in every Deployment Spec without explicit declaration. Its Deployment Root comes from GlobalConfig-backed DimOS state, and modules absent from the `modules` map run locally. +_Avoid_: Required local target declaration, implicit remote placement + +**Deployment Plan**: +An immutable, validated resolution of Deployment Spec, Blueprint modules, implementation conventions, preparation behavior, concrete target definitions, Module Deployments, and worker routes. +_Avoid_: Behavioral reconciler, mutable deployment state + +**Deployment Prepare Phase**: +The pre-worker phase that stages source and produces deployable material through target access, including native builds, cross-compilation, and artifact sync. It completes before the target-side ExternalWorker materializes module Runtime Environments. +_Avoid_: Module start, implicit build during run + +**Preparation**: +A single deployment workflow that may coordinate build and execution targets and transfer artifacts between them. Convention Presets provide normal Preparation behavior; a Deployment Spec may substitute an exceptional Preparation. +_Avoid_: One preparation per machine, ExternalWorker lifecycle + +**Runtime Environment**: +The execution-target environment materialized by ExternalWorker from staged deployment inputs before a runtime handle starts. It may install dependencies or artifacts and provision deployment-owned target resources. +_Avoid_: Build workflow, Runtime Host, ExternalWorker bootstrap environment + +**Runtime Environment Spec**: +The serializable top-level import reference and JSON-compatible configuration from which ExternalWorker constructs a Runtime Environment. It is part of a Module Deployment only when convention-derived behavior needs an exceptional override. +_Avoid_: Live coordinator object, pickled class instance + +**Shared Runtime Environment**: +A Runtime Environment reused within one deployment run when modules have the same source digest, execution-machine identity, and resolved Runtime Environment Spec. Setup is serialized and may rerun idempotently; teardown occurs only after all dependent runtime handles in that run stop. +_Avoid_: Planner-level workflow merging, per-module environment copies + +**Deployment Source Snapshot**: +A content-addressed copy of a module package staged on a target before deployment. It includes the dependency-light contract and deployment extensions as well as implementation source, and is published atomically after transfer. +_Avoid_: Mutable shared source workspace, partial in-place rsync + +**Idempotent Prepare**: +The rule that prepare executes required package-manager, build, and sync steps and relies on their caches or up-to-date checks rather than maintaining a separate DimOS freshness database. +_Avoid_: DimOS freshness cache, manual stale-state tracking + +**Worker Manager**: +A coordinator-side backend scheduler that owns worker collections, placement, parallel deployment, rollback, health aggregation, and shutdown. DimOS uses one manager instance per deployment backend per coordinator. +_Avoid_: Worker process, per-machine manager + +**Worker Manager Backend Boundary**: +The shared coordinator-facing contract implemented by WorkerManagerPython and WorkerManagerExternal. It gives ModuleCoordinator one deploy, undeploy, rollback, status, and Module Handle surface without requiring both backends to share a worker protocol. +_Avoid_: Shared worker implementation, PythonWorker subclass for external deployment + +**Module Handle**: +The coordinator-visible proxy for one deployed module instance. A Python handle may wrap an Actor in PythonWorker; an external handle may route through ExternalWorkerClient to a runtime handle on a target machine. +_Avoid_: Python object instance, machine-specific worker process + +**WorkerManagerPython**: +The manager for in-environment Python modules. It owns and schedules the local `PythonWorker` pool. +_Avoid_: External-module manager, target worker + +**WorkerManagerExternal**: +The manager for `ExternalModule` deployments. It owns Module Deployments and Target Sessions, coordinates prepare and deployment, starts one `ExternalWorker` per execution machine, and aggregates rollback and health. +_Avoid_: Per-machine manager, separate deployment reconciler + +**PythonWorker**: +A coordinator-side handle to one Python worker process that hosts one or more in-environment Python module instances. +_Avoid_: ExternalWorker, WorkerManagerPython + +**Target Session**: +The coordinator-side access path to an Execution Target. A local session executes commands and transfers files directly; an SSH session executes commands, transfers artifacts, starts the ExternalWorker, and tunnels its control RPC. +_Avoid_: Module stream transport, Runtime Host control protocol + +**ExternalWorker**: +The target-side process that owns runtime handles for all ExternalModules assigned to one execution machine for one run. It starts only after the Deployment Prepare Phase succeeds. +_Avoid_: Coordinator-side handle, one worker per module, preparation executor, persistent target agent + +**ExternalWorker Client**: +The coordinator-side RPC handle to an ExternalWorker. For an SSH target, its ordinary RPC connection is carried through the Target Session's SSH tunnel. +_Avoid_: SSH command executor, Runtime Host, stream transport + +**Runtime Host**: +The per-module runtime handle owned by an ExternalWorker. It receives a Module Launch Envelope, initializes control and stream bindings, supervises one ExternalModule implementation, and reports ready or failure. For native modules this may be only ExternalWorker state around a native subprocess; for packaged Python it may include a thin entrypoint process that imports the implementation inside the prepared Python environment. +_Avoid_: Worker manager, multi-module worker, required extra native wrapper process + +**Python Runtime Entrypoint**: +The thin process entrypoint used by a packaged Python ExternalModule to enter its prepared Python environment, import the implementation class, and attach it to the Module Launch Envelope. It is not a second worker and should contain only bootstrap logic. +_Avoid_: PythonWorker replacement, module implementation, heavyweight runtime supervisor + +**Ready Acknowledgement**: +The explicit runtime-handle signal sent after the launch envelope is parsed, the implementation is initialized, control is active, and stream bindings are ready. +_Avoid_: Treating process creation as successful module startup + +**Module Launch Envelope**: +The unified serialized handoff to a runtime handle containing module identity, implementation launch metadata, module config, stream topics, transport descriptors, and control details. It extends the current `NativeModule.stdin_config` shape. +_Avoid_: Separate user-facing config and connection payloads + +**Deployment Control Plane**: +The command and lifecycle path between ModuleCoordinator, WorkerManagerExternal, Target Sessions, ExternalWorker Clients, ExternalWorkers, and runtime handles. Target Sessions handle pre-worker bootstrap and preparation; ExternalWorker RPC handles deployed lifecycle, status, logs, health, and method calls. SSH may carry that RPC through a tunnel but never carries module stream data. +_Avoid_: Stream data transport + +**Deployment Data Plane**: +The transport path used by module streams, such as Zenoh, DDS, ROS, LCM, or SHM where applicable. +_Avoid_: Worker control protocol, lifecycle channel + +**Fail-Fast Startup Rollback**: +The rule that startup stops already-started workers and runtime handles if any module fails before the deployment reaches ready state. +_Avoid_: Partial startup, orphaned Runtime Host + +**ExternalWorker Lease**: +A coordinator heartbeat or lease that causes an ExternalWorker to stop its runtime handles if the coordinator disappears. +_Avoid_: Orphaned target processes, required persistent agent diff --git a/dimos/core/coordination/blueprints.py b/dimos/core/coordination/blueprints.py index f21ff3fe30..6bf619a7c3 100644 --- a/dimos/core/coordination/blueprints.py +++ b/dimos/core/coordination/blueprints.py @@ -28,6 +28,11 @@ from dimos.core.global_config import GlobalConfig from dimos.core.module import ModuleBase, is_module_type +from dimos.core.runtime_environment import ( + RuntimeEnvironment, + RuntimeEnvironmentRegistry, + RuntimePlacement, +) from dimos.core.stream import In, Out from dimos.core.transport import PubSubTransport from dimos.spec.utils import Spec, is_spec @@ -142,7 +147,12 @@ def create(cls, module: type[ModuleBase], kwargs: dict[str, Any]) -> Self: # These fields cannot be pickled. -_PROXY_FIELDS = ("transport_map", "global_config_overrides", "remapping_map") +_PROXY_FIELDS = ( + "transport_map", + "global_config_overrides", + "remapping_map", + "runtime_placement_map", +) @dataclass(frozen=True) @@ -156,6 +166,12 @@ class Blueprint: remapping_map: Mapping[tuple[type[ModuleBase], str], str | type[ModuleBase] | type[Spec]] = ( field(default_factory=lambda: MappingProxyType({})) ) + runtime_environment_registry: RuntimeEnvironmentRegistry = field( + default_factory=RuntimeEnvironmentRegistry + ) + runtime_placement_map: Mapping[type[ModuleBase], RuntimePlacement] = field( + default_factory=lambda: MappingProxyType({}) + ) requirement_checks: tuple[Callable[[], str | None], ...] = field(default_factory=tuple) configurator_checks: "tuple[SystemConfigurator, ...]" = field(default_factory=tuple) @@ -205,6 +221,25 @@ def remappings( remappings_dict[(module, old)] = new return replace(self, remapping_map=MappingProxyType(remappings_dict)) + def runtime_environments(self, *environments: RuntimeEnvironment) -> "Blueprint": + return replace( + self, + runtime_environment_registry=self.runtime_environment_registry.register(*environments), + ) + + def runtime_placements( + self, + placements: Mapping[type[ModuleBase], RuntimePlacement], + ) -> "Blueprint": + placement_dict = dict(self.runtime_placement_map) + for module, placement in placements.items(): + if module not in {bp.module for bp in self.blueprints}: + raise ValueError( + f"Runtime placement for {module.__name__} does not match a blueprint module" + ) + placement_dict[module] = placement + return replace(self, runtime_placement_map=MappingProxyType(placement_dict)) + def requirements(self, *checks: Callable[[], str | None]) -> "Blueprint": return replace(self, requirement_checks=self.requirement_checks + tuple(checks)) @@ -230,17 +265,40 @@ def autoconnect(*blueprints: Blueprint) -> Blueprint: all_remappings = dict( # type: ignore[var-annotated] reduce(operator.iadd, [list(x.remapping_map.items()) for x in blueprints], []) ) + runtime_environment_registry = RuntimeEnvironmentRegistry() + for blueprint in blueprints: + runtime_environment_registry = runtime_environment_registry.merge( + blueprint.runtime_environment_registry + ) + placement_owner_by_module: dict[type[ModuleBase], int] = {} + for blueprint_index, blueprint in enumerate(blueprints): + for atom in blueprint.blueprints: + placement_owner_by_module[atom.module] = blueprint_index + all_disabled_modules = tuple( + module for bp in blueprints for module in bp.disabled_modules_tuple + ) + disabled_modules = set(all_disabled_modules) + all_runtime_placements: dict[type[ModuleBase], RuntimePlacement] = {} + for blueprint_index, blueprint in enumerate(blueprints): + for module, placement in blueprint.runtime_placement_map.items(): + if placement_owner_by_module.get(module) != blueprint_index: + continue + if module not in disabled_modules: + runtime_environment_registry.resolve(placement.runtime) + if module in all_runtime_placements and all_runtime_placements[module] != placement: + raise ValueError(f"Conflicting runtime placements for {module.__name__}") + all_runtime_placements[module] = placement all_requirement_checks = tuple(check for bs in blueprints for check in bs.requirement_checks) all_configurator_checks = tuple(check for bs in blueprints for check in bs.configurator_checks) return Blueprint( blueprints=all_blueprints, - disabled_modules_tuple=tuple( - module for bp in blueprints for module in bp.disabled_modules_tuple - ), + disabled_modules_tuple=all_disabled_modules, transport_map=MappingProxyType(all_transports), global_config_overrides=MappingProxyType(all_config_overrides), remapping_map=MappingProxyType(all_remappings), + runtime_environment_registry=runtime_environment_registry, + runtime_placement_map=MappingProxyType(all_runtime_placements), requirement_checks=all_requirement_checks, configurator_checks=all_configurator_checks, ) diff --git a/dimos/core/coordination/module_coordinator.py b/dimos/core/coordination/module_coordinator.py index 849c928681..c4b109fce6 100644 --- a/dimos/core/coordination/module_coordinator.py +++ b/dimos/core/coordination/module_coordinator.py @@ -29,6 +29,11 @@ from dimos.core.global_config import GlobalConfig, global_config from dimos.core.module import ModuleBase, ModuleSpec from dimos.core.resource import Resource +from dimos.core.runtime_environment import RuntimePlacement +from dimos.core.runtime_reconciliation import ( + active_runtime_placements, + reconcile_blueprint_runtimes, +) from dimos.core.transport import ( LCMTransport, PubSubTransport, @@ -67,8 +72,10 @@ def __init__( g: GlobalConfig = global_config, ) -> None: self._global_config = g - manager_types: list[type[WorkerManager]] = [WorkerManagerPython] - self._managers = {cls.deployment_identifier: cls(g=g) for cls in manager_types} + self._python_manager = WorkerManagerPython(g=g) + self._managers: dict[str, WorkerManager] = { + self._python_manager.deployment_identifier: self._python_manager + } self._deployed_modules = {} self._deployed_atoms: dict[type[ModuleBase], BlueprintAtom] = {} self._resolved_module_refs: dict[tuple[type[ModuleBase], str], type[ModuleBase]] = {} @@ -174,38 +181,73 @@ def deploy( raise ValueError("Trying to dimos.deploy before the client has started") deployed_module = self._managers[module_class.deployment].deploy( - module_class, global_config, kwargs + module_class, + global_config, + kwargs, ) with self._modules_lock: self._deployed_modules[module_class] = deployed_module return deployed_module # type: ignore[return-value] def deploy_parallel( - self, module_specs: list[ModuleSpec], blueprint_args: Mapping[str, Mapping[str, Any]] + self, + module_specs: list[ModuleSpec], + blueprint_args: Mapping[str, Mapping[str, Any]], + *, + runtime_placement_map: Mapping[type[ModuleBase], RuntimePlacement] | None = None, ) -> list[ModuleProxy]: if not self._managers: raise ValueError("Not started") - # Group specs by deployment type, tracking original indices for reassembly + results: list[Any] = [None] * len(module_specs) indices_by_deployment: dict[str, list[int]] = {} specs_by_deployment: dict[str, list[ModuleSpec]] = {} - for index, spec in enumerate(module_specs): - # spec = (module_class, global_config, kwargs) - dep = spec[0].deployment - indices_by_deployment.setdefault(dep, []).append(index) - specs_by_deployment.setdefault(dep, []).append(spec) - - results: list[Any] = [None] * len(module_specs) + placements_by_deployment: dict[str, dict[type[ModuleBase], RuntimePlacement]] = {} + runtime_placement_map = runtime_placement_map or {} def _deploy_group(dep: str) -> None: - deployed = self._managers[dep].deploy_parallel(specs_by_deployment[dep], blueprint_args) + deployed = self._managers[dep].deploy_parallel( + specs_by_deployment[dep], + blueprint_args, + runtime_placements=placements_by_deployment.get(dep, {}), + ) for index, module in zip(indices_by_deployment[dep], deployed, strict=True): results[index] = module + def _rollback_successful_deployments() -> list[Exception]: + cleanup_errors: list[Exception] = [] + for index, module in enumerate(results): + if module is None: + continue + module_class = module_specs[index][0] + manager = self._managers.get(module_class.deployment) + if manager is None: + continue + try: + manager.undeploy(module, module_class) + except Exception as e: + cleanup_errors.append(e) + return cleanup_errors + try: + # Group specs by deployment type, tracking original indices for reassembly + for index, spec in enumerate(module_specs): + # spec = (module_class, global_config, kwargs) + dep = spec[0].deployment + indices_by_deployment.setdefault(dep, []).append(index) + specs_by_deployment.setdefault(dep, []).append(spec) + placement = runtime_placement_map.get(spec[0]) + if placement is not None: + placements_by_deployment.setdefault(dep, {})[spec[0]] = placement + safe_thread_map(list(specs_by_deployment.keys()), _deploy_group) - except: - self.stop() + except Exception as error: + cleanup_errors = _rollback_successful_deployments() + if cleanup_errors: + raise ExceptionGroup( + "Python runtime deployment failed and rollback cleanup also failed", + [error, *cleanup_errors], + ) from error raise with self._modules_lock: @@ -250,6 +292,15 @@ def _resolve_class(self, cls: type[ModuleBase]) -> type[ModuleBase]: def get_instance(self, module: type[ModuleBase]) -> ModuleProxy: return self._deployed_modules.get(self._resolve_class(module)) # type: ignore[return-value] + def transport_for(self, name: str, stream_type: type) -> PubSubTransport[Any] | None: + return self._transport_registry.get((name, stream_type)) + + def worker_ids(self, deployment_identifier: str = "python") -> set[int]: + manager = self._managers[deployment_identifier] + if not isinstance(manager, WorkerManagerPython): + raise TypeError(f"Deployment {deployment_identifier!r} does not use Python workers") + return {worker.worker_id for worker in manager.workers} + def _send_on_system_modules(self) -> None: modules = list(self._deployed_modules.values()) for module in modules: @@ -301,6 +352,7 @@ def build( _run_configurators(blueprint) _check_requirements(blueprint) _verify_no_name_conflicts(blueprint) + reconcile_blueprint_runtimes(blueprint) logger.info("Starting the modules") coordinator = cls(g=global_config) @@ -345,18 +397,19 @@ def _load_blueprint( if "g" in blueprint_args: self._global_config.update(**blueprint_args.pop("g")) - # Scale worker pool. - n_extra = int(blueprint.global_config_overrides.get("n_workers", 0)) - python_wm = cast("WorkerManagerPython", self._managers["python"]) - if n_extra: - python_wm.add_workers(n_extra) - if not python_wm.workers and blueprint.active_blueprints: - python_wm.add_workers(1) - _run_configurators(blueprint) _check_requirements(blueprint) _verify_no_name_conflicts(blueprint) _verify_no_conflicts_with_existing(blueprint, self._transport_registry) + reconcile_blueprint_runtimes(blueprint) + + # Scale worker pool only after runtime reconciliation succeeds, so this + # deployment slice does not launch any new workers before the barrier. + n_extra = int(blueprint.global_config_overrides.get("n_workers", 0)) + if n_extra: + self._python_manager.add_workers(n_extra) + if not self._python_manager.workers and blueprint.active_blueprints: + self._python_manager.add_workers(1) # Reject duplicate modules. for bp in blueprint.active_blueprints: @@ -418,9 +471,9 @@ def _unload_module(self, module_class: type[ModuleBase]) -> None: exc_info=True, ) - python_wm = cast("WorkerManagerPython", self._managers["python"]) + manager = self._managers[module_class.deployment] try: - python_wm.undeploy(proxy) + manager.undeploy(proxy, module_class) except Exception: logger.error( "Error undeploying module from worker", @@ -497,6 +550,7 @@ def _restart_module( for (consumer, ref_name), target in self._resolved_module_refs.items() if consumer is module_class ] + placement = self._python_manager.runtime_placement_for(module_class) self.unload_module(module_class) @@ -515,8 +569,9 @@ def _restart_module( self._class_aliases[old_cls] = new_class self._class_aliases[module_class] = new_class - python_wm = cast("WorkerManagerPython", self._managers["python"]) - new_proxy = python_wm.deploy_fresh(new_class, self._global_config, kwargs) + new_proxy = self._python_manager.deploy_fresh( + new_class, self._global_config, kwargs, runtime_placement=placement + ) self._deployed_modules[new_class] = new_proxy new_bp = new_class.blueprint(**kwargs) @@ -717,7 +772,14 @@ def _deploy_all_modules( for bp in blueprint.active_blueprints: module_specs.append((bp.module, gc, bp.kwargs.copy())) - module_coordinator.deploy_parallel(module_specs, blueprint_args) + module_coordinator._python_manager.register_runtime_environments( + blueprint.runtime_environment_registry + ) + module_coordinator.deploy_parallel( + module_specs, + blueprint_args, + runtime_placement_map=active_runtime_placements(blueprint), + ) for bp in blueprint.active_blueprints: module_coordinator._deployed_atoms[bp.module] = bp diff --git a/dimos/core/coordination/python_worker.py b/dimos/core/coordination/python_worker.py index 1e1ae1675c..66861770e6 100644 --- a/dimos/core/coordination/python_worker.py +++ b/dimos/core/coordination/python_worker.py @@ -14,6 +14,7 @@ from __future__ import annotations from dataclasses import dataclass +import importlib import logging import multiprocessing from multiprocessing.connection import Connection @@ -24,6 +25,7 @@ import traceback from typing import TYPE_CHECKING, Any +from dimos.core.coordination.worker_launcher import ForkserverWorkerLauncher, WorkerLauncher from dimos.core.coordination.worker_messages import ( CallMethodRequest, DeployModuleRequest, @@ -161,13 +163,14 @@ def reset_forkserver_context() -> None: class PythonWorker: - def __init__(self) -> None: + def __init__(self, launcher: WorkerLauncher | None = None) -> None: self._lock = threading.Lock() self._modules: dict[int, Actor] = {} self._reserved: int = 0 self._process: Any = None self._conn: Connection | None = None self._worker_id: int = _worker_ids.next() + self._launcher = launcher or ForkserverWorkerLauncher() self.dedicated: bool = False @property @@ -202,22 +205,16 @@ def reserve_slot(self) -> None: self._reserved += 1 def start_process(self) -> None: - ctx = get_forkserver_context() - parent_conn, child_conn = ctx.Pipe() - self._conn = parent_conn - - self._process = ctx.Process( - target=_worker_entrypoint, - args=(child_conn, self._worker_id), - daemon=True, - ) - self._process.start() + self._process = self._launcher.launch(self._worker_id) + self._conn = self._process.connection def deploy_module( self, module_class: type[ModuleBase], global_config: GlobalConfig = global_config, kwargs: dict[str, Any] | None = None, + implementation_path: str | None = None, + runtime_name: str | None = None, ) -> Actor: if self._conn is None: raise RuntimeError("Worker process not started") @@ -226,17 +223,29 @@ def deploy_module( kwargs["g"] = global_config module_id = _module_ids.next() - request = DeployModuleRequest(module_id=module_id, module_class=module_class, kwargs=kwargs) + request = DeployModuleRequest( + module_id=module_id, + module_class=module_class, + kwargs=kwargs, + implementation_path=implementation_path, + runtime_name=runtime_name, + ) try: with self._lock: self._conn.send(request) response: WorkerResponse = self._conn.recv() if response.error: - raise RuntimeError(f"Failed to deploy module: {response.error}") + context = ( + f" in runtime {runtime_name!r} using implementation {implementation_path!r}" + if runtime_name or implementation_path + else "" + ) + raise RuntimeError(f"Failed to deploy module{context}: {response.error}") actor = Actor(self._conn, module_class, self._worker_id, module_id, self._lock) - actor.set_ref(actor).result() + if self._process is None or self._process.supports_parent_actor_ref: + actor.set_ref(actor).result() self._modules[module_id] = actor logger.info( @@ -327,7 +336,7 @@ class _WorkerState: should_stop: bool = False -def _worker_entrypoint(conn: Connection, worker_id: int) -> None: +def worker_entrypoint(conn: Connection, worker_id: int) -> None: apply_library_config() signal.signal(signal.SIGINT, signal.SIG_IGN) # coordinator handles shutdown state = _WorkerState(instances={}, worker_id=worker_id) @@ -368,16 +377,33 @@ def _worker_entrypoint(conn: Connection, worker_id: int) -> None: logger.error("Error during worker shutdown", exc_info=True) +_worker_entrypoint = worker_entrypoint + + def _handle_request(request: Any, state: _WorkerState) -> WorkerResponse: match request: - case DeployModuleRequest(module_id=module_id, module_class=module_class, kwargs=kwargs): + case DeployModuleRequest( + module_id=module_id, + module_class=module_class, + kwargs=kwargs, + implementation_path=implementation_path, + runtime_name=runtime_name, + ): # Always use the same transport backend as the host. host_config = kwargs.get("g") if host_config is not None: global_config.update(transport=host_config.transport) - state.instances[module_id] = module_class(**kwargs) - + deployment_class = _resolve_deployment_class( + module_class, + implementation_path=implementation_path, + runtime_name=runtime_name, + ) + instance = deployment_class(**kwargs) + rpc = getattr(instance, "rpc", None) + if deployment_class is not module_class and rpc is not None: + rpc.serve_module_rpc(instance, name=module_class.__name__) + state.instances[module_id] = instance return WorkerResponse(result=module_id) case SetRefRequest(module_id=module_id, ref=ref): @@ -409,6 +435,30 @@ def _handle_request(request: Any, state: _WorkerState) -> WorkerResponse: return WorkerResponse(error=f"Unknown request type: {type(request)}") +def _resolve_deployment_class( + module_class: type[ModuleBase], + *, + implementation_path: str | None, + runtime_name: str | None, +) -> type[ModuleBase]: + if implementation_path is None: + return module_class + module_name, _, class_name = implementation_path.rpartition(".") + if not module_name or not class_name: + raise ValueError( + f"Invalid Runtime Implementation path {implementation_path!r} for " + f"{module_class.__module__}.{module_class.__name__} in runtime {runtime_name!r}" + ) + module = importlib.import_module(module_name) + implementation = getattr(module, class_name) + if not isinstance(implementation, type) or not issubclass(implementation, module_class): + raise TypeError( + f"Runtime Implementation {implementation_path!r} in runtime {runtime_name!r} " + f"must subclass Module Contract {module_class.__module__}.{module_class.__name__}" + ) + return implementation + + def _worker_loop(conn: Connection, state: _WorkerState) -> None: while True: try: diff --git a/dimos/core/coordination/test_blueprints.py b/dimos/core/coordination/test_blueprints.py index d8ada76b20..0cf56e2319 100644 --- a/dimos/core/coordination/test_blueprints.py +++ b/dimos/core/coordination/test_blueprints.py @@ -35,6 +35,7 @@ ) from dimos.core.core import rpc from dimos.core.module import Module +from dimos.core.runtime_environment import PythonProjectRuntimeEnvironment, RuntimePlacement from dimos.core.stream import In, Out from dimos.core.transport import LCMTransport from dimos.spec.utils import Spec @@ -251,3 +252,66 @@ def test_active_blueprints_filters_disabled() -> None: active_modules = {bp.module for bp in blueprint.active_blueprints} assert ModuleA not in active_modules assert ModuleB in active_modules + + +def test_runtime_environment_and_placement_api_composes_with_autoconnect() -> None: + runtime = PythonProjectRuntimeEnvironment(name="runtime-a", project="/tmp/runtime-a") + placement = RuntimePlacement( + runtime="runtime-a", + implementation="example_runtime_impl.ModuleAImpl", + ) + + bp_a = ModuleA.blueprint().runtime_placements({ModuleA: placement}) + bp_b = ModuleB.blueprint().runtime_environments(runtime) + merged = autoconnect(bp_a, bp_b) + + assert merged.runtime_environment_registry.resolve("runtime-a") is runtime + assert merged.runtime_placement_map[ModuleA] == placement + assert ModuleB not in merged.runtime_placement_map + + +def test_runtime_placement_rejects_unknown_or_non_member_module() -> None: + runtime = PythonProjectRuntimeEnvironment(name="runtime-a", project="/tmp/runtime-a") + blueprint = ModuleA.blueprint().runtime_environments(runtime) + + with pytest.raises(ValueError, match="does not match a blueprint module"): + blueprint.runtime_placements( + {ModuleB: RuntimePlacement(runtime="runtime-a", implementation="impl.ModuleB")} + ) + + unresolved = ModuleA.blueprint().runtime_placements( + {ModuleA: RuntimePlacement(runtime="missing", implementation="impl.ModuleA")} + ) + with pytest.raises(Exception, match="Unknown runtime environment"): + autoconnect(unresolved) + + +def test_autoconnect_allows_disabled_placement_with_unknown_runtime() -> None: + blueprint = ModuleA.blueprint().runtime_placements( + {ModuleA: RuntimePlacement(runtime="missing", implementation="impl.ModuleA")} + ) + + merged = autoconnect(blueprint.disabled_modules(ModuleA)) + + assert merged.runtime_placement_map[ModuleA] == RuntimePlacement( + runtime="missing", + implementation="impl.ModuleA", + ) + assert ModuleA not in {bp.module for bp in merged.active_blueprints} + + +def test_autoconnect_filters_placements_for_replaced_duplicate_modules() -> None: + old_runtime = PythonProjectRuntimeEnvironment(name="old-runtime", project="/tmp/old-runtime") + old = ( + ModuleA.blueprint(key1="old") + .runtime_environments(old_runtime) + .runtime_placements( + {ModuleA: RuntimePlacement(runtime="old-runtime", implementation="impl.Old")} + ) + ) + new = ModuleA.blueprint(key1="new") + + merged = autoconnect(old, new) + + assert merged.blueprints[0].kwargs == {"key1": "new"} + assert ModuleA not in merged.runtime_placement_map diff --git a/dimos/core/coordination/test_module_coordinator.py b/dimos/core/coordination/test_module_coordinator.py index c1baad17b2..6334a1028f 100644 --- a/dimos/core/coordination/test_module_coordinator.py +++ b/dimos/core/coordination/test_module_coordinator.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +import time from types import MappingProxyType from typing import Protocol @@ -33,10 +34,10 @@ _verify_no_conflicts_with_existing, _verify_no_name_conflicts, ) -from dimos.core.coordination.worker_manager_python import WorkerManagerPython from dimos.core.core import rpc from dimos.core.global_config import GlobalConfig from dimos.core.module import Module +from dimos.core.runtime_environment import PythonProjectRuntimeEnvironment, RuntimePlacement from dimos.core.stream import In, Out from dimos.msgs.sensor_msgs.Image import Image from dimos.spec.utils import Spec @@ -86,6 +87,91 @@ class ModuleC(Module): data3: In[Data3] +class RuntimeModuleA(Module): + @rpc + def stop(self) -> None: ... + + +class RuntimeModuleB(Module): + @rpc + def stop(self) -> None: ... + + +class _FakeProxy: + @rpc + def build(self) -> None: ... + + @rpc + def start(self) -> None: ... + + @rpc + def stop(self) -> None: ... + + +class _FakeManager: + deployment_identifier = "python" + + def __init__( + self, + key: str, + calls: list[str], + fail: bool = False, + fail_delay: float = 0.0, + fail_undeploy: bool = False, + fail_stop: bool = False, + ) -> None: + self.key = key + self.calls = calls + self.fail = fail + self.fail_delay = fail_delay + self.fail_undeploy = fail_undeploy + self.fail_stop = fail_stop + self.workers = [object()] + self.deployments: list[tuple[list[type[Module]], dict[type[Module], RuntimePlacement]]] = [] + self.registered_runtime_names: list[set[str]] = [] + self.runtime_placements_by_module: dict[type[Module], RuntimePlacement] = {} + + def register_runtime_environments(self, registry) -> None: + self.registered_runtime_names.append(set(registry.environments)) + + def runtime_placement_for(self, module_class) -> RuntimePlacement | None: + return self.runtime_placements_by_module.get(module_class) + + def start(self) -> None: + self.calls.append(f"start:{self.key}") + + def stop(self) -> None: + self.calls.append(f"stop:{self.key}") + if self.fail_stop: + raise RuntimeError(f"stop-boom:{self.key}") + + def deploy(self, module_class, global_config, kwargs, runtime_placement=None): + self.calls.append(f"deploy-one:{self.key}:{module_class.__name__}") + return _FakeProxy() + + def deploy_parallel(self, specs, blueprint_args, runtime_placements=None): + if self.fail: + if self.fail_delay: + time.sleep(self.fail_delay) + raise RuntimeError(f"boom:{self.key}") + placements = dict(runtime_placements or {}) + modules = [spec[0] for spec in specs] + self.runtime_placements_by_module.update(placements) + self.deployments.append((modules, placements)) + self.calls.append(f"deploy:{self.key}:{','.join(module.__name__ for module in modules)}") + return [_FakeProxy() for _ in modules] + + def undeploy(self, module: _FakeProxy) -> None: + self.calls.append(f"undeploy:{self.key}") + if self.fail_undeploy: + raise RuntimeError(f"undeploy-boom:{self.key}") + + def health_check(self) -> bool: + return True + + def suppress_console(self) -> None: ... + + class SourceModule(Module): color_image: Out[Data1] @@ -194,6 +280,87 @@ def test_build_happy_path() -> None: coordinator.stop() +def test_deploy_parallel_passes_runtime_dispatch_to_python_manager() -> None: + calls: list[str] = [] + coordinator = ModuleCoordinator(g=GlobalConfig(n_workers=1, viewer="none")) + manager = _FakeManager("python", calls) + coordinator._managers = {"python": manager} + coordinator._started = True + placement = RuntimePlacement( + runtime="runtime-a", + implementation="dimos.core.coordination.test_module_coordinator.RuntimeModuleA", + ) + + results = coordinator.deploy_parallel( + [ + (RuntimeModuleA, coordinator._global_config, {}), + (RuntimeModuleB, coordinator._global_config, {}), + ], + {}, + runtime_placement_map={RuntimeModuleA: placement}, + ) + + assert len(results) == 2 + assert calls == ["deploy:python:RuntimeModuleA,RuntimeModuleB"] + assert manager.deployments[0] == ( + [RuntimeModuleA, RuntimeModuleB], + {RuntimeModuleA: placement}, + ) + assert set(coordinator._deployed_modules) == {RuntimeModuleA, RuntimeModuleB} + + +def test_failed_python_deployment_does_not_commit_modules() -> None: + calls: list[str] = [] + coordinator = ModuleCoordinator(g=GlobalConfig(n_workers=1, viewer="none")) + coordinator._managers = {"python": _FakeManager("python", calls, fail=True)} + coordinator._started = True + placement = RuntimePlacement(runtime="runtime-a", implementation="impl.RuntimeModuleA") + + with pytest.raises(ExceptionGroup): + coordinator.deploy_parallel( + [ + (RuntimeModuleB, coordinator._global_config, {}), + (RuntimeModuleA, coordinator._global_config, {}), + ], + {}, + runtime_placement_map={RuntimeModuleA: placement}, + ) + + assert RuntimeModuleB not in coordinator._deployed_modules + assert RuntimeModuleA not in coordinator._deployed_modules + + +def test_load_blueprint_reconciles_before_runtime_manager_launch(monkeypatch) -> None: + calls: list[str] = [] + coordinator = ModuleCoordinator(g=GlobalConfig(n_workers=1, viewer="none")) + manager = _FakeManager("python", calls) + coordinator._python_manager = manager + coordinator._managers = {"python": manager} + coordinator._started = True + + def fake_reconcile(_blueprint) -> None: + calls.append("reconcile") + + monkeypatch.setattr( + "dimos.core.coordination.module_coordinator.reconcile_blueprint_runtimes", + fake_reconcile, + ) + + runtime = PythonProjectRuntimeEnvironment(name="runtime-a", project="/tmp/runtime-a") + placement = RuntimePlacement(runtime="runtime-a", implementation="impl.RuntimeModuleA") + blueprint = ( + RuntimeModuleA.blueprint() + .runtime_environments(runtime) + .runtime_placements({RuntimeModuleA: placement}) + ) + + coordinator.load_blueprint(blueprint) + + assert calls.index("reconcile") < calls.index("deploy:python:RuntimeModuleA") + assert manager.registered_runtime_names == [{"runtime-a"}] + assert manager.deployments[0] == ([RuntimeModuleA], {RuntimeModuleA: placement}) + + def test_name_conflicts_are_reported() -> None: class ModuleA(Module): shared_data: Out[Data1] @@ -613,12 +780,12 @@ def test_restart_module_preserves_stream_wiring(dynamic_coordinator) -> None: c = dynamic_coordinator.get_instance(ModuleC) assert c is not None topic_before = c.data3.transport.topic - registry_before = dynamic_coordinator._transport_registry[("data3", Data3)] + registry_before = dynamic_coordinator.transport_for("data3", Data3) dynamic_coordinator.restart_module(ModuleC, reload_source=False) # Transport in the registry is the same parent-side object. - assert dynamic_coordinator._transport_registry[("data3", Data3)] is registry_before + assert dynamic_coordinator.transport_for("data3", Data3) is registry_before c_after = dynamic_coordinator.get_instance(ModuleC) assert c_after is not None @@ -656,15 +823,13 @@ def test_restart_module_shuts_down_empty_worker(dynamic_coordinator) -> None: """Restart shuts down the old worker (when empty) and spawns a new one.""" dynamic_coordinator.load_module(ModuleA) - python_wm = dynamic_coordinator._managers["python"] - assert isinstance(python_wm, WorkerManagerPython) - old_worker_ids = {w.worker_id for w in python_wm.workers} + old_worker_ids = dynamic_coordinator.worker_ids() assert len(old_worker_ids) == 1 dynamic_coordinator.restart_module(ModuleA, reload_source=False) - new_worker_ids = {w.worker_id for w in python_wm.workers} + new_worker_ids = dynamic_coordinator.worker_ids() assert len(new_worker_ids) == 1 assert new_worker_ids.isdisjoint(old_worker_ids) diff --git a/dimos/core/coordination/test_worker.py b/dimos/core/coordination/test_worker.py index c606dcda73..730e4951b3 100644 --- a/dimos/core/coordination/test_worker.py +++ b/dimos/core/coordination/test_worker.py @@ -16,7 +16,9 @@ import pytest +from dimos.core.coordination.python_worker import _handle_request, _WorkerState from dimos.core.coordination.worker_manager_python import WorkerManagerPython +from dimos.core.coordination.worker_messages import DeployModuleRequest from dimos.core.core import rpc from dimos.core.global_config import GlobalConfig, global_config from dimos.core.module import Module @@ -64,6 +66,16 @@ def get_value(self) -> int: return self.value +class RuntimeSimpleModule(SimpleModule): + @rpc + def increment(self) -> int: + return super().increment() + 1000 + + +class NotASimpleModule(Module): + pass + + class ThirdModule(Module): multiplier: int = 1 @@ -132,6 +144,64 @@ def test_worker_manager_basic(create_worker_manager): module.stop() +def test_worker_deploy_request_imports_runtime_implementation_and_validates_subclass(): + state = _WorkerState(instances={}, worker_id=1) + response = _handle_request( + DeployModuleRequest( + module_id=10, + module_class=SimpleModule, + kwargs={"g": global_config}, + runtime_name="test-runtime", + implementation_path="dimos.core.coordination.test_worker.RuntimeSimpleModule", + ), + state, + ) + + assert response.error is None + assert isinstance(state.instances[10], RuntimeSimpleModule) + assert state.instances[10].increment() == 1001 + state.instances[10].stop() + + +def test_worker_deploy_request_runtime_implementation_error_context(): + with pytest.raises(TypeError) as exc_info: + _handle_request( + DeployModuleRequest( + module_id=10, + module_class=SimpleModule, + kwargs={"g": global_config}, + runtime_name="bad-runtime", + implementation_path="dimos.core.coordination.test_worker.NotASimpleModule", + ), + _WorkerState(instances={}, worker_id=1), + ) + + message = str(exc_info.value) + assert "bad-runtime" in message + assert "Runtime Implementation" in message + assert "Module Contract" in message + assert "NotASimpleModule" in message + + +def test_worker_deploy_request_invalid_runtime_implementation_path_context(): + with pytest.raises(ValueError) as exc_info: + _handle_request( + DeployModuleRequest( + module_id=10, + module_class=SimpleModule, + kwargs={"g": global_config}, + runtime_name="bad-runtime", + implementation_path="NotAQualifiedPath", + ), + _WorkerState(instances={}, worker_id=1), + ) + + message = str(exc_info.value) + assert "bad-runtime" in message + assert "Invalid Runtime Implementation path" in message + assert "SimpleModule" in message + + @pytest.mark.skipif_macos_bug def test_worker_manager_multiple_different_modules(create_worker_manager): worker_manager = create_worker_manager(n_workers=2) diff --git a/dimos/core/coordination/venv_worker_entrypoint.py b/dimos/core/coordination/venv_worker_entrypoint.py new file mode 100644 index 0000000000..4de578f798 --- /dev/null +++ b/dimos/core/coordination/venv_worker_entrypoint.py @@ -0,0 +1,33 @@ +# Copyright 2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +from __future__ import annotations + +import argparse +from multiprocessing.connection import Client + +from dimos.core.coordination.python_worker import worker_entrypoint + + +def main() -> None: + parser = argparse.ArgumentParser(description="Run a DimOS Python worker from a runtime env") + parser.add_argument("--address", required=True) + parser.add_argument("--authkey-hex", required=True) + parser.add_argument("--worker-id", required=True, type=int) + args = parser.parse_args() + conn = Client(args.address, family="AF_UNIX", authkey=bytes.fromhex(args.authkey_hex)) + worker_entrypoint(conn, args.worker_id) + + +if __name__ == "__main__": + main() diff --git a/dimos/core/coordination/worker_launcher.py b/dimos/core/coordination/worker_launcher.py new file mode 100644 index 0000000000..0f4951aac8 --- /dev/null +++ b/dimos/core/coordination/worker_launcher.py @@ -0,0 +1,269 @@ +# Copyright 2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +from __future__ import annotations + +from abc import ABC, abstractmethod +from multiprocessing.connection import Connection, Listener +from multiprocessing.process import BaseProcess +import os +from pathlib import Path +import queue +import secrets +import signal +import socket +import subprocess +import tempfile +import threading + +from dimos.core.runtime_environment import PythonProjectLaunchMaterial + + +class WorkerLaunchError(RuntimeError): + pass + + +class WorkerProcessHandle(ABC): + connection: Connection + supports_parent_actor_ref: bool = True + + @property + @abstractmethod + def pid(self) -> int | None: ... + + @abstractmethod + def join(self, timeout: float | None = None) -> None: ... + + @abstractmethod + def is_alive(self) -> bool: ... + + @abstractmethod + def terminate(self) -> None: ... + + +class WorkerLauncher(ABC): + @abstractmethod + def launch(self, worker_id: int) -> WorkerProcessHandle: ... + + +class ForkserverWorkerProcessHandle(WorkerProcessHandle): + def __init__(self, process: BaseProcess, connection: Connection) -> None: + self._process = process + self.connection = connection + + @property + def pid(self) -> int | None: + return self._process.pid + + def join(self, timeout: float | None = None) -> None: + self._process.join(timeout=timeout) + + def is_alive(self) -> bool: + return self._process.is_alive() + + def terminate(self) -> None: + self._process.terminate() + + +class ForkserverWorkerLauncher(WorkerLauncher): + def launch(self, worker_id: int) -> WorkerProcessHandle: + # Imported lazily to avoid the python_worker <-> worker_launcher import cycle. + from dimos.core.coordination.python_worker import get_forkserver_context, worker_entrypoint + + ctx = get_forkserver_context() + parent_conn, child_conn = ctx.Pipe() + process = ctx.Process(target=worker_entrypoint, args=(child_conn, worker_id), daemon=True) + process.start() + return ForkserverWorkerProcessHandle(process, parent_conn) + + +class SubprocessWorkerProcessHandle(WorkerProcessHandle): + supports_parent_actor_ref = False + + def __init__( + self, + process: subprocess.Popen[bytes], + connection: Connection, + *, + terminate_process_group: bool = False, + ) -> None: + self._process = process + self.connection = connection + self._terminate_process_group = terminate_process_group + + @property + def pid(self) -> int | None: + return self._process.pid + + def join(self, timeout: float | None = None) -> None: + try: + self._process.wait(timeout=timeout) + except subprocess.TimeoutExpired: + return + + def is_alive(self) -> bool: + return self._process.poll() is None + + def terminate(self) -> None: + if self._terminate_process_group and self._process.pid is not None: + _terminate_process_group(self._process.pid) + return + self._process.terminate() + + +class CommandWorkerLauncher(WorkerLauncher): + def __init__( + self, + material: PythonProjectLaunchMaterial, + *, + startup_timeout: float = 10.0, + ) -> None: + self._material = material + self._startup_timeout = startup_timeout + + def launch(self, worker_id: int) -> WorkerProcessHandle: + return _launch_subprocess_worker( + argv=( + *self._material.argv_prefix, + "-m", + "dimos.core.coordination.venv_worker_entrypoint", + ), + env=dict(self._material.env), + cwd=self._material.cwd, + worker_id=worker_id, + runtime_name=self._material.runtime_name, + startup_timeout=self._startup_timeout, + terminate_process_group=True, + ) + + +def _launch_subprocess_worker( + *, + argv: tuple[str, ...], + env: dict[str, str], + cwd: Path | None, + worker_id: int, + runtime_name: str, + startup_timeout: float, + terminate_process_group: bool, +) -> WorkerProcessHandle: + with tempfile.TemporaryDirectory(prefix="dimos-runtime-worker-") as tmpdir: + address = str(Path(tmpdir) / "worker.sock") + authkey = secrets.token_bytes(32) + listener = Listener(address, family="AF_UNIX", authkey=authkey) + process_env = {**os.environ, **env} + full_argv = ( + *argv, + "--address", + address, + "--authkey-hex", + authkey.hex(), + "--worker-id", + str(worker_id), + ) + process: subprocess.Popen[bytes] | None = None + try: + process = subprocess.Popen( + full_argv, + cwd=cwd, + env=process_env, + start_new_session=terminate_process_group, + ) + connection = _accept_worker_connection(listener, startup_timeout) + if connection is None: + _terminate_subprocess(process, terminate_process_group=terminate_process_group) + raise WorkerLaunchError( + f"Runtime {runtime_name!r} worker did not connect within {startup_timeout}s" + ) + return SubprocessWorkerProcessHandle( + process, + connection, + terminate_process_group=terminate_process_group, + ) + except Exception: + if process is not None and process.poll() is None: + _terminate_subprocess(process, terminate_process_group=terminate_process_group) + raise + finally: + listener.close() + + +def _accept_worker_connection(listener: Listener, timeout: float) -> Connection | None: + socket_listener = getattr(listener, "_listener", None) + listener_socket = getattr(socket_listener, "_socket", None) + if isinstance(listener_socket, socket.socket): + previous_timeout = listener_socket.gettimeout() + listener_socket.settimeout(timeout) + try: + return listener.accept() + except TimeoutError: + return None + finally: + listener_socket.settimeout(previous_timeout) + + results: queue.Queue[Connection | Exception] = queue.Queue(maxsize=1) + + def _accept() -> None: + try: + results.put(listener.accept()) + except Exception as error: + results.put(error) + + threading.Thread(target=_accept, daemon=True).start() + try: + result = results.get(timeout=timeout) + except queue.Empty: + return None + if isinstance(result, Exception): + raise result + return result + + +def _terminate_subprocess( + process: subprocess.Popen[bytes], + *, + terminate_process_group: bool, + timeout: float = 2.0, +) -> None: + if process.poll() is not None: + return + if terminate_process_group and process.pid is not None: + _signal_process_group(process.pid, signal.SIGTERM) + else: + process.terminate() + try: + process.wait(timeout=timeout) + return + except subprocess.TimeoutExpired: + pass + + if terminate_process_group and process.pid is not None: + _signal_process_group(process.pid, signal.SIGKILL) + else: + process.kill() + try: + process.wait(timeout=timeout) + except subprocess.TimeoutExpired: + logger_pid = process.pid + raise WorkerLaunchError(f"Worker subprocess {logger_pid} did not exit after SIGKILL") + + +def _terminate_process_group(pid: int) -> None: + _signal_process_group(pid, signal.SIGTERM) + + +def _signal_process_group(pid: int, sig: signal.Signals) -> None: + try: + os.killpg(pid, sig) + except ProcessLookupError: + return diff --git a/dimos/core/coordination/worker_manager.py b/dimos/core/coordination/worker_manager.py index f673aa21bf..4d667b0ffe 100644 --- a/dimos/core/coordination/worker_manager.py +++ b/dimos/core/coordination/worker_manager.py @@ -22,6 +22,7 @@ if TYPE_CHECKING: from dimos.core.rpc_client import ModuleProxyProtocol + from dimos.core.runtime_environment import RuntimePlacement logger = setup_logger() @@ -38,14 +39,22 @@ def deploy( module_class: type[ModuleBase], global_config: GlobalConfig, kwargs: dict[str, Any], + runtime_placement: RuntimePlacement | None = None, ) -> ModuleProxyProtocol: ... def deploy_parallel( self, specs: Sequence[ModuleSpec], blueprint_args: Mapping[str, Mapping[str, Any]], + runtime_placements: Mapping[type[ModuleBase], RuntimePlacement] | None = None, ) -> list[ModuleProxyProtocol]: ... + def undeploy( + self, + proxy: ModuleProxyProtocol, + module_class: type[ModuleBase] | None = None, + ) -> None: ... + def stop(self) -> None: ... def health_check(self) -> bool: ... diff --git a/dimos/core/coordination/worker_manager_python.py b/dimos/core/coordination/worker_manager_python.py index 4963db2dac..0efdd50536 100644 --- a/dimos/core/coordination/worker_manager_python.py +++ b/dimos/core/coordination/worker_manager_python.py @@ -15,12 +15,18 @@ from __future__ import annotations from collections.abc import Iterable, Mapping -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, cast from dimos.core.coordination.python_worker import PythonWorker +from dimos.core.coordination.worker_launcher import CommandWorkerLauncher, WorkerLauncher from dimos.core.global_config import GlobalConfig from dimos.core.module import ModuleBase, ModuleSpec from dimos.core.rpc_client import ModuleProxyProtocol, RPCClient +from dimos.core.runtime_environment import ( + PythonProjectRuntimeEnvironment, + RuntimeEnvironmentRegistry, + RuntimePlacement, +) from dimos.utils.logging_config import setup_logger from dimos.utils.safe_thread_map import safe_thread_map @@ -33,22 +39,27 @@ class WorkerManagerPython: deployment_identifier: str = "python" - def __init__(self, g: GlobalConfig) -> None: + def __init__(self, g: GlobalConfig, worker_launcher: WorkerLauncher | None = None) -> None: self._cfg = g self._n_workers = g.n_workers + self._worker_launcher = worker_launcher self._workers: list[PythonWorker] = [] + self._runtime_environment_registry = RuntimeEnvironmentRegistry() + self._runtime_workers: dict[str, list[PythonWorker]] = {} + self._runtime_launchers: dict[str, WorkerLauncher] = {} + self._runtime_placements_by_module: dict[type[ModuleBase], RuntimePlacement] = {} self._closed = False self._started = False self._stats_monitor: StatsMonitor | None = None + def register_runtime_environments(self, registry: RuntimeEnvironmentRegistry) -> None: + self._runtime_environment_registry = self._runtime_environment_registry.merge(registry) + def start(self) -> None: if self._started: return self._started = True - for _ in range(self._n_workers): - worker = PythonWorker() - worker.start_process() - self._workers.append(worker) + self._add_workers_to_pool(self._workers, self._worker_launcher, self._n_workers) logger.info("Worker pool started.", n_workers=self._n_workers) if self._cfg.dtop: @@ -63,10 +74,7 @@ def add_workers(self, n: int) -> None: raise RuntimeError("WorkerManager is closed") if not self._started: raise RuntimeError("WorkerManager not started; call start() first") - for _ in range(n): - worker = PythonWorker() - worker.start_process() - self._workers.append(worker) + self._add_workers_to_pool(self._workers, self._worker_launcher, n) self._n_workers += n logger.info("Added workers to pool.", added=n, total=self._n_workers) @@ -75,6 +83,7 @@ def deploy( module_class: type[ModuleBase], global_config: GlobalConfig, kwargs: dict[str, Any], + runtime_placement: RuntimePlacement | None = None, ) -> ModuleProxyProtocol: if self._closed: raise RuntimeError("WorkerManager is closed") @@ -82,16 +91,30 @@ def deploy( if not self._started: self.start() - self._ensure_capacity_for_dedicated([(module_class, global_config, kwargs)]) - worker = self._select_worker(dedicated=module_class.dedicated_worker) - actor = worker.deploy_module(module_class, global_config, kwargs=kwargs) - return RPCClient(actor, module_class) + workers, launcher = self._pool_for(runtime_placement) + self._ensure_capacity_for_dedicated( + [(module_class, global_config, kwargs)], + workers, + launcher, + ) + worker = self._select_worker(workers, launcher, dedicated=module_class.dedicated_worker) + actor = worker.deploy_module( + module_class, + global_config, + kwargs=kwargs, + implementation_path=(runtime_placement.implementation if runtime_placement else None), + runtime_name=(runtime_placement.runtime if runtime_placement else None), + ) + if runtime_placement is not None: + self._runtime_placements_by_module[module_class] = runtime_placement + return cast("ModuleProxyProtocol", RPCClient(actor, module_class)) def deploy_fresh( self, module_class: type[ModuleBase], global_config: GlobalConfig, kwargs: dict[str, Any], + runtime_placement: RuntimePlacement | None = None, ) -> ModuleProxyProtocol: """Spawn a brand-new worker process and deploy *module_class* on it. @@ -104,16 +127,32 @@ def deploy_fresh( if not self._started: self.start() - worker = PythonWorker() + if runtime_placement is None: + runtime_placement = self._runtime_placements_by_module.get(module_class) + workers, launcher = self._pool_for(runtime_placement) + worker = PythonWorker(launcher) worker.start_process() - self._workers.append(worker) - self._n_workers += 1 + workers.append(worker) + if runtime_placement is None: + self._n_workers += 1 if module_class.dedicated_worker: worker.dedicated = True - actor = worker.deploy_module(module_class, global_config, kwargs=kwargs) - return RPCClient(actor, module_class) - - def undeploy(self, proxy: ModuleProxyProtocol) -> None: + actor = worker.deploy_module( + module_class, + global_config, + kwargs=kwargs, + implementation_path=(runtime_placement.implementation if runtime_placement else None), + runtime_name=(runtime_placement.runtime if runtime_placement else None), + ) + if runtime_placement is not None: + self._runtime_placements_by_module[module_class] = runtime_placement + return cast("ModuleProxyProtocol", RPCClient(actor, module_class)) + + def undeploy( + self, + proxy: ModuleProxyProtocol, + module_class: type[ModuleBase] | None = None, + ) -> None: """Undeploy a module and shut down its worker if it is now empty.""" actor = getattr(proxy, "actor_instance", None) if actor is None: @@ -121,9 +160,12 @@ def undeploy(self, proxy: ModuleProxyProtocol) -> None: module_id = actor._module_id target: PythonWorker | None = None - for worker in self._workers: - if module_id in worker._modules: - target = worker + for workers in self._all_worker_pools(): + for worker in workers: + if module_id in worker._modules: + target = worker + break + if target is not None: break if target is None: raise ValueError(f"No worker holds module_id={module_id}") @@ -132,69 +174,150 @@ def undeploy(self, proxy: ModuleProxyProtocol) -> None: if not target._modules: target.shutdown() - self._workers.remove(target) - self._n_workers = max(0, self._n_workers - 1) + removed_from_default_pool = self._remove_worker_from_pool(target) + if removed_from_default_pool: + self._n_workers = max(0, self._n_workers - 1) + if module_class is not None: + self._runtime_placements_by_module.pop(module_class, None) + + def runtime_placement_for(self, module_class: type[ModuleBase]) -> RuntimePlacement | None: + return self._runtime_placements_by_module.get(module_class) def deploy_parallel( - self, specs: Iterable[ModuleSpec], blueprint_args: Mapping[str, Mapping[str, Any]] + self, + specs: Iterable[ModuleSpec], + blueprint_args: Mapping[str, Mapping[str, Any]], + runtime_placements: Mapping[type[ModuleBase], RuntimePlacement] | None = None, ) -> list[ModuleProxyProtocol]: if self._closed: raise RuntimeError("WorkerManager is closed") specs = list(specs) + runtime_placements = runtime_placements or {} if len(specs) == 0: return [] if not self._started: self.start() - self._ensure_capacity_for_dedicated(specs) + created_runtime_pools: set[str] = set() - # Pre-assign workers sequentially (so least-loaded accounting is - # correct), then deploy concurrently via threads. The per-worker lock - # serializes deploys that land on the same worker process. - # Process dedicated specs first so they claim empty workers before - # non-dedicated specs land on them; preserve input order in output. workers_by_index: dict[int, PythonWorker] = {} - order = sorted(range(len(specs)), key=lambda i: not specs[i][0].dedicated_worker) - for i in order: - module_class, _, kwargs = specs[i] - worker = self._select_worker(dedicated=module_class.dedicated_worker) - worker.reserve_slot() - kwargs.update(blueprint_args.get(module_class.name, {})) - workers_by_index[i] = worker - - assignments = [(workers_by_index[i], specs[i]) for i in range(len(specs))] + assignments: list[tuple[PythonWorker, ModuleSpec]] = [] def _deploy(item: tuple[PythonWorker, ModuleSpec]) -> ModuleProxyProtocol: worker, (module_class, global_config, kwargs) = item - return RPCClient( - worker.deploy_module(module_class, global_config, kwargs), module_class + placement = runtime_placements.get(module_class) + return cast( + "ModuleProxyProtocol", + RPCClient( + worker.deploy_module( + module_class, + global_config, + kwargs, + implementation_path=(placement.implementation if placement else None), + runtime_name=(placement.runtime if placement else None), + ), + module_class, + ), ) + def _stop_created_runtime_pools() -> list[Exception]: + cleanup_errors: list[Exception] = [] + for runtime_name in created_runtime_pools: + try: + self._stop_runtime_pool(runtime_name) + except Exception as e: + cleanup_errors.append(e) + return cleanup_errors + + def _rollback( + _outcomes: list[ + tuple[tuple[PythonWorker, ModuleSpec], ModuleProxyProtocol | Exception] + ], + successes: list[ModuleProxyProtocol], + errors: list[Exception], + ) -> list[ModuleProxyProtocol]: + cleanup_errors: list[Exception] = [] + for proxy in successes: + try: + self.undeploy(proxy) + except Exception as e: + logger.error("Error rolling back deployed module", exc_info=True) + cleanup_errors.append(e) + cleanup_errors.extend(_stop_created_runtime_pools()) + if cleanup_errors: + raise ExceptionGroup( + "Python worker deployment failed and rollback cleanup also failed", + [*errors, *cleanup_errors], + ) + raise ExceptionGroup("Python worker deployment failed", errors) + try: - return safe_thread_map(assignments, _deploy) - except: - self.stop() + # Pre-assign workers sequentially (so least-loaded accounting is + # correct), then deploy concurrently via threads. The per-worker lock + # serializes deploys that land on the same worker process. + # Process dedicated specs first so they claim empty workers before + # non-dedicated specs land on them; preserve input order in output. + order = sorted(range(len(specs)), key=lambda i: not specs[i][0].dedicated_worker) + for i in order: + module_class, _, kwargs = specs[i] + placement = runtime_placements.get(module_class) + existing_runtime_names = set(self._runtime_workers) + workers, launcher = self._pool_for(placement) + if placement is not None and placement.runtime not in existing_runtime_names: + created_runtime_pools.add(placement.runtime) + self._ensure_capacity_for_dedicated( + [specs[i]], + workers, + launcher, + ) + worker = self._select_worker( + workers, launcher, dedicated=module_class.dedicated_worker + ) + worker.reserve_slot() + kwargs.update(blueprint_args.get(module_class.name, {})) + workers_by_index[i] = worker + + assignments = [(workers_by_index[i], specs[i]) for i in range(len(specs))] + deployed_modules = safe_thread_map(assignments, _deploy, on_errors=_rollback) + except Exception as error: + if not created_runtime_pools: + raise + cleanup_errors = _stop_created_runtime_pools() + if cleanup_errors: + raise ExceptionGroup( + "Python worker deployment failed and rollback cleanup also failed", + [error, *cleanup_errors], + ) from error raise + for (module_class, _, _), _proxy in zip(specs, deployed_modules, strict=True): + placement = runtime_placements.get(module_class) + if placement is not None: + self._runtime_placements_by_module[module_class] = placement + return deployed_modules def health_check(self) -> bool: - if len(self._workers) == 0: + workers = self.workers + if len(workers) == 0: logger.error("health_check: no workers found") return False - for w in self._workers: + for w in workers: if w.pid is None: logger.error("health_check: worker died", worker_id=w.worker_id) return False return True def suppress_console(self) -> None: - for worker in self._workers: + for worker in self.workers: worker.suppress_console() @property def workers(self) -> list[PythonWorker]: - return list(self._workers) + workers = list(self._workers) + for runtime_workers in self._runtime_workers.values(): + workers.extend(runtime_workers) + return workers def stop(self) -> None: if self._closed: @@ -207,46 +330,58 @@ def stop(self) -> None: logger.info("Shutting down all workers...") - for worker in reversed(self._workers): + for worker in reversed(self.workers): try: worker.shutdown() except Exception as e: logger.error(f"Error shutting down worker: {e}", exc_info=True) self._workers.clear() + self._runtime_workers.clear() + self._runtime_launchers.clear() logger.info("All workers shut down") - def _select_worker(self, dedicated: bool = False) -> PythonWorker: + def _select_worker( + self, + workers: list[PythonWorker], + launcher: WorkerLauncher | None, + dedicated: bool = False, + ) -> PythonWorker: """Pick a worker for a new module and mark it dedicated if needed.""" if dedicated: - for w in self._workers: + for w in workers: if not w.dedicated and w.module_count == 0: w.dedicated = True return w - self.add_workers(1) - w = self._workers[-1] + self._add_workers_to_pool(workers, launcher, 1) + w = workers[-1] w.dedicated = True return w - candidates = [w for w in self._workers if not w.dedicated] + candidates = [w for w in workers if not w.dedicated] if not candidates: - self.add_workers(1) - return self._workers[-1] + self._add_workers_to_pool(workers, launcher, 1) + return workers[-1] return min(candidates, key=lambda w: w.module_count) - def _ensure_capacity_for_dedicated(self, specs: Iterable[ModuleSpec]) -> None: + def _ensure_capacity_for_dedicated( + self, + specs: Iterable[ModuleSpec], + workers: list[PythonWorker], + launcher: WorkerLauncher | None, + ) -> None: """Grow the pool so non-dedicated workers >= dedicated workers. If the total number of dedicated modules (already deployed + about to be) exceeds half the worker pool, scale up to `2 * total_dedicated` workers. """ new_dedicated = sum(1 for spec in specs if spec[0].dedicated_worker) - already_dedicated = sum(1 for w in self._workers if w.dedicated) + already_dedicated = sum(1 for w in workers if w.dedicated) total_dedicated = already_dedicated + new_dedicated if total_dedicated == 0: return - total_workers = len(self._workers) + total_workers = len(workers) if total_dedicated * 2 > total_workers: n_to_add = total_dedicated * 2 - total_workers logger.warning( @@ -255,4 +390,55 @@ def _ensure_capacity_for_dedicated(self, specs: Iterable[ModuleSpec]) -> None: before=total_workers, added=n_to_add, ) - self.add_workers(n_to_add) + self._add_workers_to_pool(workers, launcher, n_to_add) + + def _pool_for( + self, + runtime_placement: RuntimePlacement | None, + ) -> tuple[list[PythonWorker], WorkerLauncher | None]: + if runtime_placement is None: + return self._workers, self._worker_launcher + runtime_name = runtime_placement.runtime + if runtime_name not in self._runtime_workers: + runtime = self._runtime_environment_registry.resolve(runtime_name) + if not isinstance(runtime, PythonProjectRuntimeEnvironment): + raise ValueError( + f"Runtime environment {runtime.name!r} must be a Python Runtime Project" + ) + launcher = CommandWorkerLauncher(runtime.resolve_python_project()) + self._runtime_launchers[runtime_name] = launcher + self._runtime_workers[runtime_name] = [] + if self._started: + self._add_workers_to_pool( + self._runtime_workers[runtime_name], + launcher, + max(1, self._n_workers), + ) + return self._runtime_workers[runtime_name], self._runtime_launchers[runtime_name] + + def _add_workers_to_pool( + self, + workers: list[PythonWorker], + launcher: WorkerLauncher | None, + n: int, + ) -> None: + for _ in range(n): + worker = PythonWorker(launcher) + worker.start_process() + workers.append(worker) + + def _all_worker_pools(self) -> list[list[PythonWorker]]: + return [self._workers, *self._runtime_workers.values()] + + def _remove_worker_from_pool(self, worker: PythonWorker) -> bool: + for workers in self._all_worker_pools(): + if worker in workers: + workers.remove(worker) + return workers is self._workers + return False + + def _stop_runtime_pool(self, runtime_name: str) -> None: + workers = self._runtime_workers.pop(runtime_name, []) + self._runtime_launchers.pop(runtime_name, None) + for worker in reversed(workers): + worker.shutdown() diff --git a/dimos/core/coordination/worker_messages.py b/dimos/core/coordination/worker_messages.py index 8fdb1cd0f2..06725b16c0 100644 --- a/dimos/core/coordination/worker_messages.py +++ b/dimos/core/coordination/worker_messages.py @@ -25,6 +25,8 @@ class DeployModuleRequest: module_id: int module_class: type[ModuleBase] kwargs: dict[str, Any] + implementation_path: str | None = None + runtime_name: str | None = None @dataclass(frozen=True) diff --git a/dimos/core/runtime_environment.py b/dimos/core/runtime_environment.py new file mode 100644 index 0000000000..540ea05881 --- /dev/null +++ b/dimos/core/runtime_environment.py @@ -0,0 +1,190 @@ +# Copyright 2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass, field +import os +from pathlib import Path + + +@dataclass(frozen=True) +class PythonProjectLaunchMaterial: + argv_prefix: tuple[str, ...] + cwd: Path + env: Mapping[str, str] = field(default_factory=dict) + runtime_name: str = "" + project: Path = Path() + has_pixi: bool = False + prepared_python: Path = Path() + + +@dataclass(frozen=True) +class RuntimePlacement: + runtime: str + implementation: str + + +class RuntimeEnvironmentError(RuntimeError): + pass + + +class RuntimeEnvironment: + name: str + + def resolve_python_project(self) -> PythonProjectLaunchMaterial: + raise RuntimeEnvironmentError( + f"Runtime environment {self.name!r} is not a Python Runtime Project" + ) + + @property + def project_path(self) -> Path | None: + return None + + +@dataclass(frozen=True) +class PythonProjectRuntimeEnvironment(RuntimeEnvironment): + name: str + project: Path | str + env: Mapping[str, str] = field(default_factory=dict) + + @property + def project_path(self) -> Path: + return Path(self.project).expanduser().resolve() + + @property + def pyproject_path(self) -> Path: + return self.project_path / "pyproject.toml" + + @property + def uv_lock_path(self) -> Path: + return self.project_path / "uv.lock" + + @property + def pixi_toml_path(self) -> Path: + return self.project_path / "pixi.toml" + + @property + def pixi_lock_path(self) -> Path: + return self.project_path / "pixi.lock" + + @property + def prepared_python(self) -> Path: + return self.project_path / ".venv" / "bin" / "python" + + @property + def has_pixi(self) -> bool: + return self.pixi_toml_path.exists() + + def validate_project_files(self) -> None: + if not self.pyproject_path.exists(): + raise RuntimeEnvironmentError( + f"Runtime project {self.name!r} is missing {self.pyproject_path}" + ) + if not self.uv_lock_path.exists(): + raise RuntimeEnvironmentError( + f"Runtime project {self.name!r} is missing committed lockfile {self.uv_lock_path}. " + "Run a manual package-manager lock/update command or future DimOS build/update " + "command; deployment reconciliation will not rewrite lockfiles." + ) + if self.pixi_toml_path.exists() and not self.pixi_lock_path.exists(): + raise RuntimeEnvironmentError( + f"Runtime project {self.name!r} is missing committed lockfile {self.pixi_lock_path}. " + "Run `pixi lock` or a future DimOS build/update command; deployment " + "reconciliation will not rewrite lockfiles." + ) + + def resolve_python_project(self) -> PythonProjectLaunchMaterial: + self.validate_project_files() + if not self.prepared_python.exists(): + raise RuntimeEnvironmentError( + f"Runtime project {self.name!r} at {self.project_path} is not prepared: " + f"missing {self.prepared_python}. Deployment reconciliation should create or update " + "runtime environment state without changing project files." + ) + env = dict(self.env) + env["PYTHONPATH"] = _merge_pythonpath(str(_dimos_import_path()), env.get("PYTHONPATH", "")) + return PythonProjectLaunchMaterial( + argv_prefix=(str(self.prepared_python),), + cwd=self.project_path, + env=env, + runtime_name=self.name, + project=self.project_path, + has_pixi=self.has_pixi, + prepared_python=self.prepared_python, + ) + + +@dataclass(frozen=True) +class RuntimeEnvironmentRegistry: + environments: Mapping[str, RuntimeEnvironment] = field(default_factory=dict) + + def register(self, *environments: RuntimeEnvironment) -> RuntimeEnvironmentRegistry: + merged = dict(self.environments) + for environment in environments: + if environment.name in merged: + raise RuntimeEnvironmentError( + f"Runtime environment {environment.name!r} is already registered" + ) + merged[environment.name] = environment + _validate_unique_project_paths(merged) + return RuntimeEnvironmentRegistry(merged) + + def merge(self, other: RuntimeEnvironmentRegistry) -> RuntimeEnvironmentRegistry: + merged = dict(self.environments) + for name, environment in other.environments.items(): + if name in merged: + if merged[name] != environment: + raise RuntimeEnvironmentError( + f"Runtime environment {name!r} is registered more than once" + ) + continue + merged[name] = environment + _validate_unique_project_paths(merged) + return RuntimeEnvironmentRegistry(merged) + + def resolve(self, name: str) -> RuntimeEnvironment: + try: + return self.environments[name] + except KeyError as e: + known_names = ", ".join(sorted(self.environments)) or "" + raise RuntimeEnvironmentError( + f"Unknown runtime environment {name!r}. Known runtimes: {known_names}" + ) from e + + +def _validate_unique_project_paths(environments: Mapping[str, RuntimeEnvironment]) -> None: + paths: dict[Path, str] = {} + for name, environment in environments.items(): + project_path = environment.project_path + if project_path is None: + continue + if project_path in paths: + other_name = paths[project_path] + raise RuntimeEnvironmentError( + f"Runtime environments {other_name!r} and {name!r} use duplicate " + f"Runtime Project path {project_path}" + ) + paths[project_path] = name + + +def _dimos_import_path() -> Path: + return Path(__file__).resolve().parents[2] + + +def _merge_pythonpath(*values: str) -> str: + paths: list[str] = [] + for value in values: + paths.extend(path for path in value.split(os.pathsep) if path) + return os.pathsep.join(dict.fromkeys(paths)) diff --git a/dimos/core/runtime_reconciliation.py b/dimos/core/runtime_reconciliation.py new file mode 100644 index 0000000000..09a99c1517 --- /dev/null +++ b/dimos/core/runtime_reconciliation.py @@ -0,0 +1,211 @@ +# Copyright 2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +from __future__ import annotations + +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass +import os +from pathlib import Path +import subprocess + +from dimos.core.coordination.blueprints import Blueprint +from dimos.core.module import ModuleBase +from dimos.core.runtime_environment import ( + PythonProjectRuntimeEnvironment, + RuntimePlacement, +) +from dimos.utils.safe_thread_map import safe_thread_map + + +class RuntimeReconciliationError(RuntimeError): + pass + + +@dataclass(frozen=True) +class RuntimeReconciliationCommand: + argv: tuple[str, ...] + cwd: Path + env: Mapping[str, str] + description: str + + +@dataclass(frozen=True) +class RuntimeReconciliationItem: + runtime_name: str + runtime: PythonProjectRuntimeEnvironment + commands: tuple[RuntimeReconciliationCommand, ...] + + +@dataclass(frozen=True) +class RuntimeReconciliationPlan: + items: tuple[RuntimeReconciliationItem, ...] + + +class SubprocessRuntimeCommandRunner: + def run(self, command: RuntimeReconciliationCommand) -> None: + subprocess.run( + command.argv, + cwd=command.cwd, + env={**os.environ, **command.env}, + check=True, + ) + + +def select_runtime_reconciliation_plan(blueprint: Blueprint) -> RuntimeReconciliationPlan: + active_modules = {bp.module for bp in blueprint.active_blueprints} + active_runtime_names = { + placement.runtime + for module, placement in blueprint.runtime_placement_map.items() + if module in active_modules + } + items: list[RuntimeReconciliationItem] = [] + for runtime_name in sorted(active_runtime_names): + runtime = blueprint.runtime_environment_registry.resolve(runtime_name) + if isinstance(runtime, PythonProjectRuntimeEnvironment): + runtime.validate_project_files() + items.append( + RuntimeReconciliationItem( + runtime_name=runtime_name, + runtime=runtime, + commands=tuple(_commands_for_project_runtime(runtime)), + ) + ) + return RuntimeReconciliationPlan(items=tuple(items)) + + +def reconcile_blueprint_runtimes( + blueprint: Blueprint, + *, + runner: SubprocessRuntimeCommandRunner | None = None, + output: Callable[[str], None] | None = None, +) -> RuntimeReconciliationPlan: + _validate_runtime_placements(blueprint) + plan = select_runtime_reconciliation_plan(blueprint) + reconcile_runtime_plan(plan, runner=runner, output=output) + return plan + + +def reconcile_runtime_plan( + plan: RuntimeReconciliationPlan, + *, + runner: SubprocessRuntimeCommandRunner | None = None, + output: Callable[[str], None] | None = None, +) -> None: + command_runner = runner or SubprocessRuntimeCommandRunner() + + def _run_item(item: RuntimeReconciliationItem) -> None: + for command in item.commands: + if output is not None: + output(f"[{item.runtime_name}] {command.description}: {' '.join(command.argv)}") + try: + command_runner.run(command) + except subprocess.CalledProcessError as e: + raise RuntimeReconciliationError( + f"Runtime reconciliation failed for {item.runtime_name!r} at " + f"{item.runtime.project_path}: {' '.join(command.argv)} exited {e.returncode}. " + "Lockfile mutation belongs to a manual package-manager command or " + "future DimOS build/update command." + ) from e + except FileNotFoundError as e: + raise RuntimeReconciliationError( + f"Runtime reconciliation failed for {item.runtime_name!r}: " + f"command {command.argv[0]!r} was not found" + ) from e + + def _raise_grouped( + outcomes: list[tuple[RuntimeReconciliationItem, None | Exception]], + _successes: list[None], + errors: list[Exception], + ) -> None: + lines = ["Runtime reconciliation failed before worker launch:"] + for item, result in outcomes: + if isinstance(result, Exception): + lines.append(f"- {item.runtime_name} ({item.runtime.project_path}): {result}") + raise RuntimeReconciliationError("\n".join(lines)) from errors[0] + + safe_thread_map(list(plan.items), _run_item, on_errors=_raise_grouped) + + +def active_runtime_placements( + blueprint: Blueprint, +) -> Mapping[type[ModuleBase], RuntimePlacement]: + active_modules = {bp.module for bp in blueprint.active_blueprints} + return { + module: placement + for module, placement in blueprint.runtime_placement_map.items() + if module in active_modules + } + + +def _validate_runtime_placements(blueprint: Blueprint) -> None: + blueprint_modules = {bp.module for bp in blueprint.blueprints} + active_modules = {bp.module for bp in blueprint.active_blueprints} + for module, placement in blueprint.runtime_placement_map.items(): + if module not in blueprint_modules: + raise RuntimeReconciliationError( + f"Runtime placement for {module.__name__} does not match a blueprint module" + ) + if module in active_modules: + blueprint.runtime_environment_registry.resolve(placement.runtime) + + +def _commands_for_project_runtime( + runtime: PythonProjectRuntimeEnvironment, +) -> Sequence[RuntimeReconciliationCommand]: + project = runtime.project_path + env = dict(runtime.env) + if runtime.has_pixi: + return ( + RuntimeReconciliationCommand( + argv=("pixi", "install", "--locked"), + cwd=project, + env=env, + description="reconcile Pixi environment in locked mode", + ), + RuntimeReconciliationCommand( + argv=( + "pixi", + "run", + "uv", + "venv", + "-p", + ".pixi/envs/default/bin/python", + "--seed", + "--allow-existing", + ), + cwd=project, + env=env, + description="ensure uv virtualenv from Pixi Python", + ), + RuntimeReconciliationCommand( + argv=("pixi", "run", "uv", "sync", "--locked"), + cwd=project, + env=env, + description="sync uv project in locked mode", + ), + ) + return ( + RuntimeReconciliationCommand( + argv=("uv", "venv", "--seed", "--allow-existing"), + cwd=project, + env=env, + description="ensure uv virtualenv", + ), + RuntimeReconciliationCommand( + argv=("uv", "sync", "--locked"), + cwd=project, + env=env, + description="sync uv project in locked mode", + ), + ) diff --git a/dimos/core/test_runtime_environment.py b/dimos/core/test_runtime_environment.py new file mode 100644 index 0000000000..0c24b8f4e4 --- /dev/null +++ b/dimos/core/test_runtime_environment.py @@ -0,0 +1,112 @@ +# Copyright 2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +from __future__ import annotations + +import os +from pathlib import Path + +import pytest + +from dimos.core.runtime_environment import ( + PythonProjectRuntimeEnvironment, + RuntimeEnvironmentError, + RuntimeEnvironmentRegistry, +) + + +def test_register_and_merge_reject_duplicate_names() -> None: + registry = RuntimeEnvironmentRegistry() + runtime = PythonProjectRuntimeEnvironment(name="project", project="/tmp/project") + + registry = registry.register(runtime) + + conflicting_current = RuntimeEnvironmentRegistry( + {"project": PythonProjectRuntimeEnvironment(name="project", project="/tmp/other")} + ) + with pytest.raises(RuntimeEnvironmentError, match="registered more than once"): + registry.merge(conflicting_current) + + assert registry.merge(RuntimeEnvironmentRegistry({"project": runtime})) == registry + + +def test_duplicate_python_project_runtime_environment_project_path_rejected(tmp_path: Path) -> None: + first = PythonProjectRuntimeEnvironment(name="first", project=tmp_path) + second = PythonProjectRuntimeEnvironment(name="second", project=tmp_path / ".") + + with pytest.raises(RuntimeEnvironmentError, match="duplicate Runtime Project path"): + RuntimeEnvironmentRegistry().register(first, second) + + +def test_unknown_runtime_error_lists_known_runtime_names() -> None: + registry = RuntimeEnvironmentRegistry() + + with pytest.raises(RuntimeEnvironmentError, match="Known runtimes: "): + registry.resolve("missing") + + +def test_python_project_requires_pyproject_and_uv_lock(tmp_path: Path) -> None: + runtime = PythonProjectRuntimeEnvironment(name="project", project=tmp_path) + + with pytest.raises(RuntimeEnvironmentError, match="pyproject.toml"): + runtime.validate_project_files() + + runtime.pyproject_path.write_text("[project]\nname = 'example'\n", encoding="utf-8") + + with pytest.raises(RuntimeEnvironmentError, match="uv.lock"): + runtime.validate_project_files() + + +def test_pixi_toml_requires_pixi_lock(tmp_path: Path) -> None: + runtime = PythonProjectRuntimeEnvironment(name="project", project=tmp_path) + runtime.pyproject_path.write_text("[project]\nname = 'example'\n", encoding="utf-8") + runtime.uv_lock_path.write_text("version = 1\n", encoding="utf-8") + runtime.pixi_toml_path.write_text("[workspace]\n", encoding="utf-8") + + with pytest.raises(RuntimeEnvironmentError, match="pixi.lock"): + runtime.validate_project_files() + + +def test_resolve_python_project_requires_prepared_python_after_validation(tmp_path: Path) -> None: + runtime = PythonProjectRuntimeEnvironment(name="project", project=tmp_path) + runtime.pyproject_path.write_text("[project]\nname = 'example'\n", encoding="utf-8") + runtime.uv_lock_path.write_text("version = 1\n", encoding="utf-8") + + with pytest.raises(RuntimeEnvironmentError, match=r"\.venv/bin/python"): + runtime.resolve_python_project() + + runtime.prepared_python.parent.mkdir(parents=True) + runtime.prepared_python.write_text("#!/usr/bin/env python\n", encoding="utf-8") + + material = runtime.resolve_python_project() + assert material.argv_prefix == (str(runtime.prepared_python),) + assert material.has_pixi is False + assert material.prepared_python == runtime.prepared_python + assert str(Path(__file__).resolve().parents[2]) in material.env["PYTHONPATH"].split(os.pathsep) + + +def test_resolve_python_project_preserves_runtime_pythonpath(tmp_path: Path) -> None: + runtime = PythonProjectRuntimeEnvironment( + name="project", + project=tmp_path, + env={"PYTHONPATH": "/runtime/src"}, + ) + runtime.pyproject_path.write_text("[project]\nname = 'example'\n", encoding="utf-8") + runtime.uv_lock_path.write_text("version = 1\n", encoding="utf-8") + runtime.prepared_python.parent.mkdir(parents=True) + runtime.prepared_python.write_text("#!/usr/bin/env python\n", encoding="utf-8") + + entries = runtime.resolve_python_project().env["PYTHONPATH"].split(os.pathsep) + + assert entries[0] == str(Path(__file__).resolve().parents[2]) + assert "/runtime/src" in entries diff --git a/dimos/core/test_runtime_reconciliation.py b/dimos/core/test_runtime_reconciliation.py new file mode 100644 index 0000000000..aa851bc682 --- /dev/null +++ b/dimos/core/test_runtime_reconciliation.py @@ -0,0 +1,265 @@ +# Copyright 2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +from __future__ import annotations + +from dataclasses import replace +import os +from pathlib import Path +import subprocess +from types import MappingProxyType + +import pytest + +from dimos.core.coordination.blueprints import autoconnect +from dimos.core.module import ModuleBase +from dimos.core.runtime_environment import ( + PythonProjectRuntimeEnvironment, + RuntimeEnvironmentError, + RuntimePlacement, +) +from dimos.core.runtime_reconciliation import ( + RuntimeReconciliationCommand, + RuntimeReconciliationError, + RuntimeReconciliationItem, + RuntimeReconciliationPlan, + SubprocessRuntimeCommandRunner, + active_runtime_placements, + reconcile_blueprint_runtimes, + reconcile_runtime_plan, + select_runtime_reconciliation_plan, +) + + +class RuntimeModule(ModuleBase): + pass + + +class DisabledRuntimeModule(ModuleBase): + pass + + +class UnrelatedRuntimeModule(ModuleBase): + pass + + +class FakeRunner(SubprocessRuntimeCommandRunner): + def __init__(self, failing_command: tuple[str, ...] | None = None) -> None: + self.commands: list[RuntimeReconciliationCommand] = [] + self.failing_command = failing_command + + def run(self, command: RuntimeReconciliationCommand) -> None: + self.commands.append(command) + if command.argv == self.failing_command: + raise subprocess.CalledProcessError(returncode=17, cmd=command.argv) + + +def _write_project_files(tmp_path: Path, *, pixi: bool = False) -> PythonProjectRuntimeEnvironment: + (tmp_path / "pyproject.toml").write_text("[project]\nname = 'example'\n", encoding="utf-8") + (tmp_path / "uv.lock").write_text("version = 1\n", encoding="utf-8") + if pixi: + (tmp_path / "pixi.toml").write_text("[workspace]\n", encoding="utf-8") + (tmp_path / "pixi.lock").write_text("version = 1\n", encoding="utf-8") + return PythonProjectRuntimeEnvironment(name="project", project=tmp_path, env={"A": "B"}) + + +def test_runtime_placement_validation_rejects_unknown_runtime(tmp_path: Path) -> None: + runtime = _write_project_files(tmp_path) + blueprint = RuntimeModule.blueprint().runtime_environments(runtime) + + with pytest.raises(RuntimeEnvironmentError, match="Unknown runtime environment 'missing'"): + reconcile_blueprint_runtimes( + replace( + blueprint, + runtime_placement_map=MappingProxyType( + {RuntimeModule: RuntimePlacement(runtime="missing", implementation="impl.py")} + ), + ), + runner=FakeRunner(), + ) + + +def test_runtime_placement_validation_rejects_non_blueprint_module(tmp_path: Path) -> None: + runtime = _write_project_files(tmp_path) + blueprint = RuntimeModule.blueprint().runtime_environments(runtime) + + with pytest.raises(RuntimeReconciliationError, match="does not match a blueprint module"): + reconcile_blueprint_runtimes( + replace( + blueprint, + runtime_placement_map=MappingProxyType( + { + UnrelatedRuntimeModule: RuntimePlacement( + runtime="project", implementation="impl.py" + ) + } + ), + ), + runner=FakeRunner(), + ) + + +def test_active_placements_exclude_disabled_modules(tmp_path: Path) -> None: + runtime = _write_project_files(tmp_path) + blueprint = autoconnect(RuntimeModule.blueprint(), DisabledRuntimeModule.blueprint()) + blueprint = blueprint.runtime_environments(runtime).runtime_placements( + { + RuntimeModule: RuntimePlacement(runtime="project", implementation="enabled.py"), + DisabledRuntimeModule: RuntimePlacement( + runtime="project", implementation="disabled.py" + ), + } + ) + blueprint = blueprint.disabled_modules(DisabledRuntimeModule) + + assert active_runtime_placements(blueprint) == { + RuntimeModule: RuntimePlacement(runtime="project", implementation="enabled.py") + } + assert [item.runtime_name for item in select_runtime_reconciliation_plan(blueprint).items] == [ + "project" + ] + + +def test_disabled_runtime_placement_does_not_require_runtime_registration(tmp_path: Path) -> None: + runtime = _write_project_files(tmp_path) + blueprint = autoconnect(RuntimeModule.blueprint(), DisabledRuntimeModule.blueprint()) + blueprint = blueprint.runtime_environments(runtime).runtime_placements( + { + RuntimeModule: RuntimePlacement(runtime="project", implementation="enabled.py"), + DisabledRuntimeModule: RuntimePlacement( + runtime="missing", implementation="disabled.py" + ), + } + ) + blueprint = blueprint.disabled_modules(DisabledRuntimeModule) + + plan = reconcile_blueprint_runtimes(blueprint, runner=FakeRunner()) + + assert [item.runtime_name for item in plan.items] == ["project"] + + +def test_reconciliation_command_selection_for_uv(tmp_path: Path) -> None: + runtime = _write_project_files(tmp_path) + blueprint = ( + RuntimeModule.blueprint() + .runtime_environments(runtime) + .runtime_placements( + {RuntimeModule: RuntimePlacement(runtime="project", implementation="impl.py")} + ) + ) + + plan = select_runtime_reconciliation_plan(blueprint) + + assert [command.argv for command in plan.items[0].commands] == [ + ("uv", "venv", "--seed", "--allow-existing"), + ("uv", "sync", "--locked"), + ] + assert all(command.cwd == tmp_path for command in plan.items[0].commands) + + +def test_subprocess_runner_merges_command_env_with_parent_env( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + captured_env: dict[str, str] = {} + + def fake_run( + argv: tuple[str, ...], + *, + cwd: Path, + env: dict[str, str], + check: bool, + ) -> None: + assert argv == ("uv", "sync", "--locked") + assert cwd == tmp_path + assert check is True + captured_env.update(env) + + monkeypatch.setenv("DIMOS_PARENT_ENV", "parent") + monkeypatch.setattr(subprocess, "run", fake_run) + + SubprocessRuntimeCommandRunner().run( + RuntimeReconciliationCommand( + argv=("uv", "sync", "--locked"), + cwd=tmp_path, + env={"DIMOS_COMMAND_ENV": "command"}, + description="sync uv project in locked mode", + ) + ) + + assert captured_env["DIMOS_PARENT_ENV"] == "parent" + assert captured_env["DIMOS_COMMAND_ENV"] == "command" + assert captured_env["PATH"] == os.environ["PATH"] + + +def test_reconciliation_command_selection_for_pixi_backed_uv(tmp_path: Path) -> None: + runtime = _write_project_files(tmp_path, pixi=True) + blueprint = ( + RuntimeModule.blueprint() + .runtime_environments(runtime) + .runtime_placements( + {RuntimeModule: RuntimePlacement(runtime="project", implementation="impl.py")} + ) + ) + + plan = select_runtime_reconciliation_plan(blueprint) + + assert [command.argv for command in plan.items[0].commands] == [ + ("pixi", "install", "--locked"), + ( + "pixi", + "run", + "uv", + "venv", + "-p", + ".pixi/envs/default/bin/python", + "--seed", + "--allow-existing", + ), + ("pixi", "run", "uv", "sync", "--locked"), + ] + + +def test_fake_runner_records_commands_and_grouped_failure_surfaces_before_worker_launch( + tmp_path: Path, +) -> None: + runtime = _write_project_files(tmp_path) + commands = ( + RuntimeReconciliationCommand( + argv=("uv", "venv", "--seed"), + cwd=tmp_path, + env={}, + description="ensure uv virtualenv", + ), + RuntimeReconciliationCommand( + argv=("uv", "sync", "--locked"), + cwd=tmp_path, + env={}, + description="sync uv project in locked mode", + ), + ) + plan = RuntimeReconciliationPlan( + items=( + RuntimeReconciliationItem(runtime_name="project", runtime=runtime, commands=commands), + ) + ) + runner = FakeRunner(failing_command=("uv", "sync", "--locked")) + + with pytest.raises(RuntimeReconciliationError) as error: + reconcile_runtime_plan(plan, runner=runner) + + assert [command.argv for command in runner.commands] == [command.argv for command in commands] + message = str(error.value) + assert "Runtime reconciliation failed before worker launch" in message + assert "project" in message + assert "exited 17" in message diff --git a/dimos/core/test_venv_module_demo.py b/dimos/core/test_venv_module_demo.py new file mode 100644 index 0000000000..df49d608f2 --- /dev/null +++ b/dimos/core/test_venv_module_demo.py @@ -0,0 +1,123 @@ +# Copyright 2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from __future__ import annotations + +import importlib +import os +from pathlib import Path +import subprocess +import sys + +import pytest + +from dimos.core.coordination.worker_manager_python import WorkerManagerPython +from dimos.core.global_config import GlobalConfig, global_config +from dimos.core.runtime_environment import ( + PythonProjectRuntimeEnvironment, + RuntimeEnvironmentRegistry, + RuntimePlacement, +) + +REPO_ROOT = Path(__file__).resolve().parents[2] +EXAMPLE_ROOT = REPO_ROOT / "examples" / "dimos-demo-worker-module" +EXAMPLE_SRC = EXAMPLE_ROOT / "src" + + +def _example_pythonpath() -> str: + paths = [str(EXAMPLE_SRC)] + paths.extend( + path for path in sys.path if path and ("site-packages" in path or "dist-packages" in path) + ) + if existing := os.environ.get("PYTHONPATH"): + paths.append(existing) + return os.pathsep.join(dict.fromkeys(paths)) + + +@pytest.fixture +def demo_runtime_worker(monkeypatch): + monkeypatch.syspath_prepend(str(EXAMPLE_SRC)) + contract_module = importlib.import_module("dimos_demo_worker_module.contract") + demo_contract = contract_module.DemoWorkerModule + runtime = PythonProjectRuntimeEnvironment( + name="demo-worker-test-runtime", + project=EXAMPLE_ROOT, + env={"PYTHONPATH": _example_pythonpath(), "UV_PYTHON": sys.executable}, + ) + subprocess.run( + ("uv", "sync", "--locked"), + cwd=EXAMPLE_ROOT, + env={**os.environ, **runtime.env}, + check=True, + capture_output=True, + text=True, + timeout=180, + ) + placement = RuntimePlacement( + runtime=runtime.name, + implementation="dimos_demo_worker_module.runtime.DemoWorkerRuntimeModule", + ) + manager = WorkerManagerPython( + g=GlobalConfig(n_workers=1, viewer="none"), + ) + manager.register_runtime_environments(RuntimeEnvironmentRegistry().register(runtime)) + manager.start() + module = manager.deploy( + demo_contract, + global_config, + {}, + runtime_placement=placement, + ) + try: + yield module + finally: + module.stop() + manager.stop() + + +def test_demo_runtime_project_contract_import_does_not_import_runtime(monkeypatch) -> None: + monkeypatch.syspath_prepend(str(EXAMPLE_SRC)) + monkeypatch.delitem(sys.modules, "dimos_demo_worker_module.blueprint", raising=False) + monkeypatch.delitem(sys.modules, "dimos_demo_worker_module.contract", raising=False) + monkeypatch.delitem(sys.modules, "dimos_demo_worker_module.runtime", raising=False) + + blueprint_module = importlib.import_module("dimos_demo_worker_module.blueprint") + + assert "dimos_demo_worker_module.contract" in sys.modules + assert "dimos_demo_worker_module.runtime" not in sys.modules + placement = blueprint_module.demo_worker_runtime_blueprint.runtime_placement_map[ + blueprint_module.DemoWorkerModule + ] + assert placement.implementation == "dimos_demo_worker_module.runtime.DemoWorkerRuntimeModule" + + +def test_example_pythonpath_excludes_host_stdlib(monkeypatch) -> None: + site_packages = "/host/.venv/lib/python3.12/site-packages" + stdlib = "/usr/lib/python3.12" + monkeypatch.setattr(sys, "path", [stdlib, site_packages, ""]) + monkeypatch.delenv("PYTHONPATH", raising=False) + + entries = _example_pythonpath().split(os.pathsep) + + assert str(EXAMPLE_SRC) in entries + assert site_packages in entries + assert stdlib not in entries + + +@pytest.mark.skipif_macos_bug +def test_demo_runtime_project_executes_with_project_worker(demo_runtime_worker) -> None: + runtime_python = demo_runtime_worker.runtime_python() + runtime_dependency_label = demo_runtime_worker.runtime_dependency_label + assert str(EXAMPLE_ROOT / ".venv") in runtime_python + assert runtime_dependency_label() == "RuntimeDependency" diff --git a/docs/usage/index.md b/docs/usage/index.md index 6930faf96a..363f6f817d 100644 --- a/docs/usage/index.md +++ b/docs/usage/index.md @@ -9,6 +9,7 @@ This page explains general concepts. - [Modules](/docs/usage/modules.md): The primary units of deployment in DimOS, modules run in parallel and are python classes. - [Streams](/docs/usage/sensor_streams/index.md): How modules communicate, a Pub / Sub system. - [Blueprints](/docs/usage/blueprints.md): a way to group modules together and define their connections to each other. +- [Runtime Environments](/docs/usage/runtime_environments.md): place Python modules in registered Runtime Projects with locked worker dependencies. - [RPC](/docs/usage/blueprints.md#calling-the-methods-of-other-modules): how one module can call a method on another module (arguments get serialized to JSON-like binary data). - [Skills](/docs/usage/blueprints.md#defining-skills): An RPC function, except it can be called by an AI agent (a tool for an AI). - Agents: AI that has an objective, access to stream data, and is capable of calling skills as tools. diff --git a/docs/usage/module_deployment_proposal.md b/docs/usage/module_deployment_proposal.md new file mode 100644 index 0000000000..c10709a51f --- /dev/null +++ b/docs/usage/module_deployment_proposal.md @@ -0,0 +1,775 @@ +# Proposal: Module Deployment for DimOS + +Status: draft for review. + +This proposal defines one deployment model for normal Python modules, packaged Python modules, native modules, and remote execution. The coordinator retains a stable module contract, while deployment decides where to prepare and run its implementation. + +## 1. Problem / Why now + +DimOS has several deployment pressures that currently look separate: + +- Python modules sometimes need heavy or conflicting dependencies that should not live in the coordinator environment. +- Native modules need repeatable build and runtime preparation. +- Remote deployment needs code or artifact sync, target preparation, process launch, logs, health, and cleanup. +- Weak robot computers may need prepared artifacts, cross-compilation, or runtime closures built elsewhere. +- Native and packaged Python modules need a shared way to describe config, stream topics, transports, and lifecycle handoff. + +The common problem is **module deployment**. + +The current local Python path works well for in-environment Python modules. But once a module has its own runtime requirement, DimOS needs an explicit deployment layer that can prepare the requirement, launch the implementation, and keep the Blueprint-facing module identity stable. + +## 2. Current state + +### Normal Python modules + +Normal Python modules run inside the current DimOS Python worker environment. + +```text +ModuleCoordinator + -> WorkerManagerPython + -> PythonWorker + -> Python Module instance +``` + +They get the full DimOS surface: streams, RPCs, skills, module refs, lifecycle, and Blueprint wiring. This path should remain the default for local, in-environment Python modules. + +### Current NativeModule + +Today, a native module is a Python `NativeModule` wrapper deployed through the Python worker. The wrapper declares the DimOS-facing streams and config, then spawns an external executable. + +```text +PythonWorker + -> NativeModule wrapper + -> native subprocess +``` + +The wrapper owns Blueprint integration, lifecycle, topic assignment, config serialization, logs, and process supervision. The native subprocess owns computation and direct pub/sub. + +`NativeModuleConfig` already carries a proto launch recipe: + +- `cwd` +- `executable` +- `build_command` +- `extra_args` +- `extra_env` +- `stdin_config` +- `auto_build` + +When `stdin_config=True`, the wrapper sends a JSON payload to the native process: + +```json +{ + "topics": {"input": "/topic#Type", "output": "/topic#Type"}, + "config": {"field": "value"} +} +``` + +That JSON is a useful starting point for a future **Module Launch Envelope**. + +### Recent packaged Python exploration + +Recent work on isolated Python runtime modules, including PR #2704, demonstrates one backend: a Python module can keep a dependency-light contract while its implementation runs in a prepared Python project. + +It adds: + +- runtime environment registration, +- class-keyed runtime placement, +- deployment-time runtime reconciliation, +- runtime-specific Python worker pools, +- launch through the prepared `.venv/bin/python`, +- a runnable example package. + +The worker-pool design is less important than the boundary it exposes. The coordinator can import a lightweight contract without importing the implementation's dependencies. + +### Native modules expose the same boundary + +Recent native work serializes config and topics, adds native transports, and places Python wrappers beside buildable native packages. These changes expose the same boundary as packaged Python: a coordinator-visible contract backed by code that runs in another process. This proposal gives both paths one deployment model. + +## 3. Proposed model + +The proposal extends the existing manager/worker split instead of introducing a separate deployment stack. An `ExternalModule` is the lightweight contract imported by the coordinator. A `DeploymentSpec` combines a Blueprint with per-module deployment policy. A `TargetSession` provides local or SSH access during preparation. After preparation, one target-side `ExternalWorker` supervises each external implementation: native modules run as direct subprocesses, while packaged Python modules run through a thin Python entrypoint. + +```mermaid +flowchart TD + Coordinator[ModuleCoordinator] + + Coordinator --> PythonManager[WorkerManagerPython\none per coordinator] + PythonManager --> PythonWorkerA[PythonWorker] + PythonManager --> PythonWorkerB[PythonWorker] + PythonWorkerA --> ModuleA[normal Python Module] + PythonWorkerA --> ModuleB[normal Python Module] + PythonWorkerB --> ModuleC[normal Python Module] + + Coordinator --> ExternalManager[WorkerManagerExternal\none per coordinator] + ExternalManager --> LocalWorker[ExternalWorker: local machine] + ExternalManager --> RobotWorker[ExternalWorker: robot machine] + ExternalManager --> GpuWorker[ExternalWorker: GPU machine] + + LocalWorker --> LocalProcess[native subprocess] + RobotWorker --> RobotProcess[native subprocess] + RobotWorker --> RobotEntrypoint[Python entrypoint] + GpuWorker --> GpuEntrypoint[Python entrypoint] +``` + +Section 5 defines each manager and worker in operational terms. This diagram establishes only the ownership split: normal modules remain in the Python worker pool, while external implementations run under one machine-level `ExternalWorker` per deployment run. + +### 3.1 Module contract shape + +`ExternalModule` is a declarative `Module` subclass. It adds an implementation reference but no build, process, watchdog, or transport behavior. It's similar to current `NativeModule` but more "contract only". + +Packaged Python uses a class import reference: + +```python +# Contract package: safe to import in the coordinator and control environment. +class HeavyDetector(ExternalModule): + implementation = "heavy_detector.module:HeavyDetectorImpl" + + config: HeavyDetectorConfig + image: In[Image] + detections: Out[Detections] +``` + +```python +# python/src/heavy_detector/module.py +import heavy_dependency + + +class HeavyDetectorImpl(HeavyDetector): + def start(self) -> None: + ... +``` + +The thin Python entrypoint imports the class and requires: + +```python +issubclass(HeavyDetectorImpl, HeavyDetector) +``` + +Native execution uses an executable path relative to its convention-discovered implementation folder: + +```python +from pathlib import Path as FsPath + + +class MLSPlanner(ExternalModule): + implementation = FsPath("target/release/mls_planner") + + config: MLSPlannerConfig + global_map: In[PointCloud2] + goal_pose: In[PoseStamped] + path: Out[Path] +``` + +`ExternalModule.implementation` accepts `str | pathlib.Path` for ergonomic declarations. DimOS first discovers the sibling implementation folder described in section 4. A `python/` folder requires a class import reference, while `rust/` and `cpp/` require an executable path. The Python value type never selects the runtime. + +The class location anchors package discovery. `ExternalModule` should not be sent to `PythonWorker`; it requires a Deployment Spec and the external worker path. + +### 3.2 Complete Deployment Spec + +A Deployment Spec references a Blueprint, defines reusable targets, and groups each module's deployment choices in one `ModuleDeployment`: + +```python +go2_deployment = DeploymentSpec( + blueprint=go2_stack, + targets={ + "robot": SshTarget( + host="go2", + deployment_root="~/dimos-deployments/go2", + ), + "gpu": SshTarget( + host="gpu-box", + deployment_root="~/dimos-deployments/go2", + ), + }, + modules={ + MLSPlanner: ModuleDeployment( + execution_target="robot", + build_target="local", + preparation=ArmCargoPreparation(), + ), + HeavyDetector: ModuleDeployment( + execution_target="gpu", + ), + }, +) +``` + +The Blueprint remains responsible for active modules, stream wiring, module references, and configuration. The `targets` map names machines, while `modules` groups each module class's deployment policy. A module omitted from `modules` executes locally. An omitted build target resolves to the execution target, and omitted preparation or runtime-environment policy comes from the discovered convention. + +The implicit `local` target uses a `GlobalConfig`-derived deployment root under DimOS state and cannot be redefined. In-environment Python modules remain local because their existing object and RPC protocol depends on `PythonWorker`; remote execution requires an `ExternalModule` contract. Target probing rejects aliases that identify the same machine so the coordinator cannot start competing workers for one run. + +A class-keyed `ModuleDeployment` applies to every active Blueprint instance of that class. Resolved plans and launch envelopes use unique instance IDs, which prevents two instances from colliding at the worker-control boundary. Each target's `deployment_root` contains DimOS-managed control environments, source snapshots, artifacts, runtime environments, run state, logs, caches, and locks. + +### 3.3 Illustrative public model + +The following shapes describe the intended extension boundary rather than a committed compatibility contract. Resolved planning and worker protocol types remain internal so their representation can change without expanding the public API. + +```python +class ExternalModule(Module): + implementation: ClassVar[str | FsPath] +``` + +```python +@dataclass(frozen=True) +class DeploymentSpec: + blueprint: Blueprint + targets: Mapping[str, ExecutionTarget] + modules: Mapping[type[ModuleBase], ModuleDeployment] +``` + +```python +@dataclass(frozen=True) +class ModuleDeployment: + execution_target: str = "local" + build_target: str | None = None + preparation: Preparation | None = None + runtime_environment: RuntimeEnvironmentSpec | None = None +``` + +```python +@dataclass(frozen=True) +class RuntimeEnvironmentSpec: + implementation: str + config: JsonObject = field(default_factory=dict) +``` + +```python +@dataclass(frozen=True) +class SshTarget(ExecutionTarget): + host: str + deployment_root: PurePosixPath + expected_platform: Platform | None = None +``` + +`Preparation` stages source and produces deployable artifacts before `ExternalWorker` starts. `RuntimeEnvironment` turns those staged inputs into a runnable environment on the execution target: + +```python +class Preparation(ABC): + async def prepare(self, context: PreparationContext) -> None: ... + async def cleanup(self, context: PreparationContext) -> None: ... +``` + +```python +class RuntimeEnvironment(ABC): + async def setup(self, context: RuntimeEnvironmentContext) -> RuntimeLaunch: ... + async def teardown(self, context: RuntimeEnvironmentContext) -> None: ... +``` + +Convention Presets supply both objects for standard layouts. A `ModuleDeployment` overrides either one only when a deployment needs exceptional behavior. The coordinator invokes `Preparation`, which orchestrates commands and transfers through local or SSH target sessions; the commands themselves run on the selected build and execution targets. `RuntimeEnvironment` runs on the target, so its override uses a serializable `RuntimeEnvironmentSpec`: a top-level class import reference plus JSON-compatible configuration. The plan never sends a live Python object over SSH. + +Resolved plan, path, launch, environment-reference, and worker-route types remain internal. + +The distinction is operational: + +```text +DeploymentSpec user-authored deployment intent +ModuleDeployment grouped policy for one module +ExternalModule module-owned implementation declaration +DeploymentPlan validated and fully resolved actions +``` + +### 3.4 Planning, prepare, and deploy + +Deployment follows one ordered lifecycle: + +```text +resolve modules, targets, and conventions +probe build and execution targets +prepare source and artifacts through target sessions +transfer content-addressed snapshots and outputs +bootstrap one ExternalWorker per execution machine +materialize runtime environments inside ExternalWorker +start one runtime handle per ExternalModule +wait for ready acknowledgements +``` + +`TargetSession` gives the coordinator access to one machine. A local session executes commands and copies files directly. An SSH session executes remote commands, transfers files, bootstraps `ExternalWorker`, and tunnels its control RPC. One `Preparation` can use both the build and execution sessions. For example, it can cross-compile on a developer workstation and then copy the executable to a robot without splitting the workflow into unrelated per-machine hooks. + +Target probing records one platform identity for worker bootstrap, compilation, and artifact validation. `expected_platform`, when supplied, checks the detected result; it does not define a competing target platform. + +The coordinator copies each source snapshot to a content-addressed directory beneath `deployment_root` and publishes it with an atomic rename. A snapshot contains the lightweight contract, custom deployment extensions, and implementation source. `ExternalWorker` imports a custom `RuntimeEnvironment` from that snapshot before installing module dependencies, so the extension itself must remain dependency-light. + +`ExternalWorker` uses a separate, versioned control environment beneath `deployment_root/control/`. `TargetSession` transfers a pinned environment tool and provisions Python plus the control dependencies there. Remote deployment therefore does not depend on target-global Python or uv installations. + +`WorkerManagerExternal` coordinates this lifecycle and rolls back completed work if another module fails. Section 5 describes its interaction with `TargetSession`, `ExternalWorkerClient`, and `ExternalWorker` without repeating those responsibilities here. + +`DeploymentPlan` is immutable data consumed by the manager/worker path; it is not a separate behavioral reconciler layer. + +Planning validates module declarations, target references, convention resolution, and required manifests before mutation. Preparation and runtime-environment setup validate generated artifacts before launch. A native executable may be a preparation output, so planning validates its destination rather than requiring it to exist in advance. + +### 3.5 Resolved plan + +The plan shows which modules require staging, where each preparation runs, and which worker path launches the implementation: + +```text +Module Build Execute Preparation Environment Worker route +Agent local local — — WorkerManagerPython -> PythonWorker +MLSPlanner local robot Cargo cross Native WorkerManagerExternal -> ExternalWorker -> native process +HeavyDetector gpu gpu source sync uv WorkerManagerExternal -> ExternalWorker -> Python entrypoint +``` + +All deployment commands use this plan schema. Section 7 defines when the plan is previewed, persisted, and validated. + +### 3.6 Module Launch Envelope + +After preparation and connection resolution, the per-module runtime handle receives one unified envelope. For packaged Python, the thin entrypoint reads the envelope before importing the implementation. For native execution, `ExternalWorker` converts the envelope into argv, environment, stdin JSON, and control metadata for the subprocess. + +```python +@dataclass(frozen=True) +class ModuleLaunchEnvelope: + module_id: str + runtime: RuntimeLaunch + config: ModuleConfigPayload + topics: Mapping[str, TopicBinding] + control: ControlEndpoint +``` + +```json +{ + "module_id": "mls_planner-1", + "runtime": { + "executable": "target/release/mls_planner" + }, + "topics": { + "global_map": { + "channel": "/global_map", + "type": "sensor_msgs.PointCloud2", + "transport": "zenoh" + } + }, + "config": { + "world_frame": "map", + "voxel_size": 0.1 + }, + "control": { + "endpoint": "..." + } +} +``` + +This extends the current `NativeModule.stdin_config` payload. DimOS may track field provenance internally, but each implementation receives one handoff containing launch metadata, module config, stream bindings, transport descriptors, and control details. + +### 3.7 Process topology + +The plan routes normal modules to `WorkerManagerPython` and external modules to `WorkerManagerExternal`. The two managers share the coordinator-facing deployment contract, but the worker implementations stay separate because they cross different process and machine boundaries. + +```mermaid +flowchart LR + subgraph CoordinatorMachine[Coordinator machine] + subgraph CoordinatorProcess[coordinator process] + Coordinator[ModuleCoordinator] + Backend[WorkerManagerBackend contract] + PythonManager[WorkerManagerPython] + ExternalManager[WorkerManagerExternal] + Session[TargetSession\nlocal or SSH access] + Client[ExternalWorkerClient] + Envelope[ModuleLaunchEnvelope] + + Coordinator --> Backend + Backend --> PythonManager + Backend --> ExternalManager + ExternalManager --> Session + ExternalManager --> Client + ExternalManager --> Envelope + end + + subgraph PythonWorkerProcess[PythonWorker process] + PythonWorker[PythonWorker] + NormalModule[normal Python Module object] + PythonWorker --> NormalModule + end + + PythonManager --> PythonWorker + end + + subgraph TargetMachine[Execution target machine\nmay be same as coordinator] + subgraph ExternalWorkerProcess[ExternalWorker process\none per machine per run] + ExternalWorker[ExternalWorker] + PythonEnv[Python RuntimeEnvironment] + NativeEnv[Native RuntimeEnvironment] + ExternalWorker --> PythonEnv + ExternalWorker --> NativeEnv + end + + subgraph ImplProcesses[implementation processes] + PythonEntrypoint[thin Python entrypoint] + NativeProcess[native subprocess] + end + + PythonEnv --> PythonEntrypoint + NativeEnv --> NativeProcess + end + + Session -. prepare / transfer / bootstrap .-> TargetMachine + Client <-- control RPC / optional SSH tunnel --> ExternalWorker + Envelope --> PythonEntrypoint + Envelope --> ExternalWorker +``` + +Local external deployment uses the same graph with `TargetMachine` equal to `CoordinatorMachine` and a local `TargetSession`. Remote deployment inserts SSH only in the target session and optional worker-control tunnel; module stream data still uses DimOS transports. + +## 4. Package discovery convention + +The `ExternalModule` class file anchors package discovery. `ModuleDeployment` configures where that class builds and executes; it does not repeat implementation details. The sibling-directory convention keeps ordinary packages declarative without introducing a manifest that duplicates `pyproject.toml`, `Cargo.toml`, or `CMakeLists.txt`. + +The implementation directory is selected by a hardcoded sibling convention: + +```text +python/pyproject.toml packaged Python implementation +rust/Cargo.toml Rust executable implementation +cpp/CMakeLists.txt C++ executable implementation +``` + +Python may include Pixi or Nix metadata inside `python/`: + +```text +python/ + pyproject.toml + uv.lock + pixi.toml + pixi.lock + src/... +``` + +The `implementation` field is interpreted using the discovered directory, regardless of whether the declaration used `str` or `Path`: + +```text +python/ import a Python implementation class +rust/ launch an executable relative to rust/ +cpp/ launch an executable relative to cpp/ +``` + +### Existing native precedent + +Current native modules already follow a lightweight convention: + +```text +mls_planner/ + mls_planner_native.py # NativeModule wrapper and module contract + rust/ + Cargo.toml + Cargo.lock + src/... +``` + +The wrapper declares: + +```python +class MLSPlannerNativeConfig(NativeModuleConfig): + cwd = "rust" + executable = "target/release/mls_planner" + build_command = "cargo build --release" + stdin_config = True +``` + +Deployment should generalize this convention before adding another manifest. Existing build files already identify the implementation language and project root, so a second declaration would create two sources of truth. + +The new API extends the design with a parallel declarative module, leaving the existing `NativeModule` untouched until migration: + +```text +mls_planner/ + mls_planner_native.py # existing NativeModule compatibility path + mls_planner_external.py # new ExternalModule declaration + rust/ + Cargo.toml + src/... +``` + +```python +from pathlib import Path as FsPath + + +class MLSPlanner(ExternalModule): + implementation = FsPath("target/release/mls_planner") + + global_map: In[PointCloud2] + path: Out[Path] +``` + +### Packaged Python follows the same layout + +Packaged Python uses the same contract-beside-implementation structure: + +```text +detector/ + detector_module.py # ExternalModule declaration + python/ + pyproject.toml + uv.lock + src/detector_runtime/ + module.py # Runtime implementation +``` + +```python +class HeavyDetector(ExternalModule): + implementation = "detector_runtime.module:HeavyDetectorImpl" + + image: In[Image] + detections: Out[Detections] +``` + +V1 discovery rule: + +1. Start at the `ExternalModule` class file. +2. Walk to the nearest package root. +3. Look for exactly one of `python/pyproject.toml`, `rust/Cargo.toml`, or `cpp/CMakeLists.txt`. +4. Interpret `implementation` as a Python class reference or executable path according to the matched convention. +5. Fail during planning if zero or multiple implementation directories match. + +The convention can expand later, but v1 should not expose a project-root override. + +### Presets turn layouts into behavior + +Users should not need to repeat `uv sync`, `pixi install`, `cargo build`, or CMake commands for every module. DimOS should ship **Convention Presets** that recognize common layouts and select default `Preparation` and `RuntimeEnvironment` behavior. + +Initial presets: + +```text +python/pyproject.toml + python/uv.lock + -> source Preparation + uv RuntimeEnvironment + +python/pixi.toml + python/pixi.lock + python/pyproject.toml + -> source Preparation + Pixi-backed RuntimeEnvironment + +rust/Cargo.toml + -> Cargo Preparation + native RuntimeEnvironment + +cpp/CMakeLists.txt + -> CMake Preparation + native RuntimeEnvironment +``` + +Preset matching uses the most specific convention within one implementation folder: a Python directory with Pixi metadata selects the Pixi-backed environment; otherwise `pyproject.toml` plus `uv.lock` selects uv. Automatic selection requires exactly one supported implementation folder. Zero or multiple matches fail unless `ModuleDeployment` supplies explicit `Preparation` and `RuntimeEnvironment` behavior that selects and validates one implementation folder within the discovered package root. Overrides do not replace package-root discovery. + +## 5. Worker / runtime architecture + +DimOS should unify the manager boundary, not the worker implementation. `ModuleCoordinator` should see a common backend shape for deploy, undeploy, rollback, status, and module handles. Behind that boundary, `WorkerManagerPython` remains a local Python object host and `WorkerManagerExternal` remains a target-machine deployment backend. + +```mermaid +flowchart LR + subgraph Shared[shared coordinator-facing boundary] + Coordinator[ModuleCoordinator] + Backend[WorkerManagerBackend] + Handle[ModuleHandle / ModuleProxy] + Coordinator --> Backend + Backend --> Handle + end + + subgraph PythonBackend[Python backend] + PythonManager[WorkerManagerPython] + PythonWorker[PythonWorker pool] + Actor[Actor-hosted Python Module object] + PythonManager --> PythonWorker + PythonWorker --> Actor + end + + subgraph ExternalBackend[External backend] + ExternalManager[WorkerManagerExternal] + Session[TargetSession] + Client[ExternalWorkerClient] + ExternalWorker[ExternalWorker\none per target machine/run] + ExternalManager --> Session + ExternalManager --> Client + Client --> ExternalWorker + end + + Backend --> PythonManager + Backend --> ExternalManager +``` + +The shared boundary is intentionally narrow. It covers lifecycle sequencing, rollback semantics, status, logs, and the module-handle surface returned to the coordinator. It does not require the workers to share a process protocol. `PythonWorker` receives live Python classes and hosts Python objects; `ExternalWorker` receives serialized launch envelopes and supervises already-prepared target programs. + +### Normal Python stays on the existing path + +Normal Python modules stay local and use the existing `PythonWorker` path. + +```text +Normal Python Module -> PythonWorker +``` + +They are not remotely deployable in v1. If a module is assigned to a non-local target, it must be packaged Python or native. + +This proposal does not replace `PythonWorker`. Its object and RPC envelope already provide an efficient path for modules whose dependencies are available in the coordinator environment. + +### Packaged Python always uses external deployment + +Packaged Python implementations are declared through `ExternalModule` and always use `WorkerManagerExternal`, `ExternalWorker`, and a thin Python entrypoint, even on the local machine. Keeping local and remote packaged Python on the same path prevents local execution from hiding dependency, import, or lifecycle assumptions that fail after remote placement. + +```text +ExternalModule -> WorkerManagerExternal -> ExternalWorker -> thin Python entrypoint +``` + +`ExternalWorker` spawns and supervises the entrypoint without importing user implementation code. The entrypoint runs inside the prepared Python environment, imports the implementation, verifies that it subclasses the declared `ExternalModule`, initializes it, and sends an explicit ready acknowledgement. + +### Native modules reuse the external path + +Native implementations use the same `ExternalModule` contract and worker hierarchy for both local and remote targets. + +```text +ExternalModule -> WorkerManagerExternal -> ExternalWorker -> native subprocess +``` + +The existing `NativeModule` remains unchanged during initial development. New native support uses `ExternalModule`; later PRs migrate existing native modules and eventually remove `NativeModule` and `NativeModuleConfig`. + +### WorkerManagerExternal + +`WorkerManagerExternal` parallels `WorkerManagerPython`. It owns target definitions, module deployment policies, `TargetSession` and `ExternalWorkerClient` handles, rollback, health aggregation, and shutdown. It resolves grouped module policies, coordinates pre-worker `Preparation` through target sessions, and then requests runtime-environment setup and host launch through external-worker clients. + +### TargetSession and ExternalWorker + +`TargetSession` provides coordinator-side access to one target. Local sessions execute commands and transfer files directly. SSH sessions execute commands remotely, transfer content-addressed snapshots and artifacts, bootstrap `ExternalWorker`, and maintain the SSH tunnel that carries worker RPC in v1. + +`ExternalWorkerClient` is the coordinator-side RPC handle. `ExternalWorker` is the target-side process for one machine and deployment run: + +```text +one control connection +worker ID and machine identity +deployed module IDs +runtime handles on that machine +lease and shutdown state +``` + +`ExternalWorker` materializes runtime environments, starts implementation processes or handles, forwards control requests by module ID, and stops those handles during rollback or shutdown. It does not build source or transfer artifacts. Those pre-worker operations belong to `Preparation` through target sessions. + +V1 uses one ExternalWorker per machine per deployment run. A future persistent target agent can implement the same worker contract later. + +Within one deployment run, modules share a runtime environment only when their source digest, execution-machine identity, and resolved `RuntimeEnvironmentSpec` fingerprint match. The run ID scopes the path so another deployment cannot mutate or tear it down. `ExternalWorker` serializes setup with a target-side lock and may rerun idempotent commands instead of merging plans or maintaining a freshness database. Teardown begins only after every dependent runtime handle in that run stops. Cross-run reuse is limited to immutable source, artifact, and package-manager caches. + +### Runtime handles own one implementation + +The external path still needs a per-module runtime handle, but that handle is not necessarily a separate process. It owns exactly one `ExternalModule` implementation, receives one Module Launch Envelope, initializes stream and control bindings, and sends an explicit ready or failure response. + +For packaged Python, the runtime handle includes the thin prepared-Python entrypoint process. That process imports and instantiates the implementation class in-process. + +For native execution, the runtime handle can be only `ExternalWorker` state around a native subprocess. `ExternalWorker` converts the launch envelope into argv, environment, stdin JSON, and control metadata, forwards logs and control requests, supervises exit, and waits for the native SDK's ready acknowledgement. This extracts the process-management responsibilities currently implemented by `NativeModule` without adding another native wrapper process. + +## 6. Control plane vs data plane + +Deployment needs two different communication paths. + +### Deployment Control Plane + +The control plane carries spawn, stop, lifecycle, health, log, status, and supported method-call traffic. + +For `ExternalModule` implementations, control flows through: + +```text +ModuleCoordinator <-> WorkerManagerExternal <-> ExternalWorker <-> runtime handle +``` + +An SSH tunnel can disappear while the remote worker and its implementation processes continue running. V1 handles that failure with a bounded coordinator lease, configured through `GlobalConfig` and proposed to default to 30 seconds. `WorkerManagerExternal` may reconnect before expiry with the run's authenticated resume token. A successful resume replaces the old control session and rejects traffic from its previous connection epoch, preventing two coordinators from controlling the same worker. If the lease expires, `ExternalWorker` invalidates the token and stops its runtime handles. A persistent target agent remains later work. + +### Deployment Data Plane + +The data plane carries module streams such as images, point clouds, poses, paths, commands, and maps. + +Those streams continue to use transports such as Zenoh, DDS, ROS, LCM, or SHM where applicable. SSH never carries module stream data, and SHM remains machine-local. + +V1 should report cross-target transport assumptions in `dimos deploy plan`, but it should not attempt complete data-plane validation. Section 3.6 defines the Module Launch Envelope that carries resolved control and stream bindings into each runtime handle. + +## 7. End-user UX + +The safe deployment flow should be explicit: + +```bash +dimos deploy plan +dimos deploy prepare +dimos run +``` + +### `dimos deploy plan` + +This command resolves the deployment without mutating local or remote targets. + +It should show the resolved plan: + +```text +Module Contract Target Worker path +Agent Module local WorkerManagerPython -> PythonWorker +MLSPlanner ExternalModule robot WorkerManagerExternal -> ExternalWorker -> native process +HeavyDetector ExternalModule gpu WorkerManagerExternal -> ExternalWorker -> Python entrypoint +``` + +The output should include build and execution targets, selected preparation and runtime-environment behavior, source and artifact paths, transport assumptions, and validation errors. A developer should be able to see both where commands will run and which runtime path will launch each implementation. + +### `dimos deploy prepare` + +This command mutates targets but does not launch the stack. It may create deployment roots, synchronize content-addressed source snapshots and artifacts, build native outputs, or cross-compile on one target before copying the result to another. + +Prepare should be idempotent: it runs the required build or synchronization tool and relies on that tool's cache or up-to-date checks. It does not materialize module runtime environments or launch `ExternalWorker`. + +After successful preparation, the command writes a content-addressed plan manifest to the local run registry and each affected target's deployment root. The manifest binds the resolved deployment declaration to its source digests, target identities, selected policies, and staged artifact paths. `dimos run ` refuses to launch when that record is absent or no longer matches the current declaration and source. + +### `dimos run` + +`dimos run ` keeps today's local-default behavior. + +A plain Blueprint bypasses Deployment Spec planning and uses the existing coordinator and `WorkerManagerPython` path. It cannot contain `ExternalModule` declarations because those require a Deployment Spec. + +`dimos run ` verifies and loads the persisted prepared-plan manifest, bootstraps external workers, ensures module runtime environments, launches native processes or thin Python entrypoints, and registers the deployment with the same DimOS run lifecycle commands. If declarations or source no longer match the manifest, it fails before launch and asks the user to run `dimos deploy prepare` again. + +Open question: should `dimos deploy ` become shorthand for prepare plus run later? + +## 8. Lifecycle semantics + +Deployment runs participate in the existing lifecycle model. `dimos status` shows the coordinator, managers, external workers, and runtime handles. `dimos stop` stops the complete deployment rather than only the local coordinator. `dimos restart` reruns the original command and does not prepare unless that command included preparation. `dimos log` aggregates coordinator, worker, packaged-Python, and native-process logs. + +A shared runtime environment is torn down only after all dependent runtime handles stop. Immutable source snapshots and artifacts may remain cached beneath the deployment root because removing them is not part of process lifecycle cleanup. + +Startup should be fail-fast: + +```text +if any module fails before deployment is ready: + stop already-started workers and runtime handles + mark deployment failed +``` + +Runtime-handle death after startup should mark the deployment unhealthy. V1 should not restart it automatically because repeating a physical action may be unsafe; restart policy remains an explicit later feature. + +Each `ExternalWorker` needs a coordinator lease. If the coordinator disappears, the worker stops its runtime handles instead of leaving orphaned robot processes. This lease cannot reverse arbitrary system changes, so safety-related provisioning must be bounded or self-expiring. + +## 9. Implementation path / PR slicing + +This should be built as a sequence of small PRs. PR #2704 should stay as a draft/reference for the packaged-runtime exploration; it should not merge as the public packaged-Python architecture because packaged Python and native implementations should share the `ExternalModule` and external worker path. + +### PR 1: internal shape and skeletons + +This PR establishes the internal deployment layer before changing runtime behavior. It introduces planning under `dimos/core/deployment/`, manager and worker skeletons under `dimos/core/coordination/`, the declarative `ExternalModule` base, internal deployment models, `ModuleLaunchEnvelope`, and the preparation, runtime-environment, and convention interfaces. Isolated tests should exercise the planner with fake modules and backends. + +The PR does not add deployment CLI commands, alter `ModuleCoordinator`, launch packaged Python, migrate native modules, or execute remotely. Keeping those behaviors out allows review of the model without mixing it with process-management failures. + +### PR 2: local packaged Python on the unified path + +This PR validates the external path with local packaged Python before introducing native executables or SSH. It requires a `DeploymentSpec`, stages the Python project, connects `WorkerManagerExternal` to `ModuleCoordinator`, materializes the uv environment inside the local `ExternalWorker`, and launches a thin Python entrypoint. The imported implementation must subclass its `ExternalModule` contract. The work should preserve existing Python module semantics without extending the runtime-specific `PythonWorker` pools explored in PR #2704. + +Open question: what is the first runnable entrypoint before the full deployment CLI/registry exists? + +### PR 3: local native automatic prepare/build + +This PR validates that the same external path can build and launch native code without changing existing `NativeModule` wrappers. It adds an `ExternalModule` declaration beside a native package such as `rust/Cargo.toml`, runs preparation through `WorkerManagerExternal` and a local `TargetSession`, validates the result through the native runtime environment, and launches the executable as a direct child of `ExternalWorker`. + +Use the MLS planner package to exercise the sibling `rust/Cargo.toml` convention while keeping `MLSPlannerNative` available as the compatibility path. + +### PR 4: NativeModule runtime migration + +This PR begins migration only after the native path works beside the compatibility path. It moves one small example to `ExternalModule`, `ExternalWorker`, and direct native-process supervision while keeping `NativeModule` available for unmigrated packages. This avoids a repository-wide cutover before the new lifecycle has production evidence. + +After the path is stable, migrate `MLSPlannerNative` and then other native modules. Remove `NativeModule` and `NativeModuleConfig` only after all callers migrate. + +### PR 5: SSH remote packaged Python + +This PR adds SSH only after local packaged Python validates the worker protocol. It defines SSH targets, synchronizes source into the deployment root, provisions the versioned control environment, starts one remote `ExternalWorker`, tunnels its RPC, and launches thin Python entrypoints from a target-local environment. The control contract should remain usable by a future persistent target agent. + +### PR 6: SSH remote native modules + +This PR adds native artifact movement after SSH control and packaged-Python staging are stable. It synchronizes content-addressed source or prepared artifacts, supports remote builds and cross-compile transfers through target sessions, materializes the native runtime environment inside the machine's existing `ExternalWorker`, and launches native subprocesses directly. Packaged-Python and native remote support remain separate because their preparation outputs and failure recovery differ. + +## 10. Open questions + +1. When should DimOS add optional `dimos.module.toml` or another manifest? +2. What is the PR 2 run entrypoint for local packaged-Python Deployment Specs before the full CLI/registry exists? +3. When should deployment definitions get YAML/TOML after Python-first API? +4. How should shared target profiles and local overlays layer secrets and personal machine details? +5. Should `dimos deploy ` ever become shorthand for prepare plus run? +6. When should strict data-plane compatibility checks become mandatory? +7. What is the minimal resolved-plan JSON schema for tooling and CI? +8. When should DimOS add bounded garbage collection for cached deployment roots? diff --git a/docs/usage/runtime_environments.md b/docs/usage/runtime_environments.md new file mode 100644 index 0000000000..d0aeeaed86 --- /dev/null +++ b/docs/usage/runtime_environments.md @@ -0,0 +1,93 @@ +# Runtime Environments + +Runtime Environments let a blueprint place selected Python modules into a +separate interpreter or Python project without making the coordinator import the +worker-only dependency stack. + +## Runtime Environment Registration + +Register runtimes on a blueprint with `runtime_environments(...)`. Every +blueprint already has the `current` runtime for the coordinator process. A Python +Runtime Project is registered with `PythonProjectRuntimeEnvironment(name, +project)`, where `project` points at a directory containing `pyproject.toml` and +a committed `uv.lock`. + +## Runtime Placement + +Use `runtime_placements(...)` to map a Module Contract class to a +`RuntimePlacement(runtime, implementation)`. The `runtime` names the registered +environment. The `implementation` is an import path resolved inside that runtime +worker and must be a subclass of the Module Contract. + +See `examples/dimos-demo-worker-module` for a minimal contract/implementation +split. + +## Module Contract vs Runtime Implementation + +The Module Contract is the class the coordinator imports. Keep it small and +coordinator-import-safe: declare RPC methods, streams, config, and types needed +for blueprint wiring, but do not import runtime-only packages. The Runtime +Implementation lives in the Runtime Project package, subclasses the contract, +and imports the dependencies needed to do the work. + +This is different from a DimOS `Spec`. A `Spec` is a structural interface used +for module-reference injection: one module declares the methods it needs from +another module. A Module Contract is a concrete, deployable `Module` identity the +coordinator uses for blueprint graph construction, actor lookup, RPC topic +identity, streams, and lifecycle. Runtime Implementations subclass that contract +for this change; future descriptor work may loosen that requirement. + +## Runtime Project and Locked Runtime Project + +A Runtime Project is a Python project directory used by one or more worker +modules. A Locked Runtime Project has committed lockfiles so deployment can +reproduce the environment without rewriting dependency resolution state. + +- uv projects require `pyproject.toml` and `uv.lock`. +- Pixi-backed uv projects require `pyproject.toml`, `uv.lock`, `pixi.toml`, and + `pixi.lock`. + +Lockfiles are source inputs. Generate or update them manually with the package +manager (`uv lock`, `pixi lock`, or a future DimOS build/update command), then +commit them. + +## Runtime Reconciliation during deployment + +Before worker launch, DimOS selects the active Runtime Project placements and +reconciles their prepared environments in locked mode. For uv projects it runs +`uv venv --seed` and `uv sync --locked` in the Runtime Project. For Pixi-backed +projects it runs `pixi install --locked`, creates a uv venv from the Pixi +Python, and runs `pixi run uv sync --locked`. + +Reconciliation prepares `.venv` state but does not mutate lockfiles. If a lock is +missing or stale, deployment fails and asks for an explicit package-manager lock +or update step. + +## Python Runtime Worker semantics + +Python runtime workers still run DimOS modules with normal module lifecycle, +RPC, streams, and worker isolation. The coordinator sends the Module Contract and +optional Runtime Implementation path to the worker. Inside the worker, DimOS +imports the implementation path and verifies that it subclasses the contract +before instantiating it. + +## Failure modes + +- Unknown runtime name: placement references a runtime that was not registered. +- Duplicate Runtime Project path: two registered runtimes point at the same + project directory. +- Missing project files: `pyproject.toml`, `uv.lock`, or `pixi.lock` is absent. +- Reconciliation command missing or failed: `uv`/`pixi` is unavailable, locked + sync fails, or the lock does not match the project. +- Missing prepared Python: `.venv/bin/python` was not created by reconciliation. +- Invalid implementation path: import path is malformed or cannot be imported in + the worker runtime. +- Implementation type error: the imported class is not a subclass of the Module + Contract. + +## Non-goals for this change + +Runtime descriptors, remote Runtime Projects, automatic image/build systems, and +lockfile/build-update orchestration are future work. The current behavior is +local registration, explicit placement, locked reconciliation, and Python worker +launch. diff --git a/examples/dimos-demo-worker-module/README.md b/examples/dimos-demo-worker-module/README.md new file mode 100644 index 0000000000..49ce71ba69 --- /dev/null +++ b/examples/dimos-demo-worker-module/README.md @@ -0,0 +1,37 @@ +# Demo Worker Runtime Project + +This is a dependency-light example of splitting a coordinator-imported Module +Contract from a worker-only Runtime Implementation. + +- `dimos_demo_worker_module.contract.DemoWorkerModule` is safe for the + coordinator to import. It declares the module's RPC surface and avoids + runtime-only dependency imports. +- `dimos_demo_worker_module.runtime.DemoWorkerRuntimeModule` subclasses that + contract and is imported only in the selected Python Runtime Project worker. +- `dimos_demo_worker_module.blueprint.demo_worker_runtime_blueprint` shows how + to register a Runtime Project and place the contract into it. + +The project intentionally keeps dependencies minimal, but includes `inflection` +as a runtime-only dependency to prove the contract can be imported without the +worker dependency stack. DimOS automatically adds its own import path when +launching the runtime worker; the demo only adds the example package source and +compatible dependency paths. The example lockfile only captures the runtime-only +dependency closure. Sync from the checked-in lockfile before running it: + +```bash +cd examples/dimos-demo-worker-module +uv sync --locked +``` + +Run the blueprint from the repository's main environment at the repo root: + +```bash +cd ../.. +uv run python examples/dimos-demo-worker-module/demo_run_blueprint.py +``` + +The script prints the main Python executable, then calls RPC methods on the +runtime-placed module. The reported worker Python should be under this example's +`.venv/`, showing that the coordinator script ran in the main environment while +the module ran in the Runtime Project environment. It also prints a value derived +from the runtime-only dependency. diff --git a/examples/dimos-demo-worker-module/demo_run_blueprint.py b/examples/dimos-demo-worker-module/demo_run_blueprint.py new file mode 100644 index 0000000000..cdcbfe0c0e --- /dev/null +++ b/examples/dimos-demo-worker-module/demo_run_blueprint.py @@ -0,0 +1,63 @@ +# Copyright 2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Run the demo Runtime Project blueprint from the main DimOS environment. + +This script is intentionally outside ``src/``: run it with the repository's main +environment, and the blueprint will reconcile/spawn the module inside this +example project's locked Runtime Project environment. + +Run from an environment where this example package is installed, or set +``PYTHONPATH=examples/dimos-demo-worker-module/src:.``. +""" + +from __future__ import annotations + +from pathlib import Path +import sys + +EXAMPLE_ROOT = Path(__file__).resolve().parent +REPO_ROOT = EXAMPLE_ROOT.parents[1] +EXAMPLE_SRC = EXAMPLE_ROOT / "src" + +for path in (REPO_ROOT, EXAMPLE_SRC): + path_str = str(path) + if path_str not in sys.path: + sys.path.insert(0, path_str) + +from dimos_demo_worker_module.blueprint import demo_worker_runtime_blueprint +from dimos_demo_worker_module.contract import DemoWorkerModule + +from dimos.core.coordination.module_coordinator import ModuleCoordinator + + +def main() -> None: + """Build the blueprint, call the runtime worker, and stop cleanly.""" + + print(f"main python: {sys.executable}") + print(f"runtime project: {EXAMPLE_ROOT}") + print("building blueprint; this runs locked runtime reconciliation first...") + + coordinator = ModuleCoordinator.build(demo_worker_runtime_blueprint, {}) + try: + module = coordinator.get_instance(DemoWorkerModule) + print(f"worker python: {module.runtime_python()}") + print(f"runtime-only dependency: {module.runtime_dependency_label()}") + print(f"worker result: {module.transform(' hello from main venv ')}") + finally: + coordinator.stop() + + +if __name__ == "__main__": + main() diff --git a/examples/dimos-demo-worker-module/pyproject.toml b/examples/dimos-demo-worker-module/pyproject.toml new file mode 100644 index 0000000000..061ce51c49 --- /dev/null +++ b/examples/dimos-demo-worker-module/pyproject.toml @@ -0,0 +1,12 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "dimos-demo-worker-module" +version = "0.1.0" +description = "Dependency-light DimOS Runtime Project example" +requires-python = ">=3.10,<3.13" +dependencies = [ + "inflection>=0.5.1", +] diff --git a/examples/dimos-demo-worker-module/src/dimos_demo_worker_module/__init__.py b/examples/dimos-demo-worker-module/src/dimos_demo_worker_module/__init__.py new file mode 100644 index 0000000000..95cba2ffa3 --- /dev/null +++ b/examples/dimos-demo-worker-module/src/dimos_demo_worker_module/__init__.py @@ -0,0 +1,5 @@ +"""Demo Runtime Project for Python runtime workers.""" + +from dimos_demo_worker_module.contract import DemoWorkerModule + +__all__ = ["DemoWorkerModule"] diff --git a/examples/dimos-demo-worker-module/src/dimos_demo_worker_module/blueprint.py b/examples/dimos-demo-worker-module/src/dimos_demo_worker_module/blueprint.py new file mode 100644 index 0000000000..db34d59c3b --- /dev/null +++ b/examples/dimos-demo-worker-module/src/dimos_demo_worker_module/blueprint.py @@ -0,0 +1,56 @@ +# Copyright 2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Blueprint snippet demonstrating Runtime Project registration and placement.""" + +import os +from pathlib import Path +import sys + +from dimos.core.coordination.blueprints import autoconnect +from dimos.core.runtime_environment import PythonProjectRuntimeEnvironment, RuntimePlacement +from dimos_demo_worker_module.contract import DemoWorkerModule + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +EXAMPLE_SRC = PROJECT_ROOT / "src" + + +def _worker_pythonpath() -> str: + paths = [str(EXAMPLE_SRC)] + paths.extend( + path for path in sys.path if path and ("site-packages" in path or "dist-packages" in path) + ) + if existing := os.environ.get("PYTHONPATH"): + paths.append(existing) + return os.pathsep.join(dict.fromkeys(paths)) + + +demo_worker_runtime = PythonProjectRuntimeEnvironment( + name="demo-worker-runtime", + project=PROJECT_ROOT, + env={"PYTHONPATH": _worker_pythonpath(), "UV_PYTHON": sys.executable}, +) + +demo_worker_runtime_blueprint = ( + autoconnect(DemoWorkerModule.blueprint()) + .runtime_environments(demo_worker_runtime) + .runtime_placements( + { + DemoWorkerModule: RuntimePlacement( + runtime="demo-worker-runtime", + implementation="dimos_demo_worker_module.runtime.DemoWorkerRuntimeModule", + ) + } + ) +) diff --git a/examples/dimos-demo-worker-module/src/dimos_demo_worker_module/contract.py b/examples/dimos-demo-worker-module/src/dimos_demo_worker_module/contract.py new file mode 100644 index 0000000000..00199f4b23 --- /dev/null +++ b/examples/dimos-demo-worker-module/src/dimos_demo_worker_module/contract.py @@ -0,0 +1,42 @@ +# Copyright 2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Coordinator-import-safe Module Contract for the demo worker. + +Keep this module dependency-light. It should define the stable RPC/stream shape +that blueprints and the coordinator need, but avoid importing packages that only +exist in the worker Runtime Project. +""" + +from dimos.core.core import rpc +from dimos.core.module import Module + + +class DemoWorkerModule(Module): + """Contract imported by the coordinator process.""" + + @rpc + def transform(self, text: str) -> str: + """Transform text in the worker runtime.""" + raise NotImplementedError("Runtime placement must provide an implementation") + + @rpc + def runtime_python(self) -> str: + """Return the Python executable used by the worker runtime.""" + raise NotImplementedError("Runtime placement must provide an implementation") + + @rpc + def runtime_dependency_label(self) -> str: + """Return a label formatted with a runtime-only dependency.""" + raise NotImplementedError("Runtime placement must provide an implementation") diff --git a/examples/dimos-demo-worker-module/src/dimos_demo_worker_module/runtime.py b/examples/dimos-demo-worker-module/src/dimos_demo_worker_module/runtime.py new file mode 100644 index 0000000000..a9ae59497d --- /dev/null +++ b/examples/dimos-demo-worker-module/src/dimos_demo_worker_module/runtime.py @@ -0,0 +1,40 @@ +# Copyright 2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Worker-only Runtime Implementation for the demo module.""" + +import sys + +from dimos.core.core import rpc +from dimos_demo_worker_module.contract import DemoWorkerModule +from inflection import camelize + + +class DemoWorkerRuntimeModule(DemoWorkerModule): + """Concrete implementation loaded inside the selected Python Runtime Project.""" + + @rpc + def transform(self, text: str) -> str: + """Transform text in the worker runtime.""" + return f"demo-runtime:{text.strip().upper()}" + + @rpc + def runtime_python(self) -> str: + """Return the Python executable used by the worker runtime.""" + return sys.executable + + @rpc + def runtime_dependency_label(self) -> str: + """Return a label formatted with a runtime-only dependency.""" + return camelize("runtime_dependency") diff --git a/examples/dimos-demo-worker-module/uv.lock b/examples/dimos-demo-worker-module/uv.lock new file mode 100644 index 0000000000..6685c2d15a --- /dev/null +++ b/examples/dimos-demo-worker-module/uv.lock @@ -0,0 +1,43 @@ +version = 1 +revision = 3 +requires-python = ">=3.10, <3.13" +resolution-markers = [ + "python_full_version >= '3.12' and sys_platform == 'win32'", + "python_full_version >= '3.12' and sys_platform == 'emscripten'", + "python_full_version >= '3.12' and sys_platform != 'emscripten' and sys_platform != 'win32'", + "python_full_version == '3.11.*' and sys_platform == 'win32'", + "python_full_version == '3.11.*' and sys_platform == 'emscripten'", + "python_full_version == '3.11.*' and sys_platform != 'emscripten' and sys_platform != 'win32'", + "python_full_version < '3.11'", +] + +[options] +exclude-newer = "0001-01-01T00:00:00Z" # This has no effect and is included for backwards compatibility when using relative exclude-newer values. +exclude-newer-span = "P7D" + +[options.exclude-newer-package] +dimos-viewer = false +dimos-lcm = false +lcm-dimos-fork = false +pyrealsense2-extended = false +roboplan = false + +[[package]] +name = "dimos-demo-worker-module" +version = "0.1.0" +source = { editable = "." } +dependencies = [ + { name = "inflection" }, +] + +[package.metadata] +requires-dist = [{ name = "inflection", specifier = ">=0.5.1" }] + +[[package]] +name = "inflection" +version = "0.5.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/e1/7e/691d061b7329bc8d54edbf0ec22fbfb2afe61facb681f9aaa9bff7a27d04/inflection-0.5.1.tar.gz", hash = "sha256:1a29730d366e996aaacffb2f1f1cb9593dc38e2ddd30c91250c6dde09ea9b417", size = 15091, upload-time = "2020-08-22T08:16:29.139Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/59/91/aa6bde563e0085a02a435aa99b49ef75b0a4b062635e606dab23ce18d720/inflection-0.5.1-py2.py3-none-any.whl", hash = "sha256:f38b2b640938a4f35ade69ac3d053042959b62a0f1076a5bbaa1b9526605a8a2", size = 9454, upload-time = "2020-08-22T08:16:27.816Z" }, +]