Index

system

Top-level system composition: DiapasonSystem, SystemBuilder, and helpers.

Classes

SystemBuilder

SystemBuilder(
    config: Optional[DiapasonConfig] = None,
    *,
    config_path: Optional[Any] = None,
)

Config-driven fluent builder for DiapasonSystem.

Source code in src/diapason/system/builder.py

def __init__(
    self,
    config: Optional[DiapasonConfig] = None,
    *,
    config_path: Optional[Any] = None,
) -> None:
    if config is not None:
        self._config = config
    elif config_path is not None:
        from pathlib import Path

        self._config = load_config(Path(config_path))
    else:
        self._config = load_config()

    self._engine_key: Optional[str] = None
    self._engine_instance: Optional[InferenceEngine] = None
    self._engine_instance_key: Optional[str] = None
    self._model: Optional[str] = None
    self._agent_name: Optional[str] = None
    self._tool_names: Optional[List[str]] = None
    self._telemetry: Optional[bool] = None
    self._traces: Optional[bool] = None
    self._bus: Optional[EventBus] = None
    self._sandbox: Optional[bool] = None
    self._scheduler: Optional[bool] = None
    self._workflow: Optional[bool] = None
    self._sessions: Optional[bool] = None
    self._speech: Optional[bool] = None
    self._mcp_clients: List = []

Methods:

engine_instance

engine_instance(
    engine: InferenceEngine, key: str = "openai-compat"
) -> SystemBuilder

Inject a pre-built engine instance, bypassing engine discovery.

Used by callers that must target one exact endpoint (e.g. diapason eval --base-url). build() health-checks the instance and raises a loud error if it is unreachable — it never silently substitutes a different discovered engine.

Source code in src/diapason/system/builder.py

def engine_instance(
    self, engine: InferenceEngine, key: str = "openai-compat"
) -> SystemBuilder:
    """Inject a pre-built engine instance, bypassing engine discovery.

    Used by callers that must target one exact endpoint (e.g.
    ``diapason eval --base-url``). ``build()`` health-checks the instance
    and raises a loud error if it is unreachable — it never silently
    substitutes a different discovered engine.
    """
    self._engine_instance = engine
    self._engine_instance_key = key
    return self

build

build() -> DiapasonSystem

Construct a fully wired DiapasonSystem.

Source code in src/diapason/system/builder.py

