MiniMax M2.5: Agentic Refactoring

import anthropic
import subprocess

def build_refactor_context(root_dir: str, hotspot_files: list[str], 
                            max_file_tokens: int = 8000) -> str:
    """
    Package context for a multi-file refactor:
    1. Lightweight repo tree (structure, not content)
    2. Full content of hotspot files only
    3. Explicit constraints the agent must follow
    """
    # 1. Repo tree — lightweight, structure only
    tree = subprocess.run(
        ["git", "ls-files", "--others", "--exclude-standard", "-c"],
        cwd=root_dir, capture_output=True, text=True
    ).stdout
    # Filter to relevant extensions only
    relevant = [f for f in tree.strip().split("\n") 
                if f.endswith((".py", ".ts", ".go", ".rs", ".java"))]
    tree_summary = "\n".join(relevant[:80])  # cap at 80 files

    # 2. Hotspot file contents
    file_contents = []
    for path in hotspot_files:
        with open(f"{root_dir}/{path}") as f:
            content = f.read()
        tokens_est = len(content) // 4  # rough token estimate
        if tokens_est > max_file_tokens:
            content = content[:max_file_tokens * 4] + "\n... [truncated]"
        file_contents.append(f"### File: {path}\n```\n{content}\n```")

    # 3. Constraints block
    constraints = """
CONSTRAINTS — follow exactly:
- Only modify files listed in HOTSPOT FILES
- Output diffs in unified diff format (diff -u style)  
- Do not rename functions or classes — only refactor internals
- Preserve all existing test interfaces
- Flag any cross-file dependency that requires coordinated changes
"""
    return f"REPO STRUCTURE:\n{tree_summary}\n\nHOTSPOT FILES:\n" + \
           "\n\n".join(file_contents) + f"\n\n{constraints}"

SYSTEM_PROMPT = """You are a senior software engineer performing a targeted refactor.
You operate in three strict phases:
1. PLAN: Describe what needs to change and why, file by file. No code yet.
2. DIFF: Output the minimal unified diff for each file change.
3. VERIFY: List what you'd check in a code review of your own output.
Do not skip phases. Label each phase clearly."""

def run_refactor_agent(context: str, refactor_goal: str) -> str:
    client = anthropic.Anthropic()
    response = client.messages.create(
        model="MiniMax-M2.5",
        max_tokens=4096,
        system=SYSTEM_PROMPT,
        messages=[{
            "role": "user",
            "content": f"{context}\n\nREFACTOR GOAL:\n{refactor_goal}"
        }]
    )
    return response.content[0].text

PLAN:
- src/db/connection.py: Replace 3 callback-style query functions with async/await.
  Dependencies: src/api/users.py imports query_user() and query_session() from this module.
  Coordinated change required: callers in users.py must also be updated to await.
- src/api/users.py: Update 2 call sites for query_user() and query_session() to use await.
  No interface changes — external signatures unchanged.
- Cross-file risk: if session_middleware.py also imports from connection.py, it will break.
  Flagging for manual check — not in hotspot list.

--- a/src/db/connection.py
+++ b/src/db/connection.py
@@ -12,8 +12,8 @@ import psycopg2
-def query_user(user_id: str, callback):
-    conn = get_connection()
-    result = conn.execute("SELECT * FROM users WHERE id = %s", (user_id,))
-    callback(result.fetchone())
+async def query_user(user_id: str) -> dict | None:
+    async with get_async_connection() as conn:
+        result = await conn.execute("SELECT * FROM users WHERE id = %s", (user_id,))
+        return await result.fetchone()

import anthropic
import subprocess
import json

client = anthropic.Anthropic()

