orchestrata_ultimate.agents.volks.codedirector.scaffold.generator

Reference › orchestrata_ultimate › agents › volks › codedirector › scaffold › generator

generator writes the project skeleton from the Architect's design. generate_skeleton takes a project directory, the design documents and an optional project name, parses them with build_project_spec from parser, and writes the files, returning their paths. The CodeDirector's implementation module is the importer. The steps are _create_directories (all directories of the file tree), _generate_pyproject (from a Jinja template), _generate_conftest (the tests' conftest.py), _generate_init_files (an __init__.py per package) and _generate_modules (Python module stubs from the API spec, using _find_module_spec to match a file to its ModuleSpec). When the API spec carries complete code blocks, _write_code_blocks writes them as they are. _create_jinja_env builds the template environment with custom filters, _method_params and _func_params render parameter lists, and _write_file creates parent directories as needed.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/agents/volks/codedirector/scaffold/generator.py

# mypy: ignore-errors
"""Generate project skeleton from parsed design output."""

import json
import logging
import os
import subprocess

from jinja2 import Environment, FileSystemLoader

from orchestrata_ultimate.agents.volks.codedirector.scaffold.parser import (
    build_project_spec,
    parse_code_blocks,
)
from orchestrata_ultimate.agents.volks.codedirector.scaffold.types import (
    MethodSpec,
    ModuleSpec,
    ProjectSpec,
)
from orchestrata_ultimate.core.base.traces import traces

_logger = logging.getLogger(__name__)


_TEMPLATE_DIR = os.path.join(os.path.dirname(__file__), "templates")


def _method_params(method: MethodSpec) -> str:
    """Render method parameters with self/cls prefix.

    Args:
        method: MethodSpec to render parameters for.

    Returns:
        Parameter string like "self, name: str, value: int".
    """
    parts = ["self"]
    if method.is_classmethod:
        parts = ["cls"]
    for pname, ptype in method.params:
        parts.append(f"{pname}: {ptype}")
    return ", ".join(parts)


def _func_params(method: MethodSpec) -> str:
    """Render function parameters (no self/cls prefix).

    Args:
        method: MethodSpec to render parameters for.

    Returns:
        Parameter string like "name: str, value: int".
    """
    parts = [f"{pname}: {ptype}" for pname, ptype in method.params]
    return ", ".join(parts)


def _create_jinja_env() -> Environment:
    """Create Jinja2 environment with custom filters.

    Returns:
        Configured Jinja2 Environment.
    """
    env = Environment(
        loader=FileSystemLoader(_TEMPLATE_DIR),
        trim_blocks=True,
        lstrip_blocks=True,
        keep_trailing_newline=True,
    )
    env.filters["method_params"] = _method_params
    env.filters["func_params"] = _func_params
    return env


@traces(
    spec="029",
    story="Skeleton Generation",
    reqs=[
        "REQ-029-021",
        "REQ-029-022",
    ],
)
def generate_skeleton(
    project_dir: str,
    design_docs: dict[str, str],
    project_name: str = "",
) -> list[str]:
    """Parse design and generate project skeleton files.

    Creates directories, renders module stubs with correct imports,
    class definitions, method signatures, parameter validation, and
    try/except shells. The LLM only needs to fill in method bodies.

    Args:
        project_dir: Absolute path to project root directory.
        design_docs: Dict with keys: file_tree, api_spec, requirements,
                     architecture, class_diagram.
        project_name: Human-readable project name.

    Returns:
        List of absolute paths of created files.
    """
    if not project_name:
        project_name = os.path.basename(project_dir)

    spec = build_project_spec(project_name, project_dir, design_docs)
    env = _create_jinja_env()
    created_files: list[str] = []

    # Step 1: Create all directories from FILE_TREE
    _create_directories(spec, project_dir)

    # Step 2: Generate pyproject.toml
    pyproject_path = _generate_pyproject(env, spec, project_dir)
    if pyproject_path:
        created_files.append(pyproject_path)

    # Step 3: Generate conftest.py for tests
    conftest_path = _generate_conftest(env, spec, project_dir)
    if conftest_path:
        created_files.append(conftest_path)

    # Step 4: Generate __init__.py files
    init_paths = _generate_init_files(env, spec, project_dir)
    created_files.extend(init_paths)

    # Step 5: Extract code blocks (Kimi-style) or generate stubs
    api_spec_text = design_docs.get("api_spec", "")
    file_tree_text = design_docs.get("file_tree", "")
    code_blocks = parse_code_blocks(api_spec_text, file_tree_text)

    if code_blocks:
        # Kimi-style: API_SPEC has complete code — write directly
        code_paths = _write_code_blocks(code_blocks, project_dir)
        created_files.extend(code_paths)
    else:
        # Standard format: generate stubs from parsed specs
        module_paths = _generate_modules(env, spec, project_dir)
        created_files.extend(module_paths)

    # G228: Verify context.json exists
    ctx_path = os.path.join(project_dir, "context.json")
    if not os.path.exists(ctx_path):
        ctx = {
            "state": {
                "project": project_name,
                "status": "scaffold_complete",
            },
            "deferred_work": [],
            "session_history": [],
        }
        with open(ctx_path, "w") as f:
            json.dump(ctx, f, indent=2)
        created_files.append(ctx_path)

    # G228: UV sync if pyproject.toml exists
    pyproject = os.path.join(project_dir, "pyproject.toml")
    if os.path.exists(pyproject):
        try:
            subprocess.run(
                ["uv", "sync", "--python", "3.14t", "--all-extras"],
                cwd=project_dir,
                capture_output=True,
                timeout=60,
            )
            print("  [Scaffold] uv sync complete (3.14t)", flush=True)
        except (FileNotFoundError, subprocess.TimeoutExpired) as exc:
            print(f"  [Scaffold] uv sync skipped: {exc}", flush=True)

    # G228: Git add + initial commit before any review
    git_dir = os.path.join(project_dir, ".git")
    if os.path.exists(git_dir):
        try:
            subprocess.run(
                ["git", "add", "-A"],
                cwd=project_dir,
                capture_output=True,
                timeout=10,
            )
            subprocess.run(
                ["git", "commit", "-m", "Scaffold: initial project structure"],
                cwd=project_dir,
                capture_output=True,
                timeout=10,
            )
            print(
                "  [Scaffold] Git: initial commit",
                flush=True,
            )
        except (FileNotFoundError, subprocess.TimeoutExpired) as exc:
            print(f"  [Scaffold] Git commit skipped: {exc}", flush=True)

    print(
        f"  [Scaffold] Generated {len(created_files)} files in {project_dir}",
        flush=True,
    )
    return created_files


@traces(
    spec="029",
    story="Directory Creation",
    reqs=["REQ-029-018", "REQ-029-019"],
)
def _create_directories(spec: ProjectSpec, project_dir: str) -> None:
    """Create all directories needed by the file tree.

    G228: Enforces mandatory project structure:
    - tests/unit/ with per-module subdirectories
    - tests/integration/
    - docs/
    - requirements/

    Args:
        spec: Parsed project specification.
        project_dir: Root directory for the project.
    """
    # Ensure project root exists
    os.makedirs(project_dir, exist_ok=True)

    # G228: Mandatory directories
    mandatory_dirs = [
        "tests",
        "tests/unit",
        "tests/integration",
        "docs",
        "requirements",
    ]
    for d in mandatory_dirs:
        os.makedirs(os.path.join(project_dir, d), exist_ok=True)

    # G228: __init__.py for test packages
    for init_dir in ["tests", "tests/unit", "tests/integration"]:
        init_path = os.path.join(project_dir, init_dir, "__init__.py")
        if not os.path.exists(init_path):
            with open(init_path, "w") as f:
                f.write("")

    # Directories from file tree
    dirs_needed = set()
    for path in spec.file_paths:
        parent = os.path.dirname(path)
        if parent:
            dirs_needed.add(parent)

    # G228: Create per-module test subdirectories under tests/unit/
    for path in spec.file_paths:
        if path.endswith(".py") and not os.path.basename(path).startswith("test_"):
            # src/pkg/validators.py → tests/unit/validators/
            basename = os.path.basename(path).removesuffix(".py")
            if basename != "__init__":
                unit_dir = f"tests/unit/{basename}"
                dirs_needed.add(unit_dir)

    for d in sorted(dirs_needed):
        full_path = os.path.join(project_dir, d)
        os.makedirs(full_path, exist_ok=True)


@traces(
    spec="029",
    story="PyProject Generation",
    reqs=["REQ-029-020"],
)
def _generate_pyproject(
    env: Environment,
    spec: ProjectSpec,
    project_dir: str,
) -> str:
    """Generate pyproject.toml from template.

    Args:
        env: Jinja2 environment.
        spec: Parsed project specification.
        project_dir: Root directory for the project.

    Returns:
        Absolute path of created file.
    """
    template = env.get_template("pyproject.toml.j2")
    content = template.render(
        project_name=spec.name,
        python_version=spec.python_version,
        dependencies=spec.dependencies,
    )
    path = os.path.join(project_dir, "pyproject.toml")
    _write_file(path, content)
    return path


def _generate_conftest(
    env: Environment,
    spec: ProjectSpec,
    project_dir: str,
) -> str:
    """Generate tests/conftest.py from template.

    Args:
        env: Jinja2 environment.
        spec: Parsed project specification.
        project_dir: Root directory for the project.

    Returns:
        Absolute path of created file, or empty string if not needed.
    """
    tests_dir = os.path.join(project_dir, "tests")
    if not os.path.isdir(tests_dir):
        os.makedirs(tests_dir, exist_ok=True)

    template = env.get_template("conftest.py.j2")
    content = template.render(package_name=spec.package_name)
    path = os.path.join(tests_dir, "conftest.py")
    _write_file(path, content)
    return path


def _generate_init_files(
    env: Environment,
    spec: ProjectSpec,
    project_dir: str,
) -> list[str]:
    """Generate __init__.py files for all packages.

    Args:
        env: Jinja2 environment.
        spec: Parsed project specification.
        project_dir: Root directory for the project.

    Returns:
        List of absolute paths of created files.
    """
    created: list[str] = []
    template = env.get_template("init.py.j2")

    # Find all __init__.py paths
    init_paths = [p for p in spec.file_paths if p.endswith("__init__.py")]

    # Find corresponding module spec for imports
    for init_path in init_paths:
        module_spec = _find_module_spec(spec, init_path)
        imports = []
        if module_spec:
            imports = [
                i for i in module_spec.imports if i.startswith(("from ", "import "))
            ]

        content = template.render(
            package_name=spec.package_name,
            imports=imports,
        )
        full_path = os.path.join(project_dir, init_path)
        _write_file(full_path, content)
        created.append(full_path)

    # Also create __init__.py for tests/ if not already listed
    tests_init = os.path.join(project_dir, "tests", "__init__.py")
    if not os.path.exists(tests_init):
        _write_file(tests_init, "")
        created.append(tests_init)

    return created


def _generate_modules(
    env: Environment,
    spec: ProjectSpec,
    project_dir: str,
) -> list[str]:
    """Generate Python module stubs from API_SPEC.

    Each module gets correct imports, class definitions with field
    declarations, method signatures with parameter validation, and
    try/except shells with the correct exception types.

    Args:
        env: Jinja2 environment.
        spec: Parsed project specification.
        project_dir: Root directory for the project.

    Returns:
        List of absolute paths of created files.
    """
    created: list[str] = []
    template = env.get_template("module.py.j2")

    for module in spec.modules:
        # Skip __init__.py — handled separately
        if module.path.endswith("__init__.py"):
            continue

        # G226: Filter prose from imports — only real Python imports
        module.imports = [
            i for i in module.imports if i.startswith(("from ", "import "))
        ]

        content = template.render(module=module)

        # Determine output path
        full_path = os.path.join(project_dir, module.path)
        os.makedirs(os.path.dirname(full_path), exist_ok=True)
        _write_file(full_path, content)
        created.append(full_path)

    return created


def _write_code_blocks(code_blocks: dict[str, str], project_dir: str) -> list[str]:
    """Write complete code blocks extracted from Kimi-style API_SPEC.

    When the architect provides full implementations in code blocks,
    write them directly instead of generating stubs.

    Args:
        code_blocks: Dict mapping relative path to code content.
        project_dir: Root directory for the project.

    Returns:
        List of absolute paths of created files.
    """
    created: list[str] = []
    for rel_path, content in code_blocks.items():
        full_path = os.path.join(project_dir, rel_path)
        os.makedirs(os.path.dirname(full_path), exist_ok=True)
        _write_file(full_path, content)
        created.append(full_path)
        print(f"  [Scaffold] Wrote {rel_path}", flush=True)
    return created


def _find_module_spec(spec: ProjectSpec, path: str) -> ModuleSpec | None:
    """Find the ModuleSpec matching a file path.

    Args:
        spec: Parsed project specification.
        path: File path to match.

    Returns:
        Matching ModuleSpec, or None if not found.
    """
    for module in spec.modules:
        if module.path == path:
            return module
    return None


def _write_file(path: str, content: str) -> None:
    """Write content to a file, creating parent directories.

    Args:
        path: Absolute file path.
        content: File content to write.
    """
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "w") as f:
        f.write(content)