forge.core

Core runtime — module registry, lifecycle, DI container, event bus, and context management.

Provides the ForgeModule base class, ForgeRuntime coordinator, and supporting infrastructure for module initialization ordering, dependency injection, event emission, and async context propagation.

forge.core

Core runtime — module registry, lifecycle, DI container, event bus, and context management.

Provides the ForgeModule base class, ForgeRuntime coordinator, and supporting infrastructure for module initialization ordering, dependency injection, event emission, and async context propagation.

Classes

CircularDependencyError

Bases: ModuleError

Raised when a circular dependency is detected between modules.

Source code in src/forge/core/exceptions.py

class CircularDependencyError(ModuleError):
    """Raised when a circular dependency is detected between modules."""

ConfigurationError

Bases: ForgeError

Raised when configuration is missing, invalid, or misconfigured.

Source code in src/forge/core/exceptions.py

class ConfigurationError(ForgeError):
    """Raised when configuration is missing, invalid, or misconfigured."""

Container

Lightweight explicit dependency injection container.

This container holds all registered ForgeModule instances and provides type-safe retrieval. Registration is explicit (not annotation-based) per ADR-001_.

.. _ADR-001: https://github.com/kaif-ai-engineer/forge-kaif/blob/main/docs/ADR-001.md

Source code in src/forge/core/container.py

class Container:
    """
    Lightweight explicit dependency injection container.

    This container holds all registered ``ForgeModule`` instances and
    provides type-safe retrieval.  Registration is explicit (not
    annotation-based) per `ADR-001`_.

    .. _ADR-001:
       https://github.com/kaif-ai-engineer/forge-kaif/blob/main/docs/ADR-001.md
    """

    def __init__(self) -> None:
        self._modules: dict[str, ForgeModule] = {}
        self._types: dict[type[ForgeModule], ForgeModule] = {}

    def register(self, module: ForgeModule, *, replace: bool = False) -> None:
        """
        Register a module instance.

        Parameters
        ----------
        module:
            An instantiated ``ForgeModule`` subclass.
        replace:
            If ``True``, silently replace an existing registration
            with the same name.  Otherwise raise
            ``ModuleRegistrationError`` on conflict.

        Raises
        ------
        ModuleRegistrationError
            When a module with the same name is already registered and
            ``replace`` is ``False``.
        """
        if module.name in self._modules and not replace:
            raise ModuleRegistrationError(
                f"A module named '{module.name}' is already registered. "
                f"Use ``replace=True`` to override."
            )
        self._modules[module.name] = module
        self._types[type(module)] = module
        module._transition(ModuleLifecycleState.REGISTERED)

    def get(self, module_type: type[ForgeModule]) -> ForgeModule:
        """
        Retrieve a registered module by its class.

        Parameters
        ----------
        module_type:
            The class of the registered module.

        Returns
        -------
        ForgeModule
            The registered module instance.

        Raises
        ------
        ModuleNotFoundError
            When no module of the requested type is registered.
        """
        module = self._types.get(module_type)
        if module is None:
            raise ModuleNotFoundError(f"No module of type '{module_type.__name__}' is registered.")
        return module

    def get_by_name(self, name: str) -> ForgeModule:
        """
        Retrieve a registered module by its name.

        Parameters
        ----------
        name:
            The unique module name.

        Returns
        -------
        ForgeModule
            The registered module instance.

        Raises
        ------
        ModuleNotFoundError
            When no module with the given name is registered.
        """
        module = self._modules.get(name)
        if module is None:
            raise ModuleNotFoundError(
                f"No module named '{name}' is registered. "
                f"Available: {', '.join(sorted(self._modules))}"
            )
        return module

    def has(self, module_type: type[ForgeModule]) -> bool:
        """Return ``True`` if a module of the given type is registered."""
        return module_type in self._types

    def has_name(self, name: str) -> bool:
        """Return ``True`` if a module with the given name is registered."""
        return name in self._modules

    @property
    def registered(self) -> list[ForgeModule]:
        """Return all registered module instances."""
        return list(self._modules.values())

    # ------------------------------------------------------------------
    # Topological sort
    # ------------------------------------------------------------------

    def _build_dependency_graph(self) -> dict[str, set[str]]:
        graph: dict[str, set[str]] = {}
        for mod in self._modules.values():
            graph.setdefault(mod.name, set())
            for dep in mod.dependencies:
                if dep not in self._modules:
                    from forge.core.exceptions import ModuleNotFoundError

                    raise ModuleNotFoundError(
                        f"Module '{mod.name}' depends on '{dep}' but '{dep}' is not registered."
                    )
                graph[mod.name].add(dep)
        # Ensure every node is in the graph
        for name in self._modules:
            graph.setdefault(name, set())
        return graph

    def _detect_cycles(self, graph: dict[str, set[str]]) -> None:
        white, gray, black = 0, 1, 2
        color: dict[str, int] = dict.fromkeys(graph, white)
        parent: dict[str, str | None] = dict.fromkeys(graph)

        def dfs(node: str) -> None:
            color[node] = gray
            for neighbour in graph.get(node, set()):
                if color[neighbour] == gray:
                    cycle: list[str] = []
                    cur: str | None = node
                    while cur is not None:
                        cycle.append(cur)
                        if cur == neighbour:
                            break
                        cur = parent[cur]
                    cycle.reverse()
                    raise CircularDependencyError(
                        f"Circular dependency detected: {' -> '.join(cycle)} -> {neighbour}"
                    )
                if color[neighbour] == white:
                    parent[neighbour] = node
                    dfs(neighbour)
            color[node] = black

        for node in graph:
            if color[node] == white:
                dfs(node)

    def _topological_sort(self) -> list[ForgeModule]:
        graph = self._build_dependency_graph()
        self._detect_cycles(graph)

        in_degree: dict[str, int] = dict.fromkeys(graph, 0)
        for node, deps in graph.items():
            for _dep in deps:
                in_degree[node] = in_degree.get(node, 0) + 1

        queue: deque[str] = deque()
        for node, degree in in_degree.items():
            if degree == 0:
                queue.append(node)

        sorted_names: list[str] = []
        while queue:
            node = queue.popleft()
            sorted_names.append(node)
            for neighbour, deps in graph.items():
                if node in deps:
                    in_degree[neighbour] -= 1
                    if in_degree[neighbour] == 0:
                        queue.append(neighbour)

        if len(sorted_names) != len(graph):
            raise CircularDependencyError(
                "Could not resolve module initialisation order due to a circular dependency."
            )

        return [self._modules[name] for name in sorted_names]

    def initialization_order(self) -> list[ForgeModule]:
        """
        Return modules sorted by dependency order (topological sort).

        Modules with no dependencies appear first; modules that depend
        on others appear after their dependencies.
        """
        return self._topological_sort()

