orchestrata_ultimate.agents.volks.qa._autofix

Reference › orchestrata_ultimate › agents › volks › qa › _autofix

_autofix provides AutofixMixin for the QA volk (volk): mechanical repairs applied to the generated workspace before the review gates run, so the model is not asked to fix what a script can fix. _iter_workspace_py_files and _apply_py_file_fix apply a content transform to every workspace Python file and count the changes. The transforms cover Redis usage (_autofix_aioredis, _autofix_sync_redis, _autofix_redis_decode, _autofix_xadd_dict, _autofix_hset_dict, and _autofix_redis_host, which replaces hardcoded local Redis settings in test files with environment reads via _replace_redis_host_in_content), missing imports (_autofix_f821_datetime, _autofix_missing_stdlib_imports, _autofix_missing_barename_imports), relative imports (_autofix_relative_imports, with _detect_package_name), and formatting (_autofix_line_length with black, _autofix_import_order with isort, _autofix_unused_imports with ruff), plus _autofix_parse_args, which lets parse_args() accept an optional argument list.

At a glance

Class diagram

Dependency diagram

API

Classes

AutofixMixin

Bases: _MixinBase · Line 26

Source autofix routines (redis/imports/formatting) for U:Qa.

Source

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

# mypy: ignore-errors
"""Source autofix routines (redis/imports/formatting) for U:Qa."""

import logging
import os
import re
import shutil
import subprocess
from collections.abc import Callable
from typing import TYPE_CHECKING

from orchestrata_ultimate.agents.volks.qa._host import _QaHost
from orchestrata_ultimate.core.base.traces import traces

_logger = logging.getLogger(__name__)

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