# Define the tool chain
TOOLS = [
    {
        "name": "search_codebase",
        "description": "Search the repository for files or patterns. Returns matching file paths and line numbers.",
        "input_schema": {
            "type": "object",
            "properties": {
                "pattern": {"type": "string", "description": "Grep pattern to search for"},
                "file_glob": {"type": "string", "description": "File glob to limit search, e.g. '*.py'"}
            },
            "required": ["pattern"]
        }
    },
    {
        "name": "run_tests",
        "description": "Run the project test suite for a specific file or test name. Returns exit code and output.",
        "input_schema": {
            "type": "object",
            "properties": {
                "target": {"type": "string", "description": "Test file path or test function name"},
                "timeout_seconds": {"type": "integer", "default": 60}
            },
            "required": ["target"]
        }
    },
    {
        "name": "write_patch",
        "description": "Apply a unified diff patch to the repository. Returns success/failure.",
        "input_schema": {
            "type": "object",
            "properties": {
                "target_file": {"type": "string", "description": "File path to patch"},
                "patch_content": {"type": "string", "description": "Unified diff content to apply"},
                "branch": {"type": "string", "description": "Git branch to write to — must not be 'main'"}
            },
            "required": ["target_file", "patch_content", "branch"]
        }
    }
]

def handle_tool_call(tool_name: str, tool_input: dict) -> str:
    """Tool executor with permission scoping."""
    if tool_name == "search_codebase":
        result = subprocess.run(
            ["grep", "-rn", tool_input["pattern"], 
             "--include", tool_input.get("file_glob", "*")],
            capture_output=True, text=True, cwd="./repo"
        )
        return result.stdout[:4000]  # cap output

    elif tool_name == "run_tests":
        result = subprocess.run(
            ["pytest", tool_input["target"], "-v", "--tb=short"],
            capture_output=True, text=True, timeout=tool_input.get("timeout_seconds", 60)
        )
        return f"Exit code: {result.returncode}\n{result.stdout[-2000:]}"

    elif tool_name == "write_patch":
        # PERMISSION SCOPE: block writes to protected branches
        protected = {"main", "master", "develop", "release"}
        if tool_input.get("branch") in protected:
            return "ERROR: write_patch cannot target protected branches. Use a feature branch."
        # Apply the patch
        result = subprocess.run(
            ["git", "apply", "--check"],
            input=tool_input["patch_content"], capture_output=True, text=True
        )
        if result.returncode != 0:
            return f"Patch validation failed: {result.stderr}"
        subprocess.run(["git", "apply"], input=tool_input["patch_content"])
        return f"Patch applied to {tool_input['target_file']} on branch {tool_input['branch']}"

    return f"Unknown tool: {tool_name}"

MAX_TOOL_STEPS = 15  # Hard ceiling on tool calls per agent session

def run_agent_with_budget(task: str, tools: list, max_steps: int = MAX_TOOL_STEPS):
    messages = [{"role": "user", "content": task}]
    steps_used = 0

    while steps_used < max_steps:
        response = client.messages.create(
            model="MiniMax-M2.5",
            max_tokens=2048,
            tools=tools,
            messages=messages
        )

        # Append full response to maintain reasoning chain
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason == "end_turn":
            return response  # Agent finished cleanly

        if response.stop_reason == "tool_use":
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    steps_used += 1
                    if steps_used >= max_steps:
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": "BUDGET EXCEEDED: Agent step limit reached. Summarize progress and stop."
                        })
                        break
                    result = handle_tool_call(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result
                    })
            messages.append({"role": "user", "content": tool_results})

    raise RuntimeError(f"Agent exceeded {max_steps} tool steps without completing")

# .github/workflows/ai-patch-validation.yml
name: AI Patch Validation

on:
  pull_request:
    types: [opened, synchronize]
    branches: [main]

