Agent/cyber_agent/application/bootstrap.py

"""Trusted deployment bootstrap for application runtimes.

Factories are configured by the server operator, never by an API request.  Each
factory returns an object exposing ``application`` and ``services`` (or a
two-item tuple with those values).
"""

from __future__ import annotations

import importlib
import os
from typing import Any, Callable

from cyber_agent.application.runtime import (
    ApplicationRegistry,
    ApplicationRuntime,
)


APPLICATION_FACTORIES_ENV = "CYBER_AGENT_APPLICATION_FACTORIES"


def _load_factory(reference: str) -> Callable[[], Any]:
    module_name, separator, attribute = reference.strip().partition(":")
    if not separator or not module_name or not attribute:
        raise ValueError(
            "Application factory must use the trusted module:callable form"
        )
    factory = getattr(importlib.import_module(module_name), attribute, None)
    if not callable(factory):
        raise ValueError(f"Application factory is not callable: {reference}")
    return factory


def _components(value: Any) -> tuple[Any, Any]:
    if isinstance(value, tuple) and len(value) == 2:
        return value
    application = getattr(value, "application", None)
    services = getattr(value, "services", None)
    if application is None or services is None:
        raise ValueError(
            "Application factory must return (application, services) or a "
            "component object exposing both attributes"
        )
    return application, services


def build_runtime_from_environment(
    *,
    trace_store: Any,
    llm_call: Callable[..., Any],
    utility_llm_call: Callable[..., Any] | None = None,
    environ: dict[str, str] | None = None,
) -> ApplicationRuntime | None:
    """Build one shared runtime from deployment-owned factory references."""
    source = os.environ if environ is None else environ
    configured = source.get(APPLICATION_FACTORIES_ENV, "").strip()
    if not configured:
        return None
    references = [item.strip() for item in configured.split(",") if item.strip()]
    registry = ApplicationRegistry()
    for reference in references:
        application, services = _components(_load_factory(reference)())
        registry.register(application, services)
    return ApplicationRuntime(
        registry=registry,
        trace_store=trace_store,
        llm_call=llm_call,
        utility_llm_call=utility_llm_call,
    )