orchestrata_ultimate.drivers.deepseek_coder_driver

Reference › orchestrata_ultimate › drivers › deepseek_coder_driver

deepseek_coder_driver is the largest driver. It generates code directly from model output and needs no external command-line tool (SPEC-025 REQ-025-044 and REQ-025-045, SPEC-028 REQ-028-021). Despite its name it is provider-agnostic: its transport is call_llm from llm.client for the J:Junior volk, so the provider and model come from the active profile. driver_factory returns it for the deepseek and openrouter providers.

DeepSeekCoderDriver.execute builds a prompt from the task and the optional context of existing files (_build_context, truncated by _read_file_truncated), calls the model with a system prompt that asks for complete files as FILE: blocks, and writes the result with _extract_and_write_files. A failed call returns a failure tuple. If nothing is extracted, _retry_on_empty retries once with a stricter prompt. _recover_missing_files then compares the written files with the expected files listed in the design's file_tree.md (_expected_design_files, _find_missing_files) and asks again for any that are missing. After that, _autofix_lint fixes unused imports with ruff and _validate_written_files checks syntax and lint, logging problems and not failing on them.

Safeguards live in the writer: _normalize_path strips path prefixes that repeat the workspace layout, _is_destructive_replacement and _defined_names refuse to overwrite an existing implementation with a version that drops names it defines, and files with syntax errors are skipped and reported for recovery. Tests are in tests/drivers/test_deepseek_coder_driver.py.

At a glance

Class diagram

Dependency diagram

API

Classes

DeepSeekCoderDriver

Decorators: traces(spec='025', reqs=['REQ-025-044', 'REQ-025-045']) · Line 276

Async code gen driver using DeepSeek API directly.

Functions

Source

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

# mypy: ignore-errors
"""Async DeepSeek code generation driver.

Generates files directly from LLM output without requiring an external
CLI tool. Parses FILE: blocks, validates syntax, auto-fixes lint, and
retries on empty extraction or missing files.
"""

import ast
import asyncio
import logging
import os
import re
import shutil

from orchestrata_ultimate.core.traceability import traces

_logger = logging.getLogger(__name__)

CODER_SYSTEM_PROMPT = (
    "You are a senior Python developer. You write clean, "
    "production-grade code.\n\n"
    "When given a task, output ALL files as a series of FILE blocks:\n\n"
    "FILE: path/to/file.py\n```python\n<complete file content>\n```\n\n"
    "Rules:\n"
    "- Output COMPLETE file contents — never use placeholders or '...'\n"
    "- Use absolute imports (no relative imports)\n"
    "- Include proper __init__.py files for packages\n"
    "- Follow PEP 8, use type hints on public functions\n"
    "- No TODO/FIXME/HACK comments\n"
    "- Guard entry points with if __name__ == '__main__'\n"
    "- Use 'with' for file/resource management"
)

_KNOWN_ROOTS = frozenset({"src", "tests", "design", "docs"})
_SOURCE_EXTS = (".py", ".toml", ".cfg", ".ini")
_SKIP_PREFIXES = ("design/", ".env", "Dockerfile", "README")


def _defined_names(source: str) -> set[str] | None:
    """Collect every class/function name defined anywhere in the source.

    Returns:
        Set of names, or None if the source does not parse.
    """
    try:
        tree = ast.parse(source)
    except SyntaxError:
        return None
    return {
        node.name
        for node in ast.walk(tree)
        if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef | ast.ClassDef)
    }


def _is_destructive_replacement(filepath: str, new_code: str) -> bool:
    """True when writing new_code would destroy an existing implementation.

    Fix/retry rounds re-enter this driver with lint or review feedback, and
    some models (observed live: qwen3-coder-next in the openrouter-cheap
    pipeline, 2026-07-13) answer a fix prompt with a FILE: block containing
    only the corrected LINES — e.g. a body that is exactly
    ``from __future__ import annotations`` — instead of the complete file.
    That fragment compiles, so the syntax gate cannot catch it, and blindly
    writing it clobbers the finished implementation (the LRU-rebuild demo
    lost a complete, working class this way). The guard: refuse to replace
    an existing file when the replacement is smaller and preserves NONE of
    the classes/functions the current file defines.
    """
    if not filepath.endswith(".py") or not os.path.exists(filepath):
        return False
    try:
        with open(filepath, encoding="utf-8") as fh:
            existing = fh.read()
    except OSError, UnicodeDecodeError:
        return False
    existing_names = _defined_names(existing)
    if not existing_names:
        # Existing file is broken or defines nothing — replacement can
        # only help.
        return False
    new_names = _defined_names(new_code) or set()
    return not (existing_names & new_names) and len(new_code) < len(existing)