Attributes

registered property

registered: list[ForgeModule]

Return all registered module instances.

Methods:

get

get(module_type: type[ForgeModule]) -> ForgeModule

Retrieve a registered module by its class.

Parameters

module_type: The class of the registered module.

Returns

ForgeModule The registered module instance.

Raises

ModuleNotFoundError When no module of the requested type is registered.

Source code in src/forge/core/container.py

def get(self, module_type: type[ForgeModule]) -> ForgeModule:
    """
    Retrieve a registered module by its class.

    Parameters
    ----------
    module_type:
        The class of the registered module.

    Returns
    -------
    ForgeModule
        The registered module instance.

    Raises
    ------
    ModuleNotFoundError
        When no module of the requested type is registered.
    """
    module = self._types.get(module_type)
    if module is None:
        raise ModuleNotFoundError(f"No module of type '{module_type.__name__}' is registered.")
    return module

get_by_name

get_by_name(name: str) -> ForgeModule

Retrieve a registered module by its name.

Parameters

name: The unique module name.

Returns

ForgeModule The registered module instance.

Raises

ModuleNotFoundError When no module with the given name is registered.

Source code in src/forge/core/container.py

def get_by_name(self, name: str) -> ForgeModule:
    """
    Retrieve a registered module by its name.

    Parameters
    ----------
    name:
        The unique module name.

    Returns
    -------
    ForgeModule
        The registered module instance.

    Raises
    ------
    ModuleNotFoundError
        When no module with the given name is registered.
    """
    module = self._modules.get(name)
    if module is None:
        raise ModuleNotFoundError(
            f"No module named '{name}' is registered. "
            f"Available: {', '.join(sorted(self._modules))}"
        )
    return module

has

has(module_type: type[ForgeModule]) -> bool

Return True if a module of the given type is registered.

Source code in src/forge/core/container.py

def has(self, module_type: type[ForgeModule]) -> bool:
    """Return ``True`` if a module of the given type is registered."""
    return module_type in self._types

has_name

has_name(name: str) -> bool

Return True if a module with the given name is registered.

Source code in src/forge/core/container.py

def has_name(self, name: str) -> bool:
    """Return ``True`` if a module with the given name is registered."""
    return name in self._modules

initialization_order

initialization_order() -> list[ForgeModule]

Return modules sorted by dependency order (topological sort).

Modules with no dependencies appear first; modules that depend on others appear after their dependencies.

Source code in src/forge/core/container.py

def initialization_order(self) -> list[ForgeModule]:
    """
    Return modules sorted by dependency order (topological sort).

    Modules with no dependencies appear first; modules that depend
    on others appear after their dependencies.
    """
    return self._topological_sort()

register

register(module: ForgeModule, *, replace: bool = False) -> None

Register a module instance.

Parameters

module: An instantiated ForgeModule subclass. replace: If True, silently replace an existing registration with the same name. Otherwise raise ModuleRegistrationError on conflict.

Raises

ModuleRegistrationError When a module with the same name is already registered and replace is False.

Source code in src/forge/core/container.py

def register(self, module: ForgeModule, *, replace: bool = False) -> None:
    """
    Register a module instance.

    Parameters
    ----------
    module:
        An instantiated ``ForgeModule`` subclass.
    replace:
        If ``True``, silently replace an existing registration
        with the same name.  Otherwise raise
        ``ModuleRegistrationError`` on conflict.

    Raises
    ------
    ModuleRegistrationError
        When a module with the same name is already registered and
        ``replace`` is ``False``.
    """
    if module.name in self._modules and not replace:
        raise ModuleRegistrationError(
            f"A module named '{module.name}' is already registered. "
            f"Use ``replace=True`` to override."
        )
    self._modules[module.name] = module
    self._types[type(module)] = module
    module._transition(ModuleLifecycleState.REGISTERED)

EventBus

In-process async event bus for decoupled module communication.

Supports subscribing with :meth:on / :meth:off and emitting events with :meth:emit. All handlers are awaited concurrently and exceptions are caught and logged so a single failing handler never blocks other subscribers.

Source code in src/forge/core/events.py

