491 lines
18 KiB
Python
Executable File
491 lines
18 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Evaluate production gates for generated code with real compile/test execution."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Dict, List, Tuple
|
|
|
|
|
|
PLACEHOLDER_PATTERNS = [
|
|
r"\bTODO\b",
|
|
r"\bFIXME\b",
|
|
r"\bplaceholder\b",
|
|
r"auto\s*/\*\s*TODO",
|
|
r"/\*\s*missing",
|
|
]
|
|
|
|
|
|
@dataclass
|
|
class ExecResult:
|
|
passed: bool
|
|
skipped: bool
|
|
reason: str
|
|
return_code: int
|
|
stdout: str
|
|
stderr: str
|
|
command: List[str]
|
|
|
|
|
|
def truncate(text: str, limit: int = 6000) -> str:
|
|
if len(text) <= limit:
|
|
return text
|
|
return text[: limit - 32] + "\n...<truncated>...\n"
|
|
|
|
|
|
def detect_placeholders(code: str) -> Dict[str, object]:
|
|
findings: List[Dict[str, object]] = []
|
|
for pat in PLACEHOLDER_PATTERNS:
|
|
for m in re.finditer(pat, code, re.IGNORECASE):
|
|
findings.append({"pattern": pat, "start": m.start(), "end": m.end()})
|
|
return {"passed": len(findings) == 0, "count": len(findings), "findings": findings}
|
|
|
|
|
|
def run_cmd(cmd: List[str], timeout_s: int, env: Dict[str, str] | None = None) -> ExecResult:
|
|
try:
|
|
p = subprocess.run(
|
|
cmd,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
timeout=timeout_s,
|
|
check=False,
|
|
env=env,
|
|
)
|
|
return ExecResult(
|
|
passed=(p.returncode == 0),
|
|
skipped=False,
|
|
reason="ok" if p.returncode == 0 else "non_zero_exit",
|
|
return_code=p.returncode,
|
|
stdout=truncate(p.stdout),
|
|
stderr=truncate(p.stderr),
|
|
command=cmd,
|
|
)
|
|
except FileNotFoundError:
|
|
return ExecResult(
|
|
passed=False,
|
|
skipped=True,
|
|
reason="tool_missing",
|
|
return_code=127,
|
|
stdout="",
|
|
stderr="",
|
|
command=cmd,
|
|
)
|
|
except subprocess.TimeoutExpired as e:
|
|
return ExecResult(
|
|
passed=False,
|
|
skipped=False,
|
|
reason="timeout",
|
|
return_code=124,
|
|
stdout=truncate(e.stdout or ""),
|
|
stderr=truncate(e.stderr or ""),
|
|
command=cmd,
|
|
)
|
|
|
|
|
|
def language_tools(language: str) -> Dict[str, str]:
|
|
lang = language.lower()
|
|
if lang in ("cpp", "c++"):
|
|
cxx = shutil.which("g++") or shutil.which("clang++") or ""
|
|
return {"compile": cxx, "test": cxx}
|
|
if lang == "python":
|
|
py = shutil.which("python3") or shutil.which("python") or ""
|
|
return {"compile": py, "test": py}
|
|
if lang == "rust":
|
|
rustc = shutil.which("rustc") or ""
|
|
return {"compile": rustc, "test": rustc}
|
|
if lang == "go":
|
|
go = shutil.which("go") or ""
|
|
return {"compile": go, "test": go}
|
|
return {"compile": "", "test": ""}
|
|
|
|
|
|
def detect_cpp_namespace_prefix(code: str) -> str:
|
|
m = re.search(r"namespace\s+([A-Za-z_][A-Za-z0-9_]*)\s*\{", code)
|
|
if m:
|
|
return m.group(1) + "::"
|
|
q = re.search(r"([A-Za-z_][A-Za-z0-9_]*)::PriorityQueue", code)
|
|
if q:
|
|
return q.group(1) + "::"
|
|
return ""
|
|
|
|
|
|
def cpp_harness(code: str) -> str:
|
|
prefix = detect_cpp_namespace_prefix(code)
|
|
return (
|
|
code
|
|
+ "\nint main() {\n"
|
|
+ f" {prefix}PriorityQueue q;\n"
|
|
+ f" {prefix}WorkItem w(\"job\", 1, \"payload\");\n"
|
|
+ " (void)q;\n"
|
|
+ " (void)w;\n"
|
|
+ " return 0;\n"
|
|
+ "}\n"
|
|
)
|
|
|
|
|
|
def rust_harness(code: str) -> str:
|
|
return (
|
|
code
|
|
+ "\nfn main() {\n"
|
|
+ " let q = PriorityQueue::default();\n"
|
|
+ " let w = WorkItem{job_id: \"job\".to_string(), priority: 1, payload: \"payload\".to_string()};\n"
|
|
+ " let _ = q;\n"
|
|
+ " let _ = w;\n"
|
|
+ "}\n"
|
|
)
|
|
|
|
|
|
def go_harness(code: str) -> str:
|
|
return (
|
|
code
|
|
+ "\nfunc main() {\n"
|
|
+ " q := PriorityQueue{}\n"
|
|
+ " _ = q\n"
|
|
+ "}\n"
|
|
)
|
|
|
|
|
|
def python_harness(code: str) -> str:
|
|
return (
|
|
code
|
|
+ "\nif __name__ == '__main__':\n"
|
|
+ " q = PriorityQueue()\n"
|
|
+ " w = WorkItem('job', 1, 'payload')\n"
|
|
+ " _ = q\n"
|
|
+ " _ = w\n"
|
|
)
|
|
|
|
|
|
def compile_check(code: str, language: str, strict: bool) -> Dict[str, object]:
|
|
lang = language.lower()
|
|
tools = language_tools(lang)
|
|
if lang not in ("cpp", "c++", "python", "rust", "go"):
|
|
return {
|
|
"passed": False,
|
|
"skipped": True,
|
|
"reason": "unsupported_language",
|
|
"strict_blocking": bool(strict),
|
|
}
|
|
|
|
with tempfile.TemporaryDirectory(prefix="whetstone_gate_compile_") as td:
|
|
td_path = Path(td)
|
|
if lang in ("cpp", "c++"):
|
|
src = td_path / "generated.cpp"
|
|
src.write_text(code)
|
|
cmd = [tools["compile"] or "g++", "-std=c++20", "-fsyntax-only", str(src)]
|
|
elif lang == "python":
|
|
src = td_path / "generated.py"
|
|
src.write_text(code)
|
|
cmd = [tools["compile"] or "python3", "-m", "py_compile", str(src)]
|
|
elif lang == "rust":
|
|
src = td_path / "generated.rs"
|
|
src.write_text(code)
|
|
cmd = [tools["compile"] or "rustc", "--crate-type", "lib", str(src), "-o", str(td_path / "libgenerated.rlib")]
|
|
else: # go
|
|
src = td_path / "generated.go"
|
|
src.write_text(code)
|
|
go = tools["compile"] or "go"
|
|
cmd = [go, "build", str(src)]
|
|
go_env = dict()
|
|
go_env.update({"GOCACHE": str(td_path / "go-build-cache"), "GOMODCACHE": str(td_path / "go-mod-cache"), "HOME": str(td_path)})
|
|
|
|
if lang == "go":
|
|
res = run_cmd(cmd, timeout_s=20, env=go_env)
|
|
else:
|
|
res = run_cmd(cmd, timeout_s=20)
|
|
payload = {
|
|
"passed": res.passed,
|
|
"skipped": res.skipped,
|
|
"reason": res.reason,
|
|
"return_code": res.return_code,
|
|
"command": res.command,
|
|
"stdout": res.stdout,
|
|
"stderr": res.stderr,
|
|
"strict_blocking": bool(strict and (res.skipped or not res.passed)),
|
|
}
|
|
return payload
|
|
|
|
|
|
def test_check(code: str, language: str, strict: bool) -> Dict[str, object]:
|
|
lang = language.lower()
|
|
queue_spec = ("priorityqueue" in code.lower())
|
|
if not queue_spec:
|
|
return {
|
|
"passed": True,
|
|
"skipped": True,
|
|
"reason": "queue_harness_not_required",
|
|
"strict_blocking": False,
|
|
}
|
|
|
|
required = ["enqueue", "dequeue", "peek", "size", "empty", "workitem"]
|
|
text = code.lower()
|
|
missing = [r for r in required if r not in text]
|
|
if missing:
|
|
return {
|
|
"passed": False,
|
|
"skipped": False,
|
|
"reason": "missing_required_queue_members",
|
|
"missing": missing,
|
|
"strict_blocking": bool(strict),
|
|
}
|
|
|
|
required = ["enqueue", "dequeue", "peek", "size", "empty", "workitem"]
|
|
text = code.lower()
|
|
missing = [r for r in required if r not in text]
|
|
if missing:
|
|
return {
|
|
"passed": False,
|
|
"skipped": False,
|
|
"reason": "missing_required_queue_members",
|
|
"missing": missing,
|
|
"strict_blocking": bool(strict),
|
|
}
|
|
|
|
with tempfile.TemporaryDirectory(prefix="whetstone_gate_test_") as td:
|
|
td_path = Path(td)
|
|
if lang in ("cpp", "c++"):
|
|
src = td_path / "harness.cpp"
|
|
src.write_text(cpp_harness(code))
|
|
cxx = shutil.which("g++") or shutil.which("clang++")
|
|
cmd = [cxx or "g++", "-std=c++20", "-fsyntax-only", str(src)]
|
|
elif lang == "python":
|
|
src = td_path / "harness.py"
|
|
src.write_text(python_harness(code))
|
|
py = shutil.which("python3") or shutil.which("python")
|
|
cmd = [py or "python3", str(src)]
|
|
elif lang == "rust":
|
|
src = td_path / "harness.rs"
|
|
src.write_text(rust_harness(code))
|
|
rustc = shutil.which("rustc")
|
|
cmd = [rustc or "rustc", str(src), "-o", str(td_path / "harness")]
|
|
elif lang == "go":
|
|
src = td_path / "harness.go"
|
|
src.write_text(go_harness(code))
|
|
go = shutil.which("go")
|
|
cmd = [go or "go", "build", str(src)]
|
|
go_env = dict()
|
|
go_env.update({"GOCACHE": str(td_path / "go-build-cache"), "GOMODCACHE": str(td_path / "go-mod-cache"), "HOME": str(td_path)})
|
|
else:
|
|
return {
|
|
"passed": False,
|
|
"skipped": True,
|
|
"reason": "unsupported_language",
|
|
"strict_blocking": bool(strict),
|
|
}
|
|
|
|
if lang == "go":
|
|
res = run_cmd(cmd, timeout_s=25, env=go_env)
|
|
else:
|
|
res = run_cmd(cmd, timeout_s=25)
|
|
return {
|
|
"passed": res.passed,
|
|
"skipped": res.skipped,
|
|
"reason": res.reason,
|
|
"return_code": res.return_code,
|
|
"command": res.command,
|
|
"stdout": res.stdout,
|
|
"stderr": res.stderr,
|
|
"strict_blocking": bool(strict and (res.skipped or not res.passed)),
|
|
}
|
|
|
|
|
|
def lint_check(code: str, language: str, lint_strict: bool) -> Dict[str, object]:
|
|
lang = language.lower()
|
|
with tempfile.TemporaryDirectory(prefix="whetstone_gate_lint_") as td:
|
|
td_path = Path(td)
|
|
if lang in ("cpp", "c++"):
|
|
src = td_path / "generated.cpp"
|
|
src.write_text(code)
|
|
lint = shutil.which("cppcheck")
|
|
if not lint:
|
|
return {"passed": True, "skipped": True, "reason": "tool_missing", "strict_blocking": bool(lint_strict)}
|
|
cmd = [lint, "--enable=warning,style", "--quiet", str(src)]
|
|
elif lang == "python":
|
|
src = td_path / "generated.py"
|
|
src.write_text(code)
|
|
py = shutil.which("python3") or shutil.which("python")
|
|
if not py:
|
|
return {"passed": True, "skipped": True, "reason": "tool_missing", "strict_blocking": bool(lint_strict)}
|
|
probe = run_cmd([py, "-c", "import importlib.util,sys;sys.exit(0 if importlib.util.find_spec('pyflakes') else 1)"], timeout_s=10)
|
|
if not probe.passed:
|
|
return {"passed": True, "skipped": True, "reason": "tool_missing", "strict_blocking": bool(lint_strict)}
|
|
cmd = [py, "-m", "pyflakes", str(src)]
|
|
elif lang == "go":
|
|
src = td_path / "generated.go"
|
|
src.write_text(code)
|
|
gopls = shutil.which("gopls")
|
|
go = shutil.which("go")
|
|
if gopls:
|
|
cmd = [gopls, "check", str(src)]
|
|
elif go:
|
|
cmd = [go, "vet", str(src)]
|
|
else:
|
|
return {"passed": True, "skipped": True, "reason": "tool_missing", "strict_blocking": bool(lint_strict)}
|
|
go_env = dict()
|
|
go_env.update({"GOCACHE": str(td_path / "go-build-cache"), "GOMODCACHE": str(td_path / "go-mod-cache"), "HOME": str(td_path)})
|
|
elif lang == "rust":
|
|
rustc = shutil.which("rustc")
|
|
if not rustc:
|
|
return {"passed": True, "skipped": True, "reason": "tool_missing", "strict_blocking": bool(lint_strict)}
|
|
src = td_path / "generated.rs"
|
|
src.write_text(code)
|
|
cmd = [rustc, "-D", "warnings", "--crate-type", "lib", str(src), "-o", str(td_path / "liblint.rlib")]
|
|
else:
|
|
return {"passed": True, "skipped": True, "reason": "unsupported_language", "strict_blocking": False}
|
|
|
|
if lang == "go":
|
|
res = run_cmd(cmd, timeout_s=20, env=go_env)
|
|
else:
|
|
res = run_cmd(cmd, timeout_s=20)
|
|
lint_failed = (not res.passed and not res.skipped)
|
|
return {
|
|
"passed": not lint_failed,
|
|
"skipped": res.skipped,
|
|
"reason": res.reason,
|
|
"return_code": res.return_code,
|
|
"command": res.command,
|
|
"stdout": res.stdout,
|
|
"stderr": res.stderr,
|
|
"strict_blocking": bool(lint_strict and lint_failed),
|
|
}
|
|
|
|
|
|
def normalize_diagnostics(compile_res: Dict[str, object], test_res: Dict[str, object]) -> List[Dict[str, object]]:
|
|
patterns: List[Tuple[str, re.Pattern[str]]] = [
|
|
("gcc_clang", re.compile(r"^(?P<file>[^:\n]+):(\d+):(\d+):\s*(?P<sev>warning|error):\s*(?P<msg>.+)$", re.MULTILINE)),
|
|
("go", re.compile(r"^(?P<file>[^:\n]+):(?P<line>\d+):(?P<col>\d+):\s*(?P<msg>.+)$", re.MULTILINE)),
|
|
("rustc", re.compile(r"^(?P<sev>error|warning)(\[[^\]]+\])?:\s*(?P<msg>.+)$", re.MULTILINE)),
|
|
("python", re.compile(r"File \"(?P<file>[^\"]+)\", line (?P<line>\d+).*(?P<msg>SyntaxError:.*)$", re.MULTILINE)),
|
|
]
|
|
|
|
out: List[Dict[str, object]] = []
|
|
|
|
def scan(source: str, category: str, text: str) -> None:
|
|
seen = set()
|
|
for family, pat in patterns:
|
|
for m in pat.finditer(text or ""):
|
|
file_name = m.groupdict().get("file", "")
|
|
sev = m.groupdict().get("sev", "error")
|
|
msg = m.groupdict().get("msg", "parse failure")
|
|
line_txt = m.groupdict().get("line")
|
|
col_txt = m.groupdict().get("col")
|
|
line = int(line_txt) if line_txt and line_txt.isdigit() else None
|
|
col = int(col_txt) if col_txt and col_txt.isdigit() else None
|
|
if line is None and m.lastindex and m.lastindex >= 2 and m.group(2) and m.group(2).isdigit():
|
|
line = int(m.group(2))
|
|
if col is None and m.lastindex and m.lastindex >= 3 and m.group(3) and m.group(3).isdigit():
|
|
col = int(m.group(3))
|
|
key = (source, category, file_name, line, col, msg)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
out.append(
|
|
{
|
|
"source": source,
|
|
"family": family,
|
|
"severity": "warning" if sev == "warning" else "error",
|
|
"category": category,
|
|
"file": file_name,
|
|
"line": line,
|
|
"column": col,
|
|
"message": msg.strip() or "diagnostic",
|
|
"raw_excerpt": truncate(m.group(0), 300),
|
|
}
|
|
)
|
|
|
|
scan("compile", "compile", str(compile_res.get("stderr", "")))
|
|
scan("tests", "tests", str(test_res.get("stderr", "")))
|
|
if not out and (not compile_res.get("passed", True) or not test_res.get("passed", True)):
|
|
out.append(
|
|
{
|
|
"source": "gate",
|
|
"family": "fallback",
|
|
"severity": "error",
|
|
"category": "gates",
|
|
"file": "",
|
|
"line": None,
|
|
"column": None,
|
|
"message": "Gate failed without parseable diagnostic output",
|
|
"raw_excerpt": "",
|
|
}
|
|
)
|
|
return out
|
|
|
|
|
|
def main() -> None:
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--code-file", help="Path to generated code file", default="")
|
|
ap.add_argument("--code", help="Inline generated code", default="")
|
|
ap.add_argument("--language", required=True)
|
|
ap.add_argument("--strict", action="store_true", help="Strict mode: missing tools/skips are blocking")
|
|
ap.add_argument("--lint-strict", action="store_true", help="Fail gate on lint violations")
|
|
ap.add_argument("--out", default="")
|
|
args = ap.parse_args()
|
|
|
|
code = args.code
|
|
if args.code_file:
|
|
code = Path(args.code_file).read_text()
|
|
|
|
placeholders = detect_placeholders(code)
|
|
compile_res = compile_check(code, args.language, strict=args.strict)
|
|
test_res = test_check(code, args.language, strict=args.strict)
|
|
lint_res = lint_check(code, args.language, lint_strict=args.lint_strict)
|
|
diagnostics = normalize_diagnostics(compile_res, test_res)
|
|
diagnostics.extend(normalize_diagnostics({"stderr": lint_res.get("stderr", ""), "passed": lint_res.get("passed", True)},
|
|
{"stderr": "", "passed": True}))
|
|
|
|
compile_pass = bool(compile_res.get("passed", False))
|
|
test_pass = bool(test_res.get("passed", False))
|
|
if args.strict:
|
|
compile_pass = compile_pass and not bool(compile_res.get("strict_blocking", False))
|
|
test_pass = test_pass and not bool(test_res.get("strict_blocking", False))
|
|
lint_pass = True
|
|
if args.lint_strict:
|
|
lint_pass = bool(lint_res.get("passed", True))
|
|
lint_pass = lint_pass and not bool(lint_res.get("strict_blocking", False))
|
|
|
|
overall = bool(placeholders["passed"] and compile_pass and test_pass and lint_pass)
|
|
|
|
failure_reasons: List[str] = []
|
|
if not placeholders["passed"]:
|
|
failure_reasons.append("placeholder_or_todo_detected")
|
|
if not compile_pass:
|
|
failure_reasons.append(f"compile_failed:{compile_res.get('reason', 'unknown')}")
|
|
if not test_pass:
|
|
failure_reasons.append(f"tests_failed:{test_res.get('reason', 'unknown')}")
|
|
if args.lint_strict and not lint_pass:
|
|
failure_reasons.append(f"lint_failed:{lint_res.get('reason', 'unknown')}")
|
|
|
|
payload = {
|
|
"language": args.language,
|
|
"strict_mode": bool(args.strict),
|
|
"diagnostics": diagnostics,
|
|
"gates": {
|
|
"compile": compile_res,
|
|
"tests": test_res,
|
|
"lint": lint_res,
|
|
"placeholder": placeholders,
|
|
"failure_reasons": failure_reasons,
|
|
"overall_ready": overall,
|
|
},
|
|
}
|
|
|
|
text = json.dumps(payload, indent=2, sort_keys=True)
|
|
if args.out:
|
|
out = Path(args.out)
|
|
out.parent.mkdir(parents=True, exist_ok=True)
|
|
out.write_text(text + "\n")
|
|
print(text)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|