orchestrata_ultimate.agents.volks.qa._dep_install

Reference › orchestrata_ultimate › agents › volks › qa › _dep_install

_dep_install provides DepInstallMixin for the QA volk (volk): it prepares a virtual environment in which the generated project's tests can run. _get_venv_python returns the venv interpreter, creating the venv when needed, and _venv_env returns the environment for venv subprocesses with PYTHON_GIL forced to 1. _install_deps installs the workspace dependencies: it finds a requirements file or pyproject.toml (_find_requirements, _parse_pyproject_deps), scans the sources for third-party imports and installs them (_scan_and_install_imports, using _scan_source_imports, _collect_local_module_names, _top_level_import_name and _import_to_package to map import names to package names), and, when the workspace imports orchestrata, installs the runtime dependencies of the monorepo packages (_workspace_imports_orchestrata, _collect_monorepo_deps, _install_monorepo_deps). _ensure_conftest_syspath writes a conftest.py that adds the workspace and monorepo packages to the path. _validate_imports checks that generated source imports cleanly and returns the errors, and _get_import_hint extracts the failing module so the report can show its available names.

At a glance

Class diagram

Dependency diagram

API

Classes

DepInstallMixin

Bases: _MixinBase · Line 27

Virtualenv / dependency-install / import-validation helpers for U:Qa.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/agents/volks/qa/_dep_install.py

# mypy: ignore-errors
"""Virtualenv / dependency-install / import-validation helpers for U:Qa."""

import glob
import logging
import os
import re
import shutil
import subprocess
import sys
from typing import TYPE_CHECKING

from orchestrata_ultimate.agents.volks.qa._host import _QaHost
from orchestrata_ultimate.core.infra_config import REDIS_HOST, REDIS_PORT

_logger = logging.getLogger(__name__)

# W6.4b: typing-only host contract so pyright can see across mixin
# composition (see _host.py). No runtime effect — DepInstallMixin still
# inherits from plain ``object`` outside TYPE_CHECKING.
if TYPE_CHECKING:
    _MixinBase = _QaHost
else:
    _MixinBase = object


