orchestrata_ultimate.refactor.agent

Reference › orchestrata_ultimate › refactor › agent

The loop that lets a language model refactor a codebase in place through function calls (SPEC-103, requirements FR-005 to FR-007). intent_broker calls arefactor when an intent is classified as a refactor, and the package facade re-exports it for the scripts.refactor CLI. It uses llm.client for the model calls, core.logging and core.traceability, and the sandboxed RefactorTools.

arefactor(task, root, model, max_iterations) is asynchronous. It creates one RefactorTools for the root, sends the refactoring system prompt and the task together with TOOL_SCHEMAS, and loops up to max_iterations times (25 by default; the default model is deepseek-v4-pro). Each model turn that requests tools has them executed through RefactorTools.dispatch, and the string results are appended to the message history, which keeps the full assistant and tool turns. A path is recorded as changed only when an edit or write returns OK, and the run counts as verified when a run_check returns exit code zero. The loop ends when the model answers without tool calls, or when the iteration limit is reached. refactor is the synchronous wrapper; it never raises and returns a result with status error on any exception. Both return a RefactorResult holding the task, root, changed files, iteration count, tool-call trace, verified flag, final message and a status of completed, max_iterations or error. Further detail is in the refactor package narrative.

At a glance

Class diagram

Dependency diagram

API

Classes

RefactorResult (dataclass)

Decorators: dataclass, traces(spec='103', story='US-103-001', reqs=['REQ-103-023']) · Line 55

Outcome of a refactor run (SPEC-103 data model).

Functions

Source

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

"""Tool-using refactor agent (SPEC-103 FR-005/006/007).

Drives an LLM (DeepSeek) through tool calls to refactor an existing codebase in
place: OBSERVE (read / outline / grep) -> MUTATE (edit_file / write_file) ->
VERIFY (run_check). The loop carries the full assistant/tool message history so
the model has continuity, executes each requested tool call via the sandboxed
:class:`RefactorTools`, and stops when the model returns a final answer with no
tool calls, on a turn budget, or on error.
"""

from __future__ import annotations

import asyncio
import json
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.logging import get_logger
from orchestrata_ultimate.core.traceability import traces
from orchestrata_ultimate.refactor.tools import TOOL_SCHEMAS, RefactorTools

_logger = get_logger("orchestrata.refactor")

# FR-006 (REQ-103-020/021): observe-before-mutate, minimal surgical edits,
# never fabricate file contents, verify before finishing.
REFACTOR_SYSTEM_PROMPT = """\
You are a meticulous code-refactoring engineer working on an EXISTING codebase.
You modify code IN PLACE using the provided tools. You never invent file contents.

Always work in this order:
1. OBSERVE — use list_dir, grep, ast_outline and read_file to understand the real
   code BEFORE changing anything. Read every file you intend to edit.
2. PLAN — decide the minimal, surgical changes that satisfy the task. Preserve
   existing behavior and public APIs unless the task explicitly says otherwise.
3. MUTATE — use edit_file for targeted changes: copy the EXACT text you saw via
   read_file as old_string (it must match byte-for-byte and be unique; include
   surrounding context to make it unique). Use write_file only for genuinely new
   files. Change one thing at a time.
4. VERIFY — after editing, run_check with `python -m py_compile <file>` (and
   `ruff check <file>` if useful). Fix anything you broke.

Rules:
- All paths are relative to the project root; absolute paths are forbidden.
- Never fabricate code you have not read. If edit_file says old_string was not
  found or not unique, re-read the file and correct your old_string.
- Keep changes minimal and on-task; do not reformat unrelated code.
- When the task is done and verification passes, reply with a short summary of
  what you changed and make NO further tool calls.
"""


@dataclass
@traces(spec="103", story="US-103-001", reqs=["REQ-103-023"])
class RefactorResult:
    """Outcome of a refactor run (SPEC-103 data model)."""

    task: str
    root: str
    files_changed: list[str] = field(default_factory=list)
    iterations: int = 0
    tool_calls: list[dict[str, Any]] = field(default_factory=list)
    verified: bool | None = None
    final_message: str = ""
    status: str = "completed"  # completed | max_iterations | error