class EventBus:
    """
    In-process async event bus for decoupled module communication.

    Supports subscribing with :meth:`on` / :meth:`off` and emitting
    events with :meth:`emit`.  All handlers are awaited concurrently
    and exceptions are caught and logged so a single failing handler
    never blocks other subscribers.
    """

    def __init__(self) -> None:
        self._handlers: dict[str, set[Handler]] = {}

    def on(self, event: str, handler: Handler) -> None:
        """
        Register an async handler for *event*.

        Parameters
        ----------
        event:
            Dot-delimited event name (e.g. ``"runtime.ready"``).
        handler:
            An async callable that will receive the event payload.
        """
        self._handlers.setdefault(event, set()).add(handler)

    def off(self, event: str, handler: Handler) -> None:
        """
        Unregister a previously registered handler.

        Raises
        ------
        EventError
            If the handler was not registered for *event*.
        """
        handlers = self._handlers.get(event)
        if handlers is None or handler not in handlers:
            raise EventError(f"Handler {handler!r} is not registered for event '{event}'.")
        handlers.discard(handler)
        if not handlers:
            del self._handlers[event]

    async def emit(self, event: str, **kwargs: Any) -> None:
        """
        Emit *event* and await all registered handlers concurrently.

        Parameters
        ----------
        event:
            Dot-delimited event name.
        **kwargs:
            Payload passed as keyword arguments to each handler.
        """
        handlers = self._handlers.get(event, set())
        if not handlers:
            return

        results = await asyncio.gather(
            *(self._safe_dispatch(handler, event, kwargs) for handler in handlers),
            return_exceptions=True,
        )
        for exc in results:
            if isinstance(exc, BaseException):
                _log.warning(
                    "Handler for '%s' raised %s: %s",
                    event,
                    type(exc).__name__,
                    exc,
                )

    async def _safe_dispatch(
        self,
        handler: Handler,
        event: str,
        kwargs: dict[str, Any],
    ) -> None:
        try:
            await handler(**kwargs)
        except Exception:
            _log.exception("Unhandled exception in handler for '%s'", event)
            raise

    def has_listeners(self, event: str) -> bool:
        """Return ``True`` if at least one handler is registered for *event*."""
        return event in self._handlers and bool(self._handlers[event])

Methods:

emit async

emit(event: str, **kwargs: Any) -> None

Emit event and await all registered handlers concurrently.

Parameters

event: Dot-delimited event name. **kwargs: Payload passed as keyword arguments to each handler.

Source code in src/forge/core/events.py

async def emit(self, event: str, **kwargs: Any) -> None:
    """
    Emit *event* and await all registered handlers concurrently.

    Parameters
    ----------
    event:
        Dot-delimited event name.
    **kwargs:
        Payload passed as keyword arguments to each handler.
    """
    handlers = self._handlers.get(event, set())
    if not handlers:
        return

    results = await asyncio.gather(
        *(self._safe_dispatch(handler, event, kwargs) for handler in handlers),
        return_exceptions=True,
    )
    for exc in results:
        if isinstance(exc, BaseException):
            _log.warning(
                "Handler for '%s' raised %s: %s",
                event,
                type(exc).__name__,
                exc,
            )

has_listeners

has_listeners(event: str) -> bool

Return True if at least one handler is registered for event.

Source code in src/forge/core/events.py

def has_listeners(self, event: str) -> bool:
    """Return ``True`` if at least one handler is registered for *event*."""
    return event in self._handlers and bool(self._handlers[event])

off

off(event: str, handler: Handler) -> None

Unregister a previously registered handler.

Raises

EventError If the handler was not registered for event.

Source code in src/forge/core/events.py

def off(self, event: str, handler: Handler) -> None:
    """
    Unregister a previously registered handler.

    Raises
    ------
    EventError
        If the handler was not registered for *event*.
    """
    handlers = self._handlers.get(event)
    if handlers is None or handler not in handlers:
        raise EventError(f"Handler {handler!r} is not registered for event '{event}'.")
    handlers.discard(handler)
    if not handlers:
        del self._handlers[event]

on

on(event: str, handler: Handler) -> None

Register an async handler for event.

Parameters

event: Dot-delimited event name (e.g. "runtime.ready"). handler: An async callable that will receive the event payload.

Source code in src/forge/core/events.py

def on(self, event: str, handler: Handler) -> None:
    """
    Register an async handler for *event*.

    Parameters
    ----------
    event:
        Dot-delimited event name (e.g. ``"runtime.ready"``).
    handler:
        An async callable that will receive the event payload.
    """
    self._handlers.setdefault(event, set()).add(handler)

EventError

Bases: ForgeError

Raised when an event bus operation fails.

Source code in src/forge/core/exceptions.py

class EventError(ForgeError):
    """Raised when an event bus operation fails."""

ForgeError

Bases: Exception

Base exception for all forge runtime errors.

Source code in src/forge/core/exceptions.py

class ForgeError(Exception):
    """Base exception for all forge runtime errors."""

ForgeModule

Bases: ABC

Base interface for all forge modules.

Every module in the framework must implement this contract. Subclasses must provide a unique name and may optionally declare dependencies, setup, teardown, and health_check.

Source code in src/forge/core/module.py

