Minimal implementation of an LLM harness in Python, with zero dependencies.

harness.py

"""
harness.py

Minimal implementation of an LLM harness in Python, with zero dependencies.

Configure using environment variables

| Variable         | Description             | Example                   |
| ---------------- | ----------------------- | ------------------------- |
| HARNESS_ENDPOINT | Base URL of the LLM API | https://api.openai.com/v1 |
| HARNESS_API_KEY  | Your API key            | sk-...                    |
| HARNESS_MODEL    | Model identifier        | gpt-5.4                   |
"""

import os
import sys
import json
import inspect
import tempfile
import subprocess
import urllib.request
import urllib.error
from pathlib import Path

import readline
import rlcompleter



# ----------------------------------------------------------------------
# Output helpers
# ----------------------------------------------------------------------
# Dedicated printing methods so we can add better formatting later
# (colors, prefixes, log levels, etc.) without touching business logic.
# ----------------------------------------------------------------------

# ANSI colour codes
_C_RESET   = "\033[0m"
_C_GREY    = "\033[90m"
_C_YELLOW  = "\033[33m"
_C_BBLUE   = "\033[44m"   # blue background
_C_WHITE   = "\033[97m"   # white foreground

def _colors_enabled() -> bool:
    """Return True when it looks safe to emit ANSI colour."""
    if os.environ.get("NO_COLOR"):
        return False
    return sys.stdout.isatty() and sys.stderr.isatty()


class Output:
    """Centralised output helper."""

    @staticmethod
    def info(msg: str) -> None:
        """General informational message."""
        print(msg)

    @staticmethod
    def error(msg: str) -> None:
        """Error message."""
        print(f"error: {msg}", file=sys.stderr)

    @staticmethod
    def system(msg: str) -> None:
        """System / agent prompt."""
        print(msg)

    @staticmethod
    def user(msg: str) -> None:
        """User input shown in yellow."""
        if _colors_enabled():
            print(f"{_C_YELLOW}{msg}{_C_RESET}")
        else:
            print(msg)

    @staticmethod
    def assistant(msg: str) -> None:
        """Assistant response – plain text."""
        print(msg)

    @staticmethod
    def tool_call(msg: str) -> None:
        """Tool invocation shown in grey."""
        if _colors_enabled():
            print(f"{_C_GREY}{msg}{_C_RESET}")
        else:
            print(f"{msg}")

    @staticmethod
    def tool_result(msg: str) -> None:
        """Tool result shown in grey."""
        if _colors_enabled():
            print(f"{_C_GREY}{msg}{_C_RESET}")
        else:
            print(f"{msg}")

    @staticmethod
    def blank() -> None:
        """Print a blank separator line."""
        print()

    @staticmethod
    def version(version_str: str) -> None:
        """Application version / startup banner."""
        print(version_str)


# Convenience alias so existing call sites read nicely
output = Output()

# ----------------------------------------------------------------------
# Tools
# ----------------------------------------------------------------------

def tool_bash(command: str) -> str:
    """
    Runs a script using bash. Append `;echo $?` at the end of your command if you
    would like to see the exit status, and `2>&1` to capture stderr. You can chain
    multiple commands together.
    """
    result = subprocess.run(
        ["bash", "-c", command],
        capture_output=True, text=True
    )
    return result.stdout


tool_bash.schema = {
    "type": "function",
    "function": {
        "name": "bash",
        "description": inspect.cleandoc(tool_bash.__doc__),
        "parameters": {
            "type": "object",
            "properties": {
                "command": {
                    "type": "string",
                    "description": "The string to pass to `bash -c`",
                }
            }
        }
    }
}


def tool_list_files(path: str) -> str:
    """
    Gets a list of files and directories in the specified path.
    """

    p = Path(path)

    if not p.exists():
        return f"Path does not exist: {path}"
    if not p.is_dir():
        return f"Path is not a directory: {path}"

    lines = ["size | name"]
    for entry in p.iterdir():
        name = entry.name
        if entry.is_file():
            size = entry.stat().st_size
            lines.append(f"{size} | {name}")
        elif entry.is_dir():
            lines.append(f"dir | {name}")
    return '\n'.join(lines)

tool_list_files.schema = {
    "type": "function",
    "function": {
        "name": "list_files",
        "description": inspect.cleandoc(tool_list_files.__doc__),
        "parameters": {
            "type": "object",
            "properties": {
                "path": {
                    "type": "string",
                    "description": "The path of the directory",
                }
            }
        }
    }
}


