coding_agent.py

"""Minimal coding-agent harness: LLM + bash tool in a tight loop."""

from __future__ import annotations

import inspect
import json
import os
import shutil
import subprocess
from pathlib import Path
from typing import Any

import anthropic
from dotenv import load_dotenv

load_dotenv()

claude_client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

YOU_COLOR = "\u001b[94m"
ASSISTANT_COLOR = "\u001b[93m"
RESET_COLOR = "\u001b[0m"


def resolve_abs_path(path_str: str) -> Path:
    """Resolve relative paths against the current working directory."""
    path = Path(path_str).expanduser()
    if not path.is_absolute():
        path = (Path.cwd() / path).resolve()
    return path


# def read_file_tool(filename: str) -> dict[str, Any]:
#     """
#     Gets the full content of a file provided by the user.
#     :param filename: The name of the file to read.
#     :return: The full content of the file.
#     """
#     full_path = resolve_abs_path(filename)
#     content = full_path.read_text(encoding="utf-8")
#     return {"file_path": str(full_path), "content": content}
#
#
# def list_files_tool(path: str) -> dict[str, Any]:
#     """
#     Lists the files in a directory provided by the user.
#     :param path: The path to a directory to list files from.
#     :return: A list of files in the directory.
#     """
#     full_path = resolve_abs_path(path)
#     all_files: list[dict[str, str]] = []
#     for item in sorted(full_path.iterdir(), key=lambda p: (not p.is_file(), p.name.lower())):
#         all_files.append({"filename": item.name, "type": "file" if item.is_file() else "dir"})
#     return {"path": str(full_path), "files": all_files}
#
#
# def edit_file_tool(path: str, old_str: str, new_str: str) -> dict[str, Any]:
#     """
#     Replaces first occurrence of old_str with new_str in file. If old_str is empty,
#     create/overwrite file with new_str.
#     :param path: The path to the file to edit.
#     :param old_str: The string to replace.
#     :param new_str: The string to replace with.
#     :return: A dictionary with the path to the file and the action taken.
#     """
#     full_path = resolve_abs_path(path)
#     if old_str == "":
#         full_path.parent.mkdir(parents=True, exist_ok=True)
#         full_path.write_text(new_str, encoding="utf-8")
#         return {"path": str(full_path), "action": "created_file"}
#
#     original = full_path.read_text(encoding="utf-8")
#     if original.find(old_str) == -1:
#         return {"path": str(full_path), "action": "old_str not found"}
#
#     edited = original.replace(old_str, new_str, 1)
#     full_path.write_text(edited, encoding="utf-8")
#     return {"path": str(full_path), "action": "edited"}


def bash_tool(command: str, cwd: str | None = None) -> dict[str, Any]:
    """
    Run a shell command in bash. Use for reading files (cat/head), listing dirs (ls),
    editing (your editor or sed/python -c), git, tests, etc.
    :param command: The command to run as a single string (bash -lc).
    :param cwd: Optional working directory; relative paths resolve against the process cwd.
    :return: exit_code, stdout, stderr, and cwd used.
    """
    bash = shutil.which("bash")
    if not bash:
        return {
            "error": "bash not found in PATH. On Windows, install Git for Windows and ensure "
            "Git\\bin is on PATH, or use WSL.",
        }
    workdir = resolve_abs_path(cwd) if cwd else Path.cwd()
    if not workdir.is_dir():
        return {"error": f"cwd is not a directory: {workdir}"}
    try:
        proc = subprocess.run(
            [bash, "-lc", command],
            cwd=str(workdir),
            capture_output=True,
            text=True,
            timeout=300,
        )
    except subprocess.TimeoutExpired:
        return {"error": "command timed out after 300s", "cwd": str(workdir)}
    except OSError as e:
        return {"error": str(e), "cwd": str(workdir)}
    return {
        "cwd": str(workdir),
        "exit_code": proc.returncode,
        "stdout": proc.stdout,
        "stderr": proc.stderr,
    }


TOOL_REGISTRY: dict[str, Any] = {
    "bash": bash_tool,
    # "read_file": read_file_tool,
    # "list_files": list_files_tool,
    # "edit_file": edit_file_tool,
}


def get_tool_str_representation(tool_name: str) -> str:
    tool = TOOL_REGISTRY[tool_name]
    doc = inspect.getdoc(tool) or ""
    return f"""
    Name: {tool_name}
    Description: {doc}
    Signature: {inspect.signature(tool)}
    """


