orchestrata_ultimate.agents.volks.codedirector.review

Reference › orchestrata_ultimate › agents › volks › codedirector › review

review implements the CodeDirector's self-review loop and fix-cycle helpers (specification 029). volk is the only importer; the module itself uses implementation and prompts.

ReviewContext bundles the mutable state of the loop: task summary, created files, workspace, output artifacts and the callables for getting a Junior driver and calling the LLM. self_review_loop runs cycles through _run_review_cycle: each cycle reads the sources (_read_source_files), asks the LLM for a review (_run_review, parsed with parse_verdict and parse_issues), and, when issues are found, prioritises and caps them (prioritize_issues, _cap_issues) and applies fixes through a Junior driver (_apply_review_fixes, using build_fix_prompt). The fix-cycle helpers serve QA bounces: prepare_fix_delegation builds the fix prompt and the packet for the Junior and a pending-fix record; handle_fix_return processes the Junior's reply and either sends the files to QA (_build_qa_packet, subject review_and_test) or reports a failed fix (_build_fix_failed_response).

At a glance

Class diagram

Dependency diagram

API

Classes

ReviewContext

Decorators: traces(spec='029', story='Self-Review Loop', reqs=['REQ-029-065']) · Line 90

Bundle of mutable state for the review loop.

Functions

Source

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

# mypy: ignore-errors
"""Self-review loop and fix cycle helpers for CodeDirector."""

from __future__ import annotations

import inspect
import logging
from collections.abc import Awaitable, Callable
from typing import TYPE_CHECKING

from orchestrata_ultimate.agents.volks.codedirector.implementation import (
    read_file_contents,
    scan_workspace,
)
from orchestrata_ultimate.agents.volks.codedirector.prompts import (
    MAX_SELF_REVIEW_CYCLES,
    MAX_SELF_REVIEW_ISSUES,
    SELF_REVIEW_PROMPT,
)
from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import CXPPacket, WorkStatus
from orchestrata_ultimate.core.standards import CODING_STANDARDS_FIX

if TYPE_CHECKING:
    from orchestrata_ultimate.drivers.base import AsyncCodeDriver

_logger = logging.getLogger(__name__)

# Signature of BaseAgent._call_llm_with_tools (system_prompt, user_message,
# **kwargs) -> final text response, as injected by CodeDirectorVolk so this
# free-function module can drive the tool-calling loop without importing
# agents.volks.base (which would be a layering inversion: base -> ... -> this
# package is fine, the reverse is not).
CallLLMWithToolsFn = Callable[..., Awaitable[str]]


@traces(
    spec="029",
    story="Self-Review Loop",
    reqs=["REQ-029-063", "REQ-029-064", "REQ-029-065", "REQ-029-066"],
)
async def self_review_loop(
    task_summary: str,
    files_created: list[str],
    workspace: str,
    output_artifacts: list[str],
    get_junior_driver_fn: Callable[[], AsyncCodeDriver],
    call_llm_with_tools_fn: CallLLMWithToolsFn,
    graph_context: str | None = None,
) -> list[str]:
    """Review junior's code up to MAX_SELF_REVIEW_CYCLES.

    Args:
        task_summary: Short task description.
        files_created: List of relative file paths.
        workspace: Project workspace directory.
        output_artifacts: Mutable list to append artifacts.
        get_junior_driver_fn: Callable returning junior driver.
        call_llm_with_tools_fn: ``BaseAgent._call_llm_with_tools`` bound by the
            caller (CodeDirectorVolk), so the review-generation call drives the
            real tool-calling loop with U:Cd's registered tools.
        graph_context: Optional graph context string from CXP packet payload.

    Returns:
        Updated files_created list after review fixes.
    """
    ctx = ReviewContext(
        task_summary,
        files_created,
        workspace,
        output_artifacts,
        get_junior_driver_fn,
        call_llm_with_tools_fn,
    )
    for cycle in range(1, MAX_SELF_REVIEW_CYCLES + 1):
        result = await _run_review_cycle(ctx, cycle, graph_context=graph_context)
        if result is None:
            break
        ctx.files_created = result
    else:
        print(
            f"  [U:Cd] Self-review did not converge after "
            f"{MAX_SELF_REVIEW_CYCLES} cycles",
            flush=True,
        )
    return ctx.files_created


@traces(spec="029", story="Self-Review Loop", reqs=["REQ-029-065"])
class ReviewContext:
    """Bundle of mutable state for the review loop."""

    def __init__(
        self,
        task_summary: str,
        files_created: list[str],
        workspace: str,
        output_artifacts: list[str],
        get_junior_driver_fn: Callable[[], AsyncCodeDriver],
        call_llm_with_tools_fn: CallLLMWithToolsFn,
    ) -> None:
        self.task_summary = task_summary
        self.files_created = files_created
        self.workspace = workspace
        self.output_artifacts = output_artifacts
        self.get_junior_driver_fn = get_junior_driver_fn
        self.call_llm_with_tools_fn = call_llm_with_tools_fn