def tool_read_file(path: str, start: int | None = None, end: int | None = None, numbered: bool = True) -> str:
    """
    Reads a text file from the starting line number to the ending line number,
    inclusive. At most 500 lines at a time. Line endings normalized to LF.
    """

    p = Path(path)

    if start is None:
        start = 1
    if end is None:
        end = start + 500

    if not p.exists():
        return f"Path does not exist: {path}"
    if not p.is_file():
        return f"Path is not a file: {path}"
    if start > end:
        return f"end ({end}) must be >= start ({start})"
    if start < 1:
        return f"start must be >= 1, got {start}"

    if end - start > 500:
        end = start + 500

    lines = ["(reserved)"]
    current = 0

    with p.open("r", encoding="utf-8") as f:
        for line in f:
            stripped = line.rstrip()
            current += 1
            if start <= current <= end:
                if numbered:
                    lines.append(f"{current}:{stripped}")
                else:
                    lines.append(stripped)

    start = min(current, start)
    end = min(current, end)
    lines[0] = f"{path}: {current} lines total, showing line {start} to {end}"
    return '\n'.join(lines)

tool_read_file.schema = {
    "type": "function",
    "function": {
        "name": "read_file",
        "description": inspect.cleandoc(tool_read_file.__doc__),
        "parameters": {
            "type": "object",
            "properties": {
                "path": {
                    "type": "string",
                    "description": "The path of the file",
                },
                "start": {
                    "type": "number",
                    "description": "Starting line number, inclusive, default 1"
                },
                "end": {
                    "type": "number",
                    "description": "Ending line number, inclusive, default 500"
                },
                "numbered": {
                    "type": "boolean",
                    "description": "Whether to include line numbers in the response, default true"
                }
            }
        }
    }
}


def tool_edit_file(path: str, edits: list[dict]) -> str:
    """
    Applies a list of edits to a text file atomically.

    Each edit is one of:
    - {"type": "delete", "start_line": N, "end_line": M}: delete lines N through M (1-indexed, inclusive)
    - {"type": "insert", "after_line": N, "text": "..."}: insert text after line N.
      Use after_line: 0 to insert before the first line. Text may contain newlines.
    - {"type": "replace", "start_line": N, "end_line": M, "text": "..."}: replace lines N through M
      with the given text. Text may contain newlines. This is a convenient shortcut for
      delete + insert in one operation.

    All edits are applied simultaneously using **original** line numbers.
    """

    p = Path(path)

    if not p.exists():
        return f"Path does not exist: {path}"
    if not p.is_file():
        return f"Path is not a file: {path}"

    try:
        content = p.read_text(encoding="utf-8")
    except Exception as e:
        return f"Cannot read file: {e}"

    ends_with_newline = content.endswith('\n') if content else False
    file_lines = content.split('\n')
    if ends_with_newline and file_lines and file_lines[-1] == '':
        file_lines = file_lines[:-1]

    for i, edit in enumerate(edits):
        edit_type = edit.get("type")
        if edit_type == "delete":
            start = edit.get("start_line")
            end = edit.get("end_line")
            if not isinstance(start, int) or not isinstance(end, int):
                return f"Edit {i}: delete requires integer start_line and end_line"
            if start < 1:
                return f"Edit {i}: start_line ({start}) must be >= 1"
            if end < start:
                return f"Edit {i}: end_line ({end}) must be >= start_line ({start})"
            if start > len(file_lines):
                return f"Edit {i}: start_line ({start}) exceeds file length ({len(file_lines)})"
            if end > len(file_lines):
                return f"Edit {i}: end_line ({end}) exceeds file length ({len(file_lines)})"
        elif edit_type == "insert":
            after = edit.get("after_line")
            text = edit.get("text")
            if not isinstance(after, int):
                return f"Edit {i}: insert requires integer after_line"
            if not isinstance(text, str):
                return f"Edit {i}: insert requires string text"
            if after < 0:
                return f"Edit {i}: after_line ({after}) must be >= 0"
            if after > len(file_lines):
                return f"Edit {i}: after_line ({after}) exceeds file length ({len(file_lines)})"
        elif edit_type == "replace":
            start = edit.get("start_line")
            end = edit.get("end_line")
            text = edit.get("text")
            if not isinstance(start, int) or not isinstance(end, int):
                return f"Edit {i}: replace requires integer start_line and end_line"
            if not isinstance(text, str):
                return f"Edit {i}: replace requires string text"
            if start < 1:
                return f"Edit {i}: start_line ({start}) must be >= 1"
            if end < start:
                return f"Edit {i}: end_line ({end}) must be >= start_line ({start})"
            if start > len(file_lines):
                return f"Edit {i}: start_line ({start}) exceeds file length ({len(file_lines)})"
            if end > len(file_lines):
                return f"Edit {i}: end_line ({end}) exceeds file length ({len(file_lines)})"
        else:
            return f"Edit {i}: unknown type {edit_type!r}, expected 'delete', 'insert' or 'replace'"

    deleted: set[int] = set()
    inserts: dict[int, list[str]] = {}

    for edit in edits:
        if edit["type"] == "delete":
            for line_num in range(edit["start_line"], edit["end_line"] + 1):
                deleted.add(line_num)
        elif edit["type"] == "insert":
            after = edit["after_line"]
            if after not in inserts:
                inserts[after] = []
            inserts[after].append(edit["text"])
        elif edit["type"] == "replace":
            # replace deletes lines [start, end] and inserts text at start_line - 1
            for line_num in range(edit["start_line"], edit["end_line"] + 1):
                deleted.add(line_num)
            insert_after = edit["start_line"] - 1
            if insert_after not in inserts:
                inserts[insert_after] = []
            inserts[insert_after].append(edit["text"])

    new_lines: list[str] = []

    if 0 in inserts:
        for text in inserts[0]:
            new_lines.extend(text.split('\n'))

    for line_idx, line in enumerate(file_lines, start=1):
        if line_idx not in deleted:
            new_lines.append(line)
        if line_idx in inserts:
            for text in inserts[line_idx]:
                new_lines.extend(text.split('\n'))

    new_content = '\n'.join(new_lines)
    if ends_with_newline:
        new_content += '\n'

    tmpfd, tmp_path = tempfile.mkstemp(dir=p.parent, prefix=f".{p.name}.edit-")
    try:
        with os.fdopen(tmpfd, "w", encoding="utf-8") as f:
            f.write(new_content)

        udiff_process = subprocess.run(
            ["diff", "-wsu", str(p), tmp_path],
            capture_output=True, text=True
        )
        udiff = udiff_process.stdout

        os.replace(tmp_path, p)
        res = '\n'.join([
            f"{path}: applied {len(edits)} edit(s):",
            udiff
        ])
        output.tool_result(res)
        return res
    except Exception as e:
        return f"error: {e}"
    finally:
        try:
            os.unlink(tmp_path)
        except FileNotFoundError:
            pass


