orchestrata_ultimate.agents.volks.documentor.volk

Reference › orchestrata_ultimate › agents › volks › documentor › volk

DocumentorVolk (alias U:Do, role Documentor) is the technical writer of the pipeline. It extends BaseAgent (base), takes its system prompt from prompts, and is imported by the package __init__ and the kernel pipeline. DocumentorAgent is an alias subclass.

What it receives: a documentation packet (generate_docs, update_docs or documentation) whose payload carries a task summary or requirements and, for updates, existing doc paths. What it produces: a documentation_complete packet addressed to its escalation target (the Compliance volk, so documentation is captured in traceability) with doc type, doc path, summary, content, @traces references, the task summary and the output artifacts.

_handle_documentation builds a prompt that tells the model to list directories, search code, read existing docs, query the AST and write the documentation, then calls the tool loop at a low temperature (0.4) for factual output. An empty reply raises TaskExecutionError. _parse_output extracts DOC_TYPE, DOC_PATH, DOC_SUMMARY, TRACES, TASK_SUMMARY and CONTENT, and _persist_doc_content writes the content to the doc path in the workspace; if there is neither content nor a file at the path, it raises TaskExecutionError again. _build_escalation_packet assembles the outgoing packet.

At a glance

Class diagram

Dependency diagram

API

Classes

DocumentorVolk (pydantic)

Bases: BaseAgent · Line 23

Documentor agent — documentation generation and updates.

DocumentorAgent (pydantic)

Bases: DocumentorVolk · Line 248

Agent alias for DocumentorVolk.

Source

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

"""U:Do — Documentor agent for documentation generation and updates.

Reads existing code and documentation, then produces or updates documentation
files (READMEs, specs, architecture docs, flow diagrams, API docs).
"""

import logging
from typing import Any

from pydantic import Field

from orchestrata_ultimate.agents.exceptions import TaskExecutionError
from orchestrata_ultimate.agents.volks.base import BaseAgent
from orchestrata_ultimate.agents.volks.documentor.prompts import (
    DOCUMENTOR_SYSTEM_PROMPT,
)
from orchestrata_ultimate.core.graph_directives import extract_graph_context
from orchestrata_ultimate.core.packet import CXPPacket, WorkStatus

_logger = logging.getLogger(__name__)


