orchestrata_ultimate.agents.volks.architect.architect_project

Reference › orchestrata_ultimate › agents › volks › architect › architect_project

architect_project is the project-operations mixin of the Architect (volk). ArchitectProjectMixin prepares the on-disk project and records the design. _init_project creates the project directory structure, writes an initial context.json (_write_context_json), initialises a git repository (_git_init) and creates a uv virtual environment for the free-threaded Python 3.14 (_create_venv). _parse_design splits the model's reply into the ARCHITECTURE, REQUIREMENTS, FILE_TREE, API_SPEC and CLASS_DIAGRAM sections, and _core_sections_blank reports whether all of them came back empty. _write_design_docs writes them into the project's design/ folder and _git_commit commits every file in the project directory.

At a glance

Class diagram

Dependency diagram

API

Classes

ArchitectProjectMixin

Line 17

Mixin providing project initialisation and document management.

Source

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

"""Project-management operations mixin for the Architect volk.

Handles filesystem setup, design document writing, response parsing,
and git commits. Extracted from volk.py to keep class size within limits.
"""

import json
import logging
import os
import subprocess

from orchestrata_ultimate.core.base.traces import traces

_logger = logging.getLogger(__name__)


class ArchitectProjectMixin:
    """Mixin providing project initialisation and document management."""

    @traces(spec="030", story="US-030-006", reqs=["REQ-030-038"])
    def _init_project(self) -> None:
        """Create project directory structure and initialise a git repo.

        G228: Full project bootstrap — src/, tests/, docs/, requirements/,
        design/, context.json, pyproject.toml, git init, uv venv.
        """
        os.makedirs(self.project_dir, exist_ok=True)  # type: ignore[attr-defined]
        # SPEC-030 Security §Data Protection: the workspace holds generated
        # source + requirements, so restrict it to the owner (0700) instead of
        # inheriting a potentially world-readable umask default. Best-effort:
        # skip cleanly on a filesystem/OS that does not support POSIX modes.
        try:
            os.chmod(self.project_dir, 0o700)  # type: ignore[attr-defined]
        except OSError:
            pass

        dirs = [
            "design",
            "src",
            "tests",
            "tests/unit",
            "tests/integration",
            "docs",
            "requirements",
        ]
        for d in dirs:
            os.makedirs(os.path.join(self.project_dir, d), exist_ok=True)  # type: ignore[attr-defined]

        for init_dir in ["tests", "tests/unit", "tests/integration"]:
            init_path = os.path.join(self.project_dir, init_dir, "__init__.py")  # type: ignore[attr-defined]
            if not os.path.exists(init_path):
                with open(init_path, "w") as fh:
                    fh.write("")

        self._write_context_json()
        self._git_init()
        self._create_venv()

    @traces(spec="030", story="US-030-006", reqs=["REQ-030-039"])
    def _write_context_json(self) -> None:
        """Write an initial context.json to the project directory."""

        ctx_path = os.path.join(self.project_dir, "context.json")  # type: ignore[attr-defined]
        if os.path.exists(ctx_path):
            return
        ctx = {
            "state": {
                "project": os.path.basename(self.project_dir),  # type: ignore[attr-defined]
                "status": "initializing",
                "created_by": "U:Ar",
            },
            "deferred_work": [],
            "session_history": [],
        }
        with open(ctx_path, "w") as fh:
            json.dump(ctx, fh, indent=2)
        print("  [U:Ar] Created context.json", flush=True)

    @traces(spec="030", story="US-030-006", reqs=["REQ-030-040"])
    def _git_init(self) -> None:
        """Initialise a git repository in the project directory."""
        git_dir = os.path.join(self.project_dir, ".git")  # type: ignore[attr-defined]
        if os.path.exists(git_dir):
            return
        try:
            subprocess.run(
                ["git", "init"],
                cwd=self.project_dir,  # type: ignore[attr-defined]
                capture_output=True,
                timeout=10,
            )
            print(f"  [U:Ar] Git repo initialized: {self.project_dir}", flush=True)  # type: ignore[attr-defined]
        except Exception as exc:
            print(f"  [U:Ar] Git init failed: {exc}", flush=True)

    @traces(spec="030", story="US-030-006", reqs=["REQ-030-041"])
    def _create_venv(self) -> None:
        """Create a uv venv with Python 3.14t (free-threaded)."""
        venv_dir = os.path.join(self.project_dir, ".venv")  # type: ignore[attr-defined]
        if os.path.exists(venv_dir):
            return
        try:
            subprocess.run(
                ["uv", "venv", "--python", "3.14t"],
                cwd=self.project_dir,  # type: ignore[attr-defined]
                capture_output=True,
                timeout=30,
            )
            print("  [U:Ar] UV venv created (Python 3.14t)", flush=True)
        except (FileNotFoundError, subprocess.TimeoutExpired) as exc:
            _logger.debug("uv venv: %s", exc)

    @traces(spec="030", story="US-030-003", reqs=["REQ-030-024"])
    def _write_design_docs(self) -> None:
        """Write design documents to the project design/ folder."""
        design_dir = os.path.join(self.project_dir, "design")  # type: ignore[attr-defined]
        os.makedirs(design_dir, exist_ok=True)
        docs = {
            "architecture.md": self.design_docs["architecture"],  # type: ignore[attr-defined]
            "requirements.md": self.design_docs["requirements"],  # type: ignore[attr-defined]
            "file_tree.md": self.design_docs["file_tree"],  # type: ignore[attr-defined]
            "api_spec.md": self.design_docs["api_spec"],  # type: ignore[attr-defined]
            "class_diagram.md": self.design_docs["class_diagram"],  # type: ignore[attr-defined]
        }
        for filename, content in docs.items():
            if content:
                filepath = os.path.join(design_dir, filename)
                with open(filepath, "w", encoding="utf-8") as fh:
                    fh.write(content)
        print(f"  [U:Ar] Design docs written to {design_dir}", flush=True)

    @traces(
        spec="030",
        story="US-030-003",
        reqs=["REQ-030-022", "REQ-030-023"],
    )
    def _parse_design(self, response: str) -> None:
        """Parse ARCHITECTURE, REQUIREMENTS, FILE_TREE, API_SPEC, CLASS_DIAGRAM.

        Populates self.design_docs in place from section markers in *response*.

        Args:
            response: Raw LLM response containing the section markers.
        """
        sections = {
            "architecture": "ARCHITECTURE:",
            "requirements": "REQUIREMENTS:",
            "file_tree": "FILE_TREE:",
            "api_spec": "API_SPEC:",
            "class_diagram": "CLASS_DIAGRAM:",
        }
        section_order = [
            "architecture",
            "requirements",
            "file_tree",
            "api_spec",
            "class_diagram",
        ]
        for i, key in enumerate(section_order):
            marker = sections[key]
            if marker not in response:
                continue
            start = response.index(marker) + len(marker)
            end = len(response)
            for next_key in section_order[i + 1 :]:
                next_marker = sections[next_key]
                if next_marker in response:
                    next_pos = response.index(next_marker)
                    if next_pos > start:
                        end = min(end, next_pos)
                        break
            self.design_docs[key] = response[start:end].strip()  # type: ignore[attr-defined]

    @traces(spec="030", story="US-030-001", reqs=["REQ-030-007"])
    def _core_sections_blank(self) -> bool:
        """True when ALL core design sections are empty after ``_parse_design``.

        "Core" = architecture/requirements/file_tree/api_spec — the sections
        J:Junior and U:Cd actually implement against. ``class_diagram`` is
        deliberately excluded: some legitimately small tasks produce no
        diagram without that meaning the response was blank/hallucinated
        (dogfood run-2 F3 defect 2 -- Junior review found "Architecture,
        Requirements, File Tree, and API Specification sections ... are all
        blank" and the pipeline shipped it to review anyway).

        Returns:
            True if every core section is empty/whitespace-only.
        """
        return not any(
            (self.design_docs.get(key) or "").strip()  # type: ignore[attr-defined]
            for key in ("architecture", "requirements", "file_tree", "api_spec")
        )

    @traces(
        spec="030",
        story="US-030-006",
        reqs=["REQ-030-042", "REQ-030-043"],
    )
    def _git_commit(self, message: str) -> None:
        """Commit all files in the project directory.

        Args:
            message: Commit message.
        """
        project_dir = getattr(self, "project_dir", "")
        if not project_dir or not os.path.isdir(os.path.join(project_dir, ".git")):
            return
        try:
            subprocess.run(
                ["git", "add", "-A"],
                cwd=project_dir,
                capture_output=True,
                timeout=10,
            )
            subprocess.run(
                ["git", "commit", "-m", message],
                cwd=project_dir,
                capture_output=True,
                timeout=10,
            )
            print(f"  [U:Ar] Git commit: {message}", flush=True)
        except Exception as exc:
            print(f"  [U:Ar] Git commit failed: {exc}", flush=True)