diff --git a/src/lfx/src/lfx/execution/__init__.py b/src/lfx/src/lfx/execution/__init__.py index a04dcf00a6..7ee48a03aa 100644 --- a/src/lfx/src/lfx/execution/__init__.py +++ b/src/lfx/src/lfx/execution/__init__.py @@ -1,9 +1,13 @@ -"""Public entry points for the execution layer.""" +"""Public entry points for the execution layer. + +The execution coordinator and registry are owned by the :class:`ExecutorService` +(``lfx.services.executor``). The helpers below are thin convenience wrappers that +resolve the service through the standard service manager so callers do not need to +know about the service plumbing. +""" from __future__ import annotations -from dataclasses import dataclass - from lfx.execution.backends.in_process import InProcessExecutor from lfx.execution.coordinator import Coordinator from lfx.execution.executor import Executor @@ -12,37 +16,35 @@ from lfx.execution.registry import ExecutorNotFoundError, ExecutorRegistry from lfx.execution.types import RunComplete, StepResult, Unit -@dataclass -class _Defaults: - registry: ExecutorRegistry | None = None - coordinator: Coordinator | None = None +def _get_executor_service(): + from lfx.services.deps import get_service + from lfx.services.schema import ServiceType - -_defaults = _Defaults() + service = get_service(ServiceType.EXECUTOR_SERVICE) + if service is None: + msg = "ExecutorService is not available; check earlier logs for the underlying init failure." + raise RuntimeError(msg) + return service def get_default_registry() -> ExecutorRegistry: - if _defaults.registry is None: - registry = ExecutorRegistry() - registry.register(InProcessExecutor()) - _defaults.registry = registry - return _defaults.registry + return _get_executor_service().registry def get_default_coordinator() -> Coordinator: - if _defaults.coordinator is None: - _defaults.coordinator = Coordinator(registry=get_default_registry()) - return _defaults.coordinator + return _get_executor_service().coordinator def set_default_coordinator(coordinator: Coordinator) -> None: - _defaults.coordinator = coordinator + _get_executor_service().set_coordinator(coordinator) def reset_default_coordinator() -> None: - """Drop the module-level singletons; next access rebuilds them. For tests.""" - _defaults.registry = None - _defaults.coordinator = None + """Drop the executor service so the next access rebuilds registry + coordinator. For tests.""" + from lfx.services.manager import get_service_manager + from lfx.services.schema import ServiceType + + get_service_manager().update(ServiceType.EXECUTOR_SERVICE) __all__ = [ @@ -50,6 +52,7 @@ __all__ = [ "Executor", "ExecutorNotFoundError", "ExecutorRegistry", + "InProcessExecutor", "RunComplete", "StepResult", "Unit", diff --git a/src/lfx/src/lfx/execution/registry.py b/src/lfx/src/lfx/execution/registry.py index 857d0fe1dc..e9d7332085 100644 --- a/src/lfx/src/lfx/execution/registry.py +++ b/src/lfx/src/lfx/execution/registry.py @@ -12,13 +12,32 @@ class ExecutorNotFoundError(LookupError): pass +class ExecutorKindCollisionError(ValueError): + """Raised when registering an executor whose kind is already registered and replace=False.""" + + class ExecutorRegistry: def __init__(self) -> None: self._by_kind: dict[str, Executor] = {} - def register(self, executor: Executor) -> None: + def register(self, executor: Executor, *, replace: bool = True) -> None: + """Register an executor by its kind. + + Args: + executor: The executor to register. + replace: If True (default), overwrite any existing executor with the same kind. + If False, raise :class:`ExecutorKindCollisionError` instead. Discovery paths + pass ``replace=False`` so an installed package cannot silently replace a + pre-registered executor (notably the built-in ``in-process``). + """ + if not replace and executor.kind in self._by_kind: + msg = f"Executor kind={executor.kind!r} is already registered" + raise ExecutorKindCollisionError(msg) self._by_kind[executor.kind] = executor + def has(self, kind: str) -> bool: + return kind in self._by_kind + def get(self, kind: str) -> Executor: try: return self._by_kind[kind] diff --git a/src/lfx/src/lfx/services/deps.py b/src/lfx/src/lfx/services/deps.py index aa0cab4478..25a3601904 100644 --- a/src/lfx/src/lfx/services/deps.py +++ b/src/lfx/src/lfx/services/deps.py @@ -61,6 +61,10 @@ def get_service(service_type: ServiceType, default=None): try: return service_manager.get(service_type, default) except Exception: # noqa: BLE001 + # Preserve the traceback in logs so callers seeing a None return have something to grep + # for. Returning None remains the contract because several callers (e.g. get_db_service) + # treat absence as "not configured" and substitute a noop implementation. + logger.exception("Failed to resolve service %s", service_type) return None diff --git a/src/lfx/src/lfx/services/executor/__init__.py b/src/lfx/src/lfx/services/executor/__init__.py new file mode 100644 index 0000000000..57a6cf6e87 --- /dev/null +++ b/src/lfx/src/lfx/services/executor/__init__.py @@ -0,0 +1 @@ +"""Executor service module.""" diff --git a/src/lfx/src/lfx/services/executor/factory.py b/src/lfx/src/lfx/services/executor/factory.py new file mode 100644 index 0000000000..b3d02fd7ab --- /dev/null +++ b/src/lfx/src/lfx/services/executor/factory.py @@ -0,0 +1,24 @@ +"""Factory for the executor service.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from lfx.services.executor.service import ExecutorService +from lfx.services.factory import ServiceFactory +from lfx.services.schema import ServiceType + +if TYPE_CHECKING: + from lfx.services.settings.service import SettingsService + + +class ExecutorServiceFactory(ServiceFactory): + """Factory for creating ExecutorService instances.""" + + def __init__(self) -> None: + super().__init__() + self.service_class = ExecutorService + self.dependencies = [ServiceType.SETTINGS_SERVICE] + + def create(self, settings_service: SettingsService) -> ExecutorService: + return ExecutorService(settings_service=settings_service) diff --git a/src/lfx/src/lfx/services/executor/service.py b/src/lfx/src/lfx/services/executor/service.py new file mode 100644 index 0000000000..f98c8468c2 --- /dev/null +++ b/src/lfx/src/lfx/services/executor/service.py @@ -0,0 +1,118 @@ +"""Executor service: owns the executor registry and the default coordinator.""" + +from __future__ import annotations + +from importlib.metadata import entry_points +from typing import TYPE_CHECKING + +from lfx.execution.backends.in_process import InProcessExecutor +from lfx.execution.coordinator import Coordinator +from lfx.execution.registry import ExecutorKindCollisionError, ExecutorRegistry +from lfx.log.logger import logger +from lfx.services.base import Service + +if TYPE_CHECKING: + from lfx.execution.executor import Executor + from lfx.services.settings.service import SettingsService + + +ENTRY_POINT_GROUP = "lfx.executors" + + +class ExecutorService(Service): + """Service that holds the executor registry and the default coordinator.""" + + name = "executor_service" + + def __init__(self, settings_service: SettingsService) -> None: + super().__init__() + self._settings_service = settings_service + self._registry = ExecutorRegistry() + self._coordinator: Coordinator | None = None + self._populate_registry() + + def register(self, executor: Executor) -> None: + """Register an executor; replaces any prior registration for the same kind. + + Invalidates the cached coordinator so the next access rebuilds against the new + registry contents. Direct mutation of the registry through other paths bypasses + this invalidation -- always go through this method to register or replace. + """ + self._registry.register(executor) + self._coordinator = None + + def has(self, kind: str) -> bool: + """Return True if an executor with the given kind is registered.""" + return self._registry.has(kind) + + def get(self, kind: str) -> Executor: + """Return the executor registered under the given kind.""" + return self._registry.get(kind) + + @property + def registry(self) -> ExecutorRegistry: + """Return the underlying registry. + + Read access only. Mutating the registry directly (e.g. ``service.registry.register(...)``) + bypasses the cached-coordinator invalidation done by :meth:`register`; always go through + :meth:`register` for mutations. + """ + return self._registry + + @property + def coordinator(self) -> Coordinator: + if self._coordinator is None: + kind = self._settings_service.settings.executor_kind + self._coordinator = Coordinator(registry=self._registry, executor_kind=kind) + return self._coordinator + + def set_coordinator(self, coordinator: Coordinator) -> None: + """Override the default coordinator (test/extension hook).""" + self._coordinator = coordinator + + def _populate_registry(self) -> None: + """Register the built-in executor and run plugin discovery.""" + self._registry.register(InProcessExecutor()) + self._discover_entry_points() + + def _discover_entry_points(self) -> None: + try: + eps = entry_points(group=ENTRY_POINT_GROUP) + except Exception: # noqa: BLE001 + logger.exception( + "Failed to enumerate entry points for group=%r. Executor plugin discovery skipped.", + ENTRY_POINT_GROUP, + ) + return + for ep in eps: + kind: str = "" + try: + obj = ep.load() + executor = obj() if isinstance(obj, type) else obj + kind = getattr(executor, "kind", "") + self._registry.register(executor, replace=False) + logger.debug("Loaded executor from entry point: %s -> %s", ep.name, kind) + except ExecutorKindCollisionError: + logger.warning( + "Skipping entry point %r: an executor with kind=%r is already registered. " + "Entry-point discovery cannot replace existing executors; register the " + "executor explicitly via ExecutorService.register() if replacement is intended.", + ep.name, + kind, + ) + except Exception: # noqa: BLE001 + # Broad catch is intentional: a single broken plugin must not break discovery + # of others. The full traceback is preserved via logger.exception. + logger.exception("Failed to load executor entry point %s", ep.name) + + async def teardown(self) -> None: + """Reset the service to its post-init state. + + ServiceManager.teardown() does not evict the cached service instance, so we must + leave the registry usable: the built-in InProcessExecutor is re-registered and + plugin discovery re-runs so a post-teardown access still returns a working + coordinator. + """ + self._coordinator = None + self._registry = ExecutorRegistry() + self._populate_registry() diff --git a/src/lfx/src/lfx/services/schema.py b/src/lfx/src/lfx/services/schema.py index 47a53ba257..1bba5de093 100644 --- a/src/lfx/src/lfx/services/schema.py +++ b/src/lfx/src/lfx/services/schema.py @@ -27,3 +27,4 @@ class ServiceType(str, Enum): FLOW_EVENTS_SERVICE = "flow_events_service" TELEMETRY_WRITER_SERVICE = "telemetry_writer_service" EXTENSION_EVENTS_SERVICE = "extension_events_service" + EXECUTOR_SERVICE = "executor_service" diff --git a/src/lfx/src/lfx/services/settings/groups/runtime.py b/src/lfx/src/lfx/services/settings/groups/runtime.py index 14f2c9ea38..b055ab136b 100644 --- a/src/lfx/src/lfx/services/settings/groups/runtime.py +++ b/src/lfx/src/lfx/services/settings/groups/runtime.py @@ -85,6 +85,14 @@ class RuntimeSettings(BaseModel): celery_enabled: bool = False + executor_kind: str = "in-process" + """The default executor kind used by the execution coordinator. + + Must match the `kind` of an Executor registered with the executor service. The built-in + `in-process` executor runs graphs in the current process; third-party executors registered + via the `lfx.executors` entry-point group can be selected by setting this to their kind. + """ + @field_validator("event_delivery", mode="before") @classmethod def set_event_delivery(cls, value, info): diff --git a/src/lfx/tests/unit/execution/test_executor_service.py b/src/lfx/tests/unit/execution/test_executor_service.py new file mode 100644 index 0000000000..ede7b34271 --- /dev/null +++ b/src/lfx/tests/unit/execution/test_executor_service.py @@ -0,0 +1,220 @@ +"""Service-level contract tests for the ExecutorService.""" + +from __future__ import annotations + +import pytest +from lfx.execution.executor import Executor +from lfx.execution.types import RunComplete +from lfx.services.deps import get_service +from lfx.services.executor.service import ExecutorService +from lfx.services.manager import get_service_manager +from lfx.services.schema import ServiceType + + +def _get_service() -> ExecutorService: + service = get_service(ServiceType.EXECUTOR_SERVICE) + assert service is not None + return service + + +def test_executor_service_is_resolvable_via_service_manager(): + service = _get_service() + assert isinstance(service, ExecutorService) + + +def test_executor_service_preregisters_in_process(): + service = _get_service() + assert service.has("in-process") + assert service.get("in-process").kind == "in-process" + + +@pytest.mark.asyncio +async def test_coordinator_uses_settings_executor_kind(): + """Behavioral assertion: coordinator dispatches to whichever kind the settings select.""" + settings_service = get_service(ServiceType.SETTINGS_SERVICE) + settings_service.settings.executor_kind = "kind-from-settings" + + seen: list[str] = [] + + class Sentinel(Executor): + kind = "kind-from-settings" + + async def execute(self, unit): # noqa: ARG002 + seen.append(self.kind) + yield RunComplete(outputs=["ok"]) + + get_service_manager().update(ServiceType.EXECUTOR_SERVICE) + service = _get_service() + service.register(Sentinel()) + + outputs = await service.coordinator.run_to_completion(graph=object(), inputs=[]) + + assert seen == ["kind-from-settings"] + assert outputs == ["ok"] + + settings_service.settings.executor_kind = "in-process" + get_service_manager().update(ServiceType.EXECUTOR_SERVICE) + + +def test_register_replaces_and_invalidates_cached_coordinator(): + service = _get_service() + first = service.coordinator + + class Other(Executor): + kind = "in-process" + + async def execute(self, unit): # noqa: ARG002 + yield RunComplete(outputs=[]) + + service.register(Other()) + assert service.coordinator is not first + assert isinstance(service.get("in-process"), Other) + + +@pytest.mark.asyncio +async def test_teardown_preserves_builtin_in_process(): + """teardown() must leave the service usable: the built-in must still be registered. + + ServiceManager.teardown() does not evict the cached service instance, so any code + that resolves the service after teardown (e.g. background tasks racing app shutdown) + must still get a working coordinator. + """ + service = _get_service() + + custom_kinds_before = service.has("in-process") + assert custom_kinds_before + + await service.teardown() + + assert service.has("in-process") + assert service.coordinator is not None # rebuilds lazily over the fresh registry + + +def test_entry_point_cannot_replace_builtin_in_process(monkeypatch): + """A package exposing an executor with kind="in-process" must not silently replace the built-in.""" + from lfx.execution.backends.in_process import InProcessExecutor + from lfx.services.executor import service as service_module + + class Hijacker(Executor): + kind = "in-process" + + async def execute(self, unit): # noqa: ARG002 + yield RunComplete(outputs=[]) + + class _FakeEntryPoint: + name = "hijacker" + + def load(self): + return Hijacker + + monkeypatch.setattr( + service_module, + "entry_points", + lambda group: [_FakeEntryPoint()] if group == service_module.ENTRY_POINT_GROUP else [], + ) + + settings_service = get_service(ServiceType.SETTINGS_SERVICE) + fresh = ExecutorService(settings_service=settings_service) + + assert isinstance(fresh.get("in-process"), InProcessExecutor) + + +def test_entry_point_with_unique_kind_is_registered(monkeypatch): + """Entry points with non-colliding kinds are still loaded.""" + from lfx.services.executor import service as service_module + + class RemoteStub(Executor): + kind = "remote-stub" + + async def execute(self, unit): # noqa: ARG002 + yield RunComplete(outputs=[]) + + class _FakeEntryPoint: + name = "remote_stub" + + def load(self): + return RemoteStub + + monkeypatch.setattr( + service_module, + "entry_points", + lambda group: [_FakeEntryPoint()] if group == service_module.ENTRY_POINT_GROUP else [], + ) + + settings_service = get_service(ServiceType.SETTINGS_SERVICE) + fresh = ExecutorService(settings_service=settings_service) + + assert fresh.has("remote-stub") + assert isinstance(fresh.get("remote-stub"), RemoteStub) + + +def test_entry_point_load_failure_does_not_break_discovery(monkeypatch): + """A single broken entry point must not prevent the built-in or other plugins from loading.""" + from lfx.execution.backends.in_process import InProcessExecutor + from lfx.services.executor import service as service_module + + class GoodStub(Executor): + kind = "good-stub" + + async def execute(self, unit): # noqa: ARG002 + yield RunComplete(outputs=[]) + + class _BrokenEntryPoint: + name = "broken" + + def load(self): + msg = "simulated import failure" + raise ImportError(msg) + + class _GoodEntryPoint: + name = "good_stub" + + def load(self): + return GoodStub + + monkeypatch.setattr( + service_module, + "entry_points", + lambda group: [_BrokenEntryPoint(), _GoodEntryPoint()] if group == service_module.ENTRY_POINT_GROUP else [], + ) + + settings_service = get_service(ServiceType.SETTINGS_SERVICE) + fresh = ExecutorService(settings_service=settings_service) + + assert isinstance(fresh.get("in-process"), InProcessExecutor) + assert isinstance(fresh.get("good-stub"), GoodStub) + + +def test_entry_point_returning_non_executor_is_skipped(monkeypatch): + """An entry point whose loaded object lacks `kind` must not crash discovery.""" + from lfx.execution.backends.in_process import InProcessExecutor + from lfx.services.executor import service as service_module + + class NotAnExecutor: + pass + + class _BadEntryPoint: + name = "bad" + + def load(self): + return NotAnExecutor + + monkeypatch.setattr( + service_module, + "entry_points", + lambda group: [_BadEntryPoint()] if group == service_module.ENTRY_POINT_GROUP else [], + ) + + settings_service = get_service(ServiceType.SETTINGS_SERVICE) + fresh = ExecutorService(settings_service=settings_service) + + assert isinstance(fresh.get("in-process"), InProcessExecutor) + + +def test_factory_declares_settings_dependency(): + """Pin the factory contract: ExecutorService depends on settings_service only.""" + from lfx.services.executor.factory import ExecutorServiceFactory + + factory = ExecutorServiceFactory() + assert factory.service_class is ExecutorService + assert factory.dependencies == [ServiceType.SETTINGS_SERVICE] diff --git a/src/lfx/tests/unit/execution/test_registry.py b/src/lfx/tests/unit/execution/test_registry.py index 91c923e7f3..421776ada5 100644 --- a/src/lfx/tests/unit/execution/test_registry.py +++ b/src/lfx/tests/unit/execution/test_registry.py @@ -1,6 +1,6 @@ import pytest from lfx.execution.executor import Executor -from lfx.execution.registry import ExecutorNotFoundError, ExecutorRegistry +from lfx.execution.registry import ExecutorKindCollisionError, ExecutorNotFoundError, ExecutorRegistry class _Stub(Executor): @@ -29,3 +29,19 @@ def test_register_replaces_same_kind(): registry.register(a) registry.register(b) assert registry.get("stub") is b + + +def test_register_replace_false_raises_on_collision(): + registry = ExecutorRegistry() + first = _Stub() + registry.register(first) + with pytest.raises(ExecutorKindCollisionError, match="stub"): + registry.register(_Stub(), replace=False) + assert registry.get("stub") is first + + +def test_has_reflects_registration(): + registry = ExecutorRegistry() + assert not registry.has("stub") + registry.register(_Stub()) + assert registry.has("stub")