orchestrata_ultimate.agents.volks.codedirector.volk

Reference › orchestrata_ultimate › agents › volks › codedirector › volk

CodeDirectorVolk (alias U:Cd, role CodeDirector) turns the Architect's approved design into files on disk (specification 029). It is a thin coordinator: it prepares a skeleton, decides how to fill it, delegates the code writing to the Junior volk and runs a self-review on the result. It extends BaseAgent (base) and leans on three helper modules: implementation, review and prompts. CodeDirectorAgent is an alias subclass.

What it receives: an implement packet from the Architect (volk) with the design documents; implementation_complete packets from J:Junior (volk); and fix_issues packets from QA (volk). What it produces: a delegation packet for the Junior, an implementation_ready packet listing the created files, and, after a fix cycle, a review_and_test packet to U:Qa.

_handle_implement extracts the design documents, generates skeleton files from the class diagram (generate_skeleton_files), and asks Prolog for a fan-out decision (_get_fan_out_decision). If fan-out is allowed and a Postman is attached, _swarm_path runs the parallel swarm (coordinator); otherwise _serial_path builds a prompt (_build_aider_prompt) and delegates to the Junior. _handle_junior_complete records the Junior's output; if a fix is pending it returns the fix result, if the Junior reports failure it raises TaskExecutionError, and otherwise it checks the skeleton for structural escapes (_check_gen_skeleton_structural_escapes, against copies kept by _preserve_gen_skeleton_originals), runs the self-review loop (_run_self_review) and builds the ready packet with _make_ready_packet. _handle_fix_issues prepares a fix delegation for the Junior from QA's issue list and remembers the pending fix. _handle_review_edit is the deterministic floor for edit reviews. Work items are recorded through the session CLI when it is available (_record_session_work).

At a glance

Class diagram

Dependency diagram

API

Classes

CodeDirectorVolk (pydantic)

Bases: BaseAgent · Decorators: traces(spec='029', story='US-000', reqs=['REQ-029-001']) · Line 53

CodeDirector volk — thin coordinator for implementation.

CodeDirectorAgent (pydantic)

Bases: CodeDirectorVolk · Line 628

Agent alias for CodeDirectorVolk.

Functions

Source

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

# mypy: ignore-errors
"""U:Cd — CodeDirector (coder) volk with junior delegation via CXP."""

from __future__ import annotations

import logging
import shutil
import subprocess
import sys
from pathlib import Path
from typing import TYPE_CHECKING

from pydantic import Field

from orchestrata_ultimate.agents.exceptions import TaskExecutionError
from orchestrata_ultimate.agents.swarm.coordinator import _is_source_file
from orchestrata_ultimate.agents.volks._edit_review import (
    extract_edit_fields,
    has_bare_except,
    parse_fragment,
    regex_has_bare_except,
)
from orchestrata_ultimate.agents.volks.base import BaseAgent
from orchestrata_ultimate.agents.volks.codedirector.implementation import (
    append_prompt_extras,
    build_fallback_prompt,
    build_skeleton_prompt,
    execute_swarm_path,
    extract_design_docs,
    generate_skeleton_files,
    get_fan_out_decision,
)
from orchestrata_ultimate.agents.volks.codedirector.prompts import (  # noqa: F401
    CODEDIRECTOR_SYSTEM_PROMPT,
    MAX_SELF_REVIEW_CYCLES,
)
from orchestrata_ultimate.agents.volks.codedirector.review import (
    handle_fix_return,
    prepare_fix_delegation,
    self_review_loop,
)
from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.infra_config import WORKSPACE_DIR
from orchestrata_ultimate.core.packet import CXPPacket, WorkStatus

if TYPE_CHECKING:
    from orchestrata_ultimate.drivers.base import AsyncCodeDriver

_logger = logging.getLogger(__name__)


