"""what-is-a-coding-agent-harness.py — a minimal coding-agent harness loop in about 100 lines.

What it does: shows the four parts every real harness has, in the smallest runnable form:
  1. the loop (ask the model, run the action, feed the result back, repeat)
  2. the tools (a dispatch table: read_file, write_file, run)
  3. the permission gate (deny / ask / allow prefix rules, evaluated in that order)
  4. the context (a message list with a step budget and a byte budget that trims old tool output)

The "model" here is a SCRIPTED STAND-IN, not a language model: it returns a fixed sequence of
actions so the loop can be exercised without an API key. Swap `scripted_model` for a real
model call and everything else stays the same.

Inputs: none (it creates a scratch directory next to itself). Run:  python what-is-a-coding-agent-harness.py
Python 3.13, standard library only.
"""
from __future__ import annotations

import json
import os
import subprocess
import sys
import tempfile

# ---------------------------------------------------------------- 1. tools ---
ROOT = tempfile.mkdtemp(prefix="harness-demo-")


def _safe(path: str) -> str:
    full = os.path.abspath(os.path.join(ROOT, path))
    if not full.startswith(ROOT):
        raise PermissionError(f"{path} is outside the workspace")
    return full


def read_file(path: str) -> str:
    with open(_safe(path), encoding="utf-8") as f:
        return f.read()


def write_file(path: str, content: str) -> str:
    with open(_safe(path), "w", encoding="utf-8", newline="\n") as f:
        f.write(content)
    return f"wrote {len(content)} bytes to {path}"


def run(command: str) -> str:
    p = subprocess.run(command, shell=True, cwd=ROOT, capture_output=True, text=True, timeout=30)
    return (p.stdout + p.stderr).strip() or f"(exit {p.returncode}, no output)"


TOOLS = {"read_file": read_file, "write_file": write_file, "run": run}

# ------------------------------------------------------ 2. permission gate ---
# Rules are (tool, prefix) pairs, checked deny -> ask -> allow, first match wins.
DENY = [("run", "rm "), ("run", "git push")]
ASK = [("write_file", "")]            # every write asks; a real harness remembers the answer
ALLOW = [("read_file", ""), ("run", "python "), ("run", "dir"), ("run", "ls")]


def gate(tool: str, arg: str) -> str:
    for rules, verdict in ((DENY, "deny"), (ASK, "ask"), (ALLOW, "allow")):
        if any(t == tool and arg.startswith(p) for t, p in rules):
            return verdict
    return "ask"


def approve(tool: str, arg: str) -> bool:
    """Stand-in for the human prompt: auto-approve writes inside the workspace."""
    print(f"  [ask] {tool}({arg[:40]!r}) -> approved by policy (demo)")
    return True

# ------------------------------------------------------------- 3. context ---
STEP_BUDGET = 8
CONTEXT_BYTES = 1200      # tiny on purpose, so the trimmer fires in the demo


def trim(messages: list[dict]) -> None:
    """Drop the oldest tool results first, the way real harnesses do, until under budget."""
    while sum(len(m["content"]) for m in messages) > CONTEXT_BYTES:
        old = next((m for m in messages if m["role"] == "tool" and m["content"] != "[trimmed]"), None)
        if old is None:
            return
        old["content"] = "[trimmed]"
        print("  [context] trimmed the oldest tool result")

# ------------------------------------------------- 4. the scripted model ---
SCRIPT = [
    {"tool": "write_file", "arg": "hello.py", "content": "print(sum(range(10)))\n"},
    {"tool": "run", "arg": "python hello.py"},
    {"tool": "run", "arg": "rm -rf /"},               # the gate must deny this
    {"tool": "read_file", "arg": "hello.py"},
    {"tool": "run", "arg": "python -c \"print('.' * 900)\""},   # big output, forces a trim
    {"tool": "run", "arg": "python hello.py"},
    {"done": "hello.py prints 45"},
]


def scripted_model(messages: list[dict]) -> dict:
    turn = sum(1 for m in messages if m["role"] == "assistant")
    return SCRIPT[turn] if turn < len(SCRIPT) else {"done": "script exhausted"}

# ------------------------------------------------------------------ loop ---
def main() -> None:
    messages = [{"role": "system", "content": "You are a coding agent. Use tools; say done when finished."},
                {"role": "user", "content": "Write hello.py that prints the sum of 0..9 and run it."}]
    for step in range(1, STEP_BUDGET + 1):
        action = scripted_model(messages)
        messages.append({"role": "assistant", "content": json.dumps(action)})
        if "done" in action:
            print(f"step {step}: done -> {action['done']}")
            break
        tool, arg = action["tool"], action["arg"]
        verdict = gate(tool, arg)
        if verdict == "deny" or (verdict == "ask" and not approve(tool, arg)):
            result = f"permission {verdict}: {tool} {arg}"
        else:
            try:
                fn = TOOLS[tool]
                result = fn(arg, action["content"]) if tool == "write_file" else fn(arg)
            except Exception as e:  # noqa: BLE001
                result = f"error: {e}"
        print(f"step {step}: {tool}({arg[:32]!r}) [{verdict}] -> {result[:60]!r}")
        messages.append({"role": "tool", "content": result})
        trim(messages)
    else:
        print(f"stopped: step budget of {STEP_BUDGET} reached")
    print(f"context: {len(messages)} messages, {sum(len(m['content']) for m in messages)} bytes")


if __name__ == "__main__":
    sys.stdout.reconfigure(encoding="utf-8")
    main()