orchestrata_ultimate.agents.volks.junior.volk

Reference › orchestrata_ultimate › agents › volks › junior › volk

JuniorVolk (alias J:Junior, role Junior Developer) has two jobs in the pipeline (specification 027): it reviews the Architect's design before any code is written, and it generates code when the CodeDirector delegates files to it. It extends BaseAgent (base). It is imported by the CodeDirector package, the kernel pipeline and the swarm path. JuniorAgent is an alias subclass whose constructor accepts an injected code driver.

Design review. It receives a design_review packet from the Architect (volk) and produces a design_feedback packet with the verdict READY or QUESTIONS and the question list. _handle_design_review builds the prompt (_build_review_prompt), calls the LLM (_call_design_review_llm) and _build_review_response turns the reply into a feedback packet: the verdict is READY when the reply says so or yields no questions, and parse_questions extracts numbered or bulleted questions (_extract_question_line, _is_section_break). When the maximum number of design cycles is reached, _auto_ready returns READY without calling the model.

Implementation. It receives an implement packet from the CodeDirector with an aider prompt, a list of files and an is_fix flag, and produces an implementation_complete packet to U:Cd with the created files, the driver output, a success flag and dependency issues. _handle_implement obtains a driver from the driver factory for the junior alias, runs it in the workspace (set by _apply_project_dir), then scan_workspace lists created files and validate_dependencies checks Python files for imports that are neither declared dependencies nor local project modules (_collect_project_modules, _check_single_file); issues are logged by _log_deps_issues and returned, not raised.

At a glance

Class diagram

Dependency diagram

API

Classes

JuniorVolk (pydantic)