def build(self) -> DiapasonSystem:
    """Construct a fully wired DiapasonSystem."""
    config = self._config
    bus = self._bus or get_event_bus()

    engine, engine_key = self._resolve_engine(config)
    model = self._resolve_model(config, engine)

    telemetry_enabled = (
        self._telemetry if self._telemetry is not None else config.telemetry.enabled
    )
    traces_enabled = (
        self._traces if self._traces is not None else config.traces.enabled
    )
    config.traces.enabled = traces_enabled
    gpu_monitor = None
    energy_monitor = None
    if telemetry_enabled and config.telemetry.gpu_metrics:
        try:
            from diapason.telemetry.energy_monitor import (
                create_energy_monitor,
            )

            energy_monitor = create_energy_monitor(
                poll_interval_ms=config.telemetry.gpu_poll_interval_ms,
                prefer_vendor=config.telemetry.energy_vendor or None,
            )
        except ImportError:
            pass

        if energy_monitor is None:
            try:
                from diapason.telemetry.gpu_monitor import GpuMonitor

                if GpuMonitor.available():
                    gpu_monitor = GpuMonitor(
                        poll_interval_ms=config.telemetry.gpu_poll_interval_ms,
                    )
            except ImportError:
                pass

    from diapason.security import setup_security

    sec = setup_security(config, engine, bus)
    engine = sec.engine

    if telemetry_enabled:
        from diapason.telemetry.instrumented_engine import (
            InstrumentedEngine,
        )

        engine = InstrumentedEngine(
            engine,
            bus,
            gpu_monitor=gpu_monitor,
            energy_monitor=energy_monitor,
        )

    telemetry_store = None
    if telemetry_enabled:
        telemetry_store = self._setup_telemetry(config, bus)

    memory_backend = self._resolve_memory(config)
    channel_backend = self._resolve_channel(config, bus)
    tool_list = self._resolve_tools(
        config,
        engine,
        model,
        memory_backend,
        channel_backend,
    )
    executor_security = {
        "capability_policy": sec.capability_policy,
        "boundary_guard": sec.boundary_guard,
        "rate_limiter": sec.rate_limiter,
    }
    tool_executor = (
        ToolExecutor(tool_list, bus, **executor_security) if tool_list else None
    )

    skill_manager = None
    skill_few_shot_examples: List[str] = []
    if config.skills.enabled:
        try:
            from pathlib import Path

            from diapason.skills.manager import SkillManager

            skill_manager = SkillManager(
                bus, capability_policy=sec.capability_policy
            )
            skill_paths = [Path(config.skills.skills_dir).expanduser()]
            workspace_skills = Path("./skills")
            if workspace_skills.exists():
                skill_paths.insert(0, workspace_skills)
            skill_manager.discover(paths=skill_paths)
            if tool_executor:
                skill_manager.set_tool_executor(tool_executor)
            skill_tools = skill_manager.get_skill_tools(
                tool_executor=tool_executor,
            )
            tool_list.extend(skill_tools)
            if tool_list:
                tool_executor = ToolExecutor(tool_list, bus, **executor_security)
            skill_few_shot_examples = skill_manager.get_few_shot_examples()
        except Exception as exc:
            logger.warning("Failed to initialize skills: %s", exc)

    agent_name = self._agent_name or config.agent.default_agent
    container_runner = self._setup_sandbox(config)
    scheduler_store, task_scheduler = self._setup_scheduler(config, bus)
    workflow_engine = self._setup_workflow(config, bus)
    session_store = self._setup_sessions(config)

    trace_store = None
    if traces_enabled:
        try:
            from diapason.traces.store import TraceStore

            trace_store = TraceStore(config.traces.db_path)
        except Exception:
            logger.warning("Failed to initialize TraceStore", exc_info=True)

    capability_policy = sec.capability_policy
    learning_orchestrator = self._setup_learning_orchestrator(config)

    agent_manager = None
    if config.agent_manager.enabled:
        try:
            from diapason.agents.manager import AgentManager

            am_db = config.agent_manager.db_path or str(
                get_config_dir() / "agents.db"
            )
            agent_manager = AgentManager(db_path=am_db)
        except Exception as exc:
            logger.warning("Failed to initialize agent manager: %s", exc)

    agent_executor = None
    agent_scheduler = None
    if agent_manager is not None:
        try:
            from diapason.agents.executor import AgentExecutor
            from diapason.agents.scheduler import AgentScheduler

            _trace_store = None
            if config.traces.enabled:
                try:
                    from diapason.traces.store import TraceStore

                    _trace_store = TraceStore(config.traces.db_path)
                except Exception:
                    logger.warning(
                        "Failed to initialize TraceStore",
                        exc_info=True,
                    )

            agent_executor = AgentExecutor(
                manager=agent_manager,
                event_bus=bus,
                trace_store=_trace_store,
            )
            agent_scheduler = AgentScheduler(
                manager=agent_manager,
                executor=agent_executor,
            )
        except Exception:
            logger.warning("Failed to initialize agent scheduler", exc_info=True)

    speech_backend = None
    speech_enabled = self._speech if self._speech is not None else True
    if speech_enabled:
        try:
            from diapason.speech._discovery import get_speech_backend

            speech_backend = get_speech_backend(config)
        except Exception as exc:
            logger.warning("Failed to initialize speech backend: %s", exc)

    system = DiapasonSystem(
        config=config,
        bus=bus,
        engine=engine,
        engine_key=engine_key,
        model=model,
        agent_name=agent_name,
        tools=tool_list,
        tool_executor=tool_executor,
        memory_backend=memory_backend,
        channel_backend=channel_backend,
        telemetry_store=telemetry_store,
        trace_store=trace_store,
        gpu_monitor=gpu_monitor,
        scheduler_store=scheduler_store,
        scheduler=task_scheduler,
        container_runner=container_runner,
        workflow_engine=workflow_engine,
        session_store=session_store,
        capability_policy=capability_policy,
        audit_logger=sec.audit_logger,
        boundary_guard=sec.boundary_guard,
        rate_limiter=sec.rate_limiter,
        agent_manager=agent_manager,
        agent_scheduler=agent_scheduler,
        agent_executor=agent_executor,
        speech_backend=speech_backend,
        skill_manager=skill_manager,
    )
    system._learning_orchestrator = learning_orchestrator
    system._skill_few_shot_examples = skill_few_shot_examples
    system._mcp_clients = list(getattr(self, "_mcp_clients", []))
    if system.agent_executor is not None:
        system.agent_executor.set_system(system)
    if task_scheduler is not None:
        task_scheduler._system = system  # noqa: SLF001
        try:
            from diapason.heartbeat.sync import sync_heartbeat_and_routines

            if config.heartbeat.enabled or config.routines.enabled:
                sync_heartbeat_and_routines(task_scheduler, config)
        except Exception:
            logger.debug("heartbeat/routines sync skipped", exc_info=True)
    return system