class DepInstallMixin(_MixinBase):
    """Virtualenv / dependency-install / import-validation helpers for U:Qa."""

    def _get_venv_python(self) -> str:
        """Return the venv Python path, creating the venv if needed.

        Pins to python3.13 max — 3.14 breaks MCP SDK annotation inspection.
        """
        venv_dir = os.path.join(self.workspace, ".venv")
        venv_python = os.path.join(venv_dir, "bin", "python")
        if not os.path.exists(venv_python):
            uv = shutil.which("uv")
            if uv:
                cmd = [uv, "venv", venv_dir, "--python", "3.13"]
            else:
                cmd = [sys.executable, "-m", "venv", venv_dir]
            print(f"  [U:Qa] Creating workspace venv: {' '.join(cmd)}")
            subprocess.run(
                cmd, cwd=self.workspace, capture_output=True, text=True, timeout=60
            )
            # G99: pytest installed later in _install_deps() to avoid
            # version conflicts with project requirements.txt pins
        return venv_python

    def _venv_env(self) -> dict[str, str]:
        """Return os.environ with PYTHON_GIL forced to 1 for venv subprocesses.

        The parent process may run under PYTHON_GIL=0 (free-threaded Python 3.14),
        but workspace venvs use Python 3.13 which does not support GIL disabling.
        Without this override, subprocess calls to venv_python crash with:
        "Fatal Python error: config_read_gil: Disabling the GIL is not supported
        by this build".
        """
        env = os.environ.copy()
        env["PYTHON_GIL"] = "1"
        return env

    def _pip_cmd(self, venv_python: str, uv_path: str | None = None) -> list[str]:
        """Return the pip install command for the given python.

        Args:
            venv_python: Path to the Python interpreter.
            uv_path: Optional explicit path to the ``uv`` binary. When
                omitted, ``shutil.which`` is used to locate it.
        """
        uv = uv_path if uv_path is not None else shutil.which("uv")
        if uv:
            return [uv, "pip", "install", "-p", venv_python]
        return [venv_python, "-m", "pip", "install"]

    def _find_requirements(self) -> str | None:
        """Find requirements.txt or pyproject.toml in workspace or subdirs."""
        # Check workspace root first
        for name in ["pyproject.toml", "requirements.txt"]:
            path = os.path.join(self.workspace, name)
            if os.path.exists(path):
                return path
        # Search one level of subdirectories
        for entry in os.listdir(self.workspace):
            subdir = os.path.join(self.workspace, entry)
            if os.path.isdir(subdir) and not entry.startswith("."):
                for name in ["pyproject.toml", "requirements.txt"]:
                    path = os.path.join(subdir, name)
                    if os.path.exists(path):
                        return path
        return None

    def _parse_pyproject_deps(self, pyproject_path: str) -> list[str]:
        """Extract dependency names from pyproject.toml [project.dependencies].

        G103: Parse deps directly instead of editable install to avoid
        setuptools.build_meta.build_editable failures in containers.
        """
        deps = []
        in_deps = False
        try:
            with open(pyproject_path, encoding="utf-8") as f:
                for line in f:
                    stripped = line.strip()
                    if stripped == "dependencies = [":
                        in_deps = True
                        continue
                    if in_deps:
                        if stripped == "]":
                            break
                        # Extract package name from lines like '    "pydantic>=2.0.0",'
                        m = re.match(r'["\']([a-zA-Z0-9_-]+[^"\']*)["\']', stripped)
                        if m:
                            deps.append(m.group(1))
        except Exception as e:
            print(f"  [U:Qa] pyproject.toml parse warning: {e}")
        return deps

    def _scan_and_install_imports(self, pip: list[str]) -> None:
        """G201: Scan source .py files for third-party imports and install them.

        Catches cases where pyproject.toml dependencies are empty but source
        code uses third-party packages. Builds a set of top-level import
        names, filters out stdlib and local packages, then pip-installs
        the remainder.
        """
        stdlib_modules = set(sys.stdlib_module_names)
        local_modules = self._collect_local_module_names()
        third_party = self._scan_source_imports()

        # Filter to only third-party
        third_party -= stdlib_modules
        third_party -= local_modules
        third_party -= {"orchestrata", "spike", "tests", "conftest"}
        to_install = [self._import_to_package(mod) for mod in sorted(third_party)]

        if not to_install:
            return

        print(f"  [U:Qa] G201: Installing scanned imports: {' '.join(to_install)}")
        try:
            subprocess.run(
                pip + to_install + ["--quiet"],
                cwd=self.workspace,
                capture_output=True,
                text=True,
                timeout=120,
            )
        except Exception as e:
            print(f"  [U:Qa] G201 install warning: {e}")

    def _collect_local_module_names(self) -> set:
        """Collect local package/module names from workspace root and src/."""
        local_modules = set()
        for search_dir in [self.workspace, os.path.join(self.workspace, "src")]:
            if not os.path.isdir(search_dir):
                continue
            for entry in os.listdir(search_dir):
                if entry.endswith(".py") and entry != "__init__.py":
                    local_modules.add(entry[:-3])
                elif os.path.isdir(os.path.join(search_dir, entry)):
                    if not entry.startswith((".", "_")):
                        local_modules.add(entry)
        return local_modules

    def _scan_source_imports(self) -> set:
        """Scan non-test source .py files for top-level import module names."""
        third_party = set()
        for root, dirs, files in os.walk(self.workspace):
            dirs[:] = [
                d
                for d in dirs
                if d
                not in {
                    ".venv",
                    "__pycache__",
                    ".git",
                    "node_modules",
                    "tests",
                }
            ]
            for fname in files:
                if not fname.endswith(".py"):
                    continue
                if fname.startswith("test_") or fname == "conftest.py":
                    continue
                filepath = os.path.join(root, fname)
                try:
                    with open(filepath, encoding="utf-8") as f:
                        for line in f:
                            mod = self._top_level_import_name(line)
                            if mod is not None:
                                third_party.add(mod)
                except Exception:
                    continue
        return third_party

    @staticmethod
    def _top_level_import_name(line: str) -> str | None:
        """Return the top-level module name of an import line, else None."""
        stripped = line.strip()
        if stripped.startswith("import "):
            mod = stripped.split()[1].split(".")[0].strip(",")
            if mod.isidentifier():
                return mod
        elif stripped.startswith("from "):
            parts = stripped.split()
            if len(parts) >= 2:
                mod = parts[1].split(".")[0].strip(",")
                if mod.isidentifier():
                    return mod
        return None

    @staticmethod
    def _import_to_package(mod: str) -> str:
        """Map a top-level import name to its installable package name."""
        # Common import-to-package name mappings
        pkg_map = {
            "cryptography": "cryptography",
            "yaml": "pyyaml",
            "cv2": "opencv-python",
            "PIL": "pillow",
            "bs4": "beautifulsoup4",
            "attr": "attrs",
            "dotenv": "python-dotenv",
            "jwt": "pyjwt",
            "msgpack": "msgpack",
            "aiohttp": "aiohttp",
            "fastapi": "fastapi",
            "pydantic": "pydantic",
            "redis": "redis",
            "click": "click",
            "rich": "rich",
            "httpx": "httpx",
            "toml": "toml",
        }
        return pkg_map.get(mod, mod)

    def _ensure_conftest_syspath(self):
        """Write conftest.py that adds workspace and monorepo packages to sys.path.

        G103: This allows 'from package_name.module import X' to work
        without editable install — pytest adds conftest.py's directory
        to sys.path automatically, and our conftest inserts workspace root.

        G-76-01: Also adds /opt/orchestrata/packages/*/src so that
        'from orchestrata_ultimate.*' imports resolve against the installed monorepo.
        """
        syspath_block = (
            "import sys, os, glob\n"
            "import pytest\n"
            "sys.path.insert(0, os.path.dirname(__file__))\n"
            "# G-89: expose generated project's own src/ dir — must stay at front\n"
            "_src = os.path.join(os.path.dirname(__file__), 'src')\n"
            "if os.path.isdir(_src) and _src not in sys.path:\n"
            "    sys.path.insert(0, _src)\n"
            "# G-92b-01: nested package dirs (J:Aider may create workspace/pkg-name/src/)\n"
            "for _nested in glob.glob(os.path.join(os.path.dirname(__file__), '*/src')):\n"
            "    if os.path.isdir(_nested) and _nested not in sys.path:\n"
            "        sys.path.insert(0, _nested)\n"
            "# G-76-01: expose orchestrata monorepo packages as fallback\n"
            "# G-90-02: use append so workspace/src shadows monorepo on namespace conflicts\n"
            "_monorepo = '/opt/orchestrata/packages'\n"
            "if os.path.isdir(_monorepo):\n"
            "    for _pkg_src in glob.glob(f'{_monorepo}/*/src'):\n"
            "        if _pkg_src not in sys.path:\n"
            "            sys.path.append(_pkg_src)\n"
            "\n"
            "# G129-1 + G139-1: Force-set REDIS_HOST/PORT from spike config.\n"
            "# Use unconditional assignment — setdefault fails when the\n"
            "# container env has REDIS_HOST='' (empty string counts as present).\n"
            f"os.environ['REDIS_HOST'] = '{REDIS_HOST}'\n"
            f"os.environ['REDIS_PORT'] = '{REDIS_PORT}'\n"
            "\n"
            "# G124-1: Inject Redis host/port from env as fixtures\n"
            "# so ALL tests automatically get the correct Redis address\n"
            "@pytest.fixture\n"
            "def redis_host():\n"
            "    return os.environ['REDIS_HOST']\n"
            "\n"
            "@pytest.fixture\n"
            "def redis_port():\n"
            "    return int(os.environ['REDIS_PORT'])\n"
            "\n"
            "# G143-1: Reset REDIS_HOST/PORT before every test.\n"
            "# Config tests may delete or mutate env vars; this\n"
            "# autouse fixture ensures the next test always gets\n"
            "# the correct values.\n"
            "@pytest.fixture(autouse=True)\n"
            "def _reset_redis_env():\n"
            "    yield\n"
            f"    os.environ['REDIS_HOST'] = '{REDIS_HOST}'\n"
            f"    os.environ['REDIS_PORT'] = '{REDIS_PORT}'\n"
        )
        conftest_path = os.path.join(self.workspace, "conftest.py")
        if os.path.exists(conftest_path):
            # Check if it already has sys.path logic
            with open(conftest_path, encoding="utf-8") as f:
                content = f.read()
            if "sys.path" in content:
                return
            # Prepend sys.path insert to existing conftest
            with open(conftest_path, "w", encoding="utf-8") as f:
                f.write(syspath_block + "\n" + content)
            print(
                "  [U:Qa] G103/G-76-01: prepended sys.path+monorepo to existing conftest.py"
            )
        else:
            with open(conftest_path, "w", encoding="utf-8") as f:
                f.write(syspath_block)
            print(
                "  [U:Qa] G103/G-76-01: created conftest.py with sys.path+monorepo insert"
            )

    def _install_deps(self):
        """Install workspace dependencies into a venv before running tests.

        G99: Install order matters. Workspace deps first (may pin pytest),
        then pytest/pytest-asyncio last so they override any stale pins.
        G103: For pyproject.toml, parse deps directly and use conftest.py
        for sys.path instead of editable install (avoids build_editable failures).
        """
        self._ensure_init_files()
        self._rewrite_relative_imports()  # G113
        venv_python = self._get_venv_python()
        pip = self._pip_cmd(venv_python)
        req_path = self._find_requirements()
        if req_path:
            if req_path.endswith("pyproject.toml"):
                # G103: Parse deps directly, skip editable install
                deps = self._parse_pyproject_deps(req_path)
                if deps:
                    cmd = pip + deps + ["--quiet"]
                    print(f"  [U:Qa] G103: Installing parsed deps: {' '.join(deps)}")
                    try:
                        result = subprocess.run(
                            cmd,
                            cwd=self.workspace,
                            capture_output=True,
                            text=True,
                            timeout=120,
                        )
                        if result.returncode != 0:
                            print(f"  [U:Qa] Dep install stderr: {result.stderr[:200]}")
                    except Exception as e:
                        print(f"  [U:Qa] Dep install warning: {e}")
                # G103: Add sys.path conftest so package imports work
                self._ensure_conftest_syspath()
            else:
                cmd = pip + ["-r", req_path, "--quiet"]
                print(f"  [U:Qa] Installing workspace deps: {' '.join(cmd)}")
                try:
                    result = subprocess.run(
                        cmd,
                        cwd=self.workspace,
                        capture_output=True,
                        text=True,
                        timeout=120,
                    )
                    if result.returncode != 0:
                        print(f"  [U:Qa] Dep install stderr: {result.stderr[:200]}")
                except Exception as e:
                    print(f"  [U:Qa] Dep install warning: {e}")
        # G201: Scan source files for third-party imports and install them.
        # Catches cases where pyproject.toml dependencies is empty but source
        # code uses third-party packages (e.g. cryptography, aiohttp).
        self._scan_and_install_imports(pip)
        # G123-1: Always ensure conftest.py has sys.path for src/ layout,
        # even when there's no pyproject.toml
        self._ensure_conftest_syspath()
        # G-77-07: install runtime deps of monorepo packages (e.g. redis>=5.0 for kernel)
        self._install_monorepo_deps(pip)
        # G87-02: Always install python-dotenv (common env-loading dep not in monorepo)
        subprocess.run(
            pip + ["python-dotenv", "--quiet"],
            cwd=self.workspace,
            capture_output=True,
            text=True,
            timeout=60,
        )
        # G99: Always install pytest LAST — overrides any pinned versions
        # G127-3: Also install black for auto-format line length fix
        # G129-3: Also install isort for auto-fix import ordering
        subprocess.run(
            pip
            + [
                "pytest",
                "pytest-asyncio",
                "pytest-timeout",
                "black",
                "isort",
                "--quiet",
            ],
            cwd=self.workspace,
            capture_output=True,
            text=True,
            timeout=120,
        )

    def _install_monorepo_deps(self, pip: list[str]) -> None:
        """G-77-07: Install runtime deps of all monorepo packages into the test venv.

        When generated code imports from orchestrata_ultimate.kernel.event_bus, the kernel
        package's deps (e.g. redis>=5.0) must be installed even though the kernel
        itself is on sys.path via conftest. Without this, import validation fails
        with ModuleNotFoundError for transitive deps like 'redis'.

        G-256-05: Only installs when generated code actually imports from
        orchestrata.* — stdlib-only projects skip this entirely.
        """
        monorepo = "/opt/orchestrata/packages"
        if not os.path.isdir(monorepo):
            return

        # G-256-05: Check if any source file imports from orchestrata
        if not self._workspace_imports_orchestrata():
            print("  [U:Qa] G-256-05: No orchestrata imports — skipping monorepo deps")
            return
        all_deps = self._collect_monorepo_deps(monorepo)
        # Deduplicate by package name prefix
        seen = {}
        for dep in all_deps:
            name = dep.split(">=")[0].split("<=")[0].split("==")[0].strip()
            seen[name] = dep
        unique_deps = list(seen.values())
        if unique_deps:
            print(
                f"  [U:Qa] G-77-07: Installing monorepo runtime deps: "
                f"{' '.join(unique_deps[:5])}{'...' if len(unique_deps) > 5 else ''}"
            )
            try:
                subprocess.run(
                    pip + unique_deps + ["--quiet"],
                    cwd=self.workspace,
                    capture_output=True,
                    text=True,
                    timeout=180,
                )
            except Exception as e:
                print(f"  [U:Qa] G-77-07: Monorepo dep install warning: {e}")

    def _workspace_imports_orchestrata(self) -> bool:
        """G-256-05: True if any workspace src/ file imports from orchestrata."""
        src_dir = os.path.join(self.workspace, "src")
        if not os.path.isdir(src_dir):
            return False
        for root, _dirs, files in os.walk(src_dir):
            for fname in files:
                if not fname.endswith(".py"):
                    continue
                try:
                    with open(os.path.join(root, fname), encoding="utf-8") as fh:
                        content = fh.read()
                    if (
                        "from orchestrata" in content
                        or "import orchestrata_ultimate" in content
                    ):
                        return True
                except Exception:
                    continue
        return False

    def _collect_monorepo_deps(self, monorepo: str) -> list:
        """Parse [project].dependencies from every monorepo package pyproject."""
        all_deps = []
        for pkg_dir in glob.glob(os.path.join(monorepo, "*")):
            toml_path = os.path.join(pkg_dir, "pyproject.toml")
            if not os.path.isfile(toml_path):
                continue
            try:
                with open(toml_path, encoding="utf-8") as f:
                    content = f.read()
                # Extract [project] dependencies list using simple regex
                m = re.search(
                    r"\[project\].*?^dependencies\s*=\s*\[(.*?)\]",
                    content,
                    re.DOTALL | re.MULTILINE,
                )
                if not m:
                    continue
                for line in m.group(1).splitlines():
                    dep = line.strip().strip('",').strip()
                    if (
                        dep
                        and not dep.startswith("#")
                        and not dep.startswith("orchestrata")
                    ):
                        all_deps.append(dep)
            except Exception:
                continue
        return all_deps

    def _validate_imports(self) -> list[str]:
        """Check that generated source files import cleanly before running tests."""
        venv_python = self._get_venv_python()
        errors = []
        skip_files = {
            "setup.py",
            "conftest.py",
            "manage.py",
            "__main__.py",
            "run_server.py",
            "main.py",
            "app.py",
            "cli.py",
        }
        modules_to_check = []
        # Top-level .py files
        for f in os.listdir(self.workspace):
            if f.endswith(".py") and not f.startswith("test_") and f not in skip_files:
                modules_to_check.append((f, f.removesuffix(".py")))
            # Packages with __init__.py
            elif os.path.isdir(os.path.join(self.workspace, f)):
                init = os.path.join(self.workspace, f, "__init__.py")
                if os.path.exists(init) and not f.startswith((".", "_", "test")):
                    modules_to_check.append((f + "/", f))
        # G-77-02: inject monorepo sys.path + workspace root so imports resolve correctly
        monorepo_syspath = (
            "import sys, os, glob; "
            "sys.path.insert(0, os.getcwd()); "
            "_s=os.path.join(os.getcwd(),'src'); "
            "(sys.path.insert(0,_s) if os.path.isdir(_s) and _s not in sys.path else None); "
            # G-92b-01: nested package dirs (J:Aider may create workspace/pkg-name/src/)
            "[sys.path.insert(0,p) for p in glob.glob(os.path.join(os.getcwd(),'*/src')) "
            "if os.path.isdir(p) and p not in sys.path]; "
            "_m='/opt/orchestrata/packages'; "
            "[sys.path.append(p) for p in glob.glob(f'{_m}/*/src') "
            "if p not in sys.path] if os.path.isdir(_m) else None; "
        )
        # G-77-05: also validate test files in tests/ subdir (catches wrong imports)
        tests_dir = os.path.join(self.workspace, "tests")
        if os.path.isdir(tests_dir):
            for f in os.listdir(tests_dir):
                if f.startswith("test_") and f.endswith(".py"):
                    modules_to_check.append(
                        (f"tests/{f}", f"tests.{f.removesuffix('.py')}")
                    )
        for label, module in modules_to_check:
            result = subprocess.run(
                [venv_python, "-c", monorepo_syspath + f"import {module}"],
                cwd=self.workspace,
                capture_output=True,
                text=True,
                timeout=15,
                env=self._venv_env(),
            )
            if result.returncode != 0:
                stderr_lines = result.stderr.strip().splitlines()
                err_msg = stderr_lines[-1] if stderr_lines else "unknown error"
                # G53: For ImportError, probe the module's dir() to help LLM fix
                hint = self._get_import_hint(venv_python, err_msg)
                errors.append(f"{label}: {err_msg}{hint}")
        return errors

    def _get_import_hint(self, venv_python: str, err_msg: str) -> str:
        """Extract module from import/attribute errors and show dir(module)."""
        import re

        # Match "cannot import name 'X' from 'Y'" or "module 'Y' has no attribute 'X'"
        m = re.search(r"cannot import name '(\w+)' from '([\w.]+)'", err_msg)
        if not m:
            m = re.search(r"module '([\w.]+)' has no attribute '(\w+)'", err_msg)
            if m:
                # Swap groups: module is group 1, attr is group 2
                mod = m.group(1)
            else:
                return ""
        else:
            mod = m.group(2)
        # G-77-02/G-77-05: inject monorepo sys.path + workspace root for hints
        monorepo_syspath = (
            "import sys, os, glob; "
            "sys.path.insert(0, os.getcwd()); "
            "_s=os.path.join(os.getcwd(),'src'); "
            "(sys.path.insert(0,_s) if os.path.isdir(_s) and _s not in sys.path else None); "
            # G-92b-01: nested package dirs (J:Aider may create workspace/pkg-name/src/)
            "[sys.path.insert(0,p) for p in glob.glob(os.path.join(os.getcwd(),'*/src')) "
            "if os.path.isdir(p) and p not in sys.path]; "
            "_m='/opt/orchestrata/packages'; "
            "[sys.path.append(p) for p in glob.glob(f'{_m}/*/src') "
            "if p not in sys.path] if os.path.isdir(_m) else None; "
        )
        try:
            result = subprocess.run(
                [
                    venv_python,
                    "-c",
                    monorepo_syspath
                    + f"import {mod}; print([x for x in dir({mod}) if not x.startswith('_')])",
                ],
                cwd=self.workspace,
                capture_output=True,
                text=True,
                timeout=10,
                env=self._venv_env(),
            )
            if result.returncode == 0 and result.stdout.strip():
                exports = result.stdout.strip()
                return f" — available names in {mod}: {exports}"
        except Exception as exc:
            _logger.debug("module export probe failed: %s", exc)
        return ""