Index

diapason

Diapason — modular AI assistant backend with composable intelligence primitives.

Classes

Diapason

Diapason(
    *,
    config: Optional[DiapasonConfig] = None,
    config_path: Optional[str] = None,
    engine_key: Optional[str] = None,
    model: Optional[str] = None,
)

High-level Diapason SDK.

Usage::

from diapason import Diapason

with Diapason() as j:
    response = j.ask("Hello, what can you do?")
    print(response)

# Streaming:
import asyncio

async def main():
    j = Diapason()
    async for token in j.ask_stream("Tell me a joke"):
        print(token, end="", flush=True)
    j.close()

asyncio.run(main())

# Or without context manager:
j = Diapason()
response = j.ask("Hello")
j.close()

Source code in src/diapason/sdk.py

def __init__(
    self,
    *,
    config: Optional[DiapasonConfig] = None,
    config_path: Optional[str] = None,
    engine_key: Optional[str] = None,
    model: Optional[str] = None,
) -> None:
    if config is not None:
        self._config = config
    elif config_path is not None:
        self._config = load_config(Path(config_path))
    else:
        self._config = load_config()

    self._engine_key = engine_key
    self._model_override = model
    self._engine: Any = None
    self._energy_monitor: Any = None
    self._resolved_engine_key: Optional[str] = None
    self._bus = EventBus()
    self._telem_store: Optional[TelemetryStore] = None
    self._audit_logger: Any = None
    self._capability_policy: Any = None
    self.memory = MemoryHandle(self._config)

    # Set up telemetry
    if self._config.telemetry.enabled:
        try:
            self._telem_store = TelemetryStore(self._config.telemetry.db_path)
            self._telem_store.subscribe_to_bus(self._bus)
        except Exception as exc:
            logger.warning("Failed to initialize telemetry store: %s", exc)

Attributes

config property

config: DiapasonConfig

Return the active configuration.

version property

version: str

Return the Diapason version string.

Methods:

ask

ask(
    query: str,
    *,
    model: Optional[str] = None,
    agent: Optional[str] = None,
    tools: Optional[List[str]] = None,
    temperature: Optional[float] = None,
    max_tokens: Optional[int] = None,
    context: bool = True,
    channel: Optional[Any] = None,
) -> str

Send a query and return the response text.

Source code in src/diapason/sdk.py

def ask(
    self,
    query: str,
    *,
    model: Optional[str] = None,
    agent: Optional[str] = None,
    tools: Optional[List[str]] = None,
    temperature: Optional[float] = None,
    max_tokens: Optional[int] = None,
    context: bool = True,
    channel: Optional[Any] = None,
) -> str:
    """Send a query and return the response text."""
    result = self.ask_full(
        query,
        model=model,
        agent=agent,
        tools=tools,
        temperature=temperature,
        max_tokens=max_tokens,
        context=context,
        channel=channel,
    )
    return result["content"]

ask_full

ask_full(
    query: str,
    *,
    model: Optional[str] = None,
    agent: Optional[str] = None,
    tools: Optional[List[str]] = None,
    temperature: Optional[float] = None,
    max_tokens: Optional[int] = None,
    context: bool = True,
    channel: Optional[Any] = None,
) -> Dict[str, Any]

Send a query and return the full result dict.

Returns a dict with keys: content, usage, tool_results (if agent mode).

Source code in src/diapason/sdk.py