AgentRuntime dataclass

AgentRuntime(
    agent: Optional[BaseAgent] = None,
    agent_name: str = "",
    manager: Optional[AgentManager] = None,
    scheduler: Optional[AgentScheduler] = None,
    executor: Optional[AgentExecutor] = None,
)

Active agent and agent lifecycle managers.

Observability dataclass

Observability(
    telemetry_store: Optional[TelemetryStore] = None,
    trace_store: Optional[TraceStore] = None,
    trace_collector: Optional[TraceCollector] = None,
    gpu_monitor: Optional[GpuMonitor] = None,
)

Telemetry, traces, and hardware monitoring.

Scheduling dataclass

Scheduling(
    store: Optional[SchedulerStore] = None,
    runner: Optional[TaskScheduler] = None,
)

Task scheduler and its persistent store.

SecurityContext dataclass

SecurityContext(
    capability_policy: Optional[CapabilityPolicy] = None,
    audit_logger: Optional[AuditLogger] = None,
    boundary_guard: Optional[BoundaryGuard] = None,
    rate_limiter: Any = None,
)

Security policy, audit, and boundary enforcement.

DiapasonSystem dataclass

DiapasonSystem(
    config: DiapasonConfig,
    bus: EventBus,
    engine: InferenceEngine,
    engine_key: str,
    model: str,
    agent: Optional[BaseAgent] = None,
    agent_name: str = "",
    tools: List[BaseTool] = list(),
    tool_executor: Optional[ToolExecutor] = None,
    memory_backend: Optional[MemoryBackend] = None,
    channel_backend: Optional[BaseChannel] = None,
    router: Optional[RouterPolicy] = None,
    mcp_server: Optional[MCPServer] = None,
    telemetry_store: Optional[TelemetryStore] = None,
    trace_store: Optional[TraceStore] = None,
    trace_collector: Optional[TraceCollector] = None,
    gpu_monitor: Optional[GpuMonitor] = None,
    scheduler_store: Optional[SchedulerStore] = None,
    scheduler: Optional[TaskScheduler] = None,
    container_runner: Optional[ContainerRunner] = None,
    workflow_engine: Optional[WorkflowEngine] = None,
    session_store: Optional[SessionStore] = None,
    capability_policy: Optional[CapabilityPolicy] = None,
    audit_logger: Optional[AuditLogger] = None,
    boundary_guard: Optional[BoundaryGuard] = None,
    rate_limiter: Any = None,
    operator_manager: Optional[OperatorManager] = None,
    agent_manager: Optional[AgentManager] = None,
    agent_scheduler: Optional[AgentScheduler] = None,
    agent_executor: Optional[AgentExecutor] = None,
    speech_backend: Optional[SpeechBackend] = None,
    skill_manager: Optional[SkillManager] = None,
    _learning_orchestrator: Optional[
        LearningOrchestrator
    ] = None,
    _mcp_clients: List[MCPClient] = list(),
)

Fully wired system -- the single source of truth for primitive composition.

Methods:

act

act(command: str) -> Dict[str, Any]

Execute an explicit low-risk desktop command without inference.

Unlike ask(), this method deliberately authorizes the deterministic action path. Ambiguous or disallowed commands still fall back to the normal configured assistant.