class ForgeModule(ABC):
    """
    Base interface for all forge modules.

    Every module in the framework must implement this contract.
    Subclasses must provide a unique ``name`` and may optionally declare
    ``dependencies``, ``setup``, ``teardown``, and ``health_check``.
    """

    @property
    @abstractmethod
    def name(self) -> str:
        """Unique module identifier used for DI resolution."""

    def __init__(self) -> None:
        self._lifecycle_state: ModuleLifecycleState = ModuleLifecycleState.UNREGISTERED

    @property
    def dependencies(self) -> list[str]:
        """
        Names of modules this module depends on.

        The runtime guarantees that all dependencies are initialised
        before ``setup`` is called on this module.
        """
        return []

    async def setup(self, runtime: Runtime) -> None:
        """
        Called once during runtime initialisation.

        Order is guaranteed to follow a topological sort of the
        dependency graph declared across all registered modules.
        """

    async def teardown(self) -> None:
        """
        Called during graceful shutdown.

        Implementations should release any held resources.
        """

    def health_check(self) -> HealthResult:
        """
        Optional health check registered automatically by the runtime.

        Return ``HealthResult.ok()`` by default.
        """
        return HealthResult.ok()

    # ------------------------------------------------------------------
    # Internal lifecycle helpers (used by the runtime)
    # ------------------------------------------------------------------

    def _transition(self, target: ModuleLifecycleState) -> None:
        _valid: dict[ModuleLifecycleState, set[ModuleLifecycleState]] = {
            ModuleLifecycleState.UNREGISTERED: {ModuleLifecycleState.REGISTERED},
            ModuleLifecycleState.REGISTERED: {ModuleLifecycleState.INITIALIZING},
            ModuleLifecycleState.INITIALIZING: {ModuleLifecycleState.READY},
            ModuleLifecycleState.READY: {ModuleLifecycleState.TEARDOWN},
            ModuleLifecycleState.TEARDOWN: {ModuleLifecycleState.STOPPED},
        }
        allowed = _valid.get(self._lifecycle_state, set())
        if target not in allowed:
            from forge.core.exceptions import ModuleStateError

            raise ModuleStateError(
                f"Cannot transition {self.name} from "
                f"{self._lifecycle_state.value} to {target.value}"
            )
        self._lifecycle_state = target

Attributes

dependencies property

dependencies: list[str]

Names of modules this module depends on.

The runtime guarantees that all dependencies are initialised before setup is called on this module.

name abstractmethod property

name: str

Unique module identifier used for DI resolution.

Methods:

health_check

health_check() -> HealthResult

Optional health check registered automatically by the runtime.

Return HealthResult.ok() by default.

Source code in src/forge/core/module.py

def health_check(self) -> HealthResult:
    """
    Optional health check registered automatically by the runtime.

    Return ``HealthResult.ok()`` by default.
    """
    return HealthResult.ok()

setup async

setup(runtime: ForgeRuntime) -> None

Called once during runtime initialisation.

Order is guaranteed to follow a topological sort of the dependency graph declared across all registered modules.

Source code in src/forge/core/module.py

async def setup(self, runtime: Runtime) -> None:
    """
    Called once during runtime initialisation.

    Order is guaranteed to follow a topological sort of the
    dependency graph declared across all registered modules.
    """

teardown async

teardown() -> None

Called during graceful shutdown.

Implementations should release any held resources.

Source code in src/forge/core/module.py

async def teardown(self) -> None:
    """
    Called during graceful shutdown.

    Implementations should release any held resources.
    """

ForgeRuntime

Central runtime coordinator for the forge framework.

Manages module lifecycle, dependency injection, event dispatch, and graceful shutdown.

Usage::

runtime = ForgeRuntime()
runtime.register(ConfigModule())
await runtime.init()
# ... application code ...
await runtime.teardown()

Source code in src/forge/core/runtime.py

