orchestrata_ultimate.drivers.driver_factory

Reference › orchestrata_ultimate › drivers › driver_factory

driver_factory creates the right asynchronous code-generation driver for a volk (SPEC-025, REQ-025-007 to REQ-025-018). It replaces the duplicated driver selection that earlier lived in the swarm, code director and junior modules; those callers (swarm coordinator, code director volk and junior volk) now call it.

get_driver takes a volk alias (default J:Junior), a workspace and an optional environment mapping. It asks llm.config.get_volk_provider for the volk's provider, then chooses: the claude and anthropic providers give ClaudeAPIDriver; deepseek and openrouter give DeepSeekCoderDriver; aider gives AiderDriver; qwen and ollama give QwenDriver; the browser providers (qwen_browser, deepseek_browser, kimi_browser) give BrowserDriver; and claude_code gives ClaudeCodeDriver. An unknown provider logs a warning and falls back to the Claude driver. The workspace defaults to the environment variable ORCHESTRATA_WORKSPACE. The _get_*_driver helpers import and build each class lazily, and _BROWSER_PROVIDER_MAP maps browser provider names to configuration keys in browser_providers. The return type is the AsyncCodeDriver protocol. Tests are in tests/drivers/test_driver_factory.py.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Constants

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/drivers/driver_factory.py

# mypy: ignore-errors
"""Driver factory — creates the correct async code gen driver.

Consolidates the duplicated _get_driver() logic from spike's
swarm.py, codedirector.py, and junior.py into one factory function.
"""

from __future__ import annotations

import logging
import os
from collections.abc import Mapping
from typing import TYPE_CHECKING

from orchestrata_ultimate.core.traceability import traces

if TYPE_CHECKING:
    from orchestrata_ultimate.drivers.aider_driver import AiderDriver
    from orchestrata_ultimate.drivers.base import AsyncCodeDriver
    from orchestrata_ultimate.drivers.browser_driver import BrowserDriver
    from orchestrata_ultimate.drivers.claude_api_driver import ClaudeAPIDriver
    from orchestrata_ultimate.drivers.claude_code_driver import ClaudeCodeDriver
    from orchestrata_ultimate.drivers.deepseek_coder_driver import DeepSeekCoderDriver
    from orchestrata_ultimate.drivers.qwen_driver import QwenDriver

_logger = logging.getLogger(__name__)


@traces(
    spec="025",
    reqs=[
        "REQ-025-007",
        "REQ-025-008",
        "REQ-025-009",
        "REQ-025-010",
        "REQ-025-011",
        "REQ-025-012",
        "REQ-025-013",
        "REQ-025-014",
        "REQ-025-015",
        "REQ-025-016",
        "REQ-025-017",
        "REQ-025-018",
    ],
)
def get_driver(
    volk: str = "J:Junior",
    workspace: str | None = None,
    env: Mapping[str, str] | None = None,
) -> AsyncCodeDriver:
    """Create an async code generation driver for the given volk.

    Selects the driver based on the volk's configured LLM provider.
    Falls back to ClaudeAPIDriver if the provider is not recognized.

    Args:
        volk: Volk alias that determines the provider (default: J:Junior).
        workspace: Working directory for the driver.
        env: Optional environment mapping; defaults to ``os.environ``.

    Returns:
        An async code generation driver instance.
    """
    from orchestrata_ultimate.llm.config import get_volk_provider  # noqa: PLC0415

    provider = get_volk_provider(volk)
    env = env or os.environ
    ws = workspace or env.get("ORCHESTRATA_WORKSPACE", "/home/mohamed/workspace")

    if provider in ("claude", "anthropic"):
        return _get_claude_driver(ws, env)
    elif provider in ("deepseek", "openrouter"):
        # DeepSeekCoderDriver is provider-agnostic despite its name: its LLM
        # transport is call_llm(volk="J:Junior"), which resolves the actual
        # provider/model from the ACTIVE PROFILE (SPIKE_PROFILE) — an
        # openrouter profile row routes the same FILE-block mechanics through
        # OpenRouter. Without this branch, openrouter fell into the
        # unknown-provider Claude fallback: codegen silently ran on the wrong
        # model family (found wiring openrouter-cheap, 2026-07-13).
        return _get_deepseek_driver(ws)
    elif provider == "aider":
        return _get_aider_driver(ws)
    elif provider in ("qwen", "ollama"):
        return _get_qwen_driver(ws, env)
    elif provider in (
        "qwen_browser",
        "deepseek_browser",
        "kimi_browser",
    ):
        return _get_browser_driver(ws, provider, env)
    elif provider == "claude_code":
        return _get_claude_code_driver(ws, env, volk=volk)
    else:
        _logger.warning(
            "Unknown provider '%s' for %s — falling back to Claude API",
            provider,
            volk,
        )
        return _get_claude_driver(ws, env)