@traces(spec="029", story="US-000", reqs=["REQ-029-001"])
class CodeDirectorVolk(BaseAgent):
    """CodeDirector volk — thin coordinator for implementation."""

    alias: str = Field(default="U:Cd")
    role: str = "CodeDirector (Coder)"

    @traces(
        spec="029",
        story="Volk Identity",
        reqs=["REQ-029-006"],
    )
    def __init__(self):
        super().__init__()
        self.implementation_plan = None
        self.junior_output = None
        self.files_created: list[str] = []
        self.workspace = WORKSPACE_DIR
        self.project_dir = ""
        self.escalation_target = "U:Qa"
        self.task_summary = ""
        self._gen_skeleton_files: list[str] = []
        # Working directory for `session` CLI recording (None = inherit).
        # The session CLI auto-detects its project root by walking up from
        # cwd, so tests MUST point this at an isolated tmp dir or the record
        # lands in the live repository's context.json (SPEC-023).
        self.session_cwd: str | None = None

    @staticmethod
    def _repo_root() -> Path:
        """Return the repository root (the directory containing ``scripts/``)."""
        import orchestrata_ultimate

        return Path(orchestrata_ultimate.__file__).resolve().parent.parent.parent

    @traces(
        spec="029",
        story="Packet Routing",
        reqs=["REQ-029-002", "REQ-029-003", "REQ-029-004", "REQ-029-005"],
    )
    async def process(self, packet: CXPPacket) -> CXPPacket | None:
        """Route incoming CXP packets to handlers.

        Args:
            packet: Incoming CXP packet.

        Returns:
            Response CXP packet or None.
        """
        handlers = {
            "implement": self._handle_implement,
            "implementation_complete": self._handle_junior_complete,
            "fix_issues": self._handle_fix_issues,
            "review_edit": self._handle_review_edit,
        }
        handler = handlers.get(packet.subject)
        if handler:
            return await handler(packet)
        return None

    @traces(
        spec="029",
        story="Implement Handler",
        reqs=["REQ-029-007", "REQ-029-012"],
    )
    async def _handle_implement(self, pkt: CXPPacket) -> CXPPacket:
        """Handle implement — skeleton, swarm or serial path."""
        self._init_from_packet(pkt)
        print("  [U:Cd] Planning implementation...", flush=True)

        design_docs = extract_design_docs(pkt)
        skeleton_files = generate_skeleton_files(
            self.workspace, design_docs, self.task_summary
        )
        if skeleton_files:
            self.output_artifacts.append(
                "=== SKELETON FILES ===\n"
                + "\n".join(f"  - {f}" for f in skeleton_files)
            )
        source_files = [f for f in skeleton_files if _is_source_file(f)]
        prolog = getattr(self, "_prolog", None)
        decision = get_fan_out_decision(source_files, prolog, self.task_summary)
        if decision["allowed"] and self._postman is not None:
            return await self._swarm_path(
                pkt, skeleton_files, source_files, design_docs
            )
        return self._serial_path(pkt, skeleton_files, design_docs)

    @traces(
        spec="029",
        story="Packet Initialization",
        reqs=[
            "REQ-029-007",
            "REQ-029-008",
            "REQ-029-009",
            "REQ-029-010",
            "REQ-029-011",
        ],
    )
    def _init_from_packet(self, pkt: CXPPacket) -> None:
        """Initialize volk state from implement packet."""
        self.status = WorkStatus.IN_PROGRESS
        self.current_thread = pkt.thread_id
        self.summary = "Planning implementation"
        self.task_summary = pkt.payload.get("task_summary", "")
        self._original_intent = pkt.payload.get("original_intent", "")
        self.project_dir = pkt.payload.get("project_dir", "")
        if self.project_dir:
            self.workspace = self.project_dir

    @traces(
        spec="029",
        story="Swarm Path",
        reqs=["REQ-029-044", "REQ-029-045"],
    )
    async def _swarm_path(
        self,
        pkt: CXPPacket,
        skeleton_files: list[str],
        source_files: list[str],
        design_docs: dict,
    ) -> CXPPacket:
        """Execute parallel swarm coding path."""
        api_spec = design_docs.get("api_spec", "")
        self.files_created, artifact = await execute_swarm_path(
            self.workspace,
            self.task_summary,
            api_spec,
            self._original_intent,
            skeleton_files,
            source_files,
        )
        self.output_artifacts.append(artifact)
        if self.files_created:
            self.files_created = await self._run_self_review()
        self.summary = self._files_summary("swarm")
        return self._make_ready_packet(pkt, swarm=True)

    @traces(
        spec="029",
        story="Serial Path",
        reqs=["REQ-029-046"],
    )
    def _serial_path(
        self,
        pkt: CXPPacket,
        skeleton_files: list[str],
        design_docs: dict,
    ) -> CXPPacket:
        """Build aider prompt and delegate to J:Junior."""
        self._preserve_gen_skeleton_originals(skeleton_files)
        prompt = self._build_aider_prompt(pkt, skeleton_files, design_docs)
        self.implementation_plan = prompt
        self.output_artifacts.append(
            f"=== IMPLEMENTATION PROMPT ===\n{prompt[:500]}..."
        )
        print("  [U:Cd] Plan ready. Delegating to J:Junior...", flush=True)
        return CXPPacket(
            from_agent="U:Cd",
            to_agent="J:Junior",
            thread_id=pkt.thread_id,
            subject="implement",
            payload={
                "aider_prompt": prompt,
                "project_dir": self.workspace,
                "task_summary": self.task_summary,
                "skeleton_files": skeleton_files,
            },
            work_status=WorkStatus.IN_PROGRESS,
        )

    def _build_aider_prompt(
        self,
        pkt: CXPPacket,
        skeleton_files: list[str],
        design_docs: dict,
    ) -> str:
        """Build aider prompt from skeleton or fallback."""
        if skeleton_files:
            prompt = build_skeleton_prompt(
                skeleton_files, self.workspace, self.task_summary
            )
        else:
            prompt = build_fallback_prompt(
                workspace=self.workspace,
                task_summary=self.task_summary,
                requirements=design_docs.get("requirements", ""),
                architecture=design_docs.get("architecture", ""),
                file_tree=design_docs.get("file_tree", ""),
                api_spec=design_docs.get("api_spec", ""),
                class_diagram=design_docs.get("class_diagram", ""),
            )
        monorepo_src = pkt.payload.get("monorepo_src_section", "")
        return append_prompt_extras(prompt, self._original_intent, monorepo_src)

    @traces(
        spec="029",
        story="Junior Complete Handler",
        reqs=[
            "REQ-029-053",
            "REQ-029-054",
            "REQ-029-055",
            "REQ-029-056",
        ],
    )
    async def _handle_junior_complete(self, pkt: CXPPacket) -> CXPPacket:
        """Handle implementation_complete from J:Junior."""
        self.files_created = pkt.payload.get("files_created", [])
        self.junior_output = pkt.payload.get("output", "")
        self._log_junior_output()

        pending = getattr(self, "_pending_fix", None)
        if pending:
            self._pending_fix = None
            resp, self.files_created, self.summary = handle_fix_return(
                pkt, pending, self.workspace
            )
            return resp

        if not pkt.payload.get("success", True):
            # Dogfood run-1 F2 (fail-loud doctrine, mirrors the QA verdict
            # fail-open fix a8819e7d): a hard codegen driver failure must
            # never be reported as implementation_ready. `files_created`
            # here is whatever scan_workspace() happened to find in the
            # workspace (often pre-existing skeleton stubs from
            # generate_skeleton_files(), never Junior's own output when the
            # driver call itself failed) -- it is NOT a trustworthy "nothing
            # happened" signal, so `success` alone decides. Raising here
            # routes into PipelineRunner._run_stage's existing retry/
            # escalation classification (SPEC-003 `_is_retryable`) instead
            # of emitting a fabricated "ready" packet the pipeline can't
            # tell apart from success -- and skips the self-review loop
            # below entirely, so it never burns cycles reviewing stub files.
            raise TaskExecutionError(
                f"J:Junior codegen failed: {self.junior_output}",
                task_id=pkt.thread_id,
            )

        self._check_gen_skeleton_structural_escapes()

        if self.files_created:
            self.files_created = await self._run_self_review()
        self.summary = self._files_summary("generated")
        return self._make_ready_packet(pkt, full_output=True)

    @traces(
        spec="029",
        story="Fix Issues Handler",
        reqs=[
            "REQ-029-058",
            "REQ-029-059",
            "REQ-029-060",
            "REQ-029-061",
        ],
    )
    async def _handle_fix_issues(self, pkt: CXPPacket) -> CXPPacket:
        """Handle fix_issues from U:Qa — delegate to J:Junior."""
        self.status = WorkStatus.IN_PROGRESS
        self.current_thread = pkt.thread_id
        project_dir = pkt.payload.get("project_dir", "")
        if project_dir:
            self.workspace = project_dir
        issues = pkt.payload.get("issues", [])
        review_cycle = pkt.payload.get("review_cycle", 0)
        self.summary = f"Fixing {len(issues)} issues (cycle {review_cycle})"

        junior_pkt, pending = prepare_fix_delegation(pkt, self.workspace)
        self._pending_fix = pending
        self.output_artifacts.append(
            f"=== FIX PROMPT (cycle {review_cycle}) ===\n"
            f"{junior_pkt.payload.get('aider_prompt', '')}"
        )
        return junior_pkt

    @traces(
        spec="029",
        story="Ready Packet Construction",
        reqs=["REQ-029-069", "REQ-029-070", "REQ-029-071", "REQ-029-072"],
    )
    def _make_ready_packet(
        self,
        pkt: CXPPacket,
        swarm: bool = False,
        full_output: bool = False,
    ) -> CXPPacket:
        """Build implementation_ready packet for HUMAN."""
        payload = {
            "files_created": self.files_created,
            "task_summary": self.task_summary,
            "project_dir": self.project_dir,
        }
        if swarm:
            payload["swarm"] = True
        if full_output:
            payload["plan"] = self.implementation_plan
            payload["junior_output"] = self.junior_output
        return CXPPacket(
            from_agent="U:Cd",
            to_agent="HUMAN",
            thread_id=pkt.thread_id,
            subject="implementation_ready",
            payload=payload,
            work_status=WorkStatus.IN_PROGRESS,
        )

    def _preserve_gen_skeleton_originals(self, skeleton_files: list[str]) -> None:
        """Mirror gen-skeleton outputs so Gate 3 can diff after J:Junior fills.

        Gen-skeleton writes the skeleton file and a companion
        ``<file>.leafmap.json``.  J:Junior overwrites the skeleton file in
        place, so before delegating we copy each gen-skeleton ``.py`` file
        into ``<workspace>/.skeleton/<relative-path>``.  The ``.skeleton/``
        directory is hidden and is ignored by ``scan_workspace``.  Only files
        that have a leafmap are mirrored; scaffold-style skeletons are left
        alone.

        Args:
            skeleton_files: List of skeleton file paths produced by
                ``generate_skeleton_files``.
        """
        self._gen_skeleton_files: list[str] = []
        workspace_path = Path(self.workspace)
        for src in skeleton_files:
            src_path = Path(src)
            leafmap_path = Path(str(src) + ".leafmap.json")
            if not leafmap_path.exists():
                continue
            try:
                rel = src_path.relative_to(workspace_path)
            except ValueError:
                rel = src_path.name
            mirror = workspace_path / ".skeleton" / rel
            mirror.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(src_path, mirror)
            self._gen_skeleton_files.append(str(rel))

    def _check_gen_skeleton_structural_escapes(self) -> None:
        """Run Gate 3 ``skeleton-diff`` for any gen-skeleton files.

        This is observation mode: a STRUCTURAL escape is logged (and recorded
        via the session CLI if available) but the build is accepted anyway.
        Scaffold-generated skeletons are skipped.
        """
        gen_files = getattr(self, "_gen_skeleton_files", [])
        if not gen_files:
            return

        try:
            repo_root = self._repo_root()
        except Exception as exc:  # pragma: no cover - defensive
            _logger.warning("SKELETON_FLOW: cannot locate repo root for diff: %s", exc)
            return

        script_path = repo_root / "scripts" / "serialize_codebase_ast.py"
        workspace_path = Path(self.workspace)
        structural_count = 0

        for rel in gen_files:
            final_path = workspace_path / rel
            skeleton_path = workspace_path / ".skeleton" / rel
            leafmap_path = Path(str(final_path) + ".leafmap.json")
            if not skeleton_path.exists() or not leafmap_path.exists():
                continue
            try:
                result = subprocess.run(
                    [
                        sys.executable,
                        str(script_path),
                        "skeleton-diff",
                        str(skeleton_path),
                        str(final_path),
                        "--leafmap",
                        str(leafmap_path),
                    ],
                    cwd=str(repo_root),
                    capture_output=True,
                    text=True,
                    timeout=30,
                )
            except Exception as exc:  # pragma: no cover - defensive
                _logger.warning(
                    "SKELETON_FLOW: skeleton-diff failed for %s: %s", rel, exc
                )
                continue

            if result.returncode == 1 and "STRUCTURAL" in (
                result.stdout or result.stderr
            ):
                structural_count += 1
                detail = (result.stdout.strip() or result.stderr.strip())[:500]
                _logger.warning(
                    "SKELETON_FLOW: STRUCTURAL escape in %s: %s",
                    rel,
                    detail,
                )
                self._record_session_work(
                    f"SKELETON_FLOW structural escape: {rel}",
                    detail,
                )

        if structural_count:
            print(
                f"  [U:Cd] SKELETON_FLOW: {structural_count} structural escape(s) "
                "recorded (build accepted in observation mode)",
                flush=True,
            )

    def _record_session_work(self, desc: str, detail: str) -> None:
        """Record a work item via the session CLI if it is available.

        Fail-open: if the CLI is missing or the call fails, the warning log
        already captured the event.

        Args:
            desc: Short description for ``session work``.
            detail: Detail bullet.
        """
        try:
            subprocess.run(
                ["session", "work", desc, "--detail", detail],
                capture_output=True,
                timeout=10,
                check=False,
                cwd=self.session_cwd,
            )
        except Exception as exc:  # pragma: no cover - defensive
            _logger.debug("SKELETON_FLOW: session work recording failed: %s", exc)

    @traces(spec="029", reqs=["REQ-029-045", "REQ-029-055"])
    async def _run_self_review(self) -> list[str]:
        """Run self-review loop on current files.

        W4.3b: the review-generation call inside ``self_review_loop`` routes
        through ``self._call_llm_with_tools`` (SPEC-103 tool-calling loop)
        instead of a bare ``call_llm``, so U:Cd can inspect the real codebase
        while reviewing (mirrors W4.3a's Architect migration).
        """
        return await self_review_loop(
            task_summary=self.task_summary,
            files_created=self.files_created,
            workspace=self.workspace,
            output_artifacts=self.output_artifacts,
            get_junior_driver_fn=self._get_junior_driver,
            call_llm_with_tools_fn=self._call_llm_with_tools,
        )

    def _get_junior_driver(self) -> AsyncCodeDriver:
        """Get junior driver for self-review fix loop."""
        from orchestrata_ultimate.drivers.driver_factory import get_driver

        return get_driver(volk="J:Junior", workspace=self.workspace)

    @traces(
        spec="029",
        story="Junior Output Logging",
        reqs=["REQ-029-057"],
    )
    def _log_junior_output(self) -> None:
        """Log junior output and files to artifacts."""
        self.output_artifacts.append(
            f"=== JUNIOR VOLK OUTPUT ===\n{self.junior_output}"
        )
        files_display = "\n".join(f"  - {f}" for f in self.files_created) or "  (none)"
        self.output_artifacts.append(f"=== FILES CREATED ===\n{files_display}")
        print(
            f"  [U:Cd] Junior done. Files: {len(self.files_created)}",
            flush=True,
        )

    def _files_summary(self, mode: str) -> str:
        """Build summary string for current files state."""
        if self.files_created:
            return f"Code {mode} — {len(self.files_created)} files"
        return f"No files produced ({mode}) — review needed"

    def _extract_aider_prompt(self, plan: str) -> str | None:
        """Extract AIDER_PROMPT section from plan text."""
        if "AIDER_PROMPT:" in plan:
            return plan.split("AIDER_PROMPT:")[1].strip()
        return None

    def _get_fan_out_decision(self, source_files: list) -> dict:
        """Query Prolog governance for swarm fan-out decision.

        Falls back to Python heuristic if Prolog unavailable.

        Args:
            source_files: List of source file paths to shard.

        Returns:
            Dict with allowed, strategy, max_shards, reason.
        """
        prolog = getattr(self, "_prolog", None)
        return get_fan_out_decision(source_files, prolog, self.task_summary)

    @traces(spec="050", story="US-001", reqs=["REQ-050-FR1-001"])
    async def _handle_review_edit(self, packet: CXPPacket) -> CXPPacket:
        """SPEC-050 P3.1 fail-closed FLOOR — not a correctness review.

        Deterministic, no LLM call — this must be cheap and reliable so the
        CheetahClaws /edit fail-closed gate (MultiAgentCoordinator) gets a
        real APPROVE/BLOCK verdict instead of always TIMING OUT. This is a
        syntax/deletion/architecture-heuristic floor; it does not assess
        whether the edit is a *good* implementation.

        Rejects:
          - new_code that fails to parse as Python (tried as-is, dedented,
            and wrapped in a synthetic enclosing scope — see
            ``_edit_review.parse_fragment`` — since ``new_code`` is
            typically an indented FRAGMENT, not a top-level-valid module),
            when the target looks like Python.
          - edits that delete more than 80% of the original file's content.
          - edits that introduce a bare ``except:`` clause not already
            present in the original code.
        """
        file_path, old_code, new_code = _extract_edit_fields(packet.payload)
        verdict = "APPROVE"
        reason = "No correctness issues detected."

        looks_like_python = file_path.endswith(".py") or (
            not file_path and ("def " in new_code or "import " in new_code)
        )
        new_tree = None
        if looks_like_python:
            new_tree, parse_error = parse_fragment(new_code)
            if new_tree is None:
                verdict, reason = "BLOCK", f"Syntax error in new code: {parse_error}"

        if verdict == "APPROVE" and old_code and len(new_code) < 0.2 * len(old_code):
            verdict = "BLOCK"
            reason = "Edit deletes more than 80% of the file's content."

        if verdict == "APPROVE":
            if new_tree is not None:
                new_bare = has_bare_except(new_tree)
            else:
                new_bare = regex_has_bare_except(new_code)
            old_tree, _old_error = (
                parse_fragment(old_code) if old_code else (None, None)
            )
            old_bare = (
                has_bare_except(old_tree)
                if old_tree is not None
                else regex_has_bare_except(old_code)
            )
            if new_bare and not old_bare:
                verdict = "BLOCK"
                reason = "Edit introduces a bare `except:` clause."

        return CXPPacket(
            from_agent=self.alias,
            to_agent=packet.from_agent,
            thread_id=packet.thread_id,
            subject="review_edit_result",
            payload={
                "verdict": verdict,
                "reason": reason,
                "content": reason,
                "file_path": file_path,
            },
            work_status=WorkStatus.RESOLVED,
        )


def _extract_edit_fields(payload: dict) -> tuple[str, str, str]:
    """Extract (file_path, old_code, new_code) from a review_edit payload.

    Thin re-export of the shared helper (see
    ``orchestrata_ultimate.agents.volks._edit_review.extract_edit_fields``)
    kept under this name for backward compatibility with existing call
    sites/tests in this module.
    """
    return extract_edit_fields(payload)


# Class alias for orchestrata-ultimate naming
class CodeDirectorAgent(CodeDirectorVolk):
    """Agent alias for CodeDirectorVolk."""

    pass