tool_edit_file.schema = {
    "type": "function",
    "function": {
        "name": "edit_file",
        "description": inspect.cleandoc(tool_edit_file.__doc__),
        "parameters": {
            "type": "object",
            "properties": {
                "path": {
                    "type": "string",
                    "description": "The path of the file",
                },
                "edits": {
                    "type": "array",
                    "description": "List of edits to apply atomically",
                    "items": {
                        "anyOf": [
                            {
                                "type": "object",
                                "properties": {
                                    "type": {"type": "string", "enum": ["delete"]},
                                    "start_line": {"type": "number", "description": "First line to delete (1-indexed, inclusive)"},
                                    "end_line": {"type": "number", "description": "Last line to delete (1-indexed, inclusive)"}
                                },
                                "required": ["type", "start_line", "end_line"]
                            },
                            {
                                "type": "object",
                                "properties": {
                                    "type": {"type": "string", "enum": ["insert"]},
                                    "after_line": {"type": "number", "description": "Line number after which to insert. 0 means before the first line."},
                                    "text": {"type": "string", "description": "Text to insert"}
                                },
                                "required": ["type", "after_line", "text"]
                            },
                            {
                                "type": "object",
                                "properties": {
                                    "type": {"type": "string", "enum": ["replace"]},
                                    "start_line": {"type": "number", "description": "First line to replace (1-indexed, inclusive)"},
                                    "end_line": {"type": "number", "description": "Last line to replace (1-indexed, inclusive)"},
                                    "text": {"type": "string", "description": "Replacement text. May contain newlines."}
                                },
                                "required": ["type", "start_line", "end_line", "text"]
                            }
                        ]
                    }
                }
            },
            "required": ["path", "edits"]
        }
    }
}



# Registry of available tools: name -> handler function


TOOL_REGISTRY = {
    "bash": tool_bash,
    "list_files": tool_list_files,
    "read_file": tool_read_file,
    "edit_file": tool_edit_file,
}
# Schemas exposed to the LLM
TOOLS = [fn.schema for fn in TOOL_REGISTRY.values()]


# ----------------------------------------------------------------------
# Helpers
# ----------------------------------------------------------------------

def require_env(key: str) -> str:
    val = ""
    if key in os.environ:
        val = os.environ[key]

    if val == "":
        output.error(f"{key} is required")
        sys.exit(1)
    return val

def http_post_json(url: str, headers: dict[str, str], body: any) -> tuple[int, bytes]:
    req = urllib.request.Request(url)
    body_data = json.dumps(body).encode('utf-8')

    req.add_header('Content-Type', 'application/json; charset=utf-8')
    req.add_header('Content-Length', len(body_data))
    req.add_header('User-Agent', 'harness.py/1.0 (+https://www.kitsu.red/)')

    for k, v in headers.items():
        req.add_header(k, v)

    try:
        res = urllib.request.urlopen(req, body_data)
        code = res.getcode()
        res_body = res.read()
        return code, res_body
    except urllib.error.HTTPError as e:
        return e.code, e.read()