class AutofixMixin(_MixinBase):
    """Source autofix routines (redis/imports/formatting) for U:Qa."""

    @traces(spec="031", story="US-031-001", reqs=["REQ-031-015"])
    def _iter_workspace_py_files(self):
        """Yield absolute paths of workspace .py files (standard dir filter).

        Skips dot-dirs, ``__pycache__`` and ``site-packages`` — the dir
        filter shared by every per-file source auto-fix.
        """
        for root, _dirs, files in os.walk(self.workspace):
            _dirs[:] = [
                d
                for d in _dirs
                if not d.startswith(".") and d != "__pycache__" and d != "site-packages"
            ]
            for fname in files:
                if not fname.endswith(".py"):
                    continue
                yield os.path.join(root, fname)

    @traces(spec="031", story="US-031-001", reqs=["REQ-031-015"])
    def _apply_py_file_fix(
        self, transform: Callable[[str], str], report: str | None = None
    ) -> int:
        """Apply a content transform to every workspace .py file.

        Reads each file, runs ``transform(content) -> new_content`` and
        writes the result back when it changed. Files that cannot be read
        are skipped. When ``report`` is given it is printed (with the
        changed-file count) once one or more files were rewritten.

        Args:
            transform: Callable mapping file content to new content.
            report: Format string with a single ``{count}`` placeholder,
                or None to suppress the summary print.

        Returns:
            Number of files that were rewritten.
        """
        fixed_count = 0
        for fpath in self._iter_workspace_py_files():
            try:
                with open(fpath, encoding="utf-8") as f:
                    content = f.read()
            except OSError, UnicodeDecodeError:
                continue
            new_content = transform(content)
            if new_content != content:
                with open(fpath, "w", encoding="utf-8") as f:
                    f.write(new_content)
                fixed_count += 1
        if fixed_count and report is not None:
            print(report.format(count=fixed_count), flush=True)
        return fixed_count

    @traces(
        spec="031",
        story="US-031-001",
        reqs=["REQ-031-001", "REQ-031-002"],
    )
    def _autofix_aioredis(self) -> None:
        """G131-1: Replace deprecated redis with redis.asyncio.

        DeepSeek sometimes uses the deprecated `redis` package instead
        of `redis.asyncio` (included in redis>=5.0). This auto-fix
        replaces all redis imports and usage patterns.
        """
        fixed_count = 0
        for root, _dirs, files in os.walk(self.workspace):
            _dirs[:] = [
                d
                for d in _dirs
                if not d.startswith(".") and d != "__pycache__" and d != "site-packages"
            ]
            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:
                        content = f.read()
                except OSError, UnicodeDecodeError:
                    continue
                if "redis" not in content:
                    continue
                original = content
                # Replace import statements
                content = re.sub(
                    r"^import redis$",
                    "import redis.asyncio as redis",
                    content,
                    flags=re.MULTILINE,
                )
                content = re.sub(
                    r"^from redis import",
                    "from redis.asyncio import",
                    content,
                    flags=re.MULTILINE,
                )
                # Replace usage patterns
                content = content.replace("redis.from_url", "redis.from_url")
                content = content.replace("redis.Redis", "redis.Redis")
                content = content.replace(
                    "redis.ConnectionError",
                    "redis.ConnectionError",
                )
                content = content.replace(
                    "redis.ResponseError",
                    "redis.ResponseError",
                )
                # Catch any remaining redis references
                content = re.sub(
                    r"\baioredis\b",
                    "redis",
                    content,
                )
                if content != original:
                    with open(fpath, "w", encoding="utf-8") as f:
                        f.write(content)
                    fixed_count += 1
        if fixed_count:
            print(
                f"  [U:Qa] G131-1: auto-fixed redis→redis.asyncio "
                f"in {fixed_count} files",
                flush=True,
            )

    @traces(
        spec="031",
        story="US-031-001",
        reqs=["REQ-031-003"],
    )
    def _autofix_sync_redis(self) -> None:
        """G138-3: Replace sync redis.Redis with redis.asyncio.Redis.

        DeepSeek sometimes uses the synchronous redis.Redis client
        instead of redis.asyncio.Redis. This causes blocking in async
        code. Detect `import redis` (bare) + `redis.Redis(` usage and
        convert to async Redis.
        """

        @traces(
            spec="031",
            story="US-031-001",
            reqs=["REQ-031-003"],
        )
        def transform(content: str) -> str:
            # Skip if already using async redis
            if "redis.asyncio" in content:
                return content
            # Only fix files that use sync redis.Redis
            if not re.search(r"^import redis\b", content, re.MULTILINE):
                return content
            if "redis.Redis(" not in content:
                return content
            # Replace: import redis → from redis.asyncio import Redis
            content = re.sub(
                r"^import redis$",
                "from redis.asyncio import Redis",
                content,
                flags=re.MULTILINE,
            )
            # Replace: redis.Redis( → Redis(
            content = content.replace("redis.Redis(", "Redis(")
            # Replace type hints: redis.Redis → Redis
            content = re.sub(r"\bredis\.Redis\b", "Redis", content)
            # Replace exception refs
            content = content.replace(
                "redis.exceptions.",
                "redis.exceptions.",
            )
            # Add redis.exceptions import if needed
            if "redis.exceptions" in content and (
                "from redis" not in content or "exceptions" not in content
            ):
                content = re.sub(
                    r"^(from redis\.asyncio import Redis)$",
                    r"\1\nfrom redis.exceptions import RedisError",
                    content,
                    count=1,
                    flags=re.MULTILINE,
                )
            return content

        self._apply_py_file_fix(
            transform,
            report="  [U:Qa] G138-3: auto-fixed sync redis→"
            "redis.asyncio in {count} files",
        )

    @traces(
        spec="031",
        story="US-031-001",
        reqs=["REQ-031-004"],
    )
    def _autofix_f821_datetime(self) -> None:
        """G132-1: Add missing datetime import when datetime.X is used.

        DeepSeek often uses datetime.now(), datetime.fromisoformat(),
        etc. but only imports timezone from the datetime module,
        omitting the datetime class itself. This causes F821.
        """

        @traces(
            spec="031",
            story="US-031-001",
            reqs=["REQ-031-004"],
        )
        def transform(content: str) -> str:
            # Check if datetime.X is used as a standalone name
            if not re.search(
                r"\bdatetime\.(now|utcnow|fromisoformat|"
                r"strptime|combine|min|max|today)\b",
                content,
            ):
                return content
            # Check if datetime class is already imported
            if re.search(
                r"^from datetime import .*\bdatetime\b",
                content,
                re.MULTILINE,
            ):
                return content
            if re.search(r"^import datetime$", content, re.MULTILINE):
                return content
            original = content
            # Case 1: from datetime import timezone → add datetime
            content = re.sub(
                r"^(from datetime import )((?!.*\bdatetime\b).*)",
                r"\1datetime, \2",
                content,
                count=1,
                flags=re.MULTILINE,
            )
            # Case 2: no datetime import at all → add one
            if content == original:
                content = re.sub(
                    r"^(import os\b.*\n)",
                    r"\1from datetime import datetime\n",
                    content,
                    count=1,
                    flags=re.MULTILINE,
                )
            # Case 3: fallback — prepend import
            if content == original:
                content = "from datetime import datetime\n" + content
            return content

        self._apply_py_file_fix(
            transform,
            report="  [U:Qa] G132-1: auto-fixed missing datetime "
            "import in {count} files",
        )

    @traces(
        spec="031",
        story="US-031-001",
        reqs=["REQ-031-007"],
    )
    def _autofix_xadd_dict(self) -> None:
        """G134-1: Fix XADD calls that pass dicts with non-string values.

        DeepSeek often writes `xadd(key, **event.to_dict())` or passes
        dicts containing nested objects. Redis XADD requires all field
        values to be strings. This auto-fix wraps dict values with
        json.dumps for non-primitive types.
        """
        fixed_count = 0
        for root, _dirs, files in os.walk(self.workspace):
            _dirs[:] = [
                d
                for d in _dirs
                if not d.startswith(".") and d != "__pycache__" and d != "site-packages"
            ]
            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:
                        content = f.read()
                except OSError, UnicodeDecodeError:
                    continue
                if "xadd" not in content.lower():
                    continue
                original = content
                # Fix: xadd(key, **obj.to_dict()) → xadd(key, {k: str(v) for ...})
                content = re.sub(
                    r"\.xadd\(\s*(\w+)\s*,\s*\*\*(\w+)\.to_dict\(\)",
                    r".xadd(\1, {k: str(v) for k, v in \2.to_dict().items()}",
                    content,
                )
                # Fix: xadd(key, **dict_var) → xadd(key, {k: str(v)...})
                content = re.sub(
                    r"\.xadd\(\s*(\w+)\s*,\s*\*\*(\w+)\s*\)",
                    r".xadd(\1, {k: str(v) for k, v in \2.items()})",
                    content,
                )
                if content != original:
                    with open(fpath, "w", encoding="utf-8") as f:
                        f.write(content)
                    fixed_count += 1
        if fixed_count:
            print(
                f"  [U:Qa] G134-1: auto-fixed XADD dict "
                f"encoding in {fixed_count} files",
                flush=True,
            )

    @traces(
        spec="031",
        story="US-031-001",
        reqs=["REQ-031-008"],
    )
    def _autofix_hset_dict(self) -> None:
        """G136-1: Fix HSET mapping with non-string values.

        DeepSeek often writes `hset(key, mapping=obj.to_dict())` where
        to_dict() returns dicts with datetime objects, Enum values, or
        nested dicts. Redis HSET requires all field values to be
        strings, bytes, or numbers. This auto-fix wraps mapping values
        with str() conversion.
        """
        fixed_count = 0
        for root, _dirs, files in os.walk(self.workspace):
            _dirs[:] = [
                d
                for d in _dirs
                if not d.startswith(".") and d != "__pycache__" and d != "site-packages"
            ]
            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:
                        content = f.read()
                except OSError, UnicodeDecodeError:
                    continue
                if "hset" not in content.lower():
                    continue
                original = content
                # Fix: hset(key, mapping=obj.to_dict())
                content = re.sub(
                    r"\.hset\(([^,]+),\s*mapping\s*=\s*" r"(\w+)\.to_dict\(\)\s*\)",
                    r".hset(\1, mapping="
                    r"{k: str(v) for k, v in \2.to_dict().items()})",
                    content,
                )
                # Fix: hset(key, mapping={**obj.to_dict()})
                content = re.sub(
                    r"\.hset\(([^,]+),\s*mapping\s*=\s*\{"
                    r"\s*\*\*(\w+)\.to_dict\(\)\s*\}\s*\)",
                    r".hset(\1, mapping="
                    r"{k: str(v) for k, v in \2.to_dict().items()})",
                    content,
                )
                if content != original:
                    with open(fpath, "w", encoding="utf-8") as f:
                        f.write(content)
                    fixed_count += 1
        if fixed_count:
            print(
                f"  [U:Qa] G136-1: auto-fixed HSET dict "
                f"encoding in {fixed_count} files",
                flush=True,
            )

    @traces(
        spec="031",
        story="US-031-001",
        reqs=["REQ-031-009"],
    )
    def _autofix_redis_decode(self) -> None:
        """G158-1: Ensure Redis clients use decode_responses=True.

        DeepSeek often creates Redis clients without decode_responses,
        causing bytes vs string comparison failures in tests.
        """
        fixed_count = 0
        for root, _dirs, files in os.walk(self.workspace):
            _dirs[:] = [
                d for d in _dirs if d not in {".venv", "__pycache__", "site-packages"}
            ]
            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:
                        content = f.read()
                except OSError, UnicodeDecodeError:
                    continue
                if "Redis(" not in content and "redis.Redis(" not in content:
                    continue
                original = content
                # Add decode_responses=True if missing
                # Match: Redis(...) without decode_responses
                if "decode_responses" not in content:
                    content = re.sub(
                        r"(redis\.(?:async)?Redis\([^)]*)\)",
                        r"\1, decode_responses=True)",
                        content,
                    )
                    # Fix double comma from empty args
                    content = content.replace(
                        "(, decode_responses",
                        "(decode_responses",
                    )
                if content != original:
                    with open(fpath, "w", encoding="utf-8") as f:
                        f.write(content)
                    fixed_count += 1
        if fixed_count:
            print(
                f"  [U:Qa] G158-1: auto-added decode_responses=True "
                f"in {fixed_count} files",
                flush=True,
            )

    @traces(
        spec="031",
        story="US-031-001",
        reqs=["REQ-031-005"],
    )
    def _autofix_missing_stdlib_imports(self) -> None:
        """G134-2: Add missing stdlib imports when module.X is used.

        Handles common patterns where DeepSeek or QA-generated tests
        use `dataclasses.FrozenInstanceError`, `json.dumps`, etc.
        without importing the module.
        """
        # Map: module name → (usage pattern regex, import statement)
        patterns = {
            "dataclasses": (
                r"\bdataclasses\.(FrozenInstanceError|fields|" r"asdict|astuple)\b",
                "import dataclasses",
            ),
            "json": (
                r"\bjson\.(dumps|loads|dump|load)\b",
                "import json",
            ),
        }

        @traces(
            spec="031",
            story="US-031-001",
            reqs=["REQ-031-005"],
        )
        def transform(content: str) -> str:
            original = content
            for module, (usage_re, import_stmt) in patterns.items():
                if not re.search(usage_re, content):
                    continue
                # Check if already imported
                if re.search(
                    rf"^import {module}\b|^from {module} import",
                    content,
                    re.MULTILINE,
                ):
                    continue
                # Add import after first import line
                content = re.sub(
                    r"^(import \w+.*\n)",
                    rf"\1{import_stmt}\n",
                    content,
                    count=1,
                    flags=re.MULTILINE,
                )
                # Fallback: prepend
                if content == original:
                    content = import_stmt + "\n" + content
            return content

        self._apply_py_file_fix(
            transform,
            report="  [U:Qa] G134-2: auto-fixed missing stdlib "
            "imports in {count} files",
        )

    @traces(
        spec="031",
        story="US-031-001",
        reqs=["REQ-031-006"],
    )
    def _autofix_missing_barename_imports(self) -> None:
        """G150-3: Add missing bare-name imports (timedelta, Optional, etc.).

        DeepSeek frequently uses bare names like timedelta(days=7) or
        Optional[str] without importing them. These cause F821 lint
        errors that persist through fix cycles because DeepSeek's fix
        responses often fail to extract.

        This auto-fix detects bare-name usage and adds the correct
        from-import statement.
        """
        # Map: bare name → (usage regex, from-module, import name)
        bare_imports: list[tuple[str, str, str, str]] = [
            # (name, usage_regex, module, import_name)
            (
                "timedelta",
                r"\btimedelta\s*\(",
                "datetime",
                "timedelta",
            ),
            (
                "Optional",
                r"\bOptional\s*\[",
                "typing",
                "Optional",
            ),
            (
                "List",
                r"\bList\s*\[",
                "typing",
                "List",
            ),
            (
                "Dict",
                r"\bDict\s*\[",
                "typing",
                "Dict",
            ),
            (
                "Any",
                r"\bAny\b",
                "typing",
                "Any",
            ),
            (
                "Union",
                r"\bUnion\s*\[",
                "typing",
                "Union",
            ),
            (
                "Tuple",
                r"\bTuple\s*\[",
                "typing",
                "Tuple",
            ),
        ]

        @traces(
            spec="031",
            story="US-031-001",
            reqs=["REQ-031-006"],
        )
        def transform(content: str) -> str:
            original = content
            for name, usage_re, module, import_name in bare_imports:
                if not re.search(usage_re, content):
                    continue
                # Already imported?
                if re.search(
                    rf"^from {module} import .*\b{name}\b",
                    content,
                    re.MULTILINE,
                ):
                    continue
                if re.search(
                    rf"^import {module}$",
                    content,
                    re.MULTILINE,
                ):
                    continue
                # Try to extend existing from-import
                existing = re.search(
                    rf"^(from {module} import .+)$",
                    content,
                    re.MULTILINE,
                )
                if existing:
                    old_line = existing.group(1)
                    new_line = old_line + f", {import_name}"
                    content = content.replace(old_line, new_line, 1)
                else:
                    # Add new from-import after first import
                    stmt = f"from {module} import {import_name}"
                    content = re.sub(
                        r"^(import \w+.*\n)",
                        rf"\1{stmt}\n",
                        content,
                        count=1,
                        flags=re.MULTILINE,
                    )
                    if content == original:
                        content = stmt + "\n" + content
            return content

        self._apply_py_file_fix(
            transform,
            report="  [U:Qa] G150-3: auto-fixed missing bare-name "
            "imports in {count} files",
        )

    @traces(
        spec="031",
        story="US-031-001",
        reqs=["REQ-031-010"],
    )
    def _autofix_redis_host(self) -> None:
        """G125-1 + G141-1: Replace hardcoded localhost Redis in ALL test files.

        Scans both QA-generated and DeepSeek-generated test/conftest
        files for patterns like:
          host="localhost"  /  host='localhost'  /  Redis(host="localhost"
        and replaces them with os.environ['REDIS_HOST'].
        Also ensures 'import os' is present when env var reads are injected.
        """
        # G141-1: Scan ALL test/conftest files, not just QA-generated ones
        fixed_count = 0
        for fpath in self._collect_test_files():
            if not os.path.isfile(fpath):
                continue
            try:
                with open(fpath, encoding="utf-8") as f:
                    content = f.read()
            except OSError, UnicodeDecodeError:
                continue
            new_content = self._replace_redis_host_in_content(content)
            if new_content != content:
                with open(fpath, "w", encoding="utf-8") as f:
                    f.write(new_content)
                fixed_count += 1
        if fixed_count:
            print(
                f"  [U:Qa] G125-1: auto-fixed Redis host in {fixed_count} test files",
                flush=True,
            )

    @traces(spec="031", story="US-031-001", reqs=["REQ-031-010"])
    def _collect_test_files(self) -> list:
        """Collect workspace test_*.py and conftest.py paths.

        Uses the redis-host dir filter (also excludes ``.venv``).
        """
        all_test_files = []
        for root, _dirs, files in os.walk(self.workspace):
            _dirs[:] = [
                d
                for d in _dirs
                if not d.startswith(".")
                and d != "__pycache__"
                and d != "site-packages"
                and d != ".venv"
            ]
            for fname in files:
                if not fname.endswith(".py"):
                    continue
                if fname.startswith("test_") or fname == "conftest.py":
                    all_test_files.append(os.path.join(root, fname))
        return all_test_files

    @traces(spec="031", story="US-031-001", reqs=["REQ-031-010"])
    def _replace_redis_host_in_content(self, content: str) -> str:
        """Replace hardcoded localhost/port Redis config with env reads.

        Injects ``import os`` when a replacement was made and the import
        is absent. Returns content unchanged when nothing matched.
        """
        original = content
        # Replace hardcoded localhost in Redis-related contexts
        # Pattern: host="localhost" or host='localhost'
        content = re.sub(
            r"""host\s*=\s*["']localhost["']""",
            "host=os.environ['REDIS_HOST']",
            content,
        )
        # Pattern: "localhost" as standalone arg (e.g. Redis("localhost"))
        content = re.sub(
            r"""Redis\(\s*["']localhost["']""",
            "Redis(os.environ['REDIS_HOST']",
            content,
        )
        # Pattern: port=6379 or 6380 hardcoded (replace with env var)
        content = re.sub(
            r"port\s*=\s*(?:6379|6380)\b",
            "port=int(os.environ['REDIS_PORT'])",
            content,
        )
        if content != original:
            # Ensure 'import os' is present
            if "import os" not in content:
                content = "import os\n" + content
        return content

    @traces(
        spec="031",
        story="US-031-001",
        reqs=["REQ-031-011"],
    )
    def _autofix_relative_imports(self) -> None:
        """G125-2: Convert relative imports to absolute in source files.

        DeepSeek consistently generates `from orchestrata_ultimate.agents.volks.qa.module import X` which
        the deterministic review flags as IMPORTS errors. This auto-fix
        detects the package name from the src/ layout and rewrites.
        """
        pkg_name = self._detect_package_name()
        if not pkg_name:
            return
        fixed_count = 0
        for root, _dirs, files in os.walk(self.workspace):
            # Skip venv and non-source dirs
            if ".venv" in root or "__pycache__" in root:
                continue
            for fname in files:
                if not fname.endswith(".py"):
                    continue
                fpath = os.path.join(root, fname)
                with open(fpath, encoding="utf-8") as f:
                    content = f.read()
                original = content
                # Replace `from orchestrata_ultimate.agents.volks.qa.module import X` with `from pkg.module import X`
                content = re.sub(
                    r"from \.([\w.]+) import",
                    f"from {pkg_name}.\\1 import",
                    content,
                )
                # Replace `from orchestrata_ultimate.agents.volks.qa. import X` with `from pkg import X`
                content = re.sub(
                    r"from \. import",
                    f"from {pkg_name} import",
                    content,
                )
                if content != original:
                    with open(fpath, "w", encoding="utf-8") as f:
                        f.write(content)
                    fixed_count += 1
        if fixed_count:
            print(
                f"  [U:Qa] G125-2: auto-fixed relative imports "
                f"in {fixed_count} files (pkg={pkg_name})",
                flush=True,
            )

    @traces(spec="031", story="US-031-001", reqs=["REQ-031-011"])
    def _detect_package_name(self) -> str:
        """Detect the project's top-level package name.

        Prefers the ``src/`` layout (first package dir with an
        ``__init__.py``), then falls back to a flat layout. Returns an
        empty string when no package directory is found.
        """
        # Find the package name from src/ layout
        src_dir = os.path.join(self.workspace, "src")
        if os.path.isdir(src_dir):
            for entry in os.listdir(src_dir):
                entry_path = os.path.join(src_dir, entry)
                if os.path.isdir(entry_path) and not entry.startswith((".", "__")):
                    init = os.path.join(entry_path, "__init__.py")
                    if os.path.isfile(init):
                        return entry
        # Try flat layout — look for __init__.py in workspace dirs
        for entry in os.listdir(self.workspace):
            entry_path = os.path.join(self.workspace, entry)
            if os.path.isdir(entry_path) and not entry.startswith(
                (".", "__", "tests", "design", "docs")
            ):
                init = os.path.join(entry_path, "__init__.py")
                if os.path.isfile(init):
                    return entry
        return ""

    @traces(
        spec="031",
        story="US-031-001",
        reqs=["REQ-031-012"],
    )
    def _autofix_line_length(self) -> None:
        """G127-3: Auto-format source files with black to fix LINE_LENGTH.

        DeepSeek consistently generates lines >88 chars. Rather than
        burning review cycles asking DeepSeek to fix them, run black
        as post-processing — analogous to F401/Redis host auto-fixes.
        """
        venv_python = self._get_venv_python()
        try:
            subprocess.run(
                [venv_python, "-m", "black", "--line-length", "88", "."],
                cwd=self.workspace,
                capture_output=True,
                text=True,
                timeout=30,
                env=self._venv_env(),
            )
            print(
                "  [U:Qa] G127-3: auto-formatted with black",
                flush=True,
            )
        except (subprocess.TimeoutExpired, FileNotFoundError) as exc:
            _logger.debug("black autoformat skipped: %s", exc)

    @traces(
        spec="031",
        story="US-031-001",
        reqs=["REQ-031-013"],
    )
    def _autofix_import_order(self) -> None:
        """G129-3: Auto-fix import ordering with isort.

        DeepSeek consistently fails to fix import ordering
        (stdlib vs third-party). Running isort as post-processing
        fixes IMPORTS issues structurally.
        """
        venv_python = self._get_venv_python()
        try:
            subprocess.run(
                [venv_python, "-m", "isort", "--profile", "black", "."],
                cwd=self.workspace,
                capture_output=True,
                text=True,
                timeout=30,
                env=self._venv_env(),
            )
            print(
                "  [U:Qa] G129-3: auto-fixed import ordering with isort",
                flush=True,
            )
        except (subprocess.TimeoutExpired, FileNotFoundError) as exc:
            _logger.debug("isort autofix skipped: %s", exc)

    @traces(
        spec="031",
        story="US-031-001",
        reqs=["REQ-031-014"],
    )
    def _autofix_unused_imports(self) -> None:
        """G194-2: Remove unused imports with ruff after isort/black.

        isort can re-add imports that were removed. Running ruff --fix
        F401 as the final auto-fix step ensures unused imports are
        cleaned up without being re-introduced.
        """
        venv_python = self._get_venv_python()
        venv_dir = os.path.dirname(os.path.dirname(venv_python))
        ruff_bin = os.path.join(venv_dir, "bin", "ruff")
        if not os.path.isfile(ruff_bin):
            ruff_bin = shutil.which("ruff") or ""
        if not ruff_bin:
            return
        try:
            # Skip __init__.py — re-exports look unused but are
            # public API. Ruff removing them breaks the package.
            subprocess.run(
                [
                    ruff_bin,
                    "check",
                    "--fix",
                    "--select=F401",
                    "--exclude=__init__.py",
                    ".",
                ],
                cwd=self.workspace,
                capture_output=True,
                text=True,
                timeout=30,
            )
            print(
                "  [U:Qa] G194-2: auto-removed unused imports with ruff F401",
                flush=True,
            )
        except (subprocess.TimeoutExpired, FileNotFoundError) as exc:
            _logger.debug("ruff F401 autofix skipped: %s", exc)

    def _autofix_parse_args(self) -> None:
        """G129-2: Patch parse_args() to accept optional args parameter.

        QA generates tests calling parse_args(["create", ...]) but
        DeepSeek's parse_args() reads sys.argv directly. Standard
        argparse testability pattern: parse_args(args=None) where
        None reads sys.argv and a list uses that directly.
        """
        for root, _dirs, files in os.walk(self.workspace):
            _dirs[:] = [
                d
                for d in _dirs
                if not d.startswith(".") and d != "__pycache__" and d != "site-packages"
            ]
            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:
                        content = f.read()
                except OSError, UnicodeDecodeError:
                    continue
                original = content
                # Match: def parse_args(args=None) : ... parser.parse_args()
                # Replace with: def parse_args(args=None): ... parser.parse_args(args)
                content = re.sub(
                    r"def parse_args\(\)\s*(->\s*\S+\s*)?:",
                    r"def parse_args(args=None) \1:",
                    content,
                )
                content = re.sub(
                    r"(return\s+\w+\.parse_args)\(\)",
                    r"\1(args)",
                    content,
                )
                if content != original:
                    with open(fpath, "w", encoding="utf-8") as f:
                        f.write(content)
                    print(
                        f"  [U:Qa] G129-2: auto-fixed parse_args in {fname}",
                        flush=True,
                    )