@traces(
    spec="103",
    story="US-103-001",
    reqs=["REQ-103-017", "REQ-103-018", "REQ-103-019", "REQ-103-023"],
)
async def arefactor(
    task: str,
    root: str | Path,
    model: str = "deepseek-v4-pro",
    max_iterations: int = 25,
) -> RefactorResult:
    """Async entry point for the refactor loop (await from async callers).

    :func:`refactor` is the sync wrapper; callers already inside an event loop
    (e.g. the IntentBroker) await this directly.
    """
    # Local imports: keep module import cheap and avoid importing the provider
    # stack (which reads service env at import) until a refactor actually runs.
    from orchestrata_ultimate.core.protocols import Message
    from orchestrata_ultimate.llm.client import call_with_tools

    tools = RefactorTools(root)
    messages: list[Message] = [
        Message(role="system", content=REFACTOR_SYSTEM_PROMPT),
        Message(
            role="user",
            content=(
                f"Refactoring task:\n{task}\n\n"
                "The project root is '.'. Explore it with the tools, make the "
                "change in place, then verify it."
            ),
        ),
    ]
    result = RefactorResult(task=task, root=str(Path(root).resolve()))
    changed: set[str] = set()

    for i in range(max_iterations):
        result.iterations = i + 1
        resp = await call_with_tools(messages, TOOL_SCHEMAS, model=model)
        if not resp.tool_calls:
            result.final_message = resp.content
            result.status = "completed"
            _logger.info("refactor complete", extra={"cycle": i + 1})
            return result
        messages.append(
            Message(
                role="assistant",
                content=resp.content,
                tool_calls=resp.tool_calls,
                # Thinking models require their reasoning_content echoed back.
                reasoning_content=resp.metadata.get("reasoning_content"),
            )
        )
        for tc in resp.tool_calls:
            fn = tc.get("function", {})
            name = fn.get("name", "")
            try:
                args = json.loads(fn.get("arguments") or "{}")
            except json.JSONDecodeError:
                args = {}
            out = tools.dispatch(name, args)
            _logger.info(
                "refactor tool call",
                extra={"method": name, "params": str(args)[:200], "result": out[:200]},
            )
            result.tool_calls.append(
                {"name": name, "arguments": args, "result": out[:500]}
            )
            if name in ("edit_file", "write_file") and out.startswith("OK"):
                changed.add(str(args.get("path", "")))
            if name == "run_check":
                result.verified = out.startswith("exit=0")
            messages.append(
                Message(
                    role="tool",
                    content=out,
                    tool_call_id=tc.get("id", ""),
                    name=name,
                )
            )
        result.files_changed = sorted(c for c in changed if c)

    result.status = "max_iterations"
    result.files_changed = sorted(c for c in changed if c)
    _logger.warning("refactor hit max_iterations", extra={"cycle": max_iterations})
    return result


@traces(spec="103", story="US-103-001", reqs=["REQ-103-022"])
def refactor(
    task: str,
    root: str | Path,
    model: str = "deepseek-v4-pro",
    max_iterations: int = 25,
) -> RefactorResult:
    """Refactor an existing codebase in place via LLM tool calls (SPEC-103 FR-007).

    Args:
        task: The refactoring instruction.
        root: Project root the agent is sandboxed to.
        model: DeepSeek model id (default ``deepseek-v4-pro``).
        max_iterations: Tool-use loop turn budget.

    Returns:
        A :class:`RefactorResult` with files changed, the tool-call trace, and
        the verification/terminal status. Never raises — failures come back as
        ``status="error"``.
    """
    try:
        return asyncio.run(arefactor(task, root, model, max_iterations))
    except Exception as exc:  # entry point must not raise
        _logger.exception("refactor failed")
        return RefactorResult(
            task=task,
            root=str(root),
            status="error",
            final_message=f"{type(exc).__name__}: {exc}",
        )