@traces(spec="029", reqs=["REQ-029-063"])
async def _run_review_cycle(
    ctx: ReviewContext,
    cycle: int,
    graph_context: str | None = None,
) -> list[str] | None:
    """Run one self-review cycle. Returns None to stop.

    Args:
        ctx: Review loop context.
        cycle: Current cycle number.
        graph_context: Optional graph context string from CXP packet payload.

    Returns:
        Updated files list to continue, or None to stop.
    """
    print(
        f"  [U:Cd] Self-review cycle {cycle}/{MAX_SELF_REVIEW_CYCLES}...",
        flush=True,
    )
    source = _read_source_files(ctx.workspace, ctx.files_created)
    if not source:
        print("  [U:Cd] No source files — skipping", flush=True)
        return None

    verdict, issues = await _run_review(
        ctx.task_summary,
        source,
        ctx.output_artifacts,
        cycle,
        ctx.call_llm_with_tools_fn,
        graph_context=graph_context,
    )
    if verdict == "PASS" or not issues:
        print(f"  [U:Cd] Self-review PASS at cycle {cycle}", flush=True)
        return None

    issues = _cap_issues(issues, MAX_SELF_REVIEW_ISSUES)
    _print_issues(issues)
    return await _apply_review_fixes(ctx, issues, source, cycle)


@traces(spec="029", reqs=["REQ-029-066"])
async def _run_review(
    task_summary: str,
    source_listing: str,
    output_artifacts: list[str],
    cycle: int,
    call_llm_with_tools_fn: CallLLMWithToolsFn,
    graph_context: str | None = None,
) -> tuple[str, list[str]]:
    """Run a single review cycle via the tool-calling LLM loop.

    Args:
        task_summary: Short task description.
        source_listing: Current source file contents.
        output_artifacts: Mutable list to append artifacts.
        cycle: Current review cycle number.
        call_llm_with_tools_fn: ``BaseAgent._call_llm_with_tools`` bound by the
            caller, driving the tool-calling loop (SPEC-103) with U:Cd's
            registered tools instead of a single-shot text completion.
        graph_context: Optional graph context string from CXP packet payload.

    Returns:
        Tuple of (verdict, issues_list).
    """
    review_prompt = (
        f"Task: {task_summary}\n\n"
        f"Files in workspace:\n{source_listing}\n\n"
        "Review this code and provide your verdict."
    )
    from orchestrata_ultimate.agents.volks.persona import compose_persona

    # W4.3b: route the primary review-generation call through the
    # tool-calling loop (BaseAgent._call_llm_with_tools, bound in by the volk
    # as call_llm_with_tools_fn) so U:Cd can inspect the real codebase
    # (tool_registry.AGENT_TOOLS["U:Cd"]: read_file/write_file/list_dir/
    # search_code/ast_query/run_shell) while reviewing, instead of reasoning
    # only from the pasted source_listing. This replaces the direct
    # ``call_llm`` call and mirrors W4.3a's Architect migration (see
    # architect/volk.py::_handle_design).
    #
    # ``_call_llm_with_tools`` takes no ``graph_context=`` kwarg (unlike
    # ``call_llm``) -- it only forwards temperature/model to
    # ``call_with_tools`` -- so graph_context is folded directly into the
    # user turn here (Documentor/Architect precedent), or it would be
    # silently dropped before reaching the LLM.
    user_message = review_prompt
    if graph_context:
        user_message = graph_context + "\n\n---\n\n" + review_prompt
    review_response = await call_llm_with_tools_fn(
        compose_persona(SELF_REVIEW_PROMPT),
        user_message,
        temperature=0.0,
    )
    output_artifacts.append(f"=== SELF-REVIEW (cycle {cycle}) ===\n{review_response}")
    verdict = parse_verdict(review_response)
    issues = parse_issues(review_response)
    return verdict, issues


@traces(spec="029", reqs=["REQ-029-065"])
async def _apply_review_fixes(
    ctx: ReviewContext,
    issues: list[str],
    source_listing: str,
    cycle: int,
) -> list[str]:
    """Apply fixes from a review cycle via junior driver."""
    issues_text = "\n".join(f"- {issue}" for issue in issues)
    fix_prompt = build_fix_prompt(ctx.task_summary, issues_text, source_listing)
    print(f"  [U:Cd] Delegating self-review fixes (cycle {cycle})...", flush=True)
    junior = ctx.get_junior_driver_fn()
    # Typed as the union the isawaitable() check below actually handles:
    # AsyncCodeDriver.execute() is async per protocol, but real drivers are
    # accessed dynamically (driver_factory) so we defend against a
    # non-conforming sync implementation too.
    result: tuple[bool, str] | Awaitable[tuple[bool, str]] = junior.execute(
        fix_prompt, files=ctx.files_created, is_fix=True
    )
    if inspect.isawaitable(result):
        _, output = await result
    else:
        _, output = result
    ctx.output_artifacts.append(f"=== SELF-REVIEW FIX (cycle {cycle}) ===\n{output}")
    updated = scan_workspace(ctx.workspace)
    print(f"  [U:Cd] Self-review fixes applied. Files: {len(updated)}", flush=True)
    return updated