class ForgeRuntime:
    """
    Central runtime coordinator for the forge framework.

    Manages module lifecycle, dependency injection, event dispatch,
    and graceful shutdown.

    Usage::

        runtime = ForgeRuntime()
        runtime.register(ConfigModule())
        await runtime.init()
        # ... application code ...
        await runtime.teardown()
    """

    _active: ClassVar[ForgeRuntime | None] = None

    @classmethod
    def get_active(cls) -> ForgeRuntime:
        """
        Return the active initialized runtime instance.

        Raises
        ------
        RuntimeNotInitializedError
            When no runtime is currently active/initialized.
        """
        if cls._active is None or not cls._active.is_initialized:
            from forge.core.exceptions import RuntimeNotInitializedError

            raise RuntimeNotInitializedError("Runtime is not initialized.")
        return cls._active

    def __init__(self) -> None:
        self._state = _RuntimeState.IDLE
        self._shutdown_timeout: float = 30.0
        self._container = Container()
        self._events = EventBus()
        self._ready_hooks: list[ShutdownHook] = []
        self._shutdown_hooks: list[ShutdownHook] = []
        self._loop: asyncio.AbstractEventLoop | None = None

    # ------------------------------------------------------------------
    # Public API
    # ------------------------------------------------------------------

    @property
    def events(self) -> EventBus:
        """The runtime event bus."""
        return self._events

    @property
    def container(self) -> Container:
        """The dependency injection container."""
        return self._container

    @property
    def is_initialized(self) -> bool:
        """Return ``True`` after :meth:`init` completes successfully."""
        return self._state is _RuntimeState.READY

    @property
    def is_shutting_down(self) -> bool:
        """Return ``True`` during graceful shutdown."""
        return self._state is _RuntimeState.TEARING_DOWN

    # ------------------------------------------------------------------
    # Module registration
    # ------------------------------------------------------------------

    def register(self, module: ForgeModule, *, replace: bool = False) -> Self:
        """
        Register a module with the runtime.

        Parameters
        ----------
        module:
            An instantiated ``ForgeModule`` subclass.
        replace:
            If ``True``, silently replace an existing registration.

        Returns
        -------
        Self
            For method chaining.
        """
        self._container.register(module, replace=replace)
        return self

    def use_defaults(self) -> Self:
        """
        Register all default modules (config, log, retry, health, cache).

        This is a convenience method for standard applications.
        For custom configurations, register modules individually.

        Returns
        -------
        Self
            For method chaining.
        """
        from forge.config.module import ConfigModule
        from forge.log.module import LogModule
        from forge.retry.module import RetryModule

        self.register(ConfigModule())
        self.register(LogModule())
        self.register(RetryModule())

        try:
            from forge.health.module import HealthModule

            self.register(HealthModule())
        except ImportError:
            pass

        try:
            from forge.cache.module import CacheModule

            self.register(CacheModule())
        except ImportError:
            pass

        try:
            from forge.validation.module import ValidationModule

            self.register(ValidationModule())
        except ImportError:
            pass

        try:
            from forge.jobs.module import JobsModule

            self.register(JobsModule())
        except ImportError:
            pass

        try:
            from forge.ai.module import AIModule

            self.register(AIModule())
        except ImportError:
            pass

        try:
            from forge.featureflags.module import FeatureFlagsModule

            self.register(FeatureFlagsModule())
        except ImportError:
            pass

        try:
            from forge.storage.module import StorageModule

            self.register(StorageModule())
        except ImportError:
            pass

        try:
            from forge.events.module import EventsModule

            self.register(EventsModule())
        except ImportError:
            pass

        try:
            from forge.crud.module import CrudModule

            self.register(CrudModule())
        except ImportError:
            pass

        return self

    def get(self, module_type: type[ForgeModule]) -> ForgeModule:
        """
        Retrieve a registered module by its class.

        Raises ``ModuleNotFoundError`` if not registered.
        """
        return self._container.get(module_type)

    # ------------------------------------------------------------------
    # Lifecycle hooks
    # ------------------------------------------------------------------

    def on_ready(self, callback: ShutdownHook) -> None:
        """
        Register a callback invoked after all modules are initialised.

        The callback must be an async function.
        """
        self._ready_hooks.append(callback)

    def on_shutdown(self, callback: ShutdownHook) -> None:
        """
        Register a callback invoked before teardown begins.

        The callback must be an async function.
        """
        self._shutdown_hooks.append(callback)

    # ------------------------------------------------------------------
    # Initialisation & teardown
    # ------------------------------------------------------------------

    async def init(self, shutdown_timeout: float = 30.0) -> None:
        """
        Initialise all registered modules in dependency order.

        Parameters
        ----------
        shutdown_timeout:
            Seconds to wait for graceful shutdown before forcible stop.

        Raises
        ------
        ForgeError
            If the runtime is already initialized or a module fails to set up.
        """
        if self._state is not _RuntimeState.IDLE:
            raise ForgeError(f"Runtime cannot be initialised in state {self._state.value}.")

        self._state = _RuntimeState.INITIALIZING
        self._shutdown_timeout = shutdown_timeout
        self._loop = asyncio.get_running_loop()

        self._install_signal_handlers()

        modules = self._container.initialization_order()

        for module in modules:
            try:
                module._transition(ModuleLifecycleState.INITIALIZING)
                await module.setup(self)
                module._transition(ModuleLifecycleState.READY)
            except ModuleStateError:
                raise
            except Exception as exc:
                raise ForgeError(f"Failed to initialise module '{module.name}': {exc}") from exc

        await self._events.emit("runtime.ready")
        for hook in self._ready_hooks:
            await hook()

        self._state = _RuntimeState.READY
        ForgeRuntime._active = self

    async def teardown(self) -> None:
        """
        Shut down all modules in reverse dependency order.

        Safe to call multiple times; subsequent calls are no-ops.
        """
        if self._state in (_RuntimeState.STOPPED, _RuntimeState.TEARING_DOWN):
            return
        self._state = _RuntimeState.TEARING_DOWN

        await self._events.emit("runtime.shutdown")
        for hook in self._shutdown_hooks:
            try:
                await hook()
            except Exception:
                _log.exception("Shutdown hook failed")

        modules = list(reversed(self._container.initialization_order()))

        for module in modules:
            if module._lifecycle_state not in (
                ModuleLifecycleState.READY,
                ModuleLifecycleState.INITIALIZING,
            ):
                continue
            try:
                module._transition(ModuleLifecycleState.TEARDOWN)
                await module.teardown()
                module._transition(ModuleLifecycleState.STOPPED)
            except Exception:
                _log.exception("Error tearing down module '%s'", module.name)

        self._state = _RuntimeState.STOPPED
        if ForgeRuntime._active is self:
            ForgeRuntime._active = None

    # ------------------------------------------------------------------
    # Signal handling
    # ------------------------------------------------------------------

    def _install_signal_handlers(self) -> None:
        if self._loop is None:
            return
        try:

            def _make_handler(sig: signal.Signals) -> Callable[[], None]:
                def _handler() -> None:
                    self._handle_signal(sig)

                return _handler

            for sig in (signal.SIGTERM, signal.SIGINT):
                self._loop.add_signal_handler(sig, _make_handler(sig))
        except (NotImplementedError, ValueError):
            _log.warning("Signal handlers are not available on this platform.")

    def _handle_signal(self, sig: signal.Signals) -> None:
        _log.info("Received signal %s, initiating graceful shutdown...", sig.name)
        if self._loop is not None and self._state is _RuntimeState.READY:
            asyncio.ensure_future(self._shutdown_with_timeout(), loop=self._loop)  # noqa: RUF006

    async def _shutdown_with_timeout(self) -> None:
        try:
            await asyncio.wait_for(
                self.teardown(),
                timeout=self._shutdown_timeout,
            )
        except TimeoutError:
            _log.warning(
                "Shutdown timed out after %.1f seconds.",
                self._shutdown_timeout,
            )
        except Exception:
            _log.exception("Unexpected error during shutdown")

Attributes

container property

container: Container

The dependency injection container.

events property

events: EventBus

The runtime event bus.

is_initialized property

is_initialized: bool

Return True after :meth:init completes successfully.

is_shutting_down property

is_shutting_down: bool

Return True during graceful shutdown.

Methods:

get

get(module_type: type[ForgeModule]) -> ForgeModule

Retrieve a registered module by its class.

Raises ModuleNotFoundError if not registered.

Source code in src/forge/core/runtime.py

def get(self, module_type: type[ForgeModule]) -> ForgeModule:
    """
    Retrieve a registered module by its class.

    Raises ``ModuleNotFoundError`` if not registered.
    """
    return self._container.get(module_type)

