orchestrata_ultimate.verification.review.engine

Reference › orchestrata_ultimate › verification › review › engine

The driver of the deterministic, no-LLM code review. run_deterministic_review(workspace, files) is re-exported by the review package. It collects the relative paths in the workspace (_collect_workspace_files), infers the project's module names from the layout (_detect_project_modules), reads the dependency names declared in pyproject.toml, splits the requested files into source and test files, adds test files found in the workspace, and then runs _review_single_file over each. It returns a flat list of issues, each a tuple of check id, file path and description.

_review_single_file returns an empty list for a missing file and a single SYNTAX issue for a file that does not parse. Otherwise it runs the checks that apply to every file: imports, function length and arguments, error handling, mutable defaults, mocks, security, environment access, infrastructure env defaults, lint, nesting depth and god objects. Test files additionally get the test-isolation check. Source files get, through _source_only, file completeness, entry point, dependency declaration, placeholders, type hints, docstrings, TODO comments and stub methods. The checks live in checks_correctness, checks_quality, checks_safety and checks_structure. format_review_report renders the issues as human-readable text. The engine reads files and runs checks; it modifies nothing. Ordering and capping of issues is done by issue_triage.

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/verification/review/engine.py

"""Review engine: orchestrates all deterministic checks."""

import ast
import itertools
import os

from orchestrata_ultimate.verification.review.checks_correctness import (
    check_dependency_declaration,
    check_entry_point,
    check_file_completeness,
    check_imports,
)
from orchestrata_ultimate.verification.review.checks_quality import (
    check_docstrings,
    check_mutable_defaults,
    check_type_hints,
)
from orchestrata_ultimate.verification.review.checks_safety import (
    check_env_access,
    check_infra_env_defaults,
    check_lint,
    check_no_mocks,
    check_security,
    check_test_isolation,
    check_todo_comments,
)
from orchestrata_ultimate.verification.review.checks_structure import (
    check_error_handling,
    check_function_args,
    check_function_length,
    check_god_objects,
    check_nesting_depth,
    check_no_placeholder,
    check_stub_methods,
)
from orchestrata_ultimate.verification.review.helpers import (
    _get_pyproject_deps,
    _parse_file,
    _read_lines,
)
from orchestrata_ultimate.verification.review.types import Issue

_MAIN_FILES = {"cli.py", "__main__.py", "main.py"}
_SKIP_DIRS = {
    ".venv",
    "__pycache__",
    ".git",
    "site-packages",
    "node_modules",
}


def run_deterministic_review(
    workspace: str,
    files: list[str],
) -> list[Issue]:
    """Run all deterministic checks on source files.

    Args:
        workspace: Path to project workspace root.
        files: List of relative file paths to review.

    Returns:
        List of (check_id, filepath, description) tuples.
    """
    ws_files = _collect_workspace_files(workspace)
    proj_mods = _detect_project_modules(workspace)
    deps = _get_pyproject_deps(workspace)
    py_src, py_tst = _split_files(files)
    _discover_extra_tests(workspace, py_tst)
    result: list[Issue] = list(
        itertools.chain.from_iterable(
            _review_single_file(workspace, rp, ws_files, proj_mods, deps)
            for rp in py_src + py_tst
        )
    )
    return result


def _collect_workspace_files(workspace: str) -> set[str]:
    """Walk workspace and collect all relative file paths."""
    result: set[str] = {
        os.path.relpath(os.path.join(root, fn), workspace)
        for root, _, fnames in os.walk(workspace)
        for fn in fnames
    }
    return result


def _detect_project_modules(workspace: str) -> set[str]:
    """Determine project module names from workspace layout."""
    mods: set[str] = set()
    src_dir = os.path.join(workspace, "src")
    if os.path.isdir(src_dir):
        for n in os.listdir(src_dir):
            if os.path.isdir(os.path.join(src_dir, n)):
                mods.add(n)
    for n in os.listdir(workspace):
        full = os.path.join(workspace, n)
        if os.path.isdir(full) and os.path.exists(os.path.join(full, "__init__.py")):
            mods.add(n)
    return mods


