scripts.refactor

Reference › scripts › refactor

scripts/refactor.py is the command-line front end for the SPEC-103 tool-calling refactor capability: a model is given a refactoring instruction and a sandboxed project root and edits the code through tool calls. A human runs it. The work is done by refactor in orchestrata_ultimate.refactor, with logging from orchestrata_ultimate.core.logging.

Options: --task (the instruction, required), --root (the project root that acts as the sandbox, required), --model (default deepseek-v4-pro), --max-iterations (default 25) and --trace-out (a path to write the JSON result). _setup_env prepares the environment before the package is imported. After the run it prints the status, iteration count, whether the result was verified, the files changed, the first 600 characters of the final message and a numbered tool-call trace; with --trace-out it also writes the full result as JSON.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/refactor.py

#!/usr/bin/env python3
"""CLI for the SPEC-103 tool-calling refactor capability.

Refactors an existing codebase in place by driving DeepSeek through tool calls.

    python scripts/refactor.py --task "Add type hints and docstrings to foo.py" \
        --root /path/to/project [--model deepseek-v4-pro] [--max-iterations 25]
"""

import argparse
import json
import os
import sys
from pathlib import Path

# Repo root on sys.path so `scripts` (imported transitively) resolves.
_REPO_ROOT = Path(__file__).resolve().parent.parent
if str(_REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(_REPO_ROOT))


def _setup_env() -> None:
    # ProviderManager reads these at import time; refactoring itself is file ops
    # + LLM, but the provider stack still needs the service env present.
    for key, val in {
        "REDIS_PORT": "6379",
        "REDIS_DB": "0",
        # The graph address comes from .env (KG_ROUTING_URI), never a
        # literal: this pinned bolt://10.53.1.11:7687, the Neo4j primary
        # destroyed on 2026-07-27, and setdefault() meant a correctly
        # configured environment was left alone but an unconfigured one
        # got silently aimed at a host that no longer exists.
    }.items():
        os.environ.setdefault(key, val)
    # REDIS_HOST is REQUIRED (RUL-003, no infra default): this used to
    # setdefault 10.53.1.10 — the PRODUCTION Redis — so a correctly
    # configured environment was left alone but an unconfigured one got
    # silently aimed at prod. Same failure mode the KG address comment above
    # already fixed; mirrored here.
    if not os.environ.get("REDIS_HOST"):
        raise RuntimeError(
            "REDIS_HOST is not set; define it in .env — this script has no "
            "default (it used to silently point at the production Redis "
            "10.53.1.10)."
        )
    os.environ.update({"ORCHESTRATA_LLM_PROVIDER": "deepseek", "PYTHONUNBUFFERED": "1"})


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--task", required=True, help="Refactoring instruction")
    ap.add_argument("--root", required=True, help="Project root (sandbox)")
    ap.add_argument("--model", default="deepseek-v4-pro")
    ap.add_argument("--max-iterations", type=int, default=25)
    ap.add_argument(
        "--trace-out", default="", help="Optional path to write the JSON result"
    )
    args = ap.parse_args()

    _setup_env()
    from orchestrata_ultimate.core.logging import configure_logging

    configure_logging(level="INFO", json_format=False)
    from orchestrata_ultimate.refactor import refactor

    res = refactor(
        args.task, args.root, model=args.model, max_iterations=args.max_iterations
    )

    print("=" * 72)
    print(
        f"  refactor status : {res.status}  (iterations={res.iterations}, verified={res.verified})"
    )
    print(f"  root            : {res.root}")
    print(f"  files changed   : {res.files_changed}")
    print(f"  summary         : {res.final_message[:600]}")
    print("=" * 72)
    print("  tool-call trace:")
    for i, tc in enumerate(res.tool_calls):
        a = json.dumps(tc["arguments"])[:70]
        print(
            f"   {i:2} {tc['name']:11}({a}) -> {str(tc['result']).splitlines()[0][:70]}"
        )

    if args.trace_out:
        Path(args.trace_out).write_text(
            json.dumps(res.__dict__, indent=2, default=str), encoding="utf-8"
        )
        print(f"  trace written to {args.trace_out}")
    return 0 if res.status == "completed" else 1


if __name__ == "__main__":
    sys.exit(main())