synthos/coder.py

"""Agentic shell/code loop — a coding agent that runs entirely on the local GGUF.

Give it a task; it constructs a shell command/script, EXECUTES it (guarded), reads the real
output, decides if it worked, and iterates — the construct -> run -> observe -> fix loop that
makes an agent actually accomplish things instead of just describing them. Composes with the
router/MoE (any code-capable expert) and adaptive reasoning (harder tasks get more steps).

Safety: every command goes through tools.run_shell, which refuses destructive patterns and
runs inside a working directory. allow_exec must be set explicitly to actually run.
"""
from __future__ import annotations

import os
import re

from . import quality
from .tools import run_shell

_DONE = re.compile(r"\bDONE\b", re.I)
_LANGS = {"sh", "bash", "shell", "powershell", "ps", "ps1", "cmd", "python", "py", ""}

SYS = (
    "You are an agent that accomplishes tasks by running shell commands on this machine. "
    "Respond with EXACTLY ONE shell command inside a ```sh code block, and nothing else. "
    "After you see its output, either issue the next command or, if the task is fully "
    "verified complete, reply with the single word DONE. Prefer commands that TEST/VERIFY "
    "the result. Keep commands safe and idempotent."
)


def _extract_command(text: str) -> str | None:
    """Robust: handles untagged AND unclosed code fences (the bug that broke the proof)."""
    t = text or ""
    if "```" in t:
        body = t.split("```", 1)[1].split("```", 1)[0]   # after open fence, until close-or-end
        lines = [l for l in body.splitlines() if l.strip()]
        if lines and lines[0].strip().lower() in _LANGS:
            lines = lines[1:]                            # drop a stray language tag line
        if lines:
            return lines[0].strip()                      # first real command line
    # fallback: first line that looks like a command
    for line in t.splitlines():
        s = line.strip().lstrip("$ ").strip()
        if s and (not s[0].isalpha() or s.split()[0] in (
                "ls", "cat", "echo", "python", "git", "node", "dir", "type", "Get-ChildItem")):
            return s
    return None


def build_and_test(client, task: str, workdir: str = ".", max_steps: int = 6,
                   allow_exec: bool = False) -> dict:
    """Run the construct->execute->verify loop. Returns trace + success flag."""
    os.makedirs(workdir, exist_ok=True)
    history, trace = [], []
    convo = f"TASK: {task}\nWorking directory: {os.path.abspath(workdir)}"
    success = False
    for step in range(max_steps):
        prompt = convo + ("\n\nPrevious results:\n" + "\n".join(history) if history else "")
        reply = quality.clean_generate(client, prompt, system=SYS)
        if _DONE.search(reply) and "```" not in reply:
            success = True
            break
        cmd = _extract_command(reply)
        if not cmd:
            history.append(f"[no command parsed from model reply]")
            continue
        full = f"cd {workdir} && {cmd}" if os.name != "nt" else f"cd '{workdir}'; {cmd}"
        result = run_shell(full, allow_exec=allow_exec)
        trace.append({"step": step, "command": cmd, "result": result[:600]})
        history.append(f"$ {cmd}\n{result[:600]}")
    return {"task": task, "success": success, "steps": len(trace), "trace": trace}


if __name__ == "__main__":
    # offline demo of parsing + the guarded executor (no model needed)
    sample = "Here is the command:\n```sh\npython --version\n```"
    print("parsed:", _extract_command(sample))
    print("dry-run:", run_shell("python --version"))
    print("exec   :", run_shell("python --version", allow_exec=True))
    print("blocked:", run_shell("rm -rf /", allow_exec=True))