def _cap_issues(issues: list[str], max_count: int) -> list[str]:
    """Cap issue list and log if truncated.

    Args:
        issues: Full list of issues.
        max_count: Maximum issues to keep.

    Returns:
        Capped list of issues.
    """
    if len(issues) > max_count:
        print(
            f"  [U:Cd] Self-review FAIL — {len(issues)} "
            f"issues (capping to {max_count})",
            flush=True,
        )
        return issues[:max_count]
    print(
        f"  [U:Cd] Self-review FAIL — {len(issues)} issues",
        flush=True,
    )
    return issues


def _print_issues(issues: list[str]) -> None:
    """Print issue list to stdout.

    Args:
        issues: List of issue strings.
    """
    for issue in issues:
        print(f"    - {issue}", flush=True)


@traces(spec="029", story="File Content Reading", reqs=["REQ-029-075"])
def _read_source_files(workspace: str, files_created: list[str]) -> str:
    """Read all source files from workspace.

    Args:
        workspace: Project workspace directory.
        files_created: List of relative file paths.

    Returns:
        Concatenated source listing string.
    """
    return read_file_contents(workspace, files_created)


@traces(spec="029", story="Fix Issues Handler", reqs=["REQ-029-060"])
def prioritize_issues(issues: list) -> list:
    """Sort issues by severity and cap to 5.

    Args:
        issues: Raw issue strings from QA review.

    Returns:
        Sorted and capped list of issue strings.
    """
    critical_tags = ("FILE_COMPLETE", "SYNTAX", "LINT", "IMPORTS")
    cosmetic_tags = ("LINE_LENGTH", "DOCSTRING", "TODO")

    def _priority(issue_str: str) -> int:
        for i, tag in enumerate(critical_tags):
            if f"[{tag}]" in issue_str:
                return i
        for i, tag in enumerate(cosmetic_tags):
            if f"[{tag}]" in issue_str:
                return len(critical_tags) + i
        return len(critical_tags)

    issues = sorted(issues, key=_priority)
    max_fix_issues = 5
    if len(issues) > max_fix_issues:
        print(
            f"  [U:Cd] Capping {len(issues)} issues to "
            f"{max_fix_issues} (highest priority)",
            flush=True,
        )
        issues = issues[:max_fix_issues]
    return issues


def build_fix_prompt(
    task_summary: str,
    issues_text: str,
    source_listing: str,
) -> str:
    """Build a direct fix prompt without intermediate LLM call.

    Args:
        task_summary: Short description of the task.
        issues_text: Formatted issues list.
        source_listing: Current file contents.

    Returns:
        Complete fix prompt for the junior driver.
    """
    return (
        f"Fix these issues in the existing code.\n\n"
        f"Task: {task_summary}\n\n"
        f"Issues to fix:\n{issues_text}\n\n"
        f"Current files:\n{source_listing}\n\n"
        f"Output ALL modified files as complete FILE: blocks.\n"
        f"Do NOT use diffs. Output COMPLETE file contents.\n\n"
        f"{CODING_STANDARDS_FIX}"
    )


@traces(spec="029", reqs=["REQ-029-063"])
def parse_verdict(response: str) -> str:
    """Parse REVIEW_VERDICT from LLM response.

    Args:
        response: Raw LLM review response text.

    Returns:
        'PASS' or 'FAIL' string.
    """
    for line in response.splitlines():
        stripped = line.strip()
        if not stripped.startswith("REVIEW_VERDICT:"):
            continue
        verdict = stripped.split(":", 1)[1].strip().upper()
        if "PASS" in verdict:
            return "PASS"
        if "FAIL" in verdict:
            return "FAIL"
    return "FAIL"


@traces(spec="029", reqs=["REQ-029-063"])
def parse_issues(response: str) -> list:
    """Parse ISSUES list from LLM response.

    Args:
        response: Raw LLM review response text.

    Returns:
        List of issue description strings.
    """
    issues: list[str] = []
    in_issues = False
    for line in response.splitlines():
        stripped = line.strip()
        if stripped.startswith("ISSUES:"):
            in_issues = True
            rest = stripped.split(":", 1)[1].strip()
            if rest and rest != "-":
                issues.append(rest)
            continue
        if not in_issues:
            continue
        if stripped.startswith("- "):
            issue_text = stripped[2:].strip()
            if issue_text:
                issues.append(issue_text)
        elif stripped and not stripped.startswith("-"):
            break
    return issues