Source code in src/diapason/system/core.py

def act(self, command: str) -> Dict[str, Any]:
    """Execute an explicit low-risk desktop command without inference.

    Unlike ``ask()``, this method deliberately authorizes the deterministic
    action path. Ambiguous or disallowed commands still fall back to the
    normal configured assistant.
    """
    return self.ask(command, context=False, action_mode="auto")

wire_channel

wire_channel(channel_bridge: Any) -> None

Register a message handler on channel_bridge that routes every incoming message through this system (agent or engine) and replies.

Sessions are isolated per "<channel>:<conversation_id>" key so each chat retains its own history.

PARAMETER	DESCRIPTION
channel_bridge	A connected :class:~diapason.channels._stubs.BaseChannel instance whose on_message method accepts a callable. TYPE: Any

Source code in src/diapason/system/core.py

def wire_channel(self, channel_bridge: Any) -> None:
    """Register a message handler on *channel_bridge* that routes every
    incoming message through this system (agent or engine) and replies.

    Sessions are isolated per ``"<channel>:<conversation_id>"`` key so
    each chat retains its own history.

    Parameters
    ----------
    channel_bridge:
        A connected :class:`~diapason.channels._stubs.BaseChannel`
        instance whose ``on_message`` method accepts a callable.
    """
    from diapason.core.types import Message
    from diapason.sessions.session import SessionStore

    if self.session_store is None:
        from pathlib import Path

        self.session_store = SessionStore(
            db_path=Path(self.config.sessions.db_path).expanduser(),
            max_age_hours=self.config.sessions.max_age_hours,
            consolidation_threshold=self.config.sessions.consolidation_threshold,
        )

    _system = self  # capture for closure

    def _on_channel_message(cm) -> None:
        session_key = f"{cm.channel}:{cm.conversation_id}"
        session = _system.session_store.get_or_create(
            session_key,
            channel=cm.channel,
            channel_user_id=cm.sender,
        )

        prior_msgs: List[Message] = []
        for sm in session.messages:
            try:
                role = Role(sm.role)
            except ValueError:
                role = Role.USER
            prior_msgs.append(Message(role=role, content=sm.content))

        reply = ""
        try:
            if _system.agent_name and _system.agent_name != "none":
                result = _system.ask(
                    cm.content,
                    context=False,
                    agent=_system.agent_name,
                    prior_messages=prior_msgs,
                )
                reply = result.get("content", "")
            else:
                result = _system.ask(
                    cm.content,
                    context=False,
                    prior_messages=prior_msgs,
                )
                reply = result.get("content", "")
        except Exception:
            logger.exception("Channel message handler error")
            reply = "Sorry, I encountered an error processing your message."

        try:
            _system.session_store.save_message(
                session.session_id,
                "user",
                cm.content,
                channel=cm.channel,
            )
            _system.session_store.save_message(
                session.session_id,
                "assistant",
                reply,
                channel=cm.channel,
            )
        except Exception:
            logger.debug("Session save error", exc_info=True)

        if reply:
            try:
                # Canonical channel send contract (see BaseChannel.send):
                # the first positional arg is the DESTINATION id, and the
                # `conversation_id=` kwarg is the inbound message id used as
                # a reply/thread reference.  ``cm.conversation_id`` holds the
                # real per-adapter destination (Discord/Slack channel id,
                # Telegram chat id, ...) while ``cm.channel`` is only the
                # channel TYPE label ("discord", "telegram", ...).  Passing
                # the type label as the destination produced HTTP 400s
                # (#515) and using the channel id as a reply reference
                # produced MESSAGE_REFERENCE_UNKNOWN_MESSAGE (#516).
                channel_bridge.send(
                    cm.conversation_id,
                    reply,
                    conversation_id=getattr(cm, "message_id", ""),
                )
            except Exception:
                logger.exception("Channel send error")

    channel_bridge.on_message(_on_channel_message)

close

close() -> None

Release resources.

Source code in src/diapason/system/core.py

