orchestrata_ultimate.agents.volks.qa.volk

Reference › orchestrata_ultimate › agents › volks › qa › volk

QaVolk (alias U:Qa, role QA Engineer) is the gatekeeper between code generation and compliance (specification 031). It reviews the code the CodeDirector produced, generates and runs pytest suites, and either bounces problems back for repair or passes the work on. It extends BaseAgent (base) plus four mixins that hold most of the logic: AutofixMixin (_autofix), DepInstallMixin (_dep_install), TestGenMixin (_testgen) and FailureContextMixin (_context); the shared contract is in _host. QualityAssuranceAgent is an alias subclass, and the kernel pipeline imports the volk through the package __init__.

What it receives: a review packet from the CodeDirector with the created files, task summary, project directory and the Analyst's structured requirements and stories; or a re-review after a test-fix. What it produces: a compliance_check packet to U:Cp with the test verdict, test files, output and review cycles; or a fix_issues packet back to U:Cd; or, with no files, a clean acknowledgement (_no_files_ack).

_handle_review runs the auto-fix pass (_run_autofix_pass), then three deterministic gates in order through _run_review_gate: syntax, lint and style, and the code checklist. Any failing gate returns a bounce at once. Next it installs dependencies and validates that generated modules import (_bounce_for_import_errors on failure), then runs the LLM code review (_run_llm_code_review, with _parse_review_verdict and _parse_review_issues). After a test-fix cycle it re-runs existing tests (_handle_test_rerun); otherwise _handle_test_gen generates tests and runs them. Failures go to _build_test_fix_packet, which sends escalating context and signature hints to the CodeDirector, with the instruction to fix the source and not the tests. A pass is committed to git and sent by _build_compliance_packet. _handle_review_edit is a deterministic floor for test-quality regressions in edit reviews, using _review_test_edit and the helpers in _edit_review.

At a glance

Class diagram

Dependency diagram

API

Classes

QaVolk (pydantic)

Bases: AutofixMixin, DepInstallMixin, TestGenMixin, FailureContextMixin, BaseAgent · Line 179

QualityAssuranceAgent (pydantic)

Bases: QaVolk · Line 1232

Agent alias for QaVolk.

Functions

Source

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

# mypy: ignore-errors
"""U:Qa — QA volk with real LLM brain for test generation and code review."""

import logging
import os
import re
import subprocess
from collections.abc import Callable

from pydantic import Field

from orchestrata_ultimate.agents.volks._edit_review import (
    count_asserts,
    extract_edit_fields,
    is_trivial_assert,
    parse_fragment,
)
from orchestrata_ultimate.agents.volks.base import BaseAgent
from orchestrata_ultimate.agents.volks.codedirector.review_checks import (
    get_fix_instructions,
    review_gate_checklist,
    review_gate_lint,
    review_gate_syntax,
)
from orchestrata_ultimate.agents.volks.qa._autofix import AutofixMixin
from orchestrata_ultimate.agents.volks.qa._context import FailureContextMixin
from orchestrata_ultimate.agents.volks.qa._dep_install import DepInstallMixin
from orchestrata_ultimate.agents.volks.qa._testgen import TestGenMixin
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 WORKSPACE_DIR
from orchestrata_ultimate.core.packet import CXPPacket, WorkStatus

_logger = logging.getLogger(__name__)

QA_SYSTEM_PROMPT = """\
You are U:Qa, a senior QA engineer generating production-grade pytest suites.

RESPONSE FORMAT — output ONLY test files in this exact structure:

TEST_FILE: <filename.py>
```python
<complete test code>
```

Repeat TEST_FILE blocks for each file. After all files:

REVIEW_SUMMARY:
- <observation about coverage>
- <observation about quality>

TEST_RESULT: PASS or FAIL
ISSUES:
- <issue if any>

HANDOFF: On PASS, results route to U:Cp (compliance_check); on FAIL, issues route back to U:Cd (fix_issues).

RULES:
- pytest only (no unittest)
- ZERO mocking: no unittest.mock, MagicMock, patch, AsyncMock, Mock,
  no custom Mock/Fake/Stub/Dummy classes. Test REAL behavior only.
- Python 3.14+: never use asyncio.Awaitable/asyncio.Coroutine —
  use collections.abc.Awaitable or typing.Awaitable
- Use os.getenv() for env vars — never import python-dotenv
- EVERY test function MUST have a @traces decorator from the project's
  decorators module, linking to the requirement IDs from the source code.
  Copy spec/story/reqs from the source @traces of the function being tested.
"""