get_active classmethod

get_active() -> ForgeRuntime

Return the active initialized runtime instance.

Raises

RuntimeNotInitializedError When no runtime is currently active/initialized.

Source code in src/forge/core/runtime.py

@classmethod
def get_active(cls) -> ForgeRuntime:
    """
    Return the active initialized runtime instance.

    Raises
    ------
    RuntimeNotInitializedError
        When no runtime is currently active/initialized.
    """
    if cls._active is None or not cls._active.is_initialized:
        from forge.core.exceptions import RuntimeNotInitializedError

        raise RuntimeNotInitializedError("Runtime is not initialized.")
    return cls._active

init async

init(shutdown_timeout: float = 30.0) -> None

Initialise all registered modules in dependency order.

Parameters

shutdown_timeout: Seconds to wait for graceful shutdown before forcible stop.

Raises

ForgeError If the runtime is already initialized or a module fails to set up.

Source code in src/forge/core/runtime.py

async def init(self, shutdown_timeout: float = 30.0) -> None:
    """
    Initialise all registered modules in dependency order.

    Parameters
    ----------
    shutdown_timeout:
        Seconds to wait for graceful shutdown before forcible stop.

    Raises
    ------
    ForgeError
        If the runtime is already initialized or a module fails to set up.
    """
    if self._state is not _RuntimeState.IDLE:
        raise ForgeError(f"Runtime cannot be initialised in state {self._state.value}.")

    self._state = _RuntimeState.INITIALIZING
    self._shutdown_timeout = shutdown_timeout
    self._loop = asyncio.get_running_loop()

    self._install_signal_handlers()

    modules = self._container.initialization_order()

    for module in modules:
        try:
            module._transition(ModuleLifecycleState.INITIALIZING)
            await module.setup(self)
            module._transition(ModuleLifecycleState.READY)
        except ModuleStateError:
            raise
        except Exception as exc:
            raise ForgeError(f"Failed to initialise module '{module.name}': {exc}") from exc

    await self._events.emit("runtime.ready")
    for hook in self._ready_hooks:
        await hook()

    self._state = _RuntimeState.READY
    ForgeRuntime._active = self

on_ready

on_ready(callback: ShutdownHook) -> None

Register a callback invoked after all modules are initialised.

The callback must be an async function.

Source code in src/forge/core/runtime.py

def on_ready(self, callback: ShutdownHook) -> None:
    """
    Register a callback invoked after all modules are initialised.

    The callback must be an async function.
    """
    self._ready_hooks.append(callback)

on_shutdown

on_shutdown(callback: ShutdownHook) -> None

Register a callback invoked before teardown begins.

The callback must be an async function.

Source code in src/forge/core/runtime.py

def on_shutdown(self, callback: ShutdownHook) -> None:
    """
    Register a callback invoked before teardown begins.

    The callback must be an async function.
    """
    self._shutdown_hooks.append(callback)

register

register(module: ForgeModule, *, replace: bool = False) -> Self

Register a module with the runtime.

Parameters

module: An instantiated ForgeModule subclass. replace: If True, silently replace an existing registration.

Returns

Self For method chaining.

Source code in src/forge/core/runtime.py

def register(self, module: ForgeModule, *, replace: bool = False) -> Self:
    """
    Register a module with the runtime.

    Parameters
    ----------
    module:
        An instantiated ``ForgeModule`` subclass.
    replace:
        If ``True``, silently replace an existing registration.

    Returns
    -------
    Self
        For method chaining.
    """
    self._container.register(module, replace=replace)
    return self

teardown async

teardown() -> None

Shut down all modules in reverse dependency order.

Safe to call multiple times; subsequent calls are no-ops.

Source code in src/forge/core/runtime.py

async def teardown(self) -> None:
    """
    Shut down all modules in reverse dependency order.

    Safe to call multiple times; subsequent calls are no-ops.
    """
    if self._state in (_RuntimeState.STOPPED, _RuntimeState.TEARING_DOWN):
        return
    self._state = _RuntimeState.TEARING_DOWN

    await self._events.emit("runtime.shutdown")
    for hook in self._shutdown_hooks:
        try:
            await hook()
        except Exception:
            _log.exception("Shutdown hook failed")

    modules = list(reversed(self._container.initialization_order()))

    for module in modules:
        if module._lifecycle_state not in (
            ModuleLifecycleState.READY,
            ModuleLifecycleState.INITIALIZING,
        ):
            continue
        try:
            module._transition(ModuleLifecycleState.TEARDOWN)
            await module.teardown()
            module._transition(ModuleLifecycleState.STOPPED)
        except Exception:
            _log.exception("Error tearing down module '%s'", module.name)

    self._state = _RuntimeState.STOPPED
    if ForgeRuntime._active is self:
        ForgeRuntime._active = None

use_defaults

use_defaults() -> Self

Register all default modules (config, log, retry, health, cache).

This is a convenience method for standard applications. For custom configurations, register modules individually.

Returns

Self For method chaining.

Source code in src/forge/core/runtime.py

def use_defaults(self) -> Self:
    """
    Register all default modules (config, log, retry, health, cache).

    This is a convenience method for standard applications.
    For custom configurations, register modules individually.

    Returns
    -------
    Self
        For method chaining.
    """
    from forge.config.module import ConfigModule
    from forge.log.module import LogModule
    from forge.retry.module import RetryModule

    self.register(ConfigModule())
    self.register(LogModule())
    self.register(RetryModule())

    try:
        from forge.health.module import HealthModule

        self.register(HealthModule())
    except ImportError:
        pass

    try:
        from forge.cache.module import CacheModule

        self.register(CacheModule())
    except ImportError:
        pass

    try:
        from forge.validation.module import ValidationModule

        self.register(ValidationModule())
    except ImportError:
        pass

    try:
        from forge.jobs.module import JobsModule

        self.register(JobsModule())
    except ImportError:
        pass

    try:
        from forge.ai.module import AIModule

        self.register(AIModule())
    except ImportError:
        pass

    try:
        from forge.featureflags.module import FeatureFlagsModule

        self.register(FeatureFlagsModule())
    except ImportError:
        pass

    try:
        from forge.storage.module import StorageModule

        self.register(StorageModule())
    except ImportError:
        pass

    try:
        from forge.events.module import EventsModule

        self.register(EventsModule())
    except ImportError:
        pass

    try:
        from forge.crud.module import CrudModule

        self.register(CrudModule())
    except ImportError:
        pass

    return self

