orchestrata_ultimate.agents.volks._edit_review

Reference › orchestrata_ultimate › agents › volks › _edit_review

_edit_review holds deterministic checks for review_edit payloads, the edit fragments that volks review (it is imported by the architect and CodeDirector volks, among others, and has no internal imports of its own). extract_edit_fields pulls the file path, old code and new code out of a payload. parse_fragment makes a best-effort parse of a code fragment, trying successive candidates from _fragment_candidates, and returns the syntax tree or None with an error. On a parsed tree, count_asserts and is_trivial_assert find vacuous assertions, has_bare_except finds bare except: clauses, has_relative_import finds relative imports, and has_direct_driver_construction finds direct construction of graph or Redis drivers. Each of the last three has a regex fallback (regex_has_bare_except, regex_has_relative_import, regex_has_direct_driver_construction) for fragments that do not parse.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

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

"""Shared helpers for the ``review_edit`` handlers (SPEC-050 P3.1).

These are the cheap, deterministic, no-LLM checks that back the
CheetahClaws fail-closed edit gate for U:Cd/U:Qa/U:Ar. This module is
NOT a correctness review — it is a fail-closed FLOOR: syntax parsing,
deletion-ratio, test-weakening, and architecture heuristics only. Real
correctness/design review remains an LLM/human responsibility.

The one non-trivial piece of shared logic is fragment-tolerant parsing:
``new_code``/``old_code`` in a ``review_edit`` payload is usually a
FRAGMENT (e.g. an indented mid-function replacement like
``"    return x\n"``), not a full, top-level-valid module. A naive
``ast.parse`` on that fragment raises ``IndentationError`` for
perfectly ordinary edits. :func:`parse_fragment` retries with
``textwrap.dedent`` and then with a synthetic enclosing ``if True:``
block before giving up and reporting the fragment as genuinely broken.
"""

from __future__ import annotations

import ast
import re
import textwrap
from collections.abc import Iterator


def _fragment_candidates(code: str) -> Iterator[str]:
    """Yield successive parse candidates for a code fragment.

    Order: the fragment as-is, the dedented fragment, and finally the
    dedented fragment re-indented inside a synthetic ``if True:`` block
    (so a fragment that is only valid nested inside *some* enclosing
    scope — e.g. a lone ``return`` or ``break`` — still parses).
    """
    yield code
    dedented = textwrap.dedent(code)
    if dedented != code:
        yield dedented
    else:
        dedented = code
    yield "if True:\n" + textwrap.indent(dedented, "    ")


def parse_fragment(code: str) -> tuple[ast.Module | None, str | None]:
    """Best-effort parse of a ``review_edit`` code fragment.

    Tries ``ast.parse`` on the fragment as-is; on ``IndentationError``/
    ``SyntaxError`` retries with :func:`textwrap.dedent`, then retries
    wrapped in a synthetic enclosing scope. Only reports failure
    (``tree is None``) once every attempt has failed — i.e. the
    fragment is genuinely broken Python, not just an ordinarily
    indented snippet.

    Args:
        code: The code fragment to parse (may be empty).

    Returns:
        ``(tree, None)`` on success, or ``(None, error_message)`` where
        ``error_message`` is the SyntaxError from the FIRST (as-is)
        parse attempt — the most meaningful one to surface to a caller.
    """
    if not code.strip():
        return ast.parse(""), None

    first_error: str | None = None
    for candidate in _fragment_candidates(code):
        try:
            return ast.parse(candidate), None
        except (IndentationError, SyntaxError) as exc:
            if first_error is None:
                first_error = str(exc)
            continue
    return None, first_error or "unparseable fragment"


def count_asserts(tree: ast.Module | None) -> list[ast.Assert]:
    """Return every real ``ast.Assert`` statement in a parsed fragment.

    Only walks actual AST ``Assert`` nodes — a substring match on
    ``"assert "`` is bypassable by a comment or a docstring that merely
    mentions the word. ``tree`` may be ``None`` (fragment failed to
    parse even with :func:`parse_fragment`'s fallbacks), in which case
    an empty list is returned.
    """
    if tree is None:
        return []
    return [n for n in ast.walk(tree) if isinstance(n, ast.Assert)]


def is_trivial_assert(node: ast.Assert) -> bool:
    """True if an ``assert`` statement is vacuous (``assert True``/``assert 1``)."""
    test = node.test
    return isinstance(test, ast.Constant) and test.value in (True, 1)


def has_bare_except(tree: ast.Module) -> bool:
    """True if the parsed fragment contains a bare ``except:`` clause."""
    return any(
        isinstance(n, ast.ExceptHandler) and n.type is None for n in ast.walk(tree)
    )


def has_relative_import(tree: ast.Module) -> bool:
    """True if the parsed fragment contains a relative (``from .``) import."""
    return any(isinstance(n, ast.ImportFrom) and n.level > 0 for n in ast.walk(tree))


def has_direct_driver_construction(tree: ast.Module) -> bool:
    """True if the parsed fragment directly constructs a graph/Redis driver.

    Matches ``GraphDatabase.driver(...)`` and ``redis.Redis(...)`` calls —
    the two patterns the architecture rule (no direct driver construction
    outside a factory module) forbids.
    """
    for node in ast.walk(tree):
        if not isinstance(node, ast.Call):
            continue
        func = node.func
        if not isinstance(func, ast.Attribute) or not isinstance(func.value, ast.Name):
            continue
        if func.value.id == "GraphDatabase" and func.attr == "driver":
            return True
        if func.value.id == "redis" and func.attr == "Redis":
            return True
    return False


_BARE_EXCEPT_RE = re.compile(r"except\s*:")
_RELATIVE_IMPORT_RE = re.compile(r"^\s*from\s+\.+\w*", re.MULTILINE)
_DRIVER_RE = re.compile(r"GraphDatabase\.driver\(|redis\.Redis\(")


def regex_has_bare_except(code: str) -> bool:
    """Regex fallback for :func:`has_bare_except` on unparseable fragments."""
    return bool(_BARE_EXCEPT_RE.search(code))


def regex_has_relative_import(code: str) -> bool:
    """Regex fallback for :func:`has_relative_import` on unparseable fragments."""
    return bool(_RELATIVE_IMPORT_RE.search(code))


def regex_has_direct_driver_construction(code: str) -> bool:
    """Regex fallback for :func:`has_direct_driver_construction`."""
    return bool(_DRIVER_RE.search(code))


def extract_edit_fields(payload: dict) -> tuple[str, str, str]:
    """Extract ``(file_path, old_code, new_code)`` from a review_edit payload.

    Supports both the flat SPEC-050 checklist format
    (``file_path``/``old_code``/``new_code``) and the shape
    ``MultiAgentCoordinator.dispatch_review`` actually sends today
    (``{"edit": {file_path, old_string, new_string, ...}, "prompt": ...}``).
    Every field is coerced with ``or ""`` so a key present with an
    explicit ``None`` value (rather than absent) never produces a
    ``None`` field that crashes a downstream ``.startswith()``/``len()``.
    """
    edit = payload.get("edit")
    if isinstance(edit, dict):
        file_path = edit.get("file_path") or ""
        old_code = (edit.get("old_string") or edit.get("old_code")) or ""
        new_code = (edit.get("new_string") or edit.get("new_code")) or ""
    else:
        file_path = payload.get("file_path") or ""
        old_code = payload.get("old_code") or ""
        new_code = payload.get("new_code") or ""
    return file_path, old_code, new_code