CODE_REVIEW_PROMPT = """You are U:Qa, the QA Engineer performing a structured code review based on Google's engineering practices and Google Python Style Guide.

Review the source code against EVERY item in this checklist. For each item, state PASS or FAIL with a one-line explanation.

CHECKLIST:

--- Correctness ---
1. IMPORTS: All imports resolve to real, existing modules. No hallucinated packages. Import paths match the actual package API. No relative imports. Order: stdlib, third-party, local (separated by blank lines). Each import on its own line.
2. NO_MOCKING: Zero usage of unittest.mock, MagicMock, patch, AsyncMock, Mock anywhere in source or test code. ALSO: No custom mock/stub/fake/dummy classes (class MockXxx, class FakeXxx, class StubXxx). Tests MUST use real objects connected to real services. FAIL if ANY mock or stub usage exists.
3. API_CORRECTNESS: Function signatures, class constructors, and method calls match the libraries being used. No invented APIs or wrong parameter names.
4. FILE_COMPLETENESS: Every file referenced in imports exists in the file list. No missing modules. No circular imports.
5. ENTRY_POINT: The project has a clear entry point (main function, __main__.py, or CLI script) that is wired correctly. Main code guarded by `if __name__ == '__main__'`.
6. DEPENDENCY_DECLARATION: If requirements.txt or pyproject.toml exists, all imported third-party packages are listed. No undeclared dependencies.

--- Design ---
7. DESIGN: Code interactions are well-structured. No unnecessary coupling. Changes belong in the right module. No over-engineering or premature abstractions solving speculative future requirements.
8. COMPLEXITY: No unnecessarily complex code at line, function, or class level. Code is easy to understand by other developers. No clever tricks that sacrifice readability.
9. NO_DUPLICATION: No copy-pasted code blocks. Repeated logic (3+ lines appearing 2+ times) is extracted into a function.

--- Functions ---
10. FUNCTION_LENGTH: No function exceeds 40 lines of logic. Long functions are split into smaller, focused helpers without harming program structure.
11. FUNCTION_ARGS: No function has more than 5 parameters. Use dataclasses or config objects for complex parameter sets.
12. NO_PLACEHOLDER: No `pass`, `TODO`, `NotImplementedError` in critical code paths. No empty function bodies that need logic.

--- Error Handling ---
13. ERROR_HANDLING: Critical operations (file I/O, network, subprocess) have try/except or input validation. No bare `except:` or `except Exception` unless re-raising. Specific exception types used. Minimal code inside try blocks. `finally` or `with` used for cleanup.
14. EXCEPTIONS: Custom exceptions inherit from appropriate base classes. Exception names end in `Error`. No `assert` for validation of preconditions or user input (assert can be optimized away).

--- Resource Management ---
15. RESOURCE_MANAGEMENT: Files, sockets, database connections, and similar stateful resources are explicitly closed. `with` statements used for context managers. No reliance on finalizers or `__del__` for cleanup.

--- Naming ---
16. NAMING: snake_case for functions/variables/modules, PascalCase for classes/exceptions, UPPER_CASE for constants. Names are descriptive, not abbreviated. No single-letter names except loop counters (i, j, k) and established conventions (e, f). No type names in variable names (e.g., no `id_to_name_dict`).

--- Type Safety ---
17. TYPE_HINTS: All public functions have parameter and return type annotations. Use `X | None` instead of bare `Optional`. No bare `dict`, `list` — use parameterized types. `__init__` return type not annotated (always None).
18. DEFAULT_VALUES: No mutable objects (list, dict, set) as default argument values. Use `None` and initialize inside the function.

--- Documentation ---
19. DOCSTRINGS: All public classes and public functions have docstrings. Google style: summary line, blank line, Args/Returns/Raises sections. Summary line fits in 80 characters. Triple double-quotes (`\"\"\"`). Docstrings describe calling semantics, not implementation details.
20. COMMENTS: Comments explain "why", not "what". Tricky code has comments before it starts. No commented-out code. TODOs reference a bug tracker issue. Comments are proper English with punctuation.

--- Style ---
21. LINE_LENGTH: Maximum 80 characters per line. Use implicit line continuation inside parentheses/brackets. No backslash continuation. Long URLs in comments go on their own line.
22. FORMATTING: 4-space indentation, no tabs. No semicolons. One statement per line. Two blank lines between top-level definitions. One blank line between methods. No trailing whitespace. Proper whitespace around operators and after commas.
23. STRINGS: f-strings or format() for string formatting, not `+` concatenation in loops. No string accumulation with `+=` in loops (use list + join). Consistent quote character within a file. Logging uses %-placeholders, not f-strings.

--- Python Best Practices ---
24. MUTABLE_GLOBAL_STATE: No mutable global variables. Module-level constants use UPPER_CASE. Module-internal entities prefixed with `_`.
25. COMPREHENSIONS: List/dict/set comprehensions used only for simple cases. No multiple `for` clauses or complex filter expressions. Complex comprehensions rewritten as loops.
26. TRUE_FALSE: Implicit boolean evaluation used where possible (`if seq:` not `if len(seq):`). `is None` / `is not None` for None checks. Never `== None` or `== True/False`.
27. ITERATORS: Default iterators used (`for key in adict:` not `for key in adict.keys():`). No mutation of containers while iterating.
28. DECORATORS: Decorators used judiciously. No `staticmethod`. `classmethod` only for named constructors. No external dependencies (files, sockets, databases) in decorator code.
29. CONCURRENCY: If threading is used, no reliance on built-in type atomicity. `queue.Queue` for inter-thread communication. Proper locking primitives. No blocking calls in async code without `asyncio.to_thread()`.

--- Security ---
30. SECURITY: No hardcoded secrets, credentials, or API keys. No SQL/shell injection via string concatenation or unsanitized input. No use of `eval()`, `exec()`, or `__import__()` with user input.

--- Tests ---
31. TESTS: Tests are present and correct. Tests fail when code breaks. Clear assertions in each test. No false positives. Test code is maintainable and readable. Tests cover edge cases and error conditions.

Output EXACTLY in this format:

REVIEW_CHECKLIST:
1. IMPORTS: PASS|FAIL — <explanation>
2. NO_MOCKING: PASS|FAIL — <explanation>
3. API_CORRECTNESS: PASS|FAIL — <explanation>
4. FILE_COMPLETENESS: PASS|FAIL — <explanation>
5. ENTRY_POINT: PASS|FAIL — <explanation>
6. DEPENDENCY_DECLARATION: PASS|FAIL — <explanation>
7. DESIGN: PASS|FAIL — <explanation>
8. COMPLEXITY: PASS|FAIL — <explanation>
9. NO_DUPLICATION: PASS|FAIL — <explanation>
10. FUNCTION_LENGTH: PASS|FAIL — <explanation>
11. FUNCTION_ARGS: PASS|FAIL — <explanation>
12. NO_PLACEHOLDER: PASS|FAIL — <explanation>
13. ERROR_HANDLING: PASS|FAIL — <explanation>
14. EXCEPTIONS: PASS|FAIL — <explanation>
15. RESOURCE_MANAGEMENT: PASS|FAIL — <explanation>
16. NAMING: PASS|FAIL — <explanation>
17. TYPE_HINTS: PASS|FAIL — <explanation>
18. DEFAULT_VALUES: PASS|FAIL — <explanation>
19. DOCSTRINGS: PASS|FAIL — <explanation>
20. COMMENTS: PASS|FAIL — <explanation>
21. LINE_LENGTH: PASS|FAIL — <explanation>
22. FORMATTING: PASS|FAIL — <explanation>
23. STRINGS: PASS|FAIL — <explanation>
24. MUTABLE_GLOBAL_STATE: PASS|FAIL — <explanation>
25. COMPREHENSIONS: PASS|FAIL — <explanation>
26. TRUE_FALSE: PASS|FAIL — <explanation>
27. ITERATORS: PASS|FAIL — <explanation>
28. DECORATORS: PASS|FAIL — <explanation>
29. CONCURRENCY: PASS|FAIL — <explanation>
30. SECURITY: PASS|FAIL — <explanation>
31. TESTS: PASS|FAIL — <explanation>

REVIEW_VERDICT: PASS or FAIL

ISSUES:
- <specific issue with file path and line number and description>
...

If REVIEW_VERDICT is PASS, the ISSUES section must be empty.
If REVIEW_VERDICT is FAIL, ISSUES must list every actionable problem that must be fixed.
"""

# G226: No cycle cap — review loops until ALL issues resolved


