orchestrata_ultimate.verification.autofix.engine

Reference › orchestrata_ultimate › verification › autofix › engine

The entry point of the autofix pipeline (Spec 031 US-031-001, REQ-031-015). run_all_autofixes(workspace, venv_python) runs the mechanical fixers in a fixed order from the _FIX_FUNCTIONS list: six Redis fixes from redis_fixes, then relative-import, missing stdlib and missing bare-name import fixes from import_fixes, then the datetime and parse_args fixes from format_fixes. Each returns the number of files changed; non-zero counts are added to an AutofixResult. Finally _run_tool_fixes runs black and isort through the given interpreter and appends their entries to the result. The docstring states that each fix is independent and idempotent. The function edits files in place in the workspace, and it is exported by the autofix package.

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

"""Autofix engine that runs all fixes in order.

Orchestrates the full autofix pipeline: Redis fixes, import fixes,
and format fixes applied sequentially to a workspace.

Spec: 007 - Verification Agent
"""

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.verification.autofix.format_fixes import (
    autofix_f821_datetime,
    autofix_line_length,
    autofix_parse_args,
)
from orchestrata_ultimate.verification.autofix.import_fixes import (
    autofix_import_order,
    autofix_missing_barename_imports,
    autofix_missing_stdlib_imports,
    autofix_relative_imports,
)
from orchestrata_ultimate.verification.autofix.redis_fixes import (
    autofix_aioredis,
    autofix_hset_dict,
    autofix_redis_decode,
    autofix_redis_host,
    autofix_sync_redis,
    autofix_xadd_dict,
)
from orchestrata_ultimate.verification.autofix.types import AutofixResult

_FIX_FUNCTIONS = [
    ("redis", autofix_aioredis),
    ("sync_redis", autofix_sync_redis),
    ("redis_decode", autofix_redis_decode),
    ("redis_host", autofix_redis_host),
    ("xadd_dict", autofix_xadd_dict),
    ("hset_dict", autofix_hset_dict),
    ("relative_imports", autofix_relative_imports),
    ("missing_stdlib_imports", autofix_missing_stdlib_imports),
    ("missing_barename_imports", autofix_missing_barename_imports),
    ("f821_datetime", autofix_f821_datetime),
    ("parse_args", autofix_parse_args),
]


@traces(spec="031", story="US-031-001", reqs=["REQ-031-015"])
def run_all_autofixes(
    workspace: str,
    venv_python: str = "python3",
) -> AutofixResult:
    """Run all autofix functions on a workspace.

    Applies fixes in order: Redis fixes, import fixes, format
    fixes. Each fix is independent and idempotent.

    Args:
        workspace: Root directory of the project.
        venv_python: Path to the Python interpreter for
            tool-based fixes (black, isort).

    Returns:
        AutofixResult with total files fixed and fix descriptions.
    """
    result = AutofixResult()
    for name, fix_fn in _FIX_FUNCTIONS:
        count = fix_fn(workspace)
        if count:
            result.files_fixed += count
            result.fixes_applied.append(
                f"{name}: {count} files",
            )
    _run_tool_fixes(workspace, venv_python, result)
    return result


@traces(spec="031", story="US-031-001", reqs=["REQ-031-012", "REQ-031-013"])
def _run_tool_fixes(
    workspace: str,
    venv_python: str,
    result: AutofixResult,
) -> None:
    """Run tool-based fixes (black, isort)."""
    autofix_line_length(workspace, venv_python)
    result.fixes_applied.append("line_length: black formatted")
    autofix_import_order(workspace, venv_python)
    result.fixes_applied.append("import_order: isort applied")