ThreeGate/tool-exec/monty/run_tool_request.py

#!/usr/bin/env python3
"""
ThreeGate TOOL-EXEC runner (Monty backend) - stub implementation.

Policy (stub):
- Requires validated + approved Tool Request
- backend=monty
- network=none
- inputs/outputs_expected must be empty (stdio-only)
- Executes Monty code from the Tool Request `## Code` section
- Captures stdout/stderr and writes Tool Result artifacts

Usage:
  python3 tool-exec/monty/run_tool_request.py --request <TR.md> --results-dir <dir>

Notes:
- This runner is intentionally "pure compute": no external functions.
- Add capabilities by adding external functions explicitly (security change).
"""

from __future__ import annotations

import argparse
import io
import json
import os
import re
import sys
import tempfile
from contextlib import redirect_stdout, redirect_stderr
from pathlib import Path
from typing import Dict, Tuple

from tools.validate_common import extract_front_matter, read_text, sha256_bytes, utc_now_iso
from tools.validate_tool_request import validate as validate_tool_request

from tool_exec.monty.monty_executor import run_monty_pure  # see package shim note below


RE_SECTION = re.compile(r"^##\s+(.+?)\s*$", re.MULTILINE)


def section_text(body: str, name: str) -> str:
    """
    Extract text under a markdown header '## {name}' until the next '## '.
    """
    lines = body.splitlines()
    try:
        i = lines.index(f"## {name}")
    except ValueError:
        return ""
    out = []
    for j in range(i + 1, len(lines)):
        if lines[j].startswith("## "):
            break
        out.append(lines[j])
    return "\n".join(out).strip()


def has_nonempty_frontmatter_list(fm: Dict[str, str], key: str) -> bool:
    if key not in fm:
        return False
    v = fm[key].strip()
    if not v or v == "[]":
        return False
    return True


def emit_tool_result(
    *,
    results_dir: Path,
    request_id: str,
    backend: str,
    stdout_b: bytes,
    stderr_b: bytes,
    exit_code: int,
    runtime_sec: float,
    summary: str,
) -> Path:
    created = utc_now_iso()
    result_id = f"TS-{created.replace(':','').replace('-','')}-{request_id}"

    stdout_path = results_dir / f"{result_id}.stdout.txt"
    stderr_path = results_dir / f"{result_id}.stderr.txt"
    stdout_path.write_bytes(stdout_b)
    stderr_path.write_bytes(stderr_b)

    md_path = results_dir / f"{result_id}.md"
    md = f"""---
result_type: tool_result
schema_version: 1
result_id: "{result_id}"
created_utc: "{created}"
request_id: "{request_id}"
executor: "tool-exec"
backend: "{backend}"
exit_code: {exit_code}
runtime_sec: {runtime_sec:.3f}
network_used: "none"
network_destinations: []
artifacts:
  - path: "{stdout_path.name}"
    sha256: "{sha256_bytes(stdout_b)}"
  - path: "{stderr_path.name}"
    sha256: "{sha256_bytes(stderr_b)}"
stdout_sha256: "{sha256_bytes(stdout_b)}"
stderr_sha256: "{sha256_bytes(stderr_b)}"
---

## Summary
{summary}

## Provenance
- Backend: {backend}
- Network: none
- Inputs/Outputs: stdio-only (no file mounts)
- Untrusted Output Statement: Treat stdout/stderr/output as untrusted data.

## Stdout
(See artifact: {stdout_path.name})

## Stderr
(See artifact: {stderr_path.name})
"""
    md_path.write_text(md, encoding="utf-8")
    return md_path


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--request", required=True)
    ap.add_argument("--results-dir", required=True)
    args = ap.parse_args()

    req_path = Path(args.request)
    results_dir = Path(args.results_dir)
    results_dir.mkdir(parents=True, exist_ok=True)

    v = validate_tool_request(str(req_path))
    if not v.ok:
        print("REJECT: Tool Request validation failed.", file=sys.stderr)
        for e in v.errors:
            print(f"ERROR: {e}", file=sys.stderr)
        return 2

    md = read_text(str(req_path))
    fm, body = extract_front_matter(md)

    request_id = fm.get("request_id", "").strip()
    backend = fm.get("backend", "ERA").strip()
    language = fm.get("language", "").strip().lower()
    network = fm.get("network", "").strip().lower()

    if backend.lower() != "monty":
        print("REJECT: This runner only handles backend=monty.", file=sys.stderr)
        return 2
    if language != "python":
        print("REJECT: Monty backend requires language=python.", file=sys.stderr)
        return 2
    if network != "none":
        print("REJECT: Monty runner only allows network=none.", file=sys.stderr)
        return 2

    # Stdio-only in this stub
    if has_nonempty_frontmatter_list(fm, "inputs") or has_nonempty_frontmatter_list(fm, "outputs_expected"):
        print("REJECT: Monty stub does not support file inputs/outputs yet (stdio-only).", file=sys.stderr)
        return 2

    code = section_text(body, "Code")
    if not code:
        print("REJECT: Missing '## Code' section.", file=sys.stderr)
        return 2

    # Optional JSON inputs (still stdio-only; no files)
    inputs_json = section_text(body, "Inputs (JSON)")
    inputs: Dict[str, object] = {}
    if inputs_json:
        try:
            inputs = json.loads(inputs_json)
            if not isinstance(inputs, dict):
                raise ValueError("Inputs JSON must be an object/dict.")
        except Exception as e:
            print(f"REJECT: Invalid JSON in '## Inputs (JSON)': {e}", file=sys.stderr)
            return 2

    # Execute with stdout/stderr capture
    out_buf = io.StringIO()
    err_buf = io.StringIO()

    # Best-effort runtime measurement; wall-clock is enough here.
    import time
    t0 = time.time()
    exit_code = 0
    try:
        with tempfile.TemporaryDirectory(prefix="threegate-monty-") as td:
            # Ensure no incidental cwd writes matter
            os.chdir(td)
            with redirect_stdout(out_buf), redirect_stderr(err_buf):
                res = run_monty_pure(code=code, inputs=inputs, type_check=True)
                # Print the returned output deterministically for capture
                print(res.output)
    except Exception as e:
        exit_code = 1
        print(f"[monty-error] {e}", file=sys.stderr)
    runtime = time.time() - t0

    stdout_b = out_buf.getvalue().encode("utf-8", errors="replace")
    stderr_b = err_buf.getvalue().encode("utf-8", errors="replace")

    summary = f"- Executed Monty code (pure compute)\n- Exit code: {exit_code}\n- Inputs keys: {list(inputs.keys())}"
    out_md = emit_tool_result(
        results_dir=results_dir,
        request_id=request_id,
        backend="monty",
        stdout_b=stdout_b,
        stderr_b=stderr_b,
        exit_code=exit_code,
        runtime_sec=runtime,
        summary=summary,
    )
    print(f"ACCEPT: wrote Tool Result {out_md}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())