class QaVolk(
    AutofixMixin,
    DepInstallMixin,
    TestGenMixin,
    FailureContextMixin,
    BaseAgent,
):
    test_files_created: list[str] = Field(default_factory=list)
    test_results: str | None = Field(default=None)
    workspace: str = Field(default=WORKSPACE_DIR)
    review_cycle: int = Field(default=0)
    task_summary: str = Field(default="")
    alias: str = Field(default="U:Qa")
    role: str = "QA Engineer"
    _MOCK_PATTERNS = [
        (r"from\s+unittest\.mock\s+import", "unittest.mock import"),
        (r"import\s+unittest\.mock", "unittest.mock import"),
        (r"from\s+unittest\s+import\s+mock", "unittest import mock"),
        (r"Mock\s*\(", "Mock instantiation"),
        (r"MagicMock\s*\(", "MagicMock instantiation"),
        (r"AsyncMock\s*\(", "AsyncMock instantiation"),
        (r"@\s*(mock\.)?patch\s*\(", "@patch decorator"),
        (r"mocker\.(patch|spy|stub|mock)\s*\(", "pytest-mock fixture"),
        (r"create_autospec\s*\(", "create_autospec"),
        (r"patch\.object\s*\(", "patch.object()"),
        (r"patch\.dict\s*\(", "patch.dict()"),
        (r"patch\.multiple\s*\(", "patch.multiple()"),
        # G-91-04: catch custom mock/stub/fake class definitions
        (
            r"class\s+(Mock|Fake|Stub|Dummy)\w*\s*[\(:]",
            "custom mock/stub class definition",
        ),
    ]

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.test_files_created = []
        self.test_results = None
        self.workspace = WORKSPACE_DIR
        self.review_cycle = 0
        self.task_summary = ""

    async def process(self, packet: CXPPacket) -> CXPPacket | None:
        self._graph_context = extract_graph_context(packet.payload)
        if packet.subject == "review_and_test":
            return await self._handle_review(packet)
        elif packet.subject == "review_edit":
            return await self._handle_review_edit(packet)
        return None

    def _no_files_ack(self, packet: CXPPacket, task_summary: str) -> CXPPacket:
        """Build the clean review_complete ack for empty file lists."""
        self.status = WorkStatus.RESOLVED
        self.summary = "No files to review"
        return CXPPacket(
            from_agent="U:Qa",
            to_agent=packet.from_agent,
            thread_id=packet.thread_id,
            subject="review_complete",
            payload={
                "approved": True,
                "review_cycle": 0,
                "files_reviewed": [],
                "task_summary": task_summary,
                "message": "No files_created in payload — nothing to review.",
                # Forward the workspace like every other QA response payload
                # so chained stages (U:Cp) scan the TARGET project, not CWD.
                "project_dir": self.workspace,
            },
            work_status=WorkStatus.RESOLVED,
        )

    def _reset_thread_state(self, thread_id: str) -> None:
        """Reset cycle counters and fix history when the thread changes."""
        if not hasattr(self, "_review_thread") or self._review_thread != thread_id:
            self.review_cycle = 0
            self._test_fix_count = 0
            self._review_thread = thread_id
            # G229: Fix attempt history — track what was tried
            # so we can escalate with context on repeated failures
            self._fix_history: list[dict] = []

    def _read_source_files(self, files_created: list[str]) -> dict[str, str]:
        """Read existing source files into a {relpath: content} map."""
        file_contents = {}
        for f in files_created:
            filepath = os.path.join(self.workspace, f)
            if os.path.exists(filepath):
                try:
                    with open(filepath, encoding="utf-8") as fh:
                        file_contents[f] = fh.read()
                except Exception:
                    file_contents[f] = "(could not read)"
        return file_contents

    @traces(
        spec="031",
        story="US-031-002",
        reqs=[
            "REQ-031-016",
            "REQ-031-020",
            "REQ-031-024",
            "REQ-031-030",
            "REQ-031-043",
        ],
    )
    async def _handle_review(self, packet: CXPPacket) -> CXPPacket:
        self.status = WorkStatus.IN_PROGRESS
        self.current_thread = packet.thread_id

        files_created = packet.payload.get("files_created", [])
        task_summary = packet.payload.get("task_summary", "")
        self.task_summary = task_summary

        # Accept dynamic workspace from U:Ar/U:Cd pipeline. Absorbed BEFORE
        # the empty-files fast path so the ack still forwards the correct
        # project_dir downstream — U:Cp's trace scan otherwise falls back
        # to CWD-relative "src"/"tests" and analyzes the wrong tree.
        project_dir = packet.payload.get("project_dir", "")
        if project_dir:
            self.workspace = project_dir
            self.project_dir = project_dir

        # Fast path: no files to review — short-circuit with a clean ack
        # instead of spinning up the full three-gate pipeline plus
        # dependency install and test generation. This keeps simple
        # protocol smoke tests (empty-payload review packets) from
        # paying for the heavyweight review path.
        if not files_created:
            return self._no_files_ack(packet, task_summary)

        # G218: Capture requirement IDs for traceability
        self._structured_requirements = packet.payload.get(
            "structured_requirements", []
        )
        self._user_stories = packet.payload.get("user_stories", [])

        # Reset cycle counter on new thread
        self._reset_thread_state(packet.thread_id)

        # Read the source files
        file_contents = self._read_source_files(files_created)
        source_listing = ""
        for fname, content in file_contents.items():
            source_listing += f"\n=== {fname} ===\n{content}\n"

        # G226: Every code change MUST be reviewed. No exceptions.
        # Even test-fix returns get full deterministic review
        # (lint, syntax, traceability, imports, docstrings).
        from_test_fix = packet.payload.get("from_test_fix", False)

        # Phase A: Code Review — ALWAYS runs, even on test-fix returns
        self.review_cycle += 1
        self.summary = f"Code review (cycle {self.review_cycle})"
        label = "test-fix review" if from_test_fix else "code review"
        print(
            f"  [U:Qa] {label} cycle {self.review_cycle}...",
        )

        # Auto-fix pass — runs before any gates
        self._run_autofix_pass()

        # ── G227: Three-gate sequential review pipeline ──────────
        # Gate 1: SYNTAX — can every file parse?
        # Gate 2: LINT — are files clean?
        # Gate 3: CHECKLIST — quality, traceability, architecture
        # Each gate MUST pass before the next runs. No skipping.
        gates = [
            (
                1,
                "Syntax",
                review_gate_syntax,
                "  [U:Qa] GATE 1 FAIL — {n} syntax errors. STOP. Fix syntax first.",
                "GATE 1 FAIL: Syntax errors",
            ),
            (
                2,
                "Lint & style",
                review_gate_lint,
                "  [U:Qa] GATE 2 FAIL — {n} lint/style errors. STOP. Fix lint first.",
                "GATE 2 FAIL: Lint/style errors",
            ),
            (
                3,
                "Checklist",
                review_gate_checklist,
                "  [U:Qa] GATE 3 FAIL — {n} checklist issues.",
                "GATE 3 FAIL: Checklist violations",
            ),
        ]
        for number, label_text, gate_fn, fail_msg, fail_label in gates:
            bounce = self._run_review_gate(
                number,
                label_text,
                gate_fn,
                packet,
                files_created,
                task_summary,
                fail_msg,
                fail_label,
            )
            if bounce is not None:
                return bounce
        print("  [U:Qa] All 3 review gates PASSED")

        # Phase A.5: Runtime import validation (catches LLM hallucinated APIs)
        self._install_deps()
        import_errors = self._validate_imports()
        if import_errors:
            return self._bounce_for_import_errors(
                packet, import_errors, file_contents, task_summary
            )

        # Phase A.7: LLM code review — SPEC-031 US-031-003. Drives the LLM
        # through the 31-item Google-style checklist. Runs on EVERY cycle,
        # including test-fix returns (REQ-031-043). A FAIL verdict bounces to
        # U:Cd (REQ-031-042); a PASS falls through to the test phase.
        review_bounce = await self._run_llm_code_review(
            packet, source_listing, files_created, task_summary
        )
        if review_bounce is not None:
            return review_bounce

        # G100/G226: On a test-fix return whose review passed, re-run the
        # existing tests on disk instead of regenerating them.
        if from_test_fix:
            print("  [U:Qa] review passed — re-running existing tests")
            return await self._handle_test_rerun(
                packet, file_contents, source_listing, task_summary
            )

        # Phase B: Test Generation + Execution
        return await self._handle_test_gen(
            packet, file_contents, source_listing, task_summary
        )

    @traces(
        spec="031",
        story="US-031-003",
        reqs=[
            "REQ-031-039",
            "REQ-031-042",
            "REQ-031-043",
            "REQ-031-040",
            "REQ-031-041",
        ],
    )
    async def _run_llm_code_review(
        self,
        packet: CXPPacket,
        source_listing: str,
        files_created: list[str],
        task_summary: str,
    ) -> CXPPacket | None:
        """SPEC-031 US-031-003: LLM review against the 31-item checklist.

        Runs on every review cycle (including test-fix returns) after the
        deterministic gates pass. Drives the LLM through CODE_REVIEW_PROMPT
        and acts on its verdict.

        Args:
            packet: The inbound review packet.
            source_listing: Concatenated source files under review.
            files_created: Paths of the files under review.
            task_summary: Task description carried through the pipeline.

        Returns:
            A ``fix_issues`` CXP packet to U:Cd when the verdict is FAIL,
            otherwise None (review passed).
        """
        print("  [U:Qa] LLM code review (31-item checklist)...", flush=True)
        review_prompt = (
            f"Task: {task_summary}\n\n"
            "Review the following source against EVERY checklist item, then "
            "emit REVIEW_CHECKLIST, REVIEW_VERDICT, and ISSUES.\n"
            f"{source_listing}"
        )
        # W4.3c: route the primary review-generation call through the
        # tool-calling loop (SPEC-103) with U:Qa's registered allowlist
        # (tool_registry.AGENT_TOOLS["U:Qa"]: read_file/write_file/run_shell/
        # search_code/check_coverage) instead of a single-shot ``_call_llm``,
        # so U:Qa can inspect the real codebase (and even re-run coverage)
        # while reviewing, mirroring W4.3a/W4.3b (Architect/CodeDirector).
        #
        # ``_call_llm_with_tools`` takes no ``graph_context=`` kwarg (unlike
        # ``_call_llm`` -- the tool loop only forwards temperature/model to
        # ``call_with_tools``), so graph_context is folded directly into the
        # user turn here (Documentor/Architect/CodeDirector precedent) or it
        # would be silently dropped before reaching the LLM. And unlike
        # ``_call_llm`` (whose non-override branch composes the persona
        # itself), ``_call_llm_with_tools`` does not -- so ``compose_persona``
        # is applied explicitly here to keep the shared team preamble in the
        # system turn (same fix CodeDirector's ``_run_review`` needed).
        # REQ-031-039: volk="U:Qa" (self.alias) is stamped by
        # ``_call_llm_with_tools`` itself; temperature=0.0 is preserved.
        # _graph_context is set in process(); default defensively so a direct
        # _handle_review call cannot raise AttributeError.
        from orchestrata_ultimate.agents.volks.persona import (  # noqa: PLC0415
            compose_persona,
        )

        graph_context = getattr(self, "_graph_context", "")
        user_message = (
            f"{graph_context}\n\n---\n\n{review_prompt}"
            if graph_context
            else review_prompt
        )
        review_response = await self._call_llm_with_tools(
            compose_persona(CODE_REVIEW_PROMPT),
            user_message,
            temperature=0.0,
        )
        self.output_artifacts.append(
            f"=== LLM CODE REVIEW (cycle {self.review_cycle}) ===\n{review_response}"
        )

        # REQ-031-040: parse the verdict from the REVIEW_VERDICT: line.
        if self._parse_review_verdict(review_response) != "FAIL":
            print("  [U:Qa] LLM code review PASS")
            return None

        # REQ-031-041: parse actionable issues from the ISSUES: block.
        issues = self._parse_review_issues(review_response) or [
            "LLM review returned FAIL without an itemised ISSUES list"
        ]
        print(f"  [U:Qa] LLM code review FAIL — {len(issues)} issue(s)")
        for item in issues:
            print(f"    - {item}")
        fix_guide = (
            "FIX INSTRUCTIONS:\n"
            + "\n".join(f"  - {issue}" for issue in issues)
            + "\n\nResolve every issue above, then hand back for re-review."
        )
        # REQ-031-042: a FAIL verdict bounces to U:Cd with the parsed issues.
        return self._bounce_for_fix(
            packet,
            issues,
            fix_guide,
            files_created,
            task_summary,
            "LLM REVIEW FAIL: 31-item checklist",
        )

    @traces(
        spec="031",
        story="US-031-002",
        reqs=["REQ-031-015", "REQ-031-016"],
    )
    def _run_autofix_pass(self) -> None:
        """Run every source auto-fix before the review gates."""
        self._autofix_aioredis()
        self._autofix_sync_redis()
        self._autofix_f821_datetime()
        self._autofix_xadd_dict()
        self._autofix_hset_dict()
        self._autofix_redis_decode()
        self._autofix_missing_stdlib_imports()
        self._autofix_missing_barename_imports()
        self._autofix_redis_host()
        self._autofix_relative_imports()
        self._autofix_line_length()
        self._autofix_import_order()
        self._autofix_unused_imports()

    @traces(
        spec="031",
        story="US-031-002",
        reqs=[
            "REQ-031-019",
            "REQ-031-023",
            "REQ-031-027",
            "REQ-031-029",
        ],
    )
    def _run_review_gate(
        self,
        number: int,
        label_text: str,
        gate_fn: Callable[[str, list[str]], list[tuple[str, str, str]]],
        packet: CXPPacket,
        files_created: list[str],
        task_summary: str,
        fail_msg: str,
        fail_label: str,
    ) -> CXPPacket | None:
        """Run a single review gate; bounce to U:Cd on failure.

        Returns a fix_issues CXP packet when the gate found issues,
        otherwise None (gate passed).
        """
        print(f"  [U:Qa] Gate {number}/3: {label_text}...", flush=True)
        issues = gate_fn(self.workspace, files_created)
        if issues:
            issue_strs = [f"[{cid}] {fp}: {desc}" for cid, fp, desc in issues]
            print(fail_msg.format(n=len(issues)))
            for item in issue_strs:
                print(f"    - {item}")
            fix_guide = get_fix_instructions(issues)
            return self._bounce_for_fix(
                packet,
                issue_strs,
                fix_guide,
                files_created,
                task_summary,
                fail_label,
            )
        print(f"  [U:Qa] Gate {number}/3: PASS", flush=True)
        return None

    @traces(spec="031", story="US-031-002", reqs=["REQ-031-030"])
    def _bounce_for_import_errors(
        self,
        packet: CXPPacket,
        import_errors: list[str],
        file_contents: dict[str, str],
        task_summary: str,
    ) -> CXPPacket:
        """Build the fix_issues packet for runtime import-validation errors."""
        import_issues = [f"IMPORT_ERROR in {e}" for e in import_errors]
        print(f"  [U:Qa] Import validation failed — {len(import_issues)} errors")
        for ie in import_issues:
            print(f"    - {ie}")
        import_fix_guide = (
            "FIX INSTRUCTIONS:\n"
            + "\n".join(
                f"  [{ie}] → Check module exists, is installed, import path is correct."
                for ie in import_issues
            )
            + "\n\nCRITICAL: Fix ALL import errors. "
            "Do NOT hand off with broken imports."
        )
        return CXPPacket(
            from_agent="U:Qa",
            to_agent="U:Cd",
            thread_id=packet.thread_id,
            subject="fix_issues",
            payload={
                "issues": import_issues,
                "fix_instructions": import_fix_guide,
                "files_created": list(file_contents.keys()),
                "task_summary": task_summary,
                "review_cycle": self.review_cycle,
                "review_response": "Import validation failed: "
                + "; ".join(import_errors)
                + "\n\n"
                + import_fix_guide,
                "project_dir": self.workspace,
            },
            work_status=WorkStatus.IN_PROGRESS,
        )

    @traces(spec="031", story="US-031-005", reqs=["REQ-031-055"])
    async def _handle_test_rerun(
        self,
        packet: CXPPacket,
        file_contents: dict[str, str],
        source_listing: str,
        task_summary: str,
    ) -> CXPPacket:
        """G100: Re-run existing test files without regenerating them.

        During test-fix cycles, J:Aider already fixed the code and test files.
        Regenerating tests from scratch risks truncation and loses working tests.
        """
        files_created = list(file_contents.keys())

        # Discover existing test files on disk (G-77-01: check both root and tests/ subdir)
        # G202: Store relative paths so pytest can find them
        existing_tests = []
        seen = set()
        for search_dir in [self.workspace, os.path.join(self.workspace, "tests")]:
            if not os.path.isdir(search_dir):
                continue
            for fname in os.listdir(search_dir):
                if fname.startswith("test_") and fname.endswith(".py"):
                    if fname in seen:
                        continue
                    seen.add(fname)
                    rel = os.path.relpath(
                        os.path.join(search_dir, fname),
                        self.workspace,
                    )
                    existing_tests.append(rel)

        if not existing_tests:
            print("  [U:Qa] No test files on disk — falling back to test generation")
            return await self._handle_test_gen(
                packet, file_contents, source_listing, task_summary
            )

        self.test_files_created = sorted(existing_tests)
        print(
            f"  [U:Qa] Re-running {len(existing_tests)} existing test files "
            f"(G100: no regeneration)"
        )

        # Run tests
        test_passed, test_output = await self._run_tests()
        self.test_results = test_output
        self.output_artifacts.append(f"=== TEST RERUN OUTPUT ===\n{test_output}")

        # Print summary (same as _handle_test_gen)
        self._print_test_summary(test_output, test_passed)

        # Phase C: test-fix escalation — G226: no cap, no forced handoff.
        # Tests must pass. Period.
        if not test_passed:
            fail_lines = self._extract_fail_lines(test_output)
            if fail_lines:
                return self._build_test_fix_packet(
                    packet, fail_lines, files_created, task_summary
                )

        return self._build_compliance_packet(packet, test_passed, test_output)

    @traces(spec="031", story="US-031-004", reqs=["REQ-031-049"])
    def _build_import_section(self, files_created: list[str]) -> str:
        """Derive ``from <module> import ...`` hints for source files."""
        import_hints = []
        for f in files_created:
            if f.endswith(".py") and not f.startswith("test_"):
                # G94-G3: conftest adds src/ and */src/ to sys.path,
                # so strip everything up to and including 'src/' —
                # e.g. 'cxp-message-tracer/src/cxp_tracer/models.py'
                # becomes 'cxp_tracer.models'
                path = f.replace("\\", "/")
                src_idx = path.find("/src/")
                if src_idx >= 0:
                    path = path[src_idx + 5 :]
                elif path.startswith("src/"):
                    path = path[4:]
                module = path.replace("/", ".").removesuffix(".py")
                if module == "__init__" or module.endswith(".__init__"):
                    continue
                import_hints.append(f"  from {module} import ...")
        return "\n".join(import_hints) if import_hints else "  (use paths above)"

    @traces(
        spec="031",
        story="US-031-004",
        reqs=["REQ-031-048"],
    )
    def _build_trace_section(self) -> str:
        """G218: Build the traceability prompt section from requirement IDs."""
        if not self._structured_requirements:
            return ""
        req_ids = []
        for req in self._structured_requirements:
            if isinstance(req, dict):
                rid = req.get("id", req.get("req_id", ""))
                if rid:
                    req_ids.append(rid)
            elif isinstance(req, str):
                req_ids.append(req)
        if not req_ids:
            return ""
        return (
            "\n## Traceability (MANDATORY)\n"
            "Add this decorator to EVERY test function:\n"
            "  @traces(spec='001', story='US-001', "
            f"reqs={req_ids[:5]})\n"
            "Import: from orchestrata_ultimate.core.tracing.traceability "
            "import traces\n"
            "If orchestrata is not available, define a "
            "pass-through:\n"
            "  def traces(**kw): return lambda f: f\n\n"
        )

    @traces(
        spec="031",
        story="US-031-004",
        reqs=[
            "REQ-031-049",
            "REQ-031-050",
            "REQ-031-054",
        ],
    )
    async def _handle_test_gen(
        self,
        packet: CXPPacket,
        file_contents: dict[str, str],
        source_listing: str,
        task_summary: str,
    ) -> CXPPacket:
        """Phase B: Generate tests and run them."""
        files_created = list(file_contents.keys())
        self.summary = "Generating tests"
        print("  [U:Qa] Generating tests...")

        import_section = self._build_import_section(files_created)

        # G76: Build file tree so LLM understands module structure
        file_tree = "\n".join(f"  {f}" for f in sorted(files_created))

        # G218: Build traceability section from requirement IDs
        trace_section = self._build_trace_section()

        prompt = (
            f"Task: {task_summary}\n\n"
            f"{trace_section}"
            "## Phase 1: Analyze Before Writing\n"
            "Before generating ANY test code, analyze each source file:\n"
            "1. List every public class and function with its signature\n"
            "2. Identify input types, return types, and side effects\n"
            "3. List external dependencies (Redis, filesystem, network)\n"
            "4. Identify error paths (try/except, validation, edge cases)\n"
            "5. Note which functions are async vs sync\n"
            "Write this analysis as comments at the top of each test file.\n\n"
            "## Phase 2: Plan Test Categories\n"
            "For each source file, plan tests in these categories:\n"
            "- HAPPY_PATH: normal usage with valid inputs (2-3 tests)\n"
            "- EDGE_CASE: empty inputs, None, boundary values (2-3 tests)\n"
            "- ERROR_CONDITION: invalid inputs, missing resources (2-3 tests)\n"
            "- BOUNDARY: max/min values, zero, unicode, special chars (1-2)\n"
            "- INTEGRATION: multiple components working together (1-2)\n"
            "Mark each test with a comment: # Category: HAPPY_PATH etc.\n\n"
            "## Phase 3: Infrastructure Rules\n"
            "### Redis Configuration (MANDATORY)\n"
            "If ANY source uses Redis, ALL test fixtures MUST:\n"
            "  import os\n"
            "  host = os.environ['REDIS_HOST']\n"
            "  port = int(os.environ['REDIS_PORT'])\n"
            "NEVER hardcode 'localhost' or port numbers in tests.\n"
            "Use db=15 for isolation. Flush before AND after yield.\n\n"
            "### Redis-Specific Rules\n"
            "- XADD: FLAT string values only — json.dumps() nested dicts\n"
            "- SISMEMBER/EXISTS/DEL return ints (1/0), NOT bools\n"
            "- Streams: create consumer group before XREADGROUP\n"
            "- Short delays for retry tests: backoff_factor=0.01\n\n"
            "### Async Rules\n"
            "- async source -> async test + @pytest.mark.asyncio + await\n"
            "- Async fixtures: @pytest_asyncio.fixture (NOT @pytest.fixture)\n"
            "- Sync tests CANNOT use async fixtures\n\n"
            "### Fixture Patterns (use these EXACT patterns)\n"
            "```python\n"
            "# Redis async fixture\n"
            "@pytest_asyncio.fixture\n"
            "async def storage(redis_host, redis_port):\n"
            "    s = Storage(host=redis_host, port=redis_port, db=15)\n"
            "    await s.connect()\n"
            "    client = await s._get_client()\n"
            "    await client.flushdb()\n"
            "    yield s\n"
            "    await client.flushdb()\n"
            "    await s.disconnect()\n"
            "\n"
            "# Temp file fixture\n"
            "@pytest.fixture\n"
            "def engine(tmp_path):\n"
            "    f = tmp_path / 'rules.pl'\n"
            "    f.write_text('may_execute(_, _, _).')\n"
            "    yield RealEngine(str(f))\n"
            "\n"
            "# CLI test (no subprocess)\n"
            "def test_cli_help(capsys):\n"
            "    with pytest.raises(SystemExit):\n"
            "        cli_main(['--help'])\n"
            "    captured = capsys.readouterr()\n"
            "    assert 'usage' in captured.out.lower()\n"
            "```\n\n"
            "## Phase 4: Source Code\n"
            f"Workspace: {self.workspace}\n"
            f"File tree:\n{file_tree}\n\n"
            f"Import paths (use EXACTLY):\n{import_section}\n\n"
            f"Source files:\n{source_listing}\n\n"
            "## Phase 5: Output Constraints\n"
            "- Test files in workspace root, named test_<module>.py\n"
            "- Each file MAX 120 lines. Split if needed.\n"
            "- Each test MUST be self-contained (own setup + assertions)\n"
            "- Test ONLY public functions. No _private access.\n"
            "- Read ACTUAL constructor signatures from source code.\n"
            "- Read ACTUAL default values from source — do NOT guess.\n"
            "  e.g. if configure_logging defaults to INFO, test INFO not DEBUG.\n"
            "- No monkey-patching. Use REAL error conditions.\n"
            "- FIFO for queues unless explicitly stated otherwise.\n"
            "- tmp_path for temp files. Path.unlink(missing_ok=True).\n"
            "- Target: 5-10 test files, 50-100+ test functions total.\n"
            "- A TRUNCATED file is WORTHLESS. Completeness > count.\n"
            "- ZERO mocking. No Mock, Fake, Stub, Dummy classes.\n\n"
            "## Phase 6: Traceability (MANDATORY)\n"
            "Every test class and test function MUST have a @traces decorator "
            "linking it to the requirement(s) it validates.\n\n"
            "### How to add @traces\n"
            "1. Find the `decorators.py` in the source package "
            "(e.g. `from <package>.decorators import traces`). "
            "Use the SAME import path style as the source files.\n"
            "2. Read each source file's @traces decorator to find its "
            "spec, story, and reqs values.\n"
            "3. On each test function, add:\n"
            "```python\n"
            "from <package>.decorators import traces\n\n"
            '@traces(spec="001", story="US-001", reqs=["REQ-001-US-001-001"])\n'
            "def test_something():\n"
            "    ...\n"
            "```\n"
            "4. Match reqs to the source function/class being tested. "
            "If a test exercises `TaskQueue.enqueue` which has "
            '`@traces(spec="001", story="US-002", reqs=["REQ-001-US-002-003"])`, '
            "use those SAME values on the test.\n"
            "5. If a source file has NO @traces decorator, use the "
            "spec and story from the nearest related source file "
            "that does have one.\n"
            "6. EVERY test function must have exactly one @traces. "
            "No test without traceability.\n"
        )
        # _graph_context is set in process(); default defensively (same
        # pattern as _run_llm_code_review) so a direct _handle_review or
        # _handle_test_gen call cannot raise AttributeError.
        # W4.5 (backlog #14): route through the base-agent seam so this call
        # gets memory recall (SPEC-010) folded in. _call_llm stamps
        # volk="U:Qa" (self.alias) and composes persona itself.
        llm_response = await self._call_llm(
            QA_SYSTEM_PROMPT,
            prompt,
            temperature=0.0,
            graph_context=getattr(self, "_graph_context", ""),
        )
        self.output_artifacts.append(f"=== QA TEST GEN ===\n{llm_response}")

        # Extract and write test files
        self.test_files_created = await self._extract_and_write_tests(llm_response)
        print(f"  [U:Qa] Generated {len(self.test_files_created)} test files")

        # G125-1: Auto-fix hardcoded Redis localhost in test files
        self._autofix_redis_host()
        # G129-2: Auto-fix parse_args() to accept optional args parameter
        self._autofix_parse_args()
        # G194-2: Auto-fix unused imports before deterministic review
        self._autofix_unused_imports()

        # Run tests
        test_passed, test_output = await self._run_tests()
        self.test_results = test_output
        self.output_artifacts.append(f"=== TEST OUTPUT ===\n{test_output}")

        # Print test summary for observability (G29)
        self._print_test_summary(test_output, test_passed)

        # Phase C: test-fix escalation — G226: no cap, no forced handoff.
        # Tests must pass. Period.
        if not test_passed:
            fail_lines = self._extract_fail_lines(test_output)
            if fail_lines:
                return self._build_test_fix_packet(
                    packet, fail_lines, files_created, task_summary
                )

        return self._build_compliance_packet(
            packet, test_passed, test_output, review=llm_response
        )

    @traces(spec="031", reqs=["REQ-031-055"])
    def _print_test_summary(self, test_output: str, test_passed: bool) -> None:
        """Print the pytest counts line and any FAILED test names."""
        # G75: require ===== border to avoid matching pytest assertion lines
        # G126: Take LAST matching line (the counts), not first
        summary_line = ""
        for tline in test_output.splitlines():
            stripped = tline.strip()
            if stripped.startswith("=") and (
                "passed" in stripped or "failed" in stripped or "error" in stripped
            ):
                summary_line = stripped
        if summary_line:
            print(f"  [U:Qa] {summary_line}")
        else:
            print(f"  [U:Qa] Tests {'PASSED' if test_passed else 'FAILED'}")

        # G82: Print individual FAILED test names for conversation visibility
        if not test_passed:
            for tline in test_output.splitlines():
                stripped = tline.strip()
                if stripped.startswith("FAILED "):
                    print(f"    [U:Qa] {stripped}")

    @traces(spec="031", reqs=["REQ-031-055"])
    def _extract_fail_lines(self, test_output: str) -> list[str]:
        """Extract failure lines, falling back to the pytest summary line.

        G209: If structured extraction is empty but tests failed, use the
        counts summary; finally fall back to an unparseable-output marker.
        """
        fail_lines = self._extract_test_failures(test_output)
        if not fail_lines:
            for tline in test_output.splitlines():
                s = tline.strip()
                if "failed" in s and ("passed" in s or "error" in s):
                    fail_lines = [s]
                    break
            if not fail_lines:
                fail_lines = ["Tests failed (unparseable output)"]
        return fail_lines

    @traces(spec="031", reqs=["REQ-031-055"])
    def _build_test_fix_packet(
        self,
        packet: CXPPacket,
        fail_lines: list[str],
        files_created: list[str],
        task_summary: str,
    ) -> CXPPacket:
        """G226/G229: Escalate test failures to U:Cd with history + hints."""
        test_fix_count = getattr(self, "_test_fix_count", 0)
        self._test_fix_count = test_fix_count + 1
        print(
            f"  [U:Qa] {len(fail_lines)} test failures — "
            f"sending to U:Cd for fix "
            f"(test-fix cycle {self._test_fix_count})"
        )
        # G229: Escalating context with history
        escalating = self._build_escalating_context(fail_lines, "test-fix")
        source_fix_hint = (
            "\n\nFix the SOURCE CODE to make tests pass. Do NOT modify test files."
        )
        return CXPPacket(
            from_agent="U:Qa",
            to_agent="U:Cd",
            thread_id=packet.thread_id,
            subject="fix_issues",
            payload={
                "issues": fail_lines,
                "escalating_context": escalating,
                "files_created": files_created,
                "task_summary": task_summary,
                "review_cycle": self.review_cycle,
                "review_response": escalating
                + source_fix_hint
                + self._get_signature_hints(fail_lines),
                "from_test_fix": True,
                "project_dir": self.workspace,
            },
            work_status=WorkStatus.IN_PROGRESS,
        )

    @traces(spec="031", reqs=["REQ-031-055"])
    def _build_compliance_packet(
        self,
        packet: CXPPacket,
        test_passed: bool,
        test_output: str,
        review: str | None = None,
    ) -> CXPPacket:
        """Resolve status, commit, and build the U:Cp compliance packet.

        When ``review`` is given it is included in the payload (test-gen
        path); the rerun path omits it.
        """
        self.status = WorkStatus.RESOLVED
        self.summary = (
            f"Tests {'PASS' if test_passed else 'FAIL'} "
            f"— {len(self.test_files_created)} test files "
            f"(reviewed {self.review_cycle} cycle{'s' if self.review_cycle > 1 else ''})"
        )

        self._git_commit(f"QA complete: tests {'PASS' if test_passed else 'FAIL'}")

        payload = {
            "test_passed": test_passed,
            "test_files": self.test_files_created,
            "test_output": test_output,
        }
        if review is not None:
            payload["review"] = review
        payload.update(
            {
                "review_cycles": self.review_cycle,
                "project_dir": self.workspace,
                "task_summary": self.task_summary,
            }
        )
        return CXPPacket(
            from_agent="U:Qa",
            to_agent="U:Cp",
            thread_id=packet.thread_id,
            subject="compliance_check",
            payload=payload,
            work_status=WorkStatus.IN_PROGRESS,
        )

    @traces(spec="031", reqs=["REQ-031-055"])
    def _git_commit(self, message: str):
        """Commit all files in the project workspace."""
        workspace = getattr(self, "workspace", "")
        if not workspace or not os.path.isdir(os.path.join(workspace, ".git")):
            return
        # G-92c-01: containers run as root with no git identity — set it explicitly
        git_env = {
            **os.environ,
            "GIT_AUTHOR_NAME": "Orchestrata Spike",
            "GIT_AUTHOR_EMAIL": "spike@orchestrata.ai",
            "GIT_COMMITTER_NAME": "Orchestrata Spike",
            "GIT_COMMITTER_EMAIL": "spike@orchestrata.ai",
        }
        try:
            subprocess.run(
                ["git", "add", "-A"],
                cwd=workspace,
                capture_output=True,
                timeout=10,
            )
            result = subprocess.run(
                ["git", "commit", "-m", message],
                cwd=workspace,
                capture_output=True,
                timeout=10,
                env=git_env,
            )
            if result.returncode == 0:
                print(f"  [U:Qa] Git commit: {message}", flush=True)
            else:
                print(
                    f"  [U:Qa] Git commit failed (rc={result.returncode}): "
                    f"{result.stderr.decode()[:200]}",
                    flush=True,
                )
        except Exception as e:
            print(f"  [U:Qa] Git commit failed: {e}", flush=True)

    @traces(
        spec="031",
        story="US-031-003",
        reqs=["REQ-031-040"],
    )
    def _parse_review_verdict(self, review_response: str) -> str:
        """Extract PASS or FAIL from REVIEW_VERDICT line.

        FAIL is the safe default (TEST-031-030): a review whose verdict line
        is absent or malformed must not silently pass the gate — failing
        open turned every unparseable LLM response into an approval.
        """
        for line in review_response.splitlines():
            if "REVIEW_VERDICT:" in line:
                verdict_part = line.split("REVIEW_VERDICT:")[1].strip()
                if "PASS" in verdict_part.upper():
                    return "PASS"
                if "FAIL" in verdict_part.upper():
                    return "FAIL"
        return "FAIL"

    @traces(
        spec="031",
        story="US-031-003",
        reqs=["REQ-031-041"],
    )
    def _parse_review_issues(self, review_response: str) -> list:
        """Extract issues list from ISSUES: section."""
        issues = []
        in_issues = False
        for line in review_response.splitlines():
            if line.strip().startswith("ISSUES:"):
                in_issues = True
                continue
            if in_issues:
                stripped = line.strip()
                if stripped.startswith("- ") and len(stripped) > 3:
                    issues.append(stripped[2:])
                elif stripped and not stripped.startswith("-"):
                    break
        return issues

    @traces(spec="050", story="US-001", reqs=["REQ-050-FR1-001"])
    async def _handle_review_edit(self, packet: CXPPacket) -> CXPPacket:
        """SPEC-050 P3.1 fail-closed FLOOR for test-quality regressions.

        Deterministic, no LLM call — not a correctness review. Only test
        files get scrutiny — an edit to a test file is BLOCKed if it
        weakens the suite relative to the original: fewer real ``assert``
        STATEMENTS (counted via AST, not a substring match — a comment or
        docstring merely mentioning the word must never count, and must
        never be gamed by swapping a genuine assert for a vacuous
        ``assert True``/``assert 1``), a newly-added
        ``pytest.mark.skip``/``pytest.skip``, or a newly-introduced
        ``monkeypatch`` usage (this project's no-mocking/no-monkeypatch
        policy — see AGENTS.md § Test Standards). Non-test-file edits are
        approved unconditionally (U:Cd/U:Ar own correctness/architecture).
        """
        file_path, old_code, new_code = _extract_edit_fields(packet.payload)
        verdict = "APPROVE"
        reason = "Not a test file — no test-quality concerns."

        basename = os.path.basename(file_path)
        is_test_file = basename.startswith("test_") or "/tests/" in file_path.replace(
            os.sep, "/"
        )
        if is_test_file:
            verdict, reason = self._review_test_edit(old_code, new_code)

        return CXPPacket(
            from_agent=self.alias,
            to_agent=packet.from_agent,
            thread_id=packet.thread_id,
            subject="review_edit_result",
            payload={
                "verdict": verdict,
                "reason": reason,
                "content": reason,
                "file_path": file_path,
            },
            work_status=WorkStatus.RESOLVED,
        )

    @traces(spec="050", story="US-001", reqs=["REQ-050-FR1-001"])
    def _review_test_edit(self, old_code: str, new_code: str) -> tuple[str, str]:
        """Apply the test-quality checks to a test-file edit.

        Returns:
            ``(verdict, reason)``.
        """
        verdict = "APPROVE"
        reason = "No test-quality regression detected."

        old_tree, _ = parse_fragment(old_code)
        new_tree, _ = parse_fragment(new_code)
        old_asserts = count_asserts(old_tree)
        new_asserts = count_asserts(new_tree)

        if len(new_asserts) < len(old_asserts):
            verdict = "BLOCK"
            reason = (
                f"Edit removes assertions ({len(old_asserts)} -> {len(new_asserts)})."
            )
        elif (
            old_asserts
            and new_asserts
            and not all(is_trivial_assert(a) for a in old_asserts)
            and all(is_trivial_assert(a) for a in new_asserts)
        ):
            verdict = "BLOCK"
            reason = "Edit replaces genuine assertions with vacuous ones."
        elif re.search(r"pytest\.mark\.skip|pytest\.skip", new_code) and not re.search(
            r"pytest\.mark\.skip|pytest\.skip", old_code
        ):
            verdict = "BLOCK"
            reason = "Edit adds a pytest skip to a test file."
        elif "monkeypatch" in new_code and "monkeypatch" not in old_code:
            verdict = "BLOCK"
            reason = "Edit introduces monkeypatch usage (no-mocking policy)."

        return verdict, reason


@traces(spec="050", story="US-001", reqs=["REQ-050-FR1-001"])
def _extract_edit_fields(payload: dict) -> tuple[str, str, str]:
    """Extract (file_path, old_code, new_code) from a review_edit payload.

    Thin re-export of the shared helper (see
    ``orchestrata_ultimate.agents.volks._edit_review.extract_edit_fields``)
    kept under this name for backward compatibility with existing call
    sites/tests in this module.
    """
    return extract_edit_fields(payload)


class QualityAssuranceAgent(QaVolk):
    """Agent alias for QaVolk."""

    pass