orchestrata_ultimate.agents.volks.architect.architect_context

Reference › orchestrata_ultimate › agents › volks › architect › architect_context

architect_context is the context-building mixin of the Architect (volk). ArchitectContextMixin gathers existing-code context for the design prompt from the knowledge graph first and from the filesystem when the graph does not supply it. _build_full_monorepo_context combines two parts: the project tree (_build_project_tree_context and _query_project_tree, which query the graph for packages, modules and classes through a client from the graph factory) and the monorepo API sources (_build_monorepo_context). The latter picks capitalised class names mentioned in the requirements text, ignoring a set of noise words, and loads their source with _load_classes_from_graph and _load_graph_file. When the graph has nothing, _scan_filesystem_for_classes and _try_load_class_file scan the monorepo packages directory. _format_api_sources renders the result as the existing-API-sources block, and _slugify makes text directory-safe.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

ArchitectContextMixin

Line 38

Mixin providing knowledge-graph and filesystem context queries.

Source

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

# mypy: ignore-errors
"""Context-building mixin for the Architect volk.

Provides knowledge graph queries and filesystem fallback
for enriching design prompts with existing codebase context.
Extracted from volk.py to stay within the god-file/god-class limits.
"""

import logging
import os
import re
from collections.abc import Mapping

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.graph_factory import resolve_kg_database, resolve_uri

_logger = logging.getLogger(__name__)

_NOISE_WORDS: frozenset[str] = frozenset(
    {
        "Python",
        "Redis",
        "None",
        "True",
        "False",
        "READY",
        "QUESTIONS",
        "VERDICT",
        "API",
        "JSON",
        "HTTP",
        "URL",
        "ID",
    }
)


