orchestrata_ultimate.agents.volks.qa._context

Reference › orchestrata_ultimate › agents › volks › qa › _context

_context provides FailureContextMixin for the QA volk (volk): it analyses failures and builds fix guidance that gets more detailed as the same problems recur. _extract_test_failures pulls failed test names and short errors from pytest output. For constructor TypeErrors, _get_signature_hints returns real class signatures, found by mapping class names to import modules from the tests (_build_class_module_map) and inspecting the classes (_probe_class_signatures). _build_escalating_context assembles the fix context by cycle number: from cycle 3 a summary of the last three failed attempts (_escalation_history_section), from cycle 5 a flag for issues that persist across cycles (_escalation_root_cause_section), and from cycle 8 the inlined source of the first few failing files (_escalation_file_contents_section). _bounce_for_fix packages that context into a packet sent back to the CodeDirector.

At a glance

Class diagram

Dependency diagram

API

Classes

FailureContextMixin

Bases: _MixinBase · Line 24

Failure analysis and escalating-context construction for U:Qa.

Source

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

# mypy: ignore-errors
"""Failure analysis and escalating-context construction for U:Qa."""

import logging
import os
import re
import subprocess
from typing import TYPE_CHECKING

from orchestrata_ultimate.agents.volks.qa._host import _QaHost
from orchestrata_ultimate.core.packet import CXPPacket, WorkStatus

_logger = logging.getLogger(__name__)

# W6.4b: typing-only host contract so pyright can see across mixin
# composition (see _host.py). No runtime effect — FailureContextMixin
# still inherits from plain ``object`` outside TYPE_CHECKING.
if TYPE_CHECKING:
    _MixinBase = _QaHost
else:
    _MixinBase = object