def close(self) -> None:
    """Release resources."""
    if self.scheduler and hasattr(self.scheduler, "stop"):
        self.scheduler.stop()
    for resource in (
        self.scheduler_store,
        self.engine,
        self.gpu_monitor,
        self.telemetry_store,
        self.trace_store,
        self.memory_backend,
        self.session_store,
        self.channel_backend,
        self.workflow_engine,
        self.container_runner,
    ):
        if resource and hasattr(resource, "close"):
            resource.close()
    if self.agent_manager is not None:
        self.agent_manager.close()
    if self.agent_scheduler is not None:
        self.agent_scheduler.stop()
    self._close_mcp_clients()

QueryOrchestrator

QueryOrchestrator(system: OrchestratorDeps)

Source code in src/diapason/system/orchestrator.py

def __init__(self, system: OrchestratorDeps) -> None:
    self._system = system

Methods:

ask

ask(
    query: str,
    *,
    context: bool = True,
    temperature: Optional[float] = None,
    max_tokens: Optional[int] = None,
    agent: Optional[str] = None,
    tools: Optional[List[str]] = None,
    system_prompt: Optional[str] = None,
    operator_id: Optional[str] = None,
    prior_messages: Optional[List[Message]] = None,
    action_mode: str = "off",
) -> Dict[str, Any]

Execute a query through the system and return a result dict.

Source code in src/diapason/system/orchestrator.py

def ask(
    self,
    query: str,
    *,
    context: bool = True,
    temperature: Optional[float] = None,
    max_tokens: Optional[int] = None,
    agent: Optional[str] = None,
    tools: Optional[List[str]] = None,
    system_prompt: Optional[str] = None,
    operator_id: Optional[str] = None,
    prior_messages: Optional[List[Message]] = None,
    action_mode: str = "off",
) -> Dict[str, Any]:
    """Execute a query through the system and return a result dict."""
    s = self._system
    # CLI/SDK counterpart of the desktop chat fast path.  A caller that
    # explicitly chooses an agent, tools, a custom prompt or prior turns
    # retains the exact orchestration semantics they requested.
    if (
        action_mode == "auto"
        and not agent
        and not tools
        and system_prompt is None
        and not prior_messages
    ):
        try:
            from diapason.actions import LightningActionService

            action = LightningActionService(s.config).handle(query)
            if action.handled:
                return {
                    "content": action.message,
                    "usage": {},
                    "model": s.model,
                    "engine": "lightning",
                    "lightning": action.public_metadata(),
                }
        except Exception:
            logger.exception("Lightning action routing failed")
    if temperature is None:
        temperature = s.config.intelligence.temperature
    if max_tokens is None:
        max_tokens = s.config.intelligence.max_tokens

    messages = [Message(role=Role.USER, content=query)]

    if context and s.memory_backend and s.config.agent.context_from_memory:
        try:
            from diapason.tools.storage.context import (
                ContextConfig,
                inject_context,
            )

            ctx_cfg = ContextConfig(
                top_k=s.config.memory.context_top_k,
                min_score=s.config.memory.context_min_score,
                max_context_tokens=s.config.memory.context_max_tokens,
            )
            messages = inject_context(
                query,
                messages,
                s.memory_backend,
                config=ctx_cfg,
            )
        except Exception as exc:
            logger.warning("Failed to inject memory context: %s", exc)

    use_agent = agent or s.agent_name
    if not agent and use_agent != "none":
        detected = self._detect_agent_intent(query)
        if detected:
            use_agent = detected
    if use_agent and use_agent != "none":
        return self._run_agent(
            query,
            messages,
            use_agent,
            tools,
            temperature,
            max_tokens,
            system_prompt=system_prompt,
            operator_id=operator_id,
            prior_messages=prior_messages,
        )

    result = s.engine.generate(
        messages,
        model=s.model,
        temperature=temperature,
        max_tokens=max_tokens,
    )
    return {
        "content": result.get("content", ""),
        "usage": result.get("usage", {}),
        "model": s.model,
        "engine": s.engine_key,
    }

OrchestratorDeps

Bases: Protocol

Minimum surface of DiapasonSystem that QueryOrchestrator depends on.

Tests can satisfy this with a lightweight class — no need to construct the full DiapasonSystem dataclass or materialize every subsystem.