@traces(spec="029", reqs=["REQ-029-054"])
def handle_fix_return(
    packet: CXPPacket,
    pending: dict,
    workspace: str,
) -> tuple[CXPPacket, list[str], str]:
    """Process return from a fix cycle delegation.

    Args:
        packet: CXP packet from J:Junior.
        pending: Pending fix context dict.
        workspace: Project workspace directory.

    Returns:
        Tuple of (response_packet, files_created, summary).
    """
    junior_success = packet.payload.get("success", True)
    files_created = scan_workspace(workspace)

    if not junior_success and pending["from_test_fix"]:
        return _build_fix_failed_response(
            packet,
            pending,
            files_created,
            workspace,
        )
    if not junior_success:
        print(
            f"  [U:Cd] Fix cycle {pending['review_cycle']} "
            f"produced no changes — proceeding",
            flush=True,
        )

    summary = f"Fixes applied (cycle {pending['review_cycle']})"
    print(f"  [U:Cd] Fixes applied. Files: {len(files_created)}", flush=True)
    response = _build_qa_packet(packet, pending, files_created, workspace)
    return response, files_created, summary


@traces(spec="029", story="Junior Complete Handler", reqs=["REQ-029-054"])
def _build_qa_packet(
    packet: CXPPacket,
    pending: dict,
    files_created: list[str],
    workspace: str,
) -> CXPPacket:
    """Build review_and_test packet for U:Qa.

    Args:
        packet: CXP packet from J:Junior.
        pending: Pending fix context dict.
        files_created: Current file list.
        workspace: Project workspace directory.

    Returns:
        CXP packet to U:Qa.
    """
    return CXPPacket(
        from_agent="U:Cd",
        to_agent="U:Qa",
        thread_id=packet.thread_id,
        subject="review_and_test",
        payload={
            "files_created": files_created,
            "task_summary": pending["task_summary"],
            "issues": pending["issues"],
            "from_test_fix": pending["from_test_fix"],
            "project_dir": workspace,
        },
        work_status=WorkStatus.IN_PROGRESS,
    )


@traces(spec="029", story="Junior Complete Handler", reqs=["REQ-029-054"])
def _build_fix_failed_response(
    packet: CXPPacket,
    pending: dict,
    files_created: list[str],
    workspace: str,
) -> tuple[CXPPacket, list[str], str]:
    """Build response for a failed test-fix cycle.

    Args:
        packet: CXP packet from J:Junior.
        pending: Pending fix context dict.
        files_created: Current file list.
        workspace: Project workspace directory.

    Returns:
        Tuple of (response_packet, files_created, summary).
    """
    print(
        f"  [U:Cd] Test-fix cycle {pending['review_cycle']}"
        f" produced no changes — skipping re-review",
        flush=True,
    )
    response = CXPPacket(
        from_agent="U:Cd",
        to_agent="HUMAN",
        thread_id=packet.thread_id,
        subject="implementation_ready",
        payload={
            "files_created": files_created,
            "task_summary": pending["task_summary"],
            "project_dir": workspace,
            "fix_failed": True,
        },
        work_status=WorkStatus.IN_PROGRESS,
    )
    return response, files_created, "Test-fix cycle failed — no changes"


@traces(spec="029", reqs=["REQ-029-060"])
def prepare_fix_delegation(
    packet: CXPPacket,
    workspace: str,
) -> tuple[CXPPacket, dict]:
    """Build fix prompt and delegation packet for J:Junior."""
    p = packet.payload
    issues = prioritize_issues(p.get("issues", []))
    files_created = p.get("files_created", [])
    task_summary = p.get("task_summary", "")
    review_cycle = p.get("review_cycle", 0)
    print(f"  [U:Cd] Received {len(issues)} issues (cycle {review_cycle})", flush=True)
    source = read_file_contents(workspace, files_created)
    fix_prompt = build_fix_prompt(
        task_summary, "\n".join(f"- {i}" for i in issues), source
    )
    print("  [U:Cd] Delegating fixes to J:Junior via CXP...", flush=True)
    pending = {
        "task_summary": task_summary,
        "issues": issues,
        "from_test_fix": p.get("from_test_fix", False),
        "review_cycle": review_cycle,
    }
    junior_pkt = CXPPacket(
        from_agent="U:Cd",
        to_agent="J:Junior",
        thread_id=packet.thread_id,
        subject="implement",
        payload={
            "aider_prompt": fix_prompt,
            "project_dir": workspace,
            "files": files_created,
            "task_summary": task_summary,
            "is_fix": True,
        },
        work_status=WorkStatus.IN_PROGRESS,
    )
    return junior_pkt, pending