@traces(spec="025", reqs=["REQ-025-011"])
def _get_claude_driver(
    workspace: str, env: Mapping[str, str] | None = None
) -> ClaudeAPIDriver:
    """Create a ClaudeAPIDriver.

    Args:
        workspace: Working directory.
        env: Optional environment mapping to thread through to the driver.

    Returns:
        ClaudeAPIDriver instance.
    """
    from orchestrata_ultimate.drivers.claude_api_driver import ClaudeAPIDriver

    return ClaudeAPIDriver(workspace=workspace, env=env)


def _get_claude_code_driver(
    workspace: str,
    env: Mapping[str, str] | None = None,
    volk: str | None = None,
) -> ClaudeCodeDriver:
    """Create a ClaudeCodeDriver.

    Args:
        workspace: Working directory.
        env: Optional environment mapping to thread through to the driver.
        volk: Optional volk alias to thread through for --model / persona
            resolution (SPEC-025 Phase 1).

    Returns:
        ClaudeCodeDriver instance.
    """
    from orchestrata_ultimate.drivers.claude_code_driver import ClaudeCodeDriver

    return ClaudeCodeDriver(workspace=workspace, env=env, volk=volk)


@traces(spec="025", reqs=["REQ-025-012"])
def _get_deepseek_driver(workspace: str) -> DeepSeekCoderDriver:
    """Create a DeepSeekCoderDriver.

    Args:
        workspace: Working directory.

    Returns:
        DeepSeekCoderDriver instance.
    """
    from orchestrata_ultimate.drivers.deepseek_coder_driver import (
        DeepSeekCoderDriver,
    )

    return DeepSeekCoderDriver(workspace=workspace)


@traces(spec="025", reqs=["REQ-025-013"])
def _get_aider_driver(workspace: str) -> AiderDriver:
    """Create an AiderDriver.

    Args:
        workspace: Working directory.

    Returns:
        AiderDriver instance.
    """
    from orchestrata_ultimate.drivers.aider_driver import AiderDriver

    return AiderDriver(workspace=workspace)


@traces(spec="025", reqs=["REQ-025-014"])
def _get_qwen_driver(
    workspace: str, env: Mapping[str, str] | None = None
) -> QwenDriver:
    """Create a QwenDriver.

    Args:
        workspace: Working directory.
        env: Optional environment mapping to thread through to the driver.

    Returns:
        QwenDriver instance.
    """
    from orchestrata_ultimate.drivers.qwen_driver import QwenDriver

    return QwenDriver(workspace=workspace, env=env)


_BROWSER_PROVIDER_MAP: dict[str, str] = {
    "qwen_browser": "qwen_chat",
    "deepseek_browser": "deepseek_chat",
    "kimi_browser": "kimi",
}


@traces(spec="025", reqs=["REQ-025-015", "REQ-025-016", "REQ-025-017"])
def _get_browser_driver(
    workspace: str, provider: str, env: Mapping[str, str] | None = None
) -> BrowserDriver:
    """Create a BrowserDriver for a free-tier LLM web UI.

    Args:
        workspace: Working directory.
        provider: Factory provider key (e.g. "qwen_browser").
        env: Optional environment mapping to thread through to the driver.

    Returns:
        BrowserDriver instance.
    """
    from orchestrata_ultimate.drivers.browser_driver import BrowserDriver

    browser_provider = _BROWSER_PROVIDER_MAP[provider]
    return BrowserDriver(
        workspace=workspace,
        provider=browser_provider,
        environ=dict(env) if env is not None else None,
    )