def ask_full(
    self,
    query: str,
    *,
    model: Optional[str] = None,
    agent: Optional[str] = None,
    tools: Optional[List[str]] = None,
    temperature: Optional[float] = None,
    max_tokens: Optional[int] = None,
    context: bool = True,
    channel: Optional[Any] = None,
) -> Dict[str, Any]:
    """Send a query and return the full result dict.

    Returns a dict with keys: content, usage, tool_results (if agent mode).
    """
    self._ensure_engine()
    if temperature is None:
        temperature = self._config.intelligence.temperature
    if max_tokens is None:
        max_tokens = self._config.intelligence.max_tokens

    model_name = model or self._model_override

    # Resolve model via router if not specified
    if model_name is None:
        model_name = self._resolve_model(query)

    if not model_name:
        models = self._engine.list_models()
        model_name = models[0] if models else "default"

    # Agent mode
    if agent is not None:
        return self._run_agent(
            agent,
            query,
            model_name,
            tools=tools or [],
            temperature=temperature,
            max_tokens=max_tokens,
            context=context,
            channel=channel,
        )

    # Direct engine mode
    messages = [Message(role=Role.USER, content=query)]

    # Memory context injection
    if context and self._config.agent.context_from_memory:
        messages = self._inject_context(query, messages)

    # InstrumentedEngine handles telemetry + energy recording
    result = self._engine.generate(
        messages,
        model=model_name,
        temperature=temperature,
        max_tokens=max_tokens,
    )

    return {
        "content": result.get("content", ""),
        "usage": result.get("usage", {}),
        "model": model_name,
        "engine": self._resolved_engine_key,
    }

ask_stream async

ask_stream(
    query: str,
    *,
    model: Optional[str] = None,
    temperature: Optional[float] = None,
    max_tokens: Optional[int] = None,
    context: bool = True,
) -> AsyncIterator[str]

Stream tokens as they are generated. Yields token strings.

Source code in src/diapason/sdk.py

async def ask_stream(
    self,
    query: str,
    *,
    model: Optional[str] = None,
    temperature: Optional[float] = None,
    max_tokens: Optional[int] = None,
    context: bool = True,
) -> AsyncIterator[str]:
    """Stream tokens as they are generated. Yields token strings."""
    self._ensure_engine()
    if temperature is None:
        temperature = self._config.intelligence.temperature
    if max_tokens is None:
        max_tokens = self._config.intelligence.max_tokens

    model_name = model or self._model_override

    if model_name is None:
        model_name = self._resolve_model(query)

    if not model_name:
        models = self._engine.list_models()
        model_name = models[0] if models else "default"

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

    if context and self._config.agent.context_from_memory:
        messages = self._inject_context(query, messages)

    async for token in self._engine.stream(
        messages,
        model=model_name,
        temperature=temperature,
        max_tokens=max_tokens,
    ):
        yield token

ask_full_stream async

ask_full_stream(
    query: str,
    *,
    model: Optional[str] = None,
    temperature: Optional[float] = None,
    max_tokens: Optional[int] = None,
    context: bool = True,
) -> AsyncIterator[Dict[str, Any]]

Stream token dicts with metadata.

Yields dicts with token and index keys for each token. The final dict has done: True along with the full concatenated content, model, and engine keys.

Source code in src/diapason/sdk.py

async def ask_full_stream(
    self,
    query: str,
    *,
    model: Optional[str] = None,
    temperature: Optional[float] = None,
    max_tokens: Optional[int] = None,
    context: bool = True,
) -> AsyncIterator[Dict[str, Any]]:
    """Stream token dicts with metadata.

    Yields dicts with ``token`` and ``index`` keys for each token.
    The final dict has ``done: True`` along with the full concatenated
    ``content``, ``model``, and ``engine`` keys.
    """
    self._ensure_engine()
    if temperature is None:
        temperature = self._config.intelligence.temperature
    if max_tokens is None:
        max_tokens = self._config.intelligence.max_tokens

    model_name = model or self._model_override

    if model_name is None:
        model_name = self._resolve_model(query)

    if not model_name:
        models = self._engine.list_models()
        model_name = models[0] if models else "default"

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

    if context and self._config.agent.context_from_memory:
        messages = self._inject_context(query, messages)

    parts: List[str] = []
    i = 0
    async for token in self._engine.stream(
        messages,
        model=model_name,
        temperature=temperature,
        max_tokens=max_tokens,
    ):
        parts.append(token)
        yield {"token": token, "index": i}
        i += 1

    yield {
        "done": True,
        "content": "".join(parts),
        "model": model_name,
        "engine": self._resolved_engine_key,
    }