HealthCheckError

Bases: ForgeError

Raised when a health check fails critically.

Source code in src/forge/core/exceptions.py

class HealthCheckError(ForgeError):
    """Raised when a health check fails critically."""

HealthResult

Result of a single health check.

Source code in src/forge/core/module.py

class HealthResult:
    """Result of a single health check."""

    OK = "ok"
    DEGRADED = "degraded"
    ERROR = "error"

    def __init__(self, status: str = OK, message: str | None = None) -> None:
        self.status = status
        self.message = message

    @staticmethod
    def ok() -> HealthResult:
        return HealthResult(HealthResult.OK)

    @staticmethod
    def degraded(message: str = "") -> HealthResult:
        return HealthResult(HealthResult.DEGRADED, message)

    @staticmethod
    def error(message: str = "") -> HealthResult:
        return HealthResult(HealthResult.ERROR, message)

ModuleError

Bases: ForgeError

Base exception for module-related errors.

Source code in src/forge/core/exceptions.py

class ModuleError(ForgeError):
    """Base exception for module-related errors."""

ModuleLifecycleState

Bases: Enum

Lifecycle states for a forge module.

Source code in src/forge/core/module.py

class ModuleLifecycleState(enum.Enum):
    """Lifecycle states for a forge module."""

    UNREGISTERED = "UNREGISTERED"
    REGISTERED = "REGISTERED"
    INITIALIZING = "INITIALIZING"
    READY = "READY"
    TEARDOWN = "TEARDOWN"
    STOPPED = "STOPPED"

ModuleNotFoundError

Bases: ModuleError

Raised when a requested module is not registered.

Source code in src/forge/core/exceptions.py

class ModuleNotFoundError(ModuleError):
    """Raised when a requested module is not registered."""

ModuleRegistrationError

Bases: ModuleError

Raised when module registration fails (e.g., duplicate registration).

Source code in src/forge/core/exceptions.py

class ModuleRegistrationError(ModuleError):
    """Raised when module registration fails (e.g., duplicate registration)."""

ModuleStateError

Bases: ModuleError

Raised when a module operation is invalid for its current lifecycle state.

Source code in src/forge/core/exceptions.py

class ModuleStateError(ModuleError):
    """Raised when a module operation is invalid for its current lifecycle state."""

RuntimeNotInitializedError

Bases: ForgeError

Raised when an operation requires an initialized runtime but it is not.

Source code in src/forge/core/exceptions.py

class RuntimeNotInitializedError(ForgeError):
    """Raised when an operation requires an initialized runtime but it is not."""

TraceContext

Async context manager that sets trace ID, span ID, and baggage.

Usage::

async with TraceContext(trace_id="abc") as ctx:
    print(ctx.trace_id)

Source code in src/forge/core/context.py

class TraceContext:
    """
    Async context manager that sets trace ID, span ID, and baggage.

    Usage::

        async with TraceContext(trace_id="abc") as ctx:
            print(ctx.trace_id)
    """

    def __init__(
        self,
        trace_id: str | None = None,
        span_id: str | None = None,
        baggage: dict[str, Any] | None = None,
    ) -> None:
        self._trace_id = trace_id
        self._span_id = span_id
        self._baggage = baggage or {}
        self._trace_token: Token[str] | None = None
        self._span_token: Token[str] | None = None
        self._baggage_token: Token[dict[str, Any]] | None = None

    def __enter__(self) -> Self:
        self._trace_token = set_trace_id(self._trace_id)
        self._span_token = set_span_id(self._span_id)
        merged = dict(_baggage.get())
        merged.update(self._baggage)
        self._baggage_token = _baggage.set(merged)
        return self

    def __exit__(self, *args: object) -> None:
        if self._trace_token is not None:
            reset_trace_id(self._trace_token)
        if self._span_token is not None:
            reset_span_id(self._span_token)
        if self._baggage_token is not None:
            reset_baggage(self._baggage_token)

    @property
    def trace_id(self) -> str:
        return get_trace_id()

    @property
    def span_id(self) -> str:
        return get_span_id()

Functions:

generate_span_id

generate_span_id() -> str

Generate a new span ID (UUID4 hex string).

Source code in src/forge/core/context.py

def generate_span_id() -> str:
    """Generate a new span ID (UUID4 hex string)."""
    return uuid.uuid4().hex[:16]

generate_trace_id

generate_trace_id() -> str

Generate a new trace ID (UUID4 hex string).

Source code in src/forge/core/context.py

def generate_trace_id() -> str:
    """Generate a new trace ID (UUID4 hex string)."""
    return uuid.uuid4().hex

get_baggage

get_baggage() -> dict[str, Any]

Return a copy of the current baggage (mutable key-value context).

Source code in src/forge/core/context.py

def get_baggage() -> dict[str, Any]:
    """Return a copy of the current baggage (mutable key-value context)."""
    return dict(_baggage.get())

get_baggage_item

get_baggage_item(key: str, default: Any = None) -> Any

Return a single baggage item, or default if missing.

Source code in src/forge/core/context.py

