#!/usr/bin/env python3 """KAR-Reviewer: bewertet KAR-Issues per LLM auf 4 Achsen. KAR-240 Reviewer-Subagent-Loop. Klassifiziert jeden Issue in: - verdict: implement | brain_only | spike | reject - aria_lever: high | medium | low (wie sehr verbessert es Aria selbst) - effort: XS | S | M | L - quick_win: true|false (klein + hoher Hebel + klare DoD) Output: 02-Wissen/kar-review-{date}.md mit Top-10 Quick-Wins. Run: python3 aria-kar-reviewer.py --filter "video|adopt" --limit 100 """ from __future__ import annotations import sys as _sys _sys.path.insert(0, "/root/aria/lib") from aria_logging import get_logger as _get_logger _log = _get_logger("aria-kar-reviewer") import argparse, json, os, re, sys, time from datetime import datetime, timezone from pathlib import Path import requests API_KEY = os.environ.get("LINEAR_API_KEY") ANTHROPIC_API_KEY = os.environ.get("ANTHROPIC_API_KEY") URL = "https://api.linear.app/graphql" OUT_DIR = Path("/root/aria/brain/02-Wissen") _ARIA_CONFIG_PATH = Path("/root/aria/brain/youtube/.config.yaml") # ─── Statischer System-Prompt (cache_control-Kandidat) ─────────────────────── _KAR_REVIEWER_SYSTEM_PROMPT = ( "Du bist ein Aria-KAR-Reviewer. Klassifiziere Linear-Issues auf 4 Achsen. " "Antworte ausschliesslich als JSON mit den Feldern: " "verdict, aria_lever, effort, quick_win, reasoning." ) def _get_kar_reviewer_model() -> str: """Liest Modell aus ARIA_KAR_REVIEWER_MODEL env, dann .config.yaml models.kar_reviewer_model, dann fällt zurück auf claude-haiku-4-5-20251001.""" env_val = os.environ.get("ARIA_KAR_REVIEWER_MODEL") if env_val: return env_val try: import yaml as _yaml # type: ignore cfg = _yaml.safe_load(_ARIA_CONFIG_PATH.read_text(encoding="utf-8")) model = cfg.get("models", {}).get("kar_reviewer_model") if model: return model except Exception: pass return "claude-haiku-4-5-20251001" def fetch_open_kars(filter_regex: str | None = None, limit: int = 200) -> list[dict]: r = requests.post(URL, json={"query": "query { teams { nodes { id key } } }"}, headers={"Authorization": API_KEY}).json() team_id = next(t["id"] for t in r["data"]["teams"]["nodes"] if t["key"] == "KAR") q = """query($tid: String!) { team(id: $tid) { issues(first: 250) { nodes { identifier title description priority state { name type } createdAt } }}}""" r = requests.post(URL, json={"query": q, "variables": {"tid": team_id}}, headers={"Authorization": API_KEY}).json() issues = r["data"]["team"]["issues"]["nodes"] issues = [i for i in issues if i["state"]["type"] not in ("completed", "canceled")] if filter_regex: rx = re.compile(filter_regex, re.IGNORECASE) issues = [i for i in issues if rx.search(i["title"]) or (i["description"] and rx.search(i["description"]))] return issues[:limit] def review_with_claude(issue: dict, model: str) -> dict: """Send issue to Claude for fast classification. model is resolved once by caller.""" if not ANTHROPIC_API_KEY: return {"verdict": "implement", "aria_lever": "medium", "effort": "M", "quick_win": False, "reasoning": "no_api_key"} user_prompt = f"""Klassifiziere dieses Linear-KAR-Issue auf 4 Achsen. Hinweise: - "implement" = direkt umsetzbar, klare DoD, gibt Aria/Kadi/Tools-Verbesserung - "brain_only" = Wissen aufschreiben/lesen reicht, keine Code-Aenderung notwendig - "spike" = brauche erst 2-4h Recherche/Architektur-Decision bevor Code - "reject" = nicht relevant, duplikat, oder schon implizit done - aria_lever = wie viel verbessert es Aria SELBST (nicht nur Knowledge) - effort: XS=<30min S=30-90min M=1.5-4h L=>4h - quick_win = effort in (XS,S) UND aria_lever in (high,medium) UND verdict=implement Issue ID: {issue['identifier']} Title: {issue['title']} Priority: P{issue['priority']} Description: {(issue.get('description') or '')[:1500]} Antworte ausschliesslich als JSON: {{ "verdict": "implement" | "brain_only" | "spike" | "reject", "aria_lever": "high" | "medium" | "low", "effort": "XS" | "S" | "M" | "L", "quick_win": true | false, "reasoning": "ein satz auf deutsch" }}""" try: r = requests.post( "https://api.anthropic.com/v1/messages", headers={ "x-api-key": ANTHROPIC_API_KEY, "anthropic-version": "2023-06-01", "anthropic-beta": "prompt-caching-2024-07-31", "content-type": "application/json", }, json={ "model": model, "max_tokens": 200, "system": [ { "type": "text", "text": _KAR_REVIEWER_SYSTEM_PROMPT, "cache_control": {"type": "ephemeral"}, } ], "messages": [{"role": "user", "content": user_prompt}], }, timeout=30, ) data = r.json() text = data["content"][0]["text"] # Extract JSON m = re.search(r"\{[^{}]+\}", text, re.DOTALL) if m: return json.loads(m.group(0)) return {"verdict": "implement", "aria_lever": "medium", "effort": "M", "quick_win": False, "reasoning": "parse_error"} except Exception as e: _log.error("claude_call_failed", err=str(e), issue=issue["identifier"]) return {"verdict": "implement", "aria_lever": "medium", "effort": "M", "quick_win": False, "reasoning": f"err: {e}"} def main(): ap = argparse.ArgumentParser() ap.add_argument("--filter", help="Regex filter on title/description") ap.add_argument("--limit", type=int, default=200) ap.add_argument("--min-kar-id", type=int, default=386, help="Only review KAR-N where N >= this") args = ap.parse_args() model = _get_kar_reviewer_model() _log.event("reviewer_start", filter=args.filter, limit=args.limit, min_kar=args.min_kar_id, model=model) issues = fetch_open_kars(args.filter, args.limit) # Filter on min KAR-ID def kar_num(ident): m = re.match(r"KAR-(\d+)", ident) return int(m.group(1)) if m else 0 issues = [i for i in issues if kar_num(i["identifier"]) >= args.min_kar_id] print(f"[reviewer] {len(issues)} KARs to review (filtered min={args.min_kar_id})", flush=True) reviewed = [] for n, i in enumerate(issues, 1): if n % 10 == 0: print(f"[reviewer] {n}/{len(issues)}", flush=True) verdict = review_with_claude(i, model) verdict["identifier"] = i["identifier"] verdict["title"] = i["title"] verdict["priority"] = i["priority"] reviewed.append(verdict) time.sleep(0.5) # Aggregate from collections import Counter verdicts = Counter(r["verdict"] for r in reviewed) quick_wins = [r for r in reviewed if r["quick_win"]] high_lever = [r for r in reviewed if r["aria_lever"] == "high" and r["verdict"] == "implement"] # Write report date = datetime.now(timezone.utc).strftime("%Y-%m-%d") out_path = OUT_DIR / f"kar-review-{date}.md" lines = [ f"---", f"title: KAR-Review {date}", f"type: audit", f"tags: [kar-review, aria-improvement, prioritization]", f"date: {date}", f"status: aktiv", f"linear: KAR-240", f"---", "", f"# KAR-Reviewer Output — {date}", "", f"**Reviewed:** {len(reviewed)} KARs (min KAR-ID: {args.min_kar_id})", "", "## Verdict-Aggregat", "", ] for v, n in verdicts.most_common(): lines.append(f"- **{v}**: {n}") lines.extend([ "", "## Top-Quick-Wins (effort XS/S + aria_lever high/medium + verdict implement)", "", ]) for r in sorted(quick_wins, key=lambda x: (x["effort"], -ord("a" if x["aria_lever"] == "high" else "b"))): lines.append(f"- `{r['identifier']}` [{r['effort']}/{r['aria_lever']}] {r['title'][:65]}") lines.append(f" → {r['reasoning']}") lines.extend([ "", "## High-Aria-Lever (implement, lever=high, alle Effort)", "", ]) for r in sorted(high_lever, key=lambda x: x["effort"]): lines.append(f"- `{r['identifier']}` [{r['effort']}] {r['title'][:65]}") lines.append(f" → {r['reasoning']}") lines.extend([ "", "## Full Review", "", "| KAR | Verdict | Lever | Effort | QW | Title |", "|---|---|---|---|---|---|", ]) for r in reviewed: qw = "✓" if r["quick_win"] else "" lines.append(f"| `{r['identifier']}` | {r['verdict']} | {r['aria_lever']} | {r['effort']} | {qw} | {r['title'][:50]} |") out_path.write_text("\n".join(lines)) print(f"[reviewer] DONE — quick_wins={len(quick_wins)} high_lever={len(high_lever)} verdicts={dict(verdicts)}") print(f"[reviewer] Report: {out_path}") _log.event("reviewer_done", quick_wins=len(quick_wins), high_lever=len(high_lever), verdicts=dict(verdicts)) if __name__ == "__main__": _log.event("script_start") main()