def format_endpoint(base: str, path: str) -> str:
    return base.rstrip("/") + "/" + path.lstrip("/")

def chat_completions(
            base: str,
            api_key: str,
            model: str,
            messages: list[any],
            tools: list[any]
        ) -> any:
    req_data = {
        "model": model,
        "messages": messages,
        "tools": tools,
    }
    endpoint = format_endpoint(base, "chat/completions")
    headers = { "Authorization": f"Bearer {api_key}" }

    res_code, res_body = http_post_json(endpoint, headers, req_data)
    if res_code != 200:
        raise RuntimeError(f"API error {res_code}: {res_body.decode('utf-8', errors='replace')}")

    res_data = json.loads(res_body)
    return res_data["choices"][0]["message"]

def start():
    cfg_endpoint = require_env("HARNESS_ENDPOINT")
    cfg_key = require_env("HARNESS_API_KEY")
    cfg_model = require_env("HARNESS_MODEL")

    messages = []
    def append_message(role: str, content: str):
        messages.append({ "role": role, "content": content })

    def append_tc(tc_id: str, content: str):
        messages.append({ "role": "tool", "content": content, "tool_call_id": tc_id })

    system_prompt = f"""
You are `harness.py`, a coding agent. Use the provided tools to fulfill the user's request.

Environment:
- Working directory: {Path.cwd()}

Files list:
{tool_list_files(str(Path.cwd()))}
"""

    append_message("system", system_prompt)
    # Configure readline for better interactive input
    readline.parse_and_bind('tab: complete')
    readline.set_history_length(1000)

    # XDG state dir (~/.local/state by default) for persistent session data
    xdg_state = Path(os.environ.get("XDG_STATE_HOME", str(Path.home() / ".local" / "state")))
    harness_state_dir = xdg_state / "harness.py"
    harness_state_dir.mkdir(parents=True, exist_ok=True)
    history_file = harness_state_dir / "history"

    try:
        readline.read_history_file(str(history_file))
    except FileNotFoundError:
        pass

    def _save_readline_history():
        try:
            readline.write_history_file(str(history_file))
        except Exception:
            pass



    output.system(system_prompt)
    while True:
        last_message = messages[-1]
        if last_message["role"] in ["system", "assistant"]:
            try:
                input_text = input("> ")
            except KeyboardInterrupt:
                output.blank()
                continue
            except EOFError:
                output.blank()
                _save_readline_history()
                break

            # Clear the line that input() echoed so we can show our formatted version
            term_width = os.get_terminal_size().columns
            sys.stdout.write(f"\r{' ' * term_width}\r")
            sys.stdout.flush()

            # Handle slash commands
            if input_text.startswith("/"):
                cmd = input_text.strip()[1:]
                if cmd == "clear":
                    messages.clear()
                    append_message("system", system_prompt)
                    output.system(system_prompt)
                    output.blank()
                    continue
                elif cmd == "help":
                    output.info("Available commands:")
                    output.info("  /clear   - Clear conversation history and start over")
                    output.info("  /help    - Show this help message")
                    output.blank()
                    continue
                else:
                    output.error(f"Unknown command: /{cmd}. Type /help for available commands.")
                    output.blank()
                    continue
            append_message("user", input_text)
            output.user(input_text)
            output.blank()

        try:
            res = chat_completions(cfg_endpoint, cfg_key, cfg_model, messages, TOOLS)
        except Exception as e:
            output.error(str(e))
            if messages[-1]["role"] == "user":
                messages.pop()
            continue

        messages.append(res)

        if res["content"] is not None and res["content"] != "":
            output.assistant(res["content"])
            output.blank()

        if "tool_calls" in res and res["tool_calls"] is not None and len(res["tool_calls"]) > 0:
            for tc in res["tool_calls"]:
                tc_id = tc["id"]
                tc_name = tc["function"]["name"]
                tc_args = json.loads(tc["function"]["arguments"])

                output.tool_call(f"{tc_name}({json.dumps(tc_args)})")
                output.blank()
                handler = TOOL_REGISTRY.get(tc_name)
                if handler is None:
                    msg = f"Unknown tool: {tc_name}"
                    append_tc(tc_id, msg)
                    #output.tool_result(msg)
                    continue

                try:
                    result = handler(**tc_args)
                except Exception as e:
                    result = f"error: {e}"
                append_tc(tc_id, result)
                #output.tool_result(result)

def main():
    output.version("harness.py v1.0")
    start()

if __name__ == "__main__":
    main()