def get_baggage_item(key: str, default: Any = None) -> Any:
    """Return a single baggage item, or *default* if missing."""
    return _baggage.get().get(key, default)

get_meter

get_meter(name: str = 'forge', version: str = '') -> Any

Return an OTel Meter, or None if OTel is not installed.

Source code in src/forge/core/otel.py

def get_meter(name: str = "forge", version: str = "") -> Any:
    """Return an OTel ``Meter``, or ``None`` if OTel is not installed."""
    if not _OTEL_AVAILABLE:
        return None
    if not _INITIALIZED:
        init_otel()
    return cast("Any", _otel_metrics).get_meter(name, version or None)

get_span_id

get_span_id() -> str

Return the current span ID, or "" if none is set.

Source code in src/forge/core/context.py

def get_span_id() -> str:
    """Return the current span ID, or ``""`` if none is set."""
    return _current_span_id.get()

get_trace_id

get_trace_id() -> str

Return the current trace ID, or "" if none is set.

Source code in src/forge/core/context.py

def get_trace_id() -> str:
    """Return the current trace ID, or ``""`` if none is set."""
    return _current_trace_id.get()

get_tracer

get_tracer(name: str = 'forge', version: str = '') -> Any

Return an OTel Tracer, or None if OTel is not installed.

Source code in src/forge/core/otel.py

def get_tracer(name: str = "forge", version: str = "") -> Any:
    """Return an OTel ``Tracer``, or ``None`` if OTel is not installed."""
    if not _OTEL_AVAILABLE:
        return None
    if not _INITIALIZED:
        init_otel()
    return cast("Any", _otel_trace).get_tracer(name, version or None)

in_span

in_span(name: str, attributes: dict[str, Any] | None = None, tracer: Any = None) -> Generator[Any, None, None]

Context manager that creates an OTel span when OTel is available.

Yields the span (or None when OTel is not installed).

Source code in src/forge/core/otel.py

@contextmanager
def in_span(
    name: str,
    attributes: dict[str, Any] | None = None,
    tracer: Any = None,
) -> Generator[Any, None, None]:
    """
    Context manager that creates an OTel span when OTel is available.

    Yields the span (or ``None`` when OTel is not installed).
    """
    tr = tracer or get_tracer()
    if tr is None:
        yield None
        return

    with tr.start_as_current_span(name) as span:
        if attributes:
            span.set_attributes(attributes)
        yield span

init_otel

init_otel() -> None

Initialise OTel providers for internal collection only (no exporters).

Source code in src/forge/core/otel.py

def init_otel() -> None:
    """Initialise OTel providers for internal collection only (no exporters)."""
    global _INITIALIZED  # noqa: PLW0603
    if _INITIALIZED or not _OTEL_AVAILABLE:
        return

    _SdkTracerProvider()
    _otel_trace.set_tracer_provider(_SdkTracerProvider())

    _SdkMeterProvider()
    _otel_metrics.set_meter_provider(_SdkMeterProvider())

    _INITIALIZED = True
    _logger.debug("OpenTelemetry SDK initialised (internal collection, no exporters)")

is_otel_available

is_otel_available() -> bool

Return True if the optional OTel packages are installed.

Source code in src/forge/core/otel.py

def is_otel_available() -> bool:
    """Return ``True`` if the optional OTel packages are installed."""
    return _OTEL_AVAILABLE

reset_baggage

reset_baggage(token: Token[dict[str, Any]]) -> None

Restore baggage to the value it had before token was created.

Source code in src/forge/core/context.py

def reset_baggage(token: Token[dict[str, Any]]) -> None:
    """Restore baggage to the value it had before *token* was created."""
    _baggage.reset(token)

reset_span_id

reset_span_id(token: Token[str]) -> None

Restore the span ID to the value it had before token was created.

Source code in src/forge/core/context.py

def reset_span_id(token: Token[str]) -> None:
    """Restore the span ID to the value it had before *token* was created."""
    _current_span_id.reset(token)

reset_trace_id

reset_trace_id(token: Token[str]) -> None

Restore the trace ID to the value it had before token was created.

Source code in src/forge/core/context.py

def reset_trace_id(token: Token[str]) -> None:
    """Restore the trace ID to the value it had before *token* was created."""
    _current_trace_id.reset(token)

set_baggage

set_baggage(key: str, value: Any) -> None

Set a baggage key for the current async context.

Source code in src/forge/core/context.py

def set_baggage(key: str, value: Any) -> None:
    """Set a baggage key for the current async context."""
    current = dict(_baggage.get())
    current[key] = value
    _baggage.set(current)

set_span_id

set_span_id(span_id: str | None = None) -> Token[str]

Set the current span ID.

Parameters

span_id: The span ID to use. If None, a new one is generated.

Source code in src/forge/core/context.py

def set_span_id(span_id: str | None = None) -> Token[str]:
    """
    Set the current span ID.

    Parameters
    ----------
    span_id:
        The span ID to use.  If ``None``, a new one is generated.
    """
    return _current_span_id.set(span_id or generate_span_id())

set_trace_id

set_trace_id(trace_id: str | None = None) -> Token[str]

Set the current trace ID.

Parameters

trace_id: The trace ID to use. If None, a new one is generated.

Returns

Token[str] A token that can be used with reset_trace_id to restore the previous value.

Source code in src/forge/core/context.py

def set_trace_id(trace_id: str | None = None) -> Token[str]:
    """
    Set the current trace ID.

    Parameters
    ----------
    trace_id:
        The trace ID to use.  If ``None``, a new one is generated.

    Returns
    -------
    Token[str]
        A token that can be used with ``reset_trace_id`` to restore
        the previous value.
    """
    return _current_trace_id.set(trace_id or generate_trace_id())