class ArchitectContextMixin:
    """Mixin providing knowledge-graph and filesystem context queries."""

    @traces(spec="030", story="US-030-005", reqs=["REQ-030-031"])
    async def _build_project_tree_context(
        self,
        env: Mapping[str, str] | None = None,
    ) -> str:
        """Query the graph for the full project tree (packages, modules, classes).

        Args:
            env: Optional environment mapping. Defaults to ``os.environ``.

        Returns:
            A comprehensive overview of the orchestrata monorepo structure
            for the Architect to understand the existing codebase before designing.
        """
        if env is None:
            env = os.environ
        try:
            from orchestrata_ultimate.knowledge_graph.config import GraphClientConfig
            from orchestrata_ultimate.knowledge_graph.graph_client import GraphClient

            graph_cfg = GraphClientConfig(
                uri=resolve_uri(env=env),
                username="",
                password="",
                database=resolve_kg_database(env=env),
            )

            client = GraphClient(config=graph_cfg)
            ok = await client.connect()
            if not ok:
                return ""

            try:
                return await self._query_project_tree(client)
            finally:
                await client.close()

        except Exception as exc:
            _logger.debug("project tree context failed: %s", exc)
            return ""

    @traces(spec="030", story="US-030-005", reqs=["REQ-030-031"])
    async def _query_project_tree(self, client: object) -> str:
        """Run graph queries and assemble the project tree context string.

        Args:
            client: Connected GraphClient instance.

        Returns:
            Formatted project tree overview string.
        """
        pkg_cypher = (
            "MATCH (p:Package) RETURN p.name AS name, p.path AS path ORDER BY p.name"
        )
        packages = await client.query(pkg_cypher, {})  # type: ignore[attr-defined]

        mod_cypher = (
            "MATCH (p:Package)-[:HAS_MODULE]->(m:Module) "
            "RETURN p.name AS package, m.name AS name, m.path AS path "
            "ORDER BY p.name, m.name"
        )
        modules = await client.query(mod_cypher, {})  # type: ignore[attr-defined]

        class_cypher = (
            "MATCH (m:Module)-[:DEFINES]->(c:Class) "
            "WITH m.name AS module, c.name AS name, c.line_count AS lc "
            "ORDER BY module, lc DESC "
            "WITH module, collect({name: name, lc: lc}) AS classes "
            "RETURN module, classes[..10] AS top_classes"
        )
        class_data = await client.query(class_cypher, {})  # type: ignore[attr-defined]

        func_cypher = (
            "MATCH (m:Module)-[:DEFINES]->(f:Function) "
            "WITH m.name AS module, f.name AS name, f.line_count AS lc "
            "ORDER BY module, lc DESC "
            "WITH module, collect({name: name, lc: lc}) AS funcs "
            "RETURN module, funcs[..10] AS top_funcs"
        )
        func_data = await client.query(func_cypher, {})  # type: ignore[attr-defined]

        sections = ["\n\n## ORCHESTRATA PROJECT TREE\n"]
        sections.append(f"### Packages ({len(packages)}):")
        for p in packages:
            sections.append(f"- **{p['name']}**")
        sections.append("")

        sections.append("### Modules by Package:")
        current_pkg = None
        for m in modules:
            if m["package"] != current_pkg:
                current_pkg = m["package"]
                sections.append(f"\n#### {current_pkg}")
            sections.append(f"  - {m['name']}")
        sections.append("")

        sections.append("### Key Classes (by module):")
        for cd in class_data:
            sections.append(f"\n#### {cd['module']}")
            for c in cd.get("top_classes", []):
                sections.append(f"  - {c['name']} ({c.get('lc', 0)} lines)")
        sections.append("")

        sections.append("### Key Functions (by module):")
        for fd in func_data:
            sections.append(f"\n#### {fd['module']}")
            for f in fd.get("top_funcs", []):
                sections.append(f"  - {f['name']} ({f.get('lc', 0)} lines)")

        return "\n".join(sections)

    @traces(spec="030", story="US-030-005", reqs=["REQ-030-032", "REQ-030-033"])
    async def _build_monorepo_context(self, text: str) -> str:
        """Query the knowledge graph for class sources referenced in text.

        Extracts CapWords class names from the intent/requirements, queries the
        graph scoped to the 'orchestrata' project node, and returns the real
        source files so U:Ar designs against the actual API.

        Falls back to filesystem scan if the graph is unreachable.

        Args:
            text: Requirements or intent text to scan for class names.

        Returns:
            Formatted context block with source code snippets, or ''.
        """
        cap_names = set(re.findall(r"\b([A-Z][a-zA-Z][a-zA-Z0-9]*)\b", text))
        cap_names -= _NOISE_WORDS
        if not cap_names:
            return ""

        loaded: dict[str, tuple[str, str]] = {}

        try:
            await self._load_classes_from_graph(cap_names, loaded)
        except Exception as exc:
            print(
                f"  [U:Ar] graph context query failed ({exc}),"
                " falling back to filesystem"
            )

        if not loaded:
            self._scan_filesystem_for_classes(cap_names, loaded)

        if not loaded:
            return ""

        return self._format_api_sources(loaded)

    @traces(spec="030", story="US-030-005", reqs=["REQ-030-034"])
    async def _load_classes_from_graph(
        self,
        cap_names: set[str],
        loaded: dict[str, tuple[str, str]],
        env: Mapping[str, str] | None = None,
    ) -> None:
        """Query the graph for the given class names and load their source files.

        Populates *loaded* in place. Raises on connection/query failure so the
        caller can fall back to a filesystem scan.

        Args:
            cap_names: CapWords class names to look up in the graph.
            loaded: Dict to populate; keyed by absolute file path.
            env: Optional environment mapping. Defaults to ``os.environ``.
        """
        if env is None:
            env = os.environ
        from orchestrata_ultimate.knowledge_graph.config import GraphClientConfig
        from orchestrata_ultimate.knowledge_graph.graph_client import GraphClient

        graph_cfg = GraphClientConfig(
            uri=resolve_uri(env=env),
            username="",
            password="",
            database=resolve_kg_database(env=env),
        )

        async def _query_graph() -> list[dict]:
            client = GraphClient(config=graph_cfg)
            ok = await client.connect()
            if not ok:
                return []
            try:
                cypher = (
                    "MATCH (proj:Project {name: $proj})"
                    "-[:HAS_FILE]->(f:File)-[:DEFINES]->(c:Class) "
                    "WHERE c.name IN $names "
                    "RETURN c.name AS cls, f.path AS path, "
                    "c.line_start AS ls, c.line_end AS le"
                )
                return await client.query(
                    cypher,
                    {"proj": "orchestrata", "names": list(cap_names)},
                )
            finally:
                await client.close()

        rows = await _query_graph()
        for row in rows:
            fpath = row["path"]
            if fpath in loaded or not os.path.isfile(fpath):
                continue
            self._load_graph_file(fpath, loaded)

    @traces(spec="030", story="US-030-005", reqs=["REQ-030-035"])
    def _load_graph_file(
        self,
        fpath: str,
        loaded: dict[str, tuple[str, str]],
    ) -> None:
        """Read and truncate a graph-referenced source file into *loaded*.

        Args:
            fpath: Absolute path to the source file.
            loaded: Dict to populate on success; keyed by absolute path.
        """
        try:
            with open(fpath, encoding="utf-8") as f:
                content = f.read()
            lines = content.splitlines()
            if len(lines) > 200:
                content = (
                    "\n".join(lines[:200])
                    + f"\n... (truncated at 200/{len(lines)} lines)"
                )
            rel = fpath.split("/packages/", 1)[-1] if "/packages/" in fpath else fpath
            loaded[fpath] = (rel, content)
        except Exception as exc:
            _logger.debug("read file %s: %s", fpath, exc)

    @traces(spec="030", story="US-030-005", reqs=["REQ-030-030"])
    def _format_api_sources(self, loaded: dict[str, tuple[str, str]]) -> str:
        """Render the loaded source files into the EXISTING API SOURCES block.

        Args:
            loaded: Mapping of absolute path to (relative_path, source).

        Returns:
            Formatted context block for prompt injection.
        """
        sections = [
            f"### {ref}\n```python\n{content}\n```"
            for _path, (ref, content) in loaded.items()
        ]
        return (
            "\n\nEXISTING API SOURCES (read these carefully — use the ACTUAL "
            "class signatures, constructor parameters, and import paths shown "
            "below. The file path tells you the exact import):\n\n"
            + "\n\n".join(sections)
        )

    @traces(spec="030", story="US-030-005", reqs=["REQ-030-036"])
    def _scan_filesystem_for_classes(
        self,
        cap_names: set[str],
        loaded: dict[str, tuple[str, str]],
        env: Mapping[str, str] | None = None,
    ) -> None:
        """Scan the monorepo packages directory for class source files.

        Populates *loaded* in place with (relative_path, source) tuples for
        every Python file that defines any name in *cap_names*.

        Args:
            cap_names: Set of CapWords class names to search for.
            loaded: Dict to populate; keyed by absolute file path.
            env: Optional environment mapping. Defaults to ``os.environ``.
        """
        if env is None:
            env = os.environ
        monorepo_root = env.get(
            "ORCHESTRATA_MONOREPO_ROOT", "/opt/orchestrata/packages"
        )
        if not os.path.isdir(monorepo_root):
            return
        for pkg_dir in sorted(os.listdir(monorepo_root)):
            pkg_src = os.path.join(monorepo_root, pkg_dir, "src")
            if not os.path.isdir(pkg_src):
                continue
            for root, dirs, files in os.walk(pkg_src):
                dirs[:] = [
                    d for d in dirs if not d.startswith(".") and d != "__pycache__"
                ]
                for fname in files:
                    if not fname.endswith(".py"):
                        continue
                    fpath = os.path.join(root, fname)
                    if fpath in loaded:
                        continue
                    self._try_load_class_file(fpath, monorepo_root, cap_names, loaded)

    @traces(spec="030", story="US-030-005", reqs=["REQ-030-035"])
    def _try_load_class_file(
        self,
        fpath: str,
        monorepo_root: str,
        cap_names: set[str],
        loaded: dict[str, tuple[str, str]],
    ) -> None:
        """Load *fpath* into *loaded* if it defines any class in *cap_names*.

        Args:
            fpath: Absolute path to the Python file to inspect.
            monorepo_root: Monorepo packages root for building the relative path.
            cap_names: Class names to match against.
            loaded: Dict to populate on match; keyed by absolute path.
        """
        try:
            with open(fpath, encoding="utf-8") as f:
                src = f.read()
        except Exception:
            return
        has_match = any(
            re.search(r"^class\s+" + re.escape(cn) + r"\b", src, re.M)
            for cn in cap_names
        )
        if not has_match:
            return
        lines = src.splitlines()
        if len(lines) > 200:
            src = (
                "\n".join(lines[:200]) + f"\n... (truncated at 200/{len(lines)} lines)"
            )
        rel = fpath.replace(monorepo_root + "/", "")
        loaded[fpath] = (rel, src)

    @traces(spec="030", story="US-030-004", reqs=["REQ-030-029"])
    def _slugify(self, text: str) -> str:
        """Convert text to a URL/directory-safe slug.

        Args:
            text: Input text to slugify.

        Returns:
            Lowercase hyphenated slug, max 50 characters.
        """
        text = text.lower()
        text = re.sub(r"[^\w\s-]", "", text)
        text = re.sub(r"[\s_]+", "-", text)
        text = re.sub(r"-+", "-", text)
        text = text.strip("-")
        if len(text) > 50:
            text = text[:50].rstrip("-")
        return text or "project"

    @traces(spec="030", story="US-030-005", reqs=["REQ-030-030", "REQ-030-037"])
    async def _build_full_monorepo_context(self, requirements: str) -> str:
        """Assemble the combined project-tree + monorepo API source context.

        Queries the graph for the project tree overview and for real source files
        that define classes referenced in the requirements text. Escapes braces
        so that the result is safe to pass to str.format().

        Args:
            requirements: Raw requirements text used to extract class names.

        Returns:
            Combined context string ready for prompt injection, with {{ }}
            escaping applied so the output is safe for str.format() calls.
        """
        project_tree_section = await self._build_project_tree_context()
        if project_tree_section:
            project_tree_section = project_tree_section.replace("{", "{{").replace(
                "}", "}}"
            )
            print(
                "  [U:Ar] Injecting project tree from knowledge graph",
                flush=True,
            )

        # G-92-01: escape {{ }} so .format() does not crash on {variable_names}
        original_intent = getattr(self, "_original_intent", "")
        monorepo_src_section = await self._build_monorepo_context(
            requirements + "\n" + original_intent
        )
        monorepo_src_section = monorepo_src_section.replace("{", "{{").replace(
            "}", "}}"
        )
        self._monorepo_src_section = monorepo_src_section  # type: ignore[attr-defined]
        if monorepo_src_section:
            print(
                "  [U:Ar] Injecting existing API sources into design prompt",
                flush=True,
            )

        if project_tree_section and monorepo_src_section:
            return project_tree_section + "\n\n" + monorepo_src_section
        return project_tree_section or monorepo_src_section