class DocumentorVolk(BaseAgent):
    """Documentor agent — documentation generation and updates.

    Handles six work types: GENERATE_NEW, UPDATE_EXISTING, GENERATE_SPEC,
    GENERATE_FLOW, GENERATE_README, GENERATE_API_DOCS. Uses tools
    (read_file, write_file, search_code, ast_query, list_dir) to understand
    the codebase and produce accurate documentation.
    """

    alias: str = Field(default="U:Do")
    role: str = "Documentor"

    def __init__(self, **kwargs: Any) -> None:
        super().__init__(**kwargs)
        self.doc_type: str = ""
        self.doc_path: str = ""
        self.doc_summary: str = ""
        self.traces_refs: list[str] = []
        self.generated_content: str = ""
        self.doc_content: str = ""
        self.escalation_target = "U:Cp"  # Escalate to Compliance for traceability

    async def process(self, packet: CXPPacket) -> CXPPacket | None:
        """Route incoming CXP packets to the appropriate handler.

        Args:
            packet: Incoming CXP packet with documentation task description.

        Returns:
            Response CXP packet with documentation output, or None.
        """
        self._graph_context = extract_graph_context(packet.payload)
        if packet.subject in ("generate_docs", "update_docs", "documentation"):
            return await self._handle_documentation(packet)
        return None

    async def _handle_documentation(self, packet: CXPPacket) -> CXPPacket:
        """Generate or update documentation based on the task.

        Args:
            packet: CXP packet containing:
                - requirements: The task description
                - task_summary: One-line summary
                - existing_docs: List of existing doc paths (for updates)

        Returns:
            CXP packet with documentation result.
        """
        self.status = WorkStatus.IN_PROGRESS
        self.current_thread = packet.thread_id
        task = packet.payload.get(
            "task_summary", packet.payload.get("requirements", "Document task")
        )
        self.task_summary = task
        self.summary = "Generating documentation"

        _logger.info("Documentor processing: %s", task)

        # Build prompt. ``_call_llm_with_tools`` has no ``graph_context=`` kwarg
        # (unlike ``_call_llm``) — the tool-calling loop only forwards
        # temperature/model — so the graph context extracted above must be
        # folded directly into the prompt text, or it's silently dropped and
        # never reaches the LLM (every other volk forwards ``self._graph_context``
        # to its LLM call; Documentor was computing it and then discarding it).
        graph_block = (
            f"GRAPH CONTEXT:\n{self._graph_context}\n\n" if self._graph_context else ""
        )
        prompt = (
            f"{graph_block}"
            f"TASK: {task}\n\n"
            f"CONTEXT:\n{packet.payload.get('requirements', '')}\n\n"
            f"INSTRUCTIONS:\n"
            f"1. Use list_dir to understand the project structure.\n"
            f"2. Use search_code to find relevant existing code and docs.\n"
            f"3. Use read_file to read any existing documentation that needs updating.\n"
            f"4. Use ast_query to understand class structure if documenting code.\n"
            f"5. Write the documentation using write_file.\n"
            f"6. Output DOC_TYPE, DOC_PATH, DOC_SUMMARY, TRACES, TASK_SUMMARY, CONTENT.\n"
        )

        # Use tool-calling LLM for full documentation workflow
        response = await self._call_llm_with_tools(
            DOCUMENTOR_SYSTEM_PROMPT,
            prompt,
            temperature=0.4,  # Lower temperature for factual documentation
        )

        if not response or not response.strip():
            # Dogfood run-3d F3 backstop: base.py's Defect-1 finalization call
            # (a forced no-tools turn once the tool budget is exhausted)
            # already recovers most truncated-to-blank cases, but if the
            # whole tool-calling conversation still comes back with nothing,
            # RESOLVING here would hand U:Cp an empty deliverable to check
            # traceability against (which then fails downstream anyway).
            # Fail loud instead so this routes into PipelineRunner's
            # existing retry/escalation classification (SPEC-003), mirroring
            # the Architect blank-design guard. No retry here -- Defect-1's
            # finalization call already is the recovery attempt.
            raise TaskExecutionError(
                "U:Do: tool-loop response was empty -- no documentation "
                "content produced",
                task_id=packet.thread_id,
            )

        self.output_artifacts.append(f"=== DOCUMENTATION OUTPUT ===\n{response}")
        self._parse_output(response)

        # Dogfood run-3f F14: the model's authored document (the CONTENT
        # section) was parsed but never written anywhere -- U:Cp then failed
        # the run for a missing deliverable even though the LLM produced a
        # full document. Persist it now (unless the tool phase already wrote
        # a file at DOC_PATH that's at least as large -- tool-written wins
        # ties). Fail loud if neither a usable CONTENT body nor a file at
        # DOC_PATH exists afterwards, mirroring the empty-response guard
        # above: RESOLVING here would hand U:Cp a deliverable-less packet.
        doc_persisted = self._persist_doc_content()
        if not self.doc_content.strip() and not doc_persisted:
            raise TaskExecutionError(
                "U:Do: documentation produced no persistable content -- "
                "no CONTENT body and no file at DOC_PATH",
                task_id=packet.thread_id,
            )

        self.status = WorkStatus.RESOLVED
        self.summary = f"Documentation: {self.doc_type} → {self.doc_path}"

        return self._build_escalation_packet(packet)

    def _parse_output(self, response: str) -> None:
        """Parse structured fields from LLM response.

        Args:
            response: Full LLM response text.
        """
        if "DOC_TYPE:" in response:
            self.doc_type = response.split("DOC_TYPE:")[1].split("\n")[0].strip()
        if "DOC_PATH:" in response:
            self.doc_path = response.split("DOC_PATH:")[1].split("\n")[0].strip()
        if "DOC_SUMMARY:" in response:
            self.doc_summary = response.split("DOC_SUMMARY:")[1].split("\n")[0].strip()
        if "TRACES:" in response:
            traces_line = response.split("TRACES:")[1].split("\n")[0].strip()
            self.traces_refs = [t.strip() for t in traces_line.split(",") if t.strip()]
        if "TASK_SUMMARY:" in response:
            self.task_summary = (
                response.split("TASK_SUMMARY:")[1].split("\n")[0].strip()
            )
        if "CONTENT:" in response:
            # Unlike the fields above, CONTENT is the document BODY -- it
            # runs from the marker to the end of the response and may itself
            # contain colons, headings, and code fences, so it must NOT be
            # line-split like the one-line metadata fields (dogfood run-3f
            # F14: the previous parser discarded this section entirely).
            self.doc_content = response.split("CONTENT:", 1)[1].strip()

    def _persist_doc_content(self) -> bool:
        """Write ``self.doc_content`` to ``self.doc_path`` in the workspace.

        Returns:
            True if a file exists at the resolved ``doc_path`` afterwards
            (whether just written, kept from a larger pre-existing/
            tool-written file, or unreachable because ``doc_path`` was never
            set or escapes the workspace); False otherwise.
        """
        if not self.doc_path:
            return False
        # Resolve the workspace BEFORE the prefix comparison — the workspace
        # may arrive as a relative path (dogfood F17: the e2e harness exports
        # ORCHESTRATA_WORKSPACE relative to cwd), and comparing a resolved
        # absolute target against an unresolved relative base refuses every
        # legitimate write. ToolExecutor avoids this by resolving in __init__.
        workspace = self._get_workspace().resolve()
        # Same resolve-and-verify-prefix approach as ToolExecutor._safe_path
        # (tool_registry.py) -- reimplemented inline per the design brief
        # rather than importing ToolExecutor for this.
        target = (workspace / self.doc_path).resolve()
        if not str(target).startswith(str(workspace)):
            _logger.warning(
                "U:Do: DOC_PATH %r escapes the workspace -- skipping write",
                self.doc_path,
            )
            return False
        existing = target.read_text(encoding="utf-8") if target.exists() else None
        if existing is not None and len(existing) >= len(self.doc_content):
            # The tool phase already wrote an equal-or-larger file here --
            # tool-written file wins ties, never clobber it.
            return True
        if not self.doc_content.strip():
            return existing is not None
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_text(self.doc_content, encoding="utf-8")
        _logger.info(
            "U:Do: wrote documentation to %s (%d bytes)",
            target,
            len(self.doc_content.encode("utf-8")),
        )
        return True

    def _build_escalation_packet(self, packet: CXPPacket) -> CXPPacket:
        """Build escalation packet with documentation output.

        Args:
            packet: Original incoming packet.

        Returns:
            CXP packet for U:Cp with documentation artifacts.
        """
        return CXPPacket(
            from_agent="U:Do",
            to_agent=self.escalation_target,
            thread_id=packet.thread_id,
            subject="documentation_complete",
            payload={
                "doc_type": self.doc_type,
                "doc_path": self.doc_path,
                "doc_summary": self.doc_summary,
                "doc_content": self.doc_content,
                "traces": self.traces_refs,
                "task_summary": self.task_summary,
                "output_artifacts": list(self.output_artifacts),
            },
            work_status=WorkStatus.RESOLVED,
        )


class DocumentorAgent(DocumentorVolk):
    """Agent alias for DocumentorVolk."""

    pass