class FailureContextMixin(_MixinBase):
    """Failure analysis and escalating-context construction for U:Qa."""

    def _extract_test_failures(self, test_output: str) -> list:
        """Extract FAILED test names and short error from pytest output."""
        failures = []
        for line in test_output.splitlines():
            stripped = line.strip()
            if stripped.startswith("FAILED "):
                failures.append(stripped)
            elif stripped.startswith("E ") and "Error" in stripped:
                failures.append(stripped)
        # Cap at 10 to avoid overwhelming the fix prompt
        return failures

    def _get_signature_hints(self, fail_lines: list) -> str:
        """For TypeError about __init__ kwargs, return real class signatures via inspect.

        When J:Aider generates SystemEvent(topic=...) instead of SystemEvent(type=...),
        the TypeError is silently swallowed by EventBus dispatch and tests fail with
        'assert 0 == 1' — no useful signal. This method detects the TypeError, finds
        the class module from the test file imports, and returns the real constructor
        signature as a hint so J:Aider can self-correct without being told in the intent.
        """

        # Extract class names from TypeError: ClassName.__init__() lines
        classes_needed = set()
        for line in fail_lines:
            m = re.search(r"TypeError: (\w+)\.__init__\(\)", line)
            if m:
                classes_needed.add(m.group(1))
        if not classes_needed:
            return ""

        import_map = self._build_class_module_map(classes_needed)
        if not import_map:
            return ""

        hints = self._probe_class_signatures(import_map)
        if not hints:
            return ""

        return (
            "\n\nAPI SIGNATURES (real — from inspect, use ONLY these constructor args):\n"
            + "\n".join(hints)
        )

    def _build_class_module_map(self, classes_needed: set) -> dict:
        """Map needed class names to their import module via test imports.

        Scans workspace .py files for ``from MODULE import NAME`` lines and
        records the first module each wanted class is imported from.
        """
        import_map = {}
        for root, dirs, files in os.walk(self.workspace):
            dirs[:] = [
                d
                for d in dirs
                if not d.startswith(".") and d != "site-packages" and d != "__pycache__"
            ]
            for fname in files:
                if not fname.endswith(".py"):
                    continue
                fpath = os.path.join(root, fname)
                try:
                    with open(fpath, encoding="utf-8") as f:
                        for fline in f:
                            m = re.match(r"\s*from\s+([\w.]+)\s+import\s+(.+)", fline)
                            if m:
                                module = m.group(1)
                                names = [
                                    n.strip().split(" as ")[0].strip()
                                    for n in m.group(2).split(",")
                                ]
                                for name in names:
                                    if (
                                        name in classes_needed
                                        and name not in import_map
                                    ):
                                        import_map[name] = module
                except Exception as exc:
                    _logger.debug("AST parse for imports: %s", exc)
        return import_map

    def _probe_class_signatures(self, import_map: dict) -> list:
        """Probe real constructor signatures for the mapped classes.

        Runs the venv Python (with monorepo sys.path injected) to import
        each class and print ``inspect.signature``. Returns formatted hint
        lines for classes that resolved.
        """
        venv_python = self._get_venv_python()
        monorepo_syspath = (
            "import sys, os, glob; "
            "sys.path.insert(0, os.getcwd()); "
            "_s=os.path.join(os.getcwd(),'src'); "
            "(sys.path.insert(0,_s) if os.path.isdir(_s) and _s not in sys.path else None); "
            "_m='/opt/orchestrata/packages'; "
            "[sys.path.append(p) for p in glob.glob(f'{_m}/*/src') "
            "if p not in sys.path] if os.path.isdir(_m) else None; "
        )
        hints = []
        for cls_name, module in import_map.items():
            try:
                result = subprocess.run(
                    [
                        venv_python,
                        "-c",
                        monorepo_syspath
                        + f"from {module} import {cls_name}; import inspect; "
                        + f"print(f'{cls_name}' + str(inspect.signature({cls_name})))",
                    ],
                    cwd=self.workspace,
                    capture_output=True,
                    text=True,
                    timeout=10,
                    env=self._venv_env(),
                )
                if result.returncode == 0 and result.stdout.strip():
                    hints.append(f"  {result.stdout.strip()}")
            except Exception as exc:
                _logger.debug("signature probe failed: %s", exc)
        return hints

    def _build_escalating_context(
        self,
        issues: list[str],
        cycle_type: str,
    ) -> str:
        """G229: Build progressively detailed fix context.

        On cycles 1-2: just the issues + fix instructions.
        On cycle 3+: add history of previous attempts so the LLM
        knows what was already tried and failed. Include the actual
        source code of the failing file so it can see what changed.
        On cycle 5+: add root cause analysis hints.

        Args:
            issues: Current issue strings.
            cycle_type: "review" or "test-fix".

        Returns:
            Escalating context string.
        """
        history = getattr(self, "_fix_history", [])
        cycle_num = len(history) + 1

        parts: list[str] = []

        # Always include current issues
        parts.append(
            f"CURRENT ISSUES (cycle {cycle_num}):\n"
            + "\n".join(f"  {i}" for i in issues)
        )

        # Cycle 3+: previous failed attempts
        if cycle_num >= 3 and history:
            parts.extend(self._escalation_history_section(history))

        # Cycle 5+: root cause analysis hints
        if cycle_num >= 5:
            parts.extend(
                self._escalation_root_cause_section(issues, history, cycle_num)
            )

        # Cycle 8+: actual file content for context
        if cycle_num >= 8 and issues:
            parts.extend(self._escalation_file_contents_section(issues))

        # Record this attempt in history
        history.append(
            {
                "cycle": cycle_num,
                "type": cycle_type,
                "issues": list(issues[:10]),
                "issue_count": len(issues),
                "files_changed": [],
            }
        )
        self._fix_history = history

        return "\n".join(parts)

    def _escalation_history_section(self, history: list) -> list:
        """Cycle 3+ section: summarize the last 3 failed attempts."""
        parts = [f"\n\nPREVIOUS FAILED ATTEMPTS ({len(history)} cycles tried):"]
        # Show last 3 attempts
        for attempt in history[-3:]:
            parts.append(f"\n--- Attempt {attempt['cycle']} ({attempt['type']}) ---")
            parts.append(f"Issues found: {attempt['issue_count']}")
            for prev_issue in attempt["issues"][:5]:
                parts.append(f"  - {prev_issue}")
            if attempt.get("files_changed"):
                parts.append(
                    "Files changed: " + ", ".join(attempt["files_changed"][:5])
                )
        parts.append(
            "\n\nThe above attempts FAILED. Do NOT repeat "
            "the same approach. Analyze WHY the previous "
            "fixes didn't work and try a fundamentally "
            "different strategy."
        )
        return parts

    def _escalation_root_cause_section(
        self, issues: list, history: list, cycle_num: int
    ) -> list:
        """Cycle 5+ section: flag issues persisting across recent cycles."""
        # Check if same issue persists across cycles
        persistent = set()
        for attempt in history[-3:]:
            for prev_issue in attempt["issues"]:
                for current in issues:
                    # Fuzzy match — same file + same check type
                    if prev_issue.split("]")[0] == current.split("]")[0]:
                        persistent.add(current)
        if not persistent:
            return []
        parts = ["\n\nROOT CAUSE ANALYSIS REQUIRED:"]
        parts.append(
            f"{len(persistent)} issue(s) persist across "
            f"{cycle_num} cycles. The LLM fix is not "
            "addressing the root cause."
        )
        parts.append("Persistent issues:")
        for p in persistent:
            parts.append(f"  - {p}")
        parts.append(
            "\nSTRATEGY CHANGE REQUIRED:\n"
            "1. Read the ENTIRE file, not just the "
            "flagged line\n"
            "2. Trace the data flow — where does the "
            "bad value originate?\n"
            "3. Check if a decorator, import, or "
            "inheritance is altering behavior\n"
            "4. Consider rewriting the affected "
            "function from scratch\n"
            "5. If a decorator causes issues on "
            "exception classes, remove it from "
            "classes and only apply to functions"
        )
        return parts

    def _escalation_file_contents_section(self, issues: list) -> list:
        """Cycle 8+ section: inline the source of the first few failing files."""
        parts = ["\n\nFILE CONTENTS (read carefully before fixing):"]
        seen_files: set[str] = set()
        for issue in issues[:5]:
            # Extract filename from issue string
            # Format: [CHECK] filename: description
            if "]" in issue and ":" in issue:
                after_bracket = issue.split("]", 1)[1].strip()
                fname = after_bracket.split(":")[0].strip()
                if fname and fname not in seen_files:
                    fpath = os.path.join(self.workspace, fname)
                    if os.path.exists(fpath):
                        try:
                            with open(fpath, encoding="utf-8") as f:
                                content = f.read()
                            parts.append(
                                f"\n=== {fname} ===\n```python\n{content}\n```"
                            )
                            seen_files.add(fname)
                        except OSError:
                            pass
        return parts

    def _bounce_for_fix(
        self,
        packet: CXPPacket,
        issue_strs: list,
        fix_guide: str,
        files_created: list,
        task_summary: str,
        gate_label: str,
    ) -> CXPPacket:
        """G227/G229: Bounce back to CodeDirector with escalating context.

        Args:
            packet: Original CXP packet.
            issue_strs: Formatted issue strings.
            fix_guide: Fix instructions text.
            files_created: List of file paths.
            task_summary: Task description.
            gate_label: Which gate failed (for logging).

        Returns:
            CXP packet to U:Cd with fix_issues subject.
        """
        # G229: Build escalating context with history
        escalating = self._build_escalating_context(issue_strs, "review")
        return CXPPacket(
            from_agent="U:Qa",
            to_agent="U:Cd",
            thread_id=packet.thread_id,
            subject="fix_issues",
            payload={
                "issues": issue_strs,
                "fix_instructions": fix_guide,
                "escalating_context": escalating,
                "files_created": files_created,
                "task_summary": task_summary,
                "review_cycle": self.review_cycle,
                "review_response": gate_label
                + "\n\n"
                + escalating
                + "\n\n"
                + fix_guide,
                "project_dir": self.workspace,
                # Record QA's review verdict as a decision the kernel persists to
                # the semantic graph ((:Volk U:Qa)-[:MADE]->(:Decision)
                # -[:APPLIES_TO]->(:File)); makes volk_work_context.decisions and
                # the injector's decision section real.
                "decisions": [
                    {
                        "rationale": f"{gate_label}: {len(issue_strs)} "
                        f"issue(s) — {'; '.join(issue_strs[:3])}",
                        "files": files_created,
                    }
                ],
            },
            work_status=WorkStatus.IN_PROGRESS,
        )