def _split_files(files: list[str]) -> tuple[list[str], list[str]]:
    """Split files into source and test lists."""

    def _is_test(f: str) -> bool:
        return os.path.basename(f).startswith("test_") or f.startswith("tests/")

    src = [f for f in files if f.endswith(".py") and not _is_test(f)]
    tst = [f for f in files if f.endswith(".py") and _is_test(f)]
    return src, tst


def _discover_extra_tests(
    workspace: str,
    py_tests: list[str],
) -> None:
    """Find test files in workspace not already listed."""
    for root, dirs, fnames in os.walk(workspace):
        dirs[:] = [d for d in dirs if d not in _SKIP_DIRS]
        for fn in fnames:
            if fn.startswith("test_") and fn.endswith(".py"):
                rel = os.path.relpath(os.path.join(root, fn), workspace)
                if rel not in py_tests:
                    py_tests.append(rel)


def _is_test_file(rel_path: str) -> bool:
    """Check whether a relative path is a test file."""
    return os.path.basename(rel_path).startswith("test_") or rel_path.startswith(
        "tests/"
    )


def _review_single_file(
    workspace: str,
    rel_path: str,
    ws_files: set[str],
    proj_mods: set[str],
    deps: set[str],
) -> list[Issue]:
    """Run all applicable checks on a single file."""
    filepath = os.path.join(workspace, rel_path)
    if not os.path.exists(filepath):
        return []
    tree = _parse_file(filepath)
    lines = _read_lines(filepath)
    if tree is None:
        return [("SYNTAX", rel_path, "File has syntax errors -- cannot parse.")]
    issues: list[Issue] = []
    issues.extend(check_imports(filepath, tree, proj_mods))
    issues.extend(check_function_length(filepath, tree))
    issues.extend(check_function_args(filepath, tree))
    issues.extend(check_error_handling(filepath, tree))
    issues.extend(check_mutable_defaults(filepath, tree))
    issues.extend(check_no_mocks(filepath, lines))
    issues.extend(check_security(filepath, tree, lines))
    issues.extend(check_env_access(filepath, tree))
    issues.extend(check_infra_env_defaults(filepath, tree))
    issues.extend(check_lint(filepath))
    issues.extend(check_nesting_depth(tree, lines, filepath))
    issues.extend(check_god_objects(tree, lines, filepath))
    if _is_test_file(rel_path):
        issues.extend(check_test_isolation(filepath, lines))
    else:
        _source_only(filepath, tree, lines, rel_path, ws_files, proj_mods, deps, issues)
    return issues


def _source_only(
    filepath: str,
    tree: ast.Module,
    lines: list[str],
    rel_path: str,
    ws_files: set[str],
    proj_mods: set[str],
    deps: set[str],
    issues: list[Issue],
) -> None:
    """Run checks that apply only to source files."""
    is_main = os.path.basename(rel_path) in _MAIN_FILES
    issues.extend(check_file_completeness(filepath, tree, ws_files, proj_mods))
    issues.extend(check_entry_point(filepath, lines, is_main))
    issues.extend(check_dependency_declaration(filepath, tree, deps, proj_mods))
    issues.extend(check_no_placeholder(filepath, tree))
    issues.extend(check_type_hints(filepath, tree))
    issues.extend(check_docstrings(filepath, tree))
    issues.extend(check_todo_comments(filepath, lines))
    issues.extend(check_stub_methods(tree, lines, filepath))


def format_review_report(issues: list[Issue]) -> str:
    """Format issues into a human-readable review report.

    Args:
        issues: List of (check_id, filepath, description) tuples.

    Returns:
        Formatted review string.
    """
    if not issues:
        return "DETERMINISTIC_REVIEW: PASS (0 issues)"
    by_check: dict[str, list[tuple[str, str]]] = {}
    for check_id, filepath, desc in issues:
        by_check.setdefault(check_id, []).append((filepath, desc))
    lines = [
        f"DETERMINISTIC_REVIEW: FAIL ({len(issues)} issues)",
        "",
    ]
    for check_id, file_issues in sorted(by_check.items()):
        lines.append(f"[{check_id}] ({len(file_issues)} issues)")
        for filepath, desc in file_issues[:5]:
            lines.append(f"  - {desc}")
        if len(file_issues) > 5:
            lines.append(f"  ...and {len(file_issues) - 5} more")
        lines.append("")
    return "\n".join(lines)