jobs:
  validate-ai-patch:
    runs-on: ubuntu-latest
    # Only run on PRs from the AI agent branch prefix
    if: startsWith(github.head_ref, 'agent/')

    steps:
      - name: Checkout PR branch
        uses: actions/checkout@v4
        with:
          ref: ${{ github.head_ref }}

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'

      - name: Install dependencies
        run: pip install -r requirements.txt

      - name: Run full test suite
        id: tests
        run: |
          pytest tests/ -v --tb=short --json-report --json-report-file=results.json
          echo "exit_code=$?" >> $GITHUB_OUTPUT

      - name: Post test summary as PR comment
        uses: actions/github-script@v7
        with:
          script: |
            const fs = require('fs');
            const results = JSON.parse(fs.readFileSync('results.json', 'utf8'));
            const summary = `## AI Patch Validation Results
            - **Passed:** ${results.summary.passed}
            - **Failed:** ${results.summary.failed}  
            - **Duration:** ${results.duration.toFixed(2)}s
            ${results.summary.failed > 0 ? '❌ Merge blocked — fix failing tests first.' : '✅ All tests pass — ready to merge.'}`;
            github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body: summary
            });

      - name: Gate merge on test pass
        if: steps.tests.outputs.exit_code != '0'
        run: exit 1

AGENT_BRANCH_PREFIX = "agent/"

def validate_branch_name(branch: str) -> bool:
    """Agent branches must use the 'agent/' prefix and cannot be protected."""
    protected = {"main", "master", "develop", "staging", "release"}
    if branch in protected:
        return False
    if not branch.startswith(AGENT_BRANCH_PREFIX):
        return False  # Must use prefix for CI routing
    return True

import re

def parse_and_validate_agent_output(raw_output: str) -> dict:
    """
    Extract PLAN, DIFF, VERIFY sections from agent output.
    Raises ValueError with actionable message if structure is missing.
    """
    sections = {}
    
    for section in ["PLAN", "DIFF", "VERIFY"]:
        # Flexible pattern: allows markdown headers or plain labels
        pattern = rf"(?:#{1,3}\s*)?{section}:?\s*\n(.*?)(?=(?:#{1,3}\s*)?(?:PLAN|DIFF|VERIFY)|$)"
        match = re.search(pattern, raw_output, re.DOTALL | re.IGNORECASE)
        if not match:
            raise ValueError(
                f"Agent output missing '{section}' section. "
                f"Re-prompt with explicit phase labels required.\n"
                f"Output preview: {raw_output[:300]}"
            )
        sections[section] = match.group(1).strip()

    # Extract unified diffs from the DIFF section
    diff_pattern = r"```diff\n(.*?)```"
    diffs = re.findall(diff_pattern, sections["DIFF"], re.DOTALL)
    if not diffs:
        raise ValueError(
            "DIFF section found but contains no fenced diff blocks. "
            "Check that the agent is outputting `diff` code blocks."
        )
    
    sections["diffs"] = diffs
    return sections

def rank_files_by_relevance(repo_root: str, refactor_goal: str, 
                             target_files: list[str]) -> list[tuple[str, int]]:
    """
    Rank candidate files by keyword overlap with the refactor goal.
    Returns list of (file_path, relevance_score) sorted descending.
    """
    keywords = set(refactor_goal.lower().split())
    ranked = []
    
    for path in target_files:
        try:
            with open(f"{repo_root}/{path}") as f:
                content = f.read().lower()
            score = sum(content.count(kw) for kw in keywords)
            ranked.append((path, score))
        except (IOError, UnicodeDecodeError):
            continue
    
    return sorted(ranked, key=lambda x: x[1], reverse=True)

def build_sharded_context(repo_root: str, refactor_goal: str,
                           all_files: list[str], 
                           token_budget: int = 40_000) -> list[str]:
    """
    Build context shards that each stay under the token budget.
    Returns list of context strings for sequential agent calls.
    """
    ranked = rank_files_by_relevance(repo_root, refactor_goal, all_files)
    
    shards = []
    current_shard_files = []
    current_token_estimate = 0
    
    for path, score in ranked:
        with open(f"{repo_root}/{path}") as f:
            content = f.read()
        file_tokens = len(content) // 4
        
        if current_token_estimate + file_tokens > token_budget:
            if current_shard_files:
                shards.append(current_shard_files)
            current_shard_files = [path]
            current_token_estimate = file_tokens
        else:
            current_shard_files.append(path)
            current_token_estimate += file_tokens
    
    if current_shard_files:
        shards.append(current_shard_files)
    
    return shards