def _extract_and_write_files(
    response: str,
    workspace: str,
) -> tuple[list[str], list[str]]:
    """Extract FILE: blocks from LLM response and write to disk.

    Args:
        response: Raw LLM output containing FILE: blocks.
        workspace: Root directory to write files into.

    Returns:
        Tuple of (written file paths, syntax-skipped file paths).
    """
    written: list[str] = []
    syntax_skipped: list[str] = []
    parts = re.split(r"^FILE:\s*", response, flags=re.MULTILINE)

    for part in parts[1:]:
        lines = part.split("\n")
        if not lines:
            continue

        filename = _normalize_path(lines[0].strip())
        if not filename:
            continue

        code = _extract_code_block(lines[1:])
        if not code:
            continue

        if filename.endswith(".py"):
            try:
                compile(code, filename, "exec")
            except SyntaxError as exc:
                _logger.warning("SKIP %s — syntax error: %s", filename, exc)
                syntax_skipped.append(filename)
                continue

        filepath = os.path.join(workspace, filename)
        if _is_destructive_replacement(filepath, code):
            _logger.warning(
                "REFUSED %s — replacement is a fragment that would clobber "
                "the existing implementation (keeps none of its defs, "
                "%d -> %d chars)",
                filename,
                os.path.getsize(filepath),
                len(code),
            )
            continue
        os.makedirs(os.path.dirname(filepath), exist_ok=True)
        with open(filepath, "w", encoding="utf-8") as f:
            f.write(code)
        written.append(filename)

    return written, syntax_skipped


def _normalize_path(filename: str) -> str:
    """Strip leading path components that duplicate workspace structure.

    Args:
        filename: Raw filename from LLM output.

    Returns:
        Normalized relative path.
    """
    parts_f = filename.replace("\\", "/").split("/")
    for idx, seg in enumerate(parts_f):
        if seg in _KNOWN_ROOTS:
            return "/".join(parts_f[idx:])

    if len(parts_f) > 1 and parts_f[0] not in _KNOWN_ROOTS:
        candidate = "/".join(parts_f[1:])
        check = candidate.replace("\\", "/").split("/")
        if len(check) > 4:
            return parts_f[-1]
        return candidate
    return filename


def _extract_code_block(lines: list[str]) -> str:
    """Extract code content between ``` markers.

    Args:
        lines: Lines after the filename line.

    Returns:
        Extracted code string, or empty if no code block found.
    """
    code_lines: list[str] = []
    in_code = False
    for line in lines:
        stripped = line.strip()
        if stripped.startswith("```") and not in_code:
            in_code = True
        elif stripped == "```" and in_code:
            in_code = False
        elif in_code:
            code_lines.append(line)
    return "\n".join(code_lines) if code_lines else ""


async def _autofix_lint(workspace: str, files: list[str]) -> None:
    """Auto-fix F401 unused imports with ruff.

    Args:
        workspace: Root workspace directory.
        files: List of relative file paths.
    """
    ruff_bin = shutil.which("ruff")
    if not ruff_bin:
        return

    py_files = [f for f in files if f.endswith(".py")]
    for f in py_files:
        filepath = os.path.join(workspace, f)
        try:
            proc = await asyncio.create_subprocess_exec(
                ruff_bin,
                "check",
                "--fix",
                "--select=F401",
                filepath,
                stdout=asyncio.subprocess.DEVNULL,
                stderr=asyncio.subprocess.DEVNULL,
            )
            await asyncio.wait_for(proc.wait(), timeout=10)
        except (TimeoutError, FileNotFoundError) as exc:
            _logger.debug("ruff F401 cleanup skipped: %s", exc)


async def _validate_written_files(workspace: str, files: list[str]) -> dict[str, str]:
    """Validate written Python files for syntax and lint.

    Args:
        workspace: Root workspace directory.
        files: List of relative file paths.

    Returns:
        Dict mapping filename to error string (empty if clean).
    """
    errors: dict[str, str] = {}
    py_files = [f for f in files if f.endswith(".py")]
    ruff_bin = shutil.which("ruff")
    for f in py_files:
        filepath = os.path.join(workspace, f)
        try:
            with open(filepath, encoding="utf-8") as fh:
                source = fh.read()
            compile(source, f, "exec")
        except SyntaxError as exc:
            errors[f] = f"syntax: {exc}"
            continue
        if ruff_bin:
            try:
                proc = await asyncio.create_subprocess_exec(
                    ruff_bin,
                    "check",
                    "--select=F821,F811,F401",
                    filepath,
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.DEVNULL,
                )
                stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=10)
                if proc.returncode != 0 and stdout:
                    errors[f] = stdout.decode()[:200]
            except (TimeoutError, FileNotFoundError) as exc:
                _logger.debug("validation skipped for %s: %s", f, exc)

    return errors


