From e320e5df5d3c6fc85f4f0950db140e62a5fbef6b Mon Sep 17 00:00:00 2001 From: ogabrielluiz Date: Tue, 5 May 2026 10:18:39 -0300 Subject: [PATCH] feat(execution): make executor seam pluggable via lfx services Introduce an ExecutorService that owns the registry and the default coordinator, discovered through the standard lfx ServiceManager (parity with cache, database, storage, etc). Add an executor_kind setting so the default kind can be configured without touching code, and an lfx.executors entry-point group for third-party executors. Entry-point discovery refuses to overwrite an existing kind so an installed package cannot silently replace the built-in in-process executor; explicit replacement still works through ExecutorService.register(). The lfx.execution public API (get_default_coordinator, get_default_registry, set_default_coordinator, reset_default_coordinator) now resolves through the service manager but keeps the same surface, so existing call sites in Graph.arun, flow_executor, the CLI, run/base, and loop_utils are unchanged. Also surface root-cause tracebacks in lfx.services.deps.get_service: the helper still returns None on failure (callers like get_db_service rely on that), but it now logs the exception so init failures stop disappearing into the void. --- src/lfx/src/lfx/execution/__init__.py | 45 ++-- src/lfx/src/lfx/execution/registry.py | 21 +- src/lfx/src/lfx/services/deps.py | 4 + src/lfx/src/lfx/services/executor/__init__.py | 1 + src/lfx/src/lfx/services/executor/factory.py | 24 ++ src/lfx/src/lfx/services/executor/service.py | 118 ++++++++++ src/lfx/src/lfx/services/schema.py | 1 + .../lfx/services/settings/groups/runtime.py | 8 + .../unit/execution/test_executor_service.py | 220 ++++++++++++++++++ src/lfx/tests/unit/execution/test_registry.py | 18 +- 10 files changed, 437 insertions(+), 23 deletions(-) create mode 100644 src/lfx/src/lfx/services/executor/__init__.py create mode 100644 src/lfx/src/lfx/services/executor/factory.py create mode 100644 src/lfx/src/lfx/services/executor/service.py create mode 100644 src/lfx/tests/unit/execution/test_executor_service.py 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")