list_models

list_models() -> List[str]

Return a list of available model identifiers.

Source code in src/diapason/sdk.py

def list_models(self) -> List[str]:
    """Return a list of available model identifiers."""
    self._ensure_engine()
    return self._engine.list_models()

list_engines

list_engines() -> List[str]

Return a list of registered engine keys.

Source code in src/diapason/sdk.py

def list_engines(self) -> List[str]:
    """Return a list of registered engine keys."""
    from diapason.core.registry import EngineRegistry

    return list(EngineRegistry.keys())

close

close() -> None

Release all resources.

Source code in src/diapason/sdk.py

def close(self) -> None:
    """Release all resources."""
    self.memory.close()
    if self._energy_monitor is not None:
        try:
            self._energy_monitor.close()
        except Exception as exc:
            logger.debug("Error closing energy monitor: %s", exc)
        self._energy_monitor = None
    if self._telem_store is not None:
        try:
            self._telem_store.close()
        except Exception as exc:
            logger.debug("Error closing telemetry store: %s", exc)
        self._telem_store = None
    if self._audit_logger is not None:
        try:
            self._audit_logger.close()
        except Exception as exc:
            logger.debug("Error closing audit logger: %s", exc)
        self._audit_logger = None
    self._engine = None

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()

MemoryHandle

MemoryHandle(config: DiapasonConfig)

Proxy for memory operations. Lazily initializes backend.

Source code in src/diapason/sdk.py

def __init__(self, config: DiapasonConfig) -> None:
    self._config = config
    self._backend: Any = None

Methods:

index

index(
    path: str,
    *,
    chunk_size: int = 512,
    chunk_overlap: int = 64,
) -> Dict[str, Any]

Index a file or directory into memory.

Source code in src/diapason/sdk.py

def index(
    self,
    path: str,
    *,
    chunk_size: int = 512,
    chunk_overlap: int = 64,
) -> Dict[str, Any]:
    """Index a file or directory into memory."""
    from diapason.tools.storage.chunking import ChunkConfig
    from diapason.tools.storage.ingest import ingest_path

    backend = self._get_backend()
    cfg = ChunkConfig(chunk_size=chunk_size, chunk_overlap=chunk_overlap)
    chunks = ingest_path(Path(path), config=cfg)

    doc_ids: List[str] = []
    for chunk in chunks:
        doc_id = backend.store(
            chunk.content,
            source=chunk.source,
            metadata={"index": chunk.index},
        )
        doc_ids.append(doc_id)

    return {
        "chunks": len(chunks),
        "doc_ids": doc_ids,
        "path": path,
    }

search

search(
    query: str, *, top_k: int = 5
) -> List[Dict[str, Any]]

Search memory for relevant chunks.

Source code in src/diapason/sdk.py

def search(self, query: str, *, top_k: int = 5) -> List[Dict[str, Any]]:
    """Search memory for relevant chunks."""
    backend = self._get_backend()
    results = backend.retrieve(query, top_k=top_k)
    return [
        {
            "content": r.content,
            "score": r.score,
            "source": r.source,
            "metadata": r.metadata,
        }
        for r in results
    ]

stats

stats() -> Dict[str, Any]

Return memory backend statistics.

Source code in src/diapason/sdk.py

def stats(self) -> Dict[str, Any]:
    """Return memory backend statistics."""
    backend = self._get_backend()
    if hasattr(backend, "count"):
        return {
            "count": backend.count(),
            "backend": self._config.memory.default_backend,
        }
    return {"backend": self._config.memory.default_backend}

close

close() -> None

Release the memory backend.

Source code in src/diapason/sdk.py

def close(self) -> None:
    """Release the memory backend."""
    if self._backend is not None:
        if hasattr(self._backend, "close"):
            self._backend.close()
        self._backend = None

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