orchestrata_ultimate.drivers.qwen_driver

Reference › orchestrata_ultimate › drivers › qwen_driver

qwen_driver drives the Qwen Code command-line tool as an agentic coder that writes files directly into the workspace, running qwen --yolo -p - with the task on standard input. It mirrors claude_api_driver. driver_factory returns QwenDriver for the qwen and ollama providers. execute finds the binary with _find_qwen, initialises the project if no QWEN.md exists (_init_project), runs the task (_run_qwen) under a timeout (1200 seconds by default, from _get_timeout_seconds), and lists the created files with _scan_source_files, which skips virtual environments, caches, .git, the .qwen folder and the design folder. A missing binary or a timeout returns a failure result. It satisfies the AsyncCodeDriver protocol; tests are in tests/drivers/test_qwen_driver.py.

At a glance

Class diagram

Dependency diagram

API

Classes

QwenDriver

Line 112

Async code gen driver using Qwen Code CLI.

Functions

Source

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

# mypy: ignore-errors
"""Async Qwen Code CLI driver.

Uses Qwen Code as an agentic coding tool that writes files directly
to disk. Runs ``qwen --yolo -p -`` with stdin pipe.
"""

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

from orchestrata_ultimate.core.traceability import traces

_logger = logging.getLogger(__name__)

TIMEOUT_SECONDS = 1200

_SKIP_DIRS = frozenset(
    {
        ".venv",
        "__pycache__",
        ".git",
        "node_modules",
        ".qwen",
        "design",
    }
)
_SKIP_PREFIXES = ("Module:",)
_SOURCE_EXTS = frozenset(
    {
        ".py",
        ".toml",
        ".cfg",
        ".ini",
        ".yaml",
        ".yml",
    }
)


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

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

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

    Returns:
        Timeout in seconds.
    """
    env = env or os.environ
    return int(env.get("ORCHESTRATA_QWEN_TIMEOUT_SECONDS", str(TIMEOUT_SECONDS)))


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

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

    Returns:
        Path to qwen binary, or empty string if not found.
    """
    env = env or os.environ
    found = shutil.which("qwen", path=env.get("PATH"))
    if found:
        return found
    home = env.get("HOME")
    candidates: list[str] = []
    if home:
        candidates.append(os.path.join(home, ".local", "bin", "qwen"))
    candidates.extend(["/usr/bin/qwen", "/usr/local/bin/qwen"])
    for candidate in candidates:
        if os.path.isfile(candidate):
            return candidate
    return ""


def _scan_source_files(workspace: str) -> list[str]:
    """Scan workspace for Python/config source files.

    Args:
        workspace: Root directory to scan.

    Returns:
        Sorted list of relative file paths.
    """
    files: list[str] = []
    for root, dirs, filenames in os.walk(workspace):
        dirs[:] = [
            d
            for d in dirs
            if d not in _SKIP_DIRS and not any(d.startswith(p) for p in _SKIP_PREFIXES)
        ]
        for fname in filenames:
            if fname.startswith("."):
                continue
            ext = os.path.splitext(fname)[1]
            if ext in _SOURCE_EXTS:
                rel = os.path.relpath(os.path.join(root, fname), workspace)
                files.append(rel)
    return sorted(files)


class QwenDriver:
    """Async code gen driver using Qwen Code CLI.

    Runs ``qwen --yolo`` in the project directory and lets Qwen Code
    write files directly to disk. Scans workspace afterward.

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

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

        Args:
            workspace: Working directory. Created if needed.
            env: Optional environment mapping. Defaults to ``os.environ``.
        """
        self._env = env or os.environ
        self.workspace = workspace or self._env.get(
            "ORCHESTRATA_WORKSPACE", "/home/mohamed/workspace"
        )
        os.makedirs(self.workspace, exist_ok=True)

    @traces(
        spec="025",
        story="US-001",
        reqs=["REQ-025-046", "REQ-025-048", "REQ-025-050"],
    )
    async def execute(
        self,
        task: str,
        files: list[str] | None = None,
        is_fix: bool = False,
    ) -> tuple[bool, str]:
        """Generate code by running Qwen Code in the project dir.

        Args:
            task: The coding task/prompt.
            files: Optional file paths (unused — Qwen reads workspace).
            is_fix: True if this is a repair prompt.

        Returns:
            Tuple of (success, output_or_file_count).
        """
        qwen_bin = _find_qwen(self._env)
        if not qwen_bin:
            msg = "qwen CLI not found"
            _logger.error(msg)
            return False, msg

        _logger.info("Generating code in %s", self.workspace)
        await self._init_project(qwen_bin)

        timeout_seconds = _get_timeout_seconds(self._env)
        try:
            output = await asyncio.wait_for(
                self._run_qwen(qwen_bin, task),
                timeout=timeout_seconds,
            )
        except TimeoutError:
            msg = f"Timed out after {timeout_seconds}s"
            _logger.warning(msg)
            return False, msg
        except Exception as exc:
            msg = f"Error: {exc}"
            _logger.error(msg)
            return False, msg

        created = await asyncio.to_thread(_scan_source_files, self.workspace)
        if not created:
            _logger.warning("No source files created")
            return False, output or "No files generated"

        for f in created:
            _logger.info("Wrote %s", f)

        _logger.info("Generated %d files", len(created))
        return True, output or f"Created {len(created)} files"

    async def _init_project(self, qwen_bin: str) -> None:
        """Run qwen init if QWEN.md doesn't exist yet.

        Args:
            qwen_bin: Path to qwen binary.
        """
        qwen_md = os.path.join(self.workspace, "QWEN.md")
        if os.path.exists(qwen_md):
            return
        try:
            proc = await asyncio.create_subprocess_exec(
                qwen_bin,
                "init",
                "--yes",
                cwd=self.workspace,
                stdout=asyncio.subprocess.DEVNULL,
                stderr=asyncio.subprocess.DEVNULL,
            )
            await asyncio.wait_for(proc.wait(), timeout=30)
            _logger.info("Initialized project")
        except (TimeoutError, FileNotFoundError) as exc:
            _logger.debug("project init skipped: %s", exc)

    async def _run_qwen(self, qwen_bin: str, task: str) -> str:
        """Run qwen with task via stdin pipe.

        Args:
            qwen_bin: Path to qwen binary.
            task: The coding task prompt.

        Returns:
            Qwen Code stdout output.
        """
        proc = await asyncio.create_subprocess_exec(
            qwen_bin,
            "--yolo",
            "--max-session-turns",
            "50",
            "-o",
            "text",
            "-p",
            "-",
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.STDOUT,
            cwd=self.workspace,
        )
        assert proc.stdin is not None  # guaranteed by stdin=PIPE above
        assert proc.stdout is not None  # guaranteed by stdout=PIPE above

        proc.stdin.write(task.encode("utf-8"))
        await proc.stdin.drain()
        proc.stdin.close()

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

        await proc.wait()
        return "".join(output_lines)