#!/usr/bin/env python3 """Aria Pre-Output-Critic Lint (KAR-65). Reads a JSON tool-input payload from stdin (Claude Code hook contract) or from a file via --input, runs the declarative cases in eval-lint-cases.yaml against the relevant text fields, and exits with code 2 if any case of severity 'high' matches. Severity 'medium' prints warnings to stderr; 'low' is log-only. CLI: eval-lint.py --tool telegram_reply # stdin JSON eval-lint.py --tool telegram_reply --text "..." # ad-hoc text eval-lint.py --selftest # runs builtin test cases """ from __future__ import annotations import argparse import json import re import sys from dataclasses import dataclass from pathlib import Path import yaml CASES_PATH = Path("/root/aria/config/eval-lint-cases.yaml") LOG_PATH = Path("/root/.claude/hooks/eval-lint.log") CODE_INLINE_RE = re.compile(r"`[^`\n]+`") CODE_BLOCK_RE = re.compile(r"```.*?```", re.DOTALL) URL_RE = re.compile(r"https?://\S+") @dataclass class Hit: case: str severity: str match: str why: str fix_hint: str def load_cases(path: Path = CASES_PATH) -> list[dict]: data = yaml.safe_load(path.read_text()) return data.get("cases", []) def strip_code(text: str) -> str: """Remove code blocks, inline code, and URLs so patterns scoped outside code don't false-positive on legitimate content inside backticks.""" out = CODE_BLOCK_RE.sub("", text) out = CODE_INLINE_RE.sub("", out) out = URL_RE.sub("", out) return out # Tiny, safe expression evaluator for the YAML 'applies_when' field. # Supports only: IDENT == "STRING" joined by and / or . # This is intentionally narrow — any other syntax falls back to "true". COND_TERM_RE = re.compile(r'\s*(\w+)\s*==\s*"([^"]*)"\s*') def case_applies(case: dict, ctx: dict) -> bool: cond = case.get("applies_when") if not cond: return True # Split on top-level "or" first, then on "and" inside each disjunct. # No parens supported. def eval_term(term: str) -> bool: m = COND_TERM_RE.fullmatch(term) if not m: return False name, value = m.group(1), m.group(2) return ctx.get(name) == value for or_part in re.split(r"\bor\b", cond): terms = re.split(r"\band\b", or_part) if all(eval_term(t) for t in terms if t.strip()): return True return False def find_hits(text: str, ctx: dict, cases: list[dict]) -> list[Hit]: hits: list[Hit] = [] for case in cases: if not case_applies(case, ctx): continue scoped_text = strip_code(text) if case.get("scope_outside_code") else text patterns = case.get("patterns") or ([case["pattern"]] if case.get("pattern") else []) for pat in patterns: try: m = re.search(pat, scoped_text) except re.error: continue if m: hits.append( Hit( case=case["name"], severity=case.get("severity", "low"), match=m.group(0)[:80], why=case.get("why", ""), fix_hint=case.get("fix_hint", ""), ) ) break # one match per case is enough return hits def log_line(line: str) -> None: LOG_PATH.parent.mkdir(parents=True, exist_ok=True) with LOG_PATH.open("a") as f: f.write(line.rstrip() + "\n") def report(hits: list[Hit], severity_min: str) -> int: rank = {"low": 0, "medium": 1, "high": 2} min_rank = rank[severity_min] blocking = [h for h in hits if rank[h.severity] >= rank["high"]] warning = [h for h in hits if h.severity == "medium"] low = [h for h in hits if h.severity == "low"] for h in low: log_line(f"[low] {h.case}: match={h.match!r}") for h in warning: if min_rank <= rank["medium"]: print(f"[warn] {h.case}: {h.why}", file=sys.stderr) print(f" match: {h.match!r}", file=sys.stderr) print(f" fix: {h.fix_hint}", file=sys.stderr) if blocking: print("Pre-Output-Critic blocked the call. Fix the following before re-trying:", file=sys.stderr) for h in blocking: print(f" [{h.severity}] {h.case}: {h.why}", file=sys.stderr) print(f" match: {h.match!r}", file=sys.stderr) print(f" fix: {h.fix_hint}", file=sys.stderr) return 2 return 0 def extract_text(payload: dict, tool: str) -> tuple[str, dict]: """Pull text + format out of the Claude Code hook payload (tool_input).""" ti = payload.get("tool_input") or {} if tool == "telegram_reply": return ti.get("text", ""), { "tool": "telegram_reply", "format": ti.get("format", "text"), "language": "de", } if tool == "edit": return ti.get("new_string", ""), {"tool": "edit", "format": "text", "language": "de"} if tool == "write": return ti.get("content", ""), {"tool": "write", "format": "text", "language": "de"} return "", {"tool": tool, "format": "text", "language": "de"} SELFTEST_CASES = [ # (text, format, expect_block) ("Hallo Kais.", "markdownv2", True), # unescaped dot ("Hallo Kais\\.", "markdownv2", False), # escaped dot ("Test \\~15k Tokens", "markdownv2", False), # escaped tilde ("Test ~15k Tokens", "markdownv2", False), # tilde now medium -> warn only ("Strikethrough ~weg~ jetzt", "markdownv2", False), # tilde medium -> warn only ("ok geh", "markdownv2", False), # nothing reserved ("Manu hat heute", "text", True), # forbidden brand ("Em-Dash test — hier", "text", False), # medium severity -> not block ("Lass uns explorieren", "text", False), # low severity -> not block ("`Wert mit . inside backticks`", "markdownv2", False), # in code -> ok ("*Bold-Header*", "markdownv2", False), # bold * legitim ("_italic_ text", "markdownv2", False), # italic _ legitim ("Drei #Optionen", "markdownv2", True), # unescaped hash ("Commit + Push", "markdownv2", True), # unescaped plus ("Drei \\#Optionen", "markdownv2", False), # escaped hash ok ("Commit \\+ Push", "markdownv2", False), # escaped plus ok ] def selftest() -> int: cases = load_cases() failures = 0 for text, fmt, expect_block in SELFTEST_CASES: ctx = {"tool": "telegram_reply", "format": fmt, "language": "de"} hits = find_hits(text, ctx, cases) any_high = any(h.severity == "high" for h in hits) ok = (any_high == expect_block) status = "OK" if ok else "FAIL" if not ok: failures += 1 print(f"[{status}] expect_block={expect_block} got_high={any_high} text={text!r}") for h in hits: print(f" -> [{h.severity}] {h.case} match={h.match!r}") print(f"\n{len(SELFTEST_CASES) - failures}/{len(SELFTEST_CASES)} passed") return 1 if failures else 0 def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--tool", default="telegram_reply", help="Tool name (telegram_reply, edit, write)") parser.add_argument("--text", default=None, help="Direct text (skips stdin)") parser.add_argument("--format", default="markdownv2", help="Output format") parser.add_argument("--severity-min", default="medium", choices=["low", "medium", "high"]) parser.add_argument("--input", default=None, help="JSON payload file") parser.add_argument("--selftest", action="store_true") args = parser.parse_args() if args.selftest: return selftest() cases = load_cases() if args.text is not None: text = args.text ctx = {"tool": args.tool, "format": args.format, "language": "de"} else: if args.input: payload = json.loads(Path(args.input).read_text()) else: raw = sys.stdin.read().strip() if not raw: return 0 try: payload = json.loads(raw) except json.JSONDecodeError: return 0 text, ctx = extract_text(payload, args.tool) if not text: return 0 hits = find_hits(text, ctx, cases) return report(hits, args.severity_min) if __name__ == "__main__": sys.exit(main())