orchestrata_ultimate.drivers.aider_driver

Reference › orchestrata_ultimate › drivers › aider_driver

aider_driver runs the Aider command-line tool as an asynchronous subprocess with a DeepSeek backend (SPEC-025, REQ-025-035 to REQ-025-040). AiderDriver is created on a workspace directory by driver_factory when a volk's provider is aider, and satisfies the AsyncCodeDriver protocol. execute sends a task, with optional files, an is_fix flag and an optional environment mapping, to aider --message --yes-always --no-auto-commits and returns a success flag with the output. _run_process collects the output. _find_aider looks for the binary on PATH, then under the home directory's .local/bin, /usr/bin and /usr/local/bin, and returns an empty string when none is found. _get_timeout_seconds returns 300 seconds by default (REQ-025-038), overridable through the environment variable ORCHESTRATA_AIDER_TIMEOUT_SECONDS so tests can exercise the timeout branch quickly. Tests are in tests/drivers/test_aider_driver.py. The curated story is on the drivers page.

At a glance

Class diagram

Dependency diagram

API

Classes

AiderDriver

Decorators: traces(spec='025', reqs=['REQ-025-035', 'REQ-025-036', 'REQ-025-037', 'REQ-025-038', 'REQ-025-039', 'REQ-025-040']) · Line 77

Async CLI driver for Aider with DeepSeek backend.

Functions

Source

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

# mypy: ignore-errors
"""Async Aider CLI driver.

Runs ``aider --message --yes-always --no-auto-commits`` as an async
subprocess with DeepSeek backend.
"""

import asyncio
import logging
import os
import shutil
from collections.abc import Mapping

from orchestrata_ultimate.core.base.traces import traces

_logger = logging.getLogger(__name__)

TIMEOUT_SECONDS = 300


@traces(spec="025", reqs=["REQ-025-038"])
def _get_timeout_seconds(env: Mapping[str, str] | None = None) -> int:
    """Resolve the subprocess timeout, in seconds.

    Defaults to the spec-mandated ``TIMEOUT_SECONDS`` (300, REQ-025-038).
    Overridable via ``ORCHESTRATA_AIDER_TIMEOUT_SECONDS`` so the timeout
    branch can be exercised in tests without waiting out the real 300s.

    Args:
        env: Optional environment mapping. Defaults to ``os.environ``.

    Returns:
        Timeout in seconds.
    """
    if env is None:
        env = os.environ
    return int(env.get("ORCHESTRATA_AIDER_TIMEOUT_SECONDS", str(TIMEOUT_SECONDS)))


@traces(spec="025", reqs=["REQ-025-035"])
def _find_aider(env: Mapping[str, str] | None = None) -> str:
    """Find the aider CLI binary.

    Args:
        env: Optional environment mapping. Defaults to ``os.environ``.

    Returns:
        Path to aider binary, or empty string if not found.
    """
    if env is None:
        env = os.environ
    found = shutil.which("aider", path=env.get("PATH"))
    if found:
        return found
    home_aider = (
        os.path.join(env["HOME"], ".local", "bin", "aider")
        if "HOME" in env
        else os.path.expanduser("~/.local/bin/aider")
    )
    for candidate in [home_aider, "/usr/bin/aider", "/usr/local/bin/aider"]:
        if os.path.isfile(candidate):
            return candidate
    return ""


@traces(
    spec="025",
    reqs=[
        "REQ-025-035",
        "REQ-025-036",
        "REQ-025-037",
        "REQ-025-038",
        "REQ-025-039",
        "REQ-025-040",
    ],
)
class AiderDriver:
    """Async CLI driver for Aider with DeepSeek backend.

    Runs aider in non-interactive mode. Aider with DeepSeek typically
    completes in 30-60s.

    Command (REQ-025-036): aider --message <task> --yes-always --no-auto-commits

    Attributes:
        workspace: Directory where code is generated.
    """

    def __init__(self, workspace: str | None = None) -> None:
        """Initialize with target workspace directory.

        Args:
            workspace: Working directory for the subprocess.
        """
        self.workspace = workspace or os.environ.get(
            "ORCHESTRATA_WORKSPACE", "/home/mohamed/workspace"
        )
        os.makedirs(self.workspace, exist_ok=True)

    @traces(
        spec="025",
        story="US-025-001",
        reqs=["REQ-025-036", "REQ-025-037", "REQ-025-038", "REQ-025-040"],
    )
    async def execute(
        self,
        task: str,
        files: list[str] | None = None,
        is_fix: bool = False,
        env: Mapping[str, str] | None = None,
    ) -> tuple[bool, str]:
        """Send a task to Aider and return (success, output).

        Args:
            task: The prompt to send to Aider.
            files: Optional file paths to pass via --file.
            is_fix: True if this is a repair prompt.
            env: Optional environment mapping. Defaults to ``os.environ``.

        Returns:
            Tuple of (success, output_text).
        """
        if env is None:
            env = os.environ
        aider_bin = _find_aider(env)
        if not aider_bin:
            msg = "Aider CLI not found"
            _logger.error(msg)
            return False, msg

        _logger.info("Executing task in %s", self.workspace)

        cmd = [
            aider_bin,
            "--message",
            task,
            "--yes-always",
            "--no-auto-commits",
            "--no-git",
        ]

        if files:
            for f in files:
                filepath = os.path.join(self.workspace, f)
                if os.path.exists(filepath):
                    cmd.extend(["--file", filepath])

        aider_model = env.get("AIDER_MODEL", "")
        if aider_model:
            cmd.extend(["--model", aider_model])

        env = dict(env)
        deepseek_key = env.get("DEEPSEEK_API_KEY", "")
        if deepseek_key:
            env["DEEPSEEK_API_KEY"] = deepseek_key

        timeout_seconds = _get_timeout_seconds(env)
        try:
            return await asyncio.wait_for(
                self._run_process(cmd, env),
                timeout=timeout_seconds,
            )
        except TimeoutError:
            msg = f"Timed out after {timeout_seconds}s"
            _logger.warning(msg)
            return False, msg
        except FileNotFoundError:
            msg = f"Aider not found at {aider_bin}"
            _logger.error(msg)
            return False, msg

    @traces(spec="025", story="US-025-001", reqs=["REQ-025-039"])
    async def _run_process(
        self,
        cmd: list[str],
        env: dict[str, str],
    ) -> tuple[bool, str]:
        """Run subprocess and collect output.

        Args:
            cmd: Command and arguments.
            env: Environment variables.

        Returns:
            Tuple of (success, output_text).
        """
        proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.STDOUT,
            cwd=self.workspace,
            env=env,
        )
        assert proc.stdout is not None  # guaranteed by stdout=PIPE above

        output_lines: list[str] = []
        while True:
            line = await proc.stdout.readline()
            if not line:
                break
            decoded = line.decode("utf-8", errors="replace")
            _logger.debug("[Aider] %s", decoded.rstrip())
            output_lines.append(decoded)

        await proc.wait()
        output = "".join(output_lines)

        if proc.returncode != 0:
            _logger.warning("Exited with code %d", proc.returncode)
            return False, output

        _logger.info("Task completed successfully")
        return True, output