def _read_file_truncated(filepath: str, max_chars: int = 5000) -> str | None:
    """Read a file, truncating at max_chars if needed."""
    try:
        with open(filepath, encoding="utf-8") as fh:
            content = fh.read()
        if len(content) > max_chars:
            return content[:max_chars] + "\n# ... truncated"
        return content
    except (OSError, UnicodeDecodeError) as exc:
        _logger.debug("read file %s: %s", filepath, exc)
        return None


@traces(
    spec="025",
    reqs=["REQ-025-044", "REQ-025-045"],
)
class DeepSeekCoderDriver:
    """Async code gen driver using DeepSeek API directly.

    Generates files from LLM output by parsing FILE: blocks.
    Includes retry on empty extraction, missing file recovery,
    and automatic lint fixing.

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

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

        Args:
            workspace: Directory to write generated files into.
        """
        self.workspace = workspace or os.environ.get(
            "ORCHESTRATA_WORKSPACE", "/home/mohamed/workspace"
        )
        os.makedirs(self.workspace, exist_ok=True)

    @traces(spec="028", reqs=["REQ-028-021"])
    async def execute(
        self,
        task: str,
        files: list[str] | None = None,
        is_fix: bool = False,
    ) -> tuple[bool, str]:
        """Generate code for a task and write files to workspace.

        Args:
            task: The coding task/prompt.
            files: Optional existing files to include as context.
            is_fix: When True, skip file-reading (caller embedded content).

        Returns:
            Tuple of (success, output_text).
        """
        _logger.info("Generating code in %s", self.workspace)

        context = ""
        if not is_fix and files:
            context = await asyncio.to_thread(self._build_context, files)

        prompt = self._build_prompt(task, context)

        from orchestrata_ultimate.llm.client import call_llm  # noqa: PLC0415

        try:
            response = await call_llm(
                CODER_SYSTEM_PROMPT, prompt, temperature=0.2, volk="J:Junior"
            )
        except Exception as exc:
            msg = f"LLM call failed: {exc}"
            _logger.error(msg)
            return False, msg

        written, syntax_skipped = await asyncio.to_thread(
            _extract_and_write_files, response, self.workspace
        )

        if not written:
            _logger.info("No files extracted — retrying with strict prompt")
            written, syntax_skipped = await self._retry_on_empty(
                task, context, response
            )

        if not written:
            _logger.warning("No files extracted after retry")
            return False, "No files generated"

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

        written = await self._recover_missing_files(task, written, syntax_skipped)

        await _autofix_lint(self.workspace, written)

        lint_errors = await _validate_written_files(self.workspace, written)
        for f, errs in lint_errors.items():
            _logger.warning("LINT %s: %s", f, errs)

        _logger.info("Generated %d files", len(written))
        return True, response

    def _build_context(self, files: list[str]) -> str:
        """Build context string from existing source files."""
        context = ""
        for f in files:
            if not any(f.endswith(ext) for ext in _SOURCE_EXTS):
                continue
            if any(f.startswith(pfx) for pfx in _SKIP_PREFIXES):
                continue
            content = _read_file_truncated(os.path.join(self.workspace, f))
            if content is not None:
                context += f"\n=== {f} ===\n{content}\n"
        return context

    def _build_prompt(self, task: str, context: str) -> str:
        """Build the full prompt with format reminder.

        Args:
            task: The coding task.
            context: Existing file context (may be empty).

        Returns:
            The complete prompt string.
        """
        format_reminder = (
            "\n\nIMPORTANT: Output ALL modified files as FILE: blocks. "
            "Each file must be complete — no partial snippets.\n"
            "FILE: path/to/file.py\n```python\n<full content>\n```"
        )
        if context:
            return (
                f"Existing files in the project:\n{context}\n\n"
                f"Task:\n{task}{format_reminder}"
            )
        return task + format_reminder

    @traces(spec="028", reqs=["REQ-028-021"])
    async def _retry_on_empty(
        self,
        task: str,
        context: str,
        original_response: str,
    ) -> tuple[list[str], list[str]]:
        """Retry generation with stricter FILE: block prompt.

        Args:
            task: Original task prompt.
            context: Existing file context string.
            original_response: The response that failed extraction.

        Returns:
            Tuple of (written file paths, syntax-skipped file paths).
        """
        failed_snippet = original_response[:500]
        strict_prompt = (
            "Your previous response could NOT be parsed because it "
            "did not use the required FILE: block format.\n\n"
            f"Your response started with:\n{failed_snippet}\n\n"
            "You MUST output files using EXACTLY this format:\n\n"
            "FILE: path/to/file.py\n```python\n<full content>\n```\n\n"
            "Do NOT explain. Do NOT use diffs. Output COMPLETE files "
            "in FILE: blocks ONLY.\n\n"
        )
        if context:
            strict_prompt += f"Existing files:\n{context}\n\n"
        strict_prompt += f"Task:\n{task}"

        from orchestrata_ultimate.llm.client import call_llm  # noqa: PLC0415

        try:
            response = await call_llm(
                CODER_SYSTEM_PROMPT,
                strict_prompt,
                temperature=0.3,
                volk="J:Junior",
            )
        except Exception as exc:
            _logger.warning("Retry LLM call failed: %s", exc)
            return [], []

        return await asyncio.to_thread(
            _extract_and_write_files, response, self.workspace
        )

    @traces(spec="028", reqs=["REQ-028-021"])
    async def _recover_missing_files(
        self,
        task: str,
        written: list[str],
        syntax_skipped: list[str] | None = None,
    ) -> list[str]:
        """Detect missing files from design and retry generation.

        Args:
            task: Original task prompt.
            written: List of files already written.
            syntax_skipped: Files skipped due to syntax errors.

        Returns:
            Updated list of all written files.
        """
        missing = await asyncio.to_thread(
            self._find_missing_files, written, syntax_skipped
        )
        if not missing:
            return written

        _logger.info(
            "%d files missing from design — retrying: %s",
            len(missing),
            ", ".join(missing),
        )

        context = await asyncio.to_thread(self._build_recovery_context, written)
        retry_prompt = (
            "The following files from the design are MISSING and "
            "must be created:\n"
            + "\n".join(f"- {f}" for f in missing)
            + f"\n\nExisting files for reference:\n{context}"
            + f"\n\nOriginal task:\n{task}"
            + "\n\nIMPORTANT: Output ALL missing files as FILE: blocks."
        )

        from orchestrata_ultimate.llm.client import call_llm  # noqa: PLC0415

        try:
            retry_response = await call_llm(
                CODER_SYSTEM_PROMPT,
                retry_prompt,
                temperature=0.2,
                volk="J:Junior",
            )
        except Exception as exc:
            _logger.warning("Recovery retry failed: %s", exc)
            return written

        retry_written, _ = await asyncio.to_thread(
            _extract_and_write_files, retry_response, self.workspace
        )
        for f in retry_written:
            _logger.info("Wrote %s (recovery)", f)

        return written + retry_written

    def _find_missing_files(
        self,
        written: list[str],
        syntax_skipped: list[str] | None,
    ) -> list[str]:
        """Find files expected by design but not written.

        Args:
            written: Files already written.
            syntax_skipped: Files skipped due to syntax errors.

        Returns:
            List of missing file names.
        """
        expected_files = self._expected_design_files()
        if not expected_files:
            return []

        written_basenames = {os.path.basename(f) for f in written}
        missing: list[str] = []

        for expected in expected_files:
            basename = os.path.basename(expected)
            if basename == "__init__.py":
                continue
            if basename not in written_basenames:
                if not self._file_exists_on_disk(basename):
                    missing.append(expected)

        self._merge_syntax_skipped(missing, syntax_skipped)
        return missing

    def _expected_design_files(self) -> set[str]:
        """Parse design/file_tree.md for the set of expected (non-test) .py files."""
        file_tree_path = os.path.join(self.workspace, "design", "file_tree.md")
        if not os.path.exists(file_tree_path):
            return set()

        try:
            with open(file_tree_path, encoding="utf-8") as fh:
                tree_content = fh.read()
        except OSError, UnicodeDecodeError:
            return set()

        expected_files: set[str] = set()
        for match in re.finditer(r"[─├└│\s]*(\S+\.py)", tree_content):
            fname = match.group(1).strip().strip("`'\"")
            if fname.startswith("test_") or fname == "conftest.py":
                continue
            expected_files.add(fname)
        return expected_files

    def _merge_syntax_skipped(
        self, missing: list[str], syntax_skipped: list[str] | None
    ) -> None:
        """Append syntax-skipped files to *missing*, de-duplicating by basename."""
        if not syntax_skipped:
            return
        for skipped in syntax_skipped:
            basename = os.path.basename(skipped)
            existing_bases = {os.path.basename(m) for m in missing}
            if basename not in existing_bases:
                missing.append(skipped)

    def _file_exists_on_disk(self, basename: str) -> bool:
        """Check if a file exists anywhere in the workspace.

        Args:
            basename: File name to search for.

        Returns:
            True if the file exists.
        """
        for root, _dirs, files_on_disk in os.walk(self.workspace):
            if ".venv" in root or "__pycache__" in root:
                continue
            if basename in files_on_disk:
                return True
        return False

    def _build_recovery_context(self, written: list[str]) -> str:
        """Build context from already-written files for recovery."""
        context = ""
        for f in written:
            content = _read_file_truncated(os.path.join(self.workspace, f))
            if content is not None:
                context += f"\n=== {f} ===\n{content}\n"
        return context