Bases: BaseAgent · Decorators: traces(spec='027', story='Volk Identity and Lifecycle', reqs=['REQ-027-001', 'REQ-027-002', 'REQ-027-003', 'REQ-027-004… · Line 101

Junior developer volk -- reviews designs and generates code.

JuniorAgent (pydantic)

Bases: JuniorVolk · Line 712

Agent alias for JuniorVolk with driver injection support.

Functions

Source

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

# mypy: ignore-errors
"""J:Junior -- Junior developer volk with design review and code generation.

Reviews architecture designs, asks blocking questions (max 3 cycles),
then generates code using the configured driver backend. Validates
dependencies after generation and scans workspace for created files.
"""

from __future__ import annotations

import logging
import os
from typing import TYPE_CHECKING

from pydantic import Field

from orchestrata_ultimate.agents.volks.base import BaseAgent
from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.graph_directives import extract_graph_context
from orchestrata_ultimate.core.infra_config import PROJECT_WORKSPACE_ROOT
from orchestrata_ultimate.core.packet import CXPPacket, WorkStatus
from orchestrata_ultimate.core.standards import CODING_STANDARDS

if TYPE_CHECKING:
    from orchestrata_ultimate.drivers.base import AsyncCodeDriver

_logger = logging.getLogger(__name__)

MAX_DESIGN_REVIEW_CYCLES = 3

DESIGN_REVIEW_PROMPT = (
    "You are J:Junior, a junior developer reviewing an architecture "
    "design before you implement it. This is design review cycle {cycle} "
    "of maximum {max_cycles}.\n\n"
    "YOUR GOAL: Determine whether you can implement this design RIGHT "
    "NOW without guessing. If yes, say READY. If no, ask the minimum "
    "questions needed to unblock yourself.\n\n"
    "TASK FIDELITY CHECK (do this BEFORE the checklist below, on every "
    'cycle): Re-read the "Original Task" and "Requirements" sections '
    "of the design below. Does the ARCHITECTURE/FILE_TREE/API_SPEC "
    "actually address THAT task and THOSE requirements? If the design "
    "describes a different system, solves an unrelated problem, or "
    "ignores requirements that were explicitly stated, that is a "
    "blocking gap regardless of how complete the design otherwise looks "
    "-- respond VERDICT: QUESTIONS and name the exact mismatch (e.g. "
    '"the design implements X but the task asks for Y"). Do not say '
    "READY on a design that does not match the stated task.\n\n"
    "REVIEW CHECKLIST -- walk through each item mentally:\n"
    "  C1. IMPORT PATHS: Can you write every import statement? For each "
    "file in FILE_TREE, do you know the exact package path?\n"
    "  C2. FUNCTION SIGNATURES: Does API_SPEC provide the complete "
    "signature (parameters, types, return type) for every public "
    "function?\n"
    "  C3. DEPENDENCY GRAPH: Is the dependency order clear? Can you "
    "determine which file to implement first?\n"
    "  C4. PYPROJECT.TOML: Are all third-party dependencies named?\n"
    "  C5. FILE_TREE vs API_SPEC CONSISTENCY: Does every file in "
    "FILE_TREE have a corresponding entry in API_SPEC?\n"
    "  C6. ERROR HANDLING: For I/O functions, does the spec tell you "
    "what exception to raise? (NOT blocking.)\n"
    "  C7. TEST STRATEGY: Is it clear what each test file should test? "
    "(NOT blocking.)\n\n"
    "CONVERGENCE RULES (strictly enforced):\n"
    "- Cycle 1: Ask up to 5 questions about items C1-C5 ONLY.\n"
    "- Cycle 2: Ask up to 3 questions. Only if C1-C5 still have gaps.\n"
    "- Cycle {max_cycles}+: Say READY regardless.\n"
    "- NEVER ask about C6 or C7.\n"
    "- NEVER re-ask a question that was already answered.\n"
    "- If the design is good enough to start coding, say READY.\n\n"
    "OUTPUT FORMAT -- respond with EXACTLY one of:\n\n"
    "Option A (no blocking gaps):\n"
    "VERDICT: READY\n\n"
    "Option B (blocking gaps found):\n"
    "VERDICT: QUESTIONS\n"
    "QUESTIONS:\n"
    "1. [C<n>] <specific blocking question>\n"
    "2. [C<n>] <specific blocking question>\n"
    "...\n\n"
    "IMPORTANT: Each question MUST reference which checklist item "
    "(C1-C5) it addresses.\n\n"
    'HANDOFF: Your VERDICT is returned to U:Ar as a "design_feedback" '
    "packet (READY or QUESTIONS).\n"
    # REQ-027-031: append the shared CODING_STANDARDS, brace-escaped so
    # DESIGN_REVIEW_PROMPT.format(cycle=, max_cycles=) leaves them intact.
) + CODING_STANDARDS.replace("{", "{{").replace("}", "}}")


@traces(
    spec="027",
    story="Volk Identity and Lifecycle",
    reqs=[
        "REQ-027-001",  # JuniorVolk defined as subclass of BaseAgent
        "REQ-027-002",  # JuniorVolk alias set to J:Junior
        "REQ-027-003",  # JuniorVolk role set to Junior Developer
        "REQ-027-004",  # JuniorVolk initializes workspace from PROJECT_WORKSPACE_ROOT
        "REQ-027-005",  # JuniorVolk initializes files_created as empty list
        "REQ-027-006",  # JuniorVolk initializes junior_output as None
        "REQ-027-070",  # JuniorVolk decorated with @traces
    ],
)
class JuniorVolk(BaseAgent):
    """Junior developer volk -- reviews designs and generates code.

    Handles two CXP subjects:
    - design_review: Reviews architecture, asks blocking questions
    - implement: Generates code via configured driver backend
    """

    alias: str = Field(default="J:Junior")
    role: str = "Junior Developer"

    def __init__(self) -> None:
        """Initialize Junior volk with workspace and artifact tracking."""
        super().__init__()
        self.workspace: str = PROJECT_WORKSPACE_ROOT
        self.files_created: list[str] = []
        self.junior_output: str | None = None

    @traces(
        spec="027",
        story="CXP Routing",
        reqs=[
            "REQ-027-007",  # process() routes design_review subject
            "REQ-027-008",  # process() routes implement subject
            "REQ-027-009",  # process() returns None for unrecognised subjects
            "REQ-027-010",  # process() extracts graph context from packet payload
        ],
    )
    async def process(self, packet: CXPPacket) -> CXPPacket | None:
        """Route incoming CXP packet to the appropriate handler.

        Args:
            packet: Incoming CXP packet to process.

        Returns:
            Response CXP packet, or None if subject is unhandled.
        """
        self._graph_context = extract_graph_context(packet.payload)
        if packet.subject == "design_review":
            return await self._handle_design_review(packet)
        elif packet.subject == "implement":
            return await self._handle_implement(packet)
        return None

    @traces(
        spec="027",
        story="Design Review — Core Flow",
        reqs=[
            "REQ-027-011",  # _handle_design_review() sets status = IN_PROGRESS
            "REQ-027-012",  # _handle_design_review() reads design_cycle from payload (default 1)
            "REQ-027-013",  # _handle_design_review() sets current_thread from packet.thread_id
            "REQ-027-014",  # When design_cycle >= MAX_DESIGN_REVIEW_CYCLES, _auto_ready() called
            "REQ-027-015",  # When design_cycle < MAX_DESIGN_REVIEW_CYCLES, LLM called
            "REQ-027-016",  # LLM response appended to output_artifacts
        ],
    )
    async def _handle_design_review(self, packet: CXPPacket) -> CXPPacket:
        """Review architecture and ask questions or signal READY.

        Args:
            packet: CXP packet with design payload.

        Returns:
            Feedback packet addressed to U:Ar.
        """
        self.status = WorkStatus.IN_PROGRESS
        self.current_thread = packet.thread_id
        cycle = packet.payload.get("design_cycle", 1)
        self.summary = f"Reviewing design (cycle {cycle})"
        _logger.info("Reviewing design (cycle %d)", cycle)

        if cycle >= MAX_DESIGN_REVIEW_CYCLES:
            return self._auto_ready(packet, cycle)

        response = await self._call_design_review_llm(packet, cycle)
        self.output_artifacts.append(
            f"=== DESIGN REVIEW (cycle {cycle}) ===\n{response}"
        )
        return self._build_review_response(packet, response, cycle)

    @traces(
        spec="027",
        story="Implementation — Core Flow",
        reqs=[
            "REQ-027-045",  # _handle_implement() sets status = IN_PROGRESS
            "REQ-027-046",  # _handle_implement() calls _apply_project_dir()
            "REQ-027-047",  # _handle_implement() resolves driver via get_driver()
            "REQ-027-048",  # Driver execute() receives aider_prompt, files, is_fix
            "REQ-027-049",  # Driver output stored in junior_output and output_artifacts
            "REQ-027-050",  # scan_workspace() called after driver execution
            "REQ-027-051",  # validate_dependencies() called after scan
            "REQ-027-052",  # Dependency issues logged via _log_deps_issues()
            "REQ-027-053",  # implementation_complete packet addressed to U:Cd
            "REQ-027-054",  # implementation_complete payload contains files_created, output, success, deps_issues
            "REQ-027-055",  # status set to RESOLVED after implementation completes
        ],
    )
    async def _handle_implement(self, packet: CXPPacket) -> CXPPacket:
        """Generate code using the configured driver backend.

        Args:
            packet: CXP packet with implementation payload.

        Returns:
            Completion packet addressed to U:Cd.
        """
        self.status = WorkStatus.IN_PROGRESS
        self.current_thread = packet.thread_id
        self._apply_project_dir(packet)
        self.summary = "Generating code"
        _logger.info("Implementing in %s", self.workspace)

        from orchestrata_ultimate.drivers.driver_factory import get_driver

        driver = get_driver(volk="J:Junior", workspace=self.workspace)
        success, output = await driver.execute(
            packet.payload.get("aider_prompt", ""),
            files=packet.payload.get("files", []),
            is_fix=packet.payload.get("is_fix", False),
        )

        self.junior_output = output
        self.output_artifacts.append(f"=== JUNIOR OUTPUT ===\n{output}")
        self.files_created = scan_workspace(self.workspace)
        deps_issues = validate_dependencies(self.workspace, self.files_created)
        self._log_deps_issues(deps_issues)

        self.summary = f"Code generated -- {len(self.files_created)} files"
        self.status = WorkStatus.RESOLVED
        return CXPPacket(
            from_agent="J:Junior",
            to_agent="U:Cd",
            thread_id=packet.thread_id,
            subject="implementation_complete",
            payload={
                "files_created": self.files_created,
                "output": output,
                "success": success,
                "deps_issues": deps_issues,
            },
            work_status=WorkStatus.IN_PROGRESS,
        )

    @traces(
        spec="027",
        story="auto-READY",
        reqs=[
            "REQ-027-041",  # _auto_ready() sets status = RESOLVED
            "REQ-027-042",  # _auto_ready() sets current_thread = None
            "REQ-027-043",  # _auto_ready() emits design_feedback packet with verdict = READY
            "REQ-027-044",  # _auto_ready() payload response contains human-readable auto-READY message
        ],
    )
    def _auto_ready(self, packet: CXPPacket, cycle: int) -> CXPPacket:
        """Return auto-READY when max design cycles reached.

        Args:
            packet: The original design review packet.
            cycle: Current cycle number.

        Returns:
            READY feedback packet to U:Ar.
        """
        _logger.info("Design review: READY (auto, last cycle)")
        self.summary = "Design reviewed -- ready to implement"
        self.status = WorkStatus.RESOLVED
        self.current_thread = None
        return CXPPacket(
            from_agent="J:Junior",
            to_agent="U:Ar",
            thread_id=packet.thread_id,
            subject="design_feedback",
            payload={
                "verdict": "READY",
                "response": "Auto-READY: max design cycles reached",
                "design_cycle": cycle,
            },
            work_status=WorkStatus.IN_PROGRESS,
        )

    @traces(
        spec="027",
        story="Design Review — DESIGN_REVIEW_PROMPT",
        reqs=[
            "REQ-027-030",  # LLM called with temperature=0.0 for deterministic review
        ],
    )
    async def _call_design_review_llm(self, packet: CXPPacket, cycle: int) -> str:
        """Call LLM to review the design.

        Args:
            packet: CXP packet containing design documents.
            cycle: Current review cycle number.

        Returns:
            Raw LLM response string.
        """
        system_prompt = DESIGN_REVIEW_PROMPT.format(
            cycle=cycle, max_cycles=MAX_DESIGN_REVIEW_CYCLES
        )
        review_prompt = _build_review_prompt(packet.payload)
        # W4.5 (backlog #14): route through the base-agent seam so this call
        # gets memory recall (SPEC-010) folded in, same as persona composition
        # already did. _call_llm stamps volk="J:Junior" (self.alias) itself.
        return await self._call_llm(
            system_prompt,
            review_prompt,
            temperature=0.0,
            graph_context=self._graph_context,
        )

    @traces(
        spec="027",
        story="Design Review — Response Parsing and Packet Construction",
        reqs=[
            "REQ-027-035",  # _build_review_response() returns READY when VERDICT: READY found
            "REQ-027-036",  # _build_review_response() returns READY when no questions parsed
            "REQ-027-037",  # _build_review_response() returns QUESTIONS when questions parsed
        ],
    )
    def _build_review_response(
        self,
        packet: CXPPacket,
        response: str,
        cycle: int,
    ) -> CXPPacket:
        """Parse LLM response and build feedback packet.

        Args:
            packet: Original design review packet.
            response: Raw LLM response text.
            cycle: Current review cycle number.

        Returns:
            Feedback packet to U:Ar with verdict and questions.
        """
        is_ready = "VERDICT: READY" in response
        questions = parse_questions(response)

        if is_ready or not questions:
            _logger.info("Design review: READY")
            self.summary = "Design reviewed -- ready to implement"
            self.status = WorkStatus.RESOLVED
            self.current_thread = None
            return _make_feedback_packet(packet.thread_id, "READY", response, cycle)

        _logger.info("Design review: %d questions", len(questions))
        self.summary = f"Asked {len(questions)} design questions"
        self.status = WorkStatus.RESOLVED
        self.current_thread = None
        return _make_feedback_packet(
            packet.thread_id,
            "QUESTIONS",
            response,
            cycle,
            questions=questions,
        )

    @traces(
        spec="027",
        story="Implementation — Core Flow",
        reqs=[
            "REQ-027-046",  # _handle_implement() calls _apply_project_dir()
        ],
    )
    def _apply_project_dir(self, packet: CXPPacket) -> None:
        """Set workspace from packet if provided.

        Args:
            packet: CXP packet that may contain project_dir.
        """
        project_dir = packet.payload.get("project_dir", "")
        if project_dir:
            self.workspace = project_dir

    @traces(
        spec="027",
        story="Implementation — Core Flow",
        reqs=[
            "REQ-027-052",  # Dependency issues logged via _log_deps_issues()
        ],
    )
    def _log_deps_issues(self, deps_issues: list[str]) -> None:
        """Log dependency validation issues.

        Args:
            deps_issues: List of dependency issue strings.
        """
        if deps_issues:
            _logger.warning("Found %d dependency issues", len(deps_issues))
            for issue in deps_issues:
                _logger.warning("  - %s", issue)


def _build_review_prompt(payload: dict) -> str:
    """Assemble the user prompt for design review from payload fields.

    Args:
        payload: CXP packet payload containing design documents.

    Returns:
        Formatted review prompt string.
    """
    task_summary = payload.get("task_summary", "")
    architecture = payload.get("architecture", "")
    requirements = payload.get("requirements", "")
    file_tree = payload.get("file_tree", "")
    api_spec = payload.get("api_spec", "")
    class_diagram = payload.get("class_diagram", "")

    # Dogfood run-2 F3 defect 4: U:Ar's design_review packet already carries
    # task_summary (ArchitectVolk._make_review_packet), but this prompt
    # builder never surfaced it, so the reviewer had no way to check the
    # design actually addresses the stated task (DESIGN_REVIEW_PROMPT's
    # TASK FIDELITY CHECK references this exact "Original Task" section).
    task_section = f"Original Task:\n{task_summary}\n\n" if task_summary else ""
    class_section = f"Class Diagram:\n{class_diagram}\n\n" if class_diagram else ""
    return (
        f"{task_section}"
        f"Architecture:\n{architecture}\n\n"
        f"Requirements:\n{requirements}\n\n"
        f"File Tree:\n{file_tree}\n\n"
        f"API Specification:\n{api_spec}\n\n"
        f"{class_section}"
        "Review this design carefully. Do you have everything "
        "you need to implement this? Ask questions or say READY."
    )


@traces(
    spec="027",
    story="Design Review — Response Parsing and Packet Construction",
    reqs=[
        "REQ-027-038",  # design_feedback packet addressed to U:Ar
        "REQ-027-039",  # design_feedback payload contains verdict, response, design_cycle
        "REQ-027-040",  # design_feedback payload contains questions only when verdict is QUESTIONS
    ],
)
def _make_feedback_packet(
    thread_id: str,
    verdict: str,
    response: str,
    cycle: int,
    questions: list[str] | None = None,
) -> CXPPacket:
    """Create a design_feedback CXP packet for U:Ar.

    Args:
        thread_id: Thread ID for the packet.
        verdict: READY or QUESTIONS.
        response: Full LLM response text.
        cycle: Current design cycle number.
        questions: Optional list of blocking questions.

    Returns:
        Feedback CXP packet addressed to U:Ar.
    """
    payload: dict = {
        "verdict": verdict,
        "response": response,
        "design_cycle": cycle,
    }
    if questions:
        payload["questions"] = questions
    return CXPPacket(
        from_agent="J:Junior",
        to_agent="U:Ar",
        thread_id=thread_id,
        subject="design_feedback",
        payload=payload,
        work_status=WorkStatus.IN_PROGRESS,
    )


@traces(
    spec="027",
    story="Design Review — Response Parsing",
    reqs=[
        "REQ-027-032",  # parse_questions() extracts numbered lines
        "REQ-027-033",  # parse_questions() extracts bulleted lines
        "REQ-027-034",  # parse_questions() stops on first non-question
    ],
)
def parse_questions(response: str) -> list[str]:
    """Parse QUESTIONS list from LLM response.

    Extracts numbered or bulleted questions from the QUESTIONS:
    section of the LLM response.

    Args:
        response: Raw LLM response text.

    Returns:
        List of question strings.
    """
    questions: list[str] = []
    in_questions = False
    for line in response.splitlines():
        stripped = line.strip()
        if stripped.startswith("QUESTIONS:"):
            in_questions = True
            continue
        if not in_questions:
            continue
        question = _extract_question_line(stripped)
        if question is not None:
            questions.append(question)
        elif _is_section_break(stripped):
            break
    return questions


@traces(
    spec="027",
    story="Design Review — Response Parsing",
    reqs=[
        "REQ-027-033",  # _extract_question_line extracts bulleted lines
    ],
)
def _extract_question_line(stripped: str) -> str | None:
    """Extract a question from a numbered or bulleted line.

    Args:
        stripped: Stripped line text.

    Returns:
        Question text, or None if line is not a question.
    """
    if stripped and stripped[0].isdigit() and ". " in stripped:
        question = stripped.split(". ", 1)[1].strip()
        return question if question else None
    if stripped.startswith("- "):
        question = stripped[2:].strip()
        return question if question else None
    return None


@traces(
    spec="027",
    story="Design Review — Response Parsing",
    reqs=[
        "REQ-027-034",  # _is_section_break stops extraction on first non-question
    ],
)
def _is_section_break(stripped: str) -> bool:
    """Check if a line breaks the QUESTIONS section.

    Args:
        stripped: Stripped line text.

    Returns:
        True if this line terminates the questions section.
    """
    if not stripped:
        return False
    if stripped[0].isdigit():
        return False
    if stripped.startswith("-"):
        return False
    return True


@traces(
    spec="027",
    story="scan_workspace",
    reqs=[
        "REQ-027-056",  # scan_workspace() returns empty list when workspace doesn't exist
        "REQ-027-057",  # scan_workspace() excludes hidden directories
        "REQ-027-058",  # scan_workspace() excludes __pycache__
        "REQ-027-059",  # scan_workspace() excludes Module: prefixed directories
        "REQ-027-060",  # scan_workspace() excludes hidden files
        "REQ-027-061",  # scan_workspace() returns relative paths
        "REQ-027-062",  # scan_workspace() returns sorted paths
    ],
)
def scan_workspace(workspace: str) -> list[str]:
    """Scan workspace directory for created files.

    Walks the workspace tree, skipping hidden dirs, __pycache__,
    and Module: prefixed directories. Returns sorted relative paths.

    Args:
        workspace: Absolute path to the workspace directory.

    Returns:
        Sorted list of relative file paths.
    """
    files: list[str] = []
    if not os.path.exists(workspace):
        return files
    for root, dirs, filenames in os.walk(workspace):
        dirs[:] = [
            d
            for d in dirs
            if not d.startswith(".")
            and d != "__pycache__"
            and not d.startswith("Module:")
        ]
        for fname in filenames:
            if not fname.startswith("."):
                rel = os.path.relpath(os.path.join(root, fname), workspace)
                files.append(rel)
    return sorted(files)


@traces(
    spec="027",
    story="validate_dependencies",
    reqs=[
        "REQ-027-063",  # validate_dependencies() returns empty when no deps declared
        "REQ-027-064",  # validate_dependencies() skips non-.py files
        "REQ-027-065",  # validate_dependencies() uses _parse_file() for AST
        "REQ-027-066",  # validate_dependencies() skips unparseable files
        "REQ-027-067",  # validate_dependencies() collects local modules from src/
        "REQ-027-068",  # validate_dependencies() calls check_dependency_declaration()
        "REQ-027-069",  # Issue strings formatted as [CHECK_ID] rel_path: description
    ],
)
def validate_dependencies(workspace: str, files_created: list[str]) -> list[str]:
    """Run dependency check on generated Python files.

    Scans each generated .py source file against the project's
    pyproject.toml to find third-party imports not declared
    as dependencies.

    Args:
        workspace: Absolute path to the workspace directory.
        files_created: List of relative file paths to check.

    Returns:
        List of human-readable dependency issue strings.
    """
    from orchestrata_ultimate.verification.review.helpers import _get_pyproject_deps

    declared_deps = _get_pyproject_deps(workspace)
    if not declared_deps:
        return []

    project_modules = _collect_project_modules(workspace)
    issues: list[str] = []
    for rel_path in files_created:
        if not rel_path.endswith(".py"):
            continue
        file_issues = _check_single_file(
            workspace, rel_path, declared_deps, project_modules
        )
        issues.extend(file_issues)
    return issues


@traces(spec="027", story="validate_dependencies", reqs=["REQ-027-067"])
def _collect_project_modules(workspace: str) -> set[str]:
    """Collect local project module names from src/ directory.

    Args:
        workspace: Absolute path to workspace.

    Returns:
        Set of local module names.
    """
    project_modules: set[str] = set()
    src_dir = os.path.join(workspace, "src")
    if os.path.isdir(src_dir):
        for name in os.listdir(src_dir):
            if os.path.isdir(os.path.join(src_dir, name)):
                project_modules.add(name)
    return project_modules


@traces(
    spec="027",
    story="validate_dependencies",
    reqs=["REQ-027-065", "REQ-027-066", "REQ-027-068", "REQ-027-069"],
)
def _check_single_file(
    workspace: str,
    rel_path: str,
    declared_deps: set[str],
    project_modules: set[str],
) -> list[str]:
    """Check a single file for undeclared dependency imports.

    Args:
        workspace: Absolute path to workspace.
        rel_path: Relative path to the Python file.
        declared_deps: Set of declared dependency names.
        project_modules: Set of local project module names.

    Returns:
        List of formatted issue strings.
    """
    filepath = os.path.join(workspace, rel_path)
    if not os.path.exists(filepath):
        return []
    from orchestrata_ultimate.verification.review.checks_correctness import (
        check_dependency_declaration,
    )
    from orchestrata_ultimate.verification.review.helpers import _parse_file

    tree = _parse_file(filepath)
    if tree is None:
        return []
    file_issues = check_dependency_declaration(
        filepath, tree, declared_deps, project_modules
    )
    return [
        f"[{check_id}] {rel_path}: {desc}" for check_id, _fname, desc in file_issues
    ]


# Class alias for orchestrata-ultimate naming
class JuniorAgent(JuniorVolk):
    """Agent alias for JuniorVolk with driver injection support."""

    def __init__(self, driver: AsyncCodeDriver | None = None) -> None:
        super().__init__()
        self._driver_override = driver