SYSTEM_PROMPT = """
You are a coding assistant whose goal it is to help us solve coding tasks.
You have access to a series of tools you can execute. Here are the tools you can execute:

{tool_list_repr}

When you want to use a tool, reply with exactly one line per tool in the format: 'tool: TOOL_NAME({{JSON_ARGS}})' and nothing else on that line.
Use compact single-line JSON with double quotes. After tool execution you receive one message tool_result(...) which may be a JSON array when multiple tools ran.
If no tool is needed, respond normally.
"""


def get_full_system_prompt() -> str:
    tool_str_repr = ""
    for tool_name in TOOL_REGISTRY:
        tool_str_repr += "TOOL\n===" + get_tool_str_representation(tool_name)
        tool_str_repr += f"\n{'=' * 15}\n"
    return SYSTEM_PROMPT.format(tool_list_repr=tool_str_repr)


def extract_tool_invocations(text: str) -> list[tuple[str, dict[str, Any]]]:
    """
    Return list of (tool_name, args) requested in 'tool: name({...})' lines.
    The parser expects single-line, compact JSON in parentheses.
    """
    invocations: list[tuple[str, dict[str, Any]]] = []
    for raw_line in text.splitlines():
        line = raw_line.strip()
        if not line.startswith("tool:"):
            continue
        try:
            after = line[len("tool:") :].strip()
            name, rest = after.split("(", 1)
            name = name.strip()
            if not rest.endswith(")"):
                continue
            json_str = rest[:-1].strip()
            args = json.loads(json_str)
            if not isinstance(args, dict):
                continue
            invocations.append((name, args))
        except (json.JSONDecodeError, ValueError):
            continue
    return invocations


def execute_llm_call(conversation: list[dict[str, str]]) -> str:
    system_content = ""
    messages: list[dict[str, str]] = []

    for msg in conversation:
        if msg["role"] == "system":
            system_content = msg["content"]
        else:
            messages.append(msg)

    response = claude_client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2000,
        system=system_content,
        messages=messages,
    )
    block = response.content[0]
    if block.type != "text":
        raise RuntimeError(f"Unexpected content block type: {block.type}")
    return block.text


def dispatch_tool(name: str, args: dict[str, Any]) -> dict[str, Any]:
    if name not in TOOL_REGISTRY:
        return {"error": f"unknown tool: {name}"}

    tool = TOOL_REGISTRY[name]
    if name == "bash":
        cmd = args.get("command", "")
        if not isinstance(cmd, str):
            return {"error": "bash requires string 'command'"}
        cwd_arg = args.get("cwd")
        if cwd_arg is not None and not isinstance(cwd_arg, str):
            return {"error": "bash 'cwd' must be a string or omitted"}
        return tool(cmd, cwd_arg)
    return tool(**args)


def run_coding_agent_loop() -> None:
    system_prompt = get_full_system_prompt()
    conversation: list[dict[str, str]] = [{"role": "system", "content": system_prompt}]

    print(f"{ASSISTANT_COLOR}Harness ready.{RESET_COLOR} Type exit or Ctrl+C to quit.\n")

    while True:
        try:
            user_input = input(f"{YOU_COLOR}You:{RESET_COLOR} ").strip()
        except (KeyboardInterrupt, EOFError):
            print()
            break

        if user_input.lower() in {"exit", "quit"}:
            break

        conversation.append({"role": "user", "content": user_input})

        while True:
            assistant_response = execute_llm_call(conversation)
            tool_invocations = extract_tool_invocations(assistant_response)

            conversation.append({"role": "assistant", "content": assistant_response})

            if not tool_invocations:
                print(f"{ASSISTANT_COLOR}Assistant:{RESET_COLOR} {assistant_response}\n")
                break

            batch: list[dict[str, Any]] = []
            for name, args in tool_invocations:
                print(f"{ASSISTANT_COLOR}[tool]{RESET_COLOR} {name} {json.dumps(args)}")
                try:
                    resp = dispatch_tool(name, args)
                except OSError as e:
                    resp = {"error": str(e)}
                batch.append({"tool": name, "args": args, "result": resp})
            conversation.append({"role": "user", "content": f"tool_result({json.dumps(batch)})"})


if __name__ == "__main__":
    run_coding_agent_loop()