#!/usr/bin/env python3 """ KAR-72: scan brain-notes for un-promoted "Nächste Schritte" sections. Output JSON to /tmp/aria-triage-YYYY-MM-DD.json (or path via --out). Aria reads this, runs Multi-Level-Challenge per candidate (see brain/05-Referenzen/promotion-challenge-protocol.md), proposes to Kais. Frontmatter convention (KAR-72): linear_issues: [KAR-XX, KAR-YY] # already-promoted issues promotion_status: promoted | killed | deferred | (absent = pending) defer_until: 2026-06-01 # if deferred kill_reason: "..." # if killed """ from __future__ import annotations import argparse import json import re import sys from datetime import datetime from pathlib import Path BRAIN_ROOT = Path("/root/aria/brain") SCAN_DIRS = ["00-Inbox", "02-Wissen", "03-Recherchen", "06-Daily"] SECTION_HEADING_PATTERNS = [ r"^#{2,4}\s*(?:\d+\.\s*)?(?:Nächste Schritte|Naechste Schritte)\b", r"^#{2,4}\s*(?:\d+\.\s*)?Next Steps\b", r"^#{2,4}\s*Action Items\b", r"^#{2,4}\s*Aufgabe(?:n)?\b", r"^#{2,4}\s*TODO\b", ] FRONTMATTER_RE = re.compile(r"^---\n(.*?)\n---\n", re.DOTALL) LINEAR_REF_RE = re.compile(r"\bKAR-\d+\b") def parse_frontmatter(text: str) -> dict: m = FRONTMATTER_RE.match(text) if not m: return {} fm_text = m.group(1) fm: dict = {} for line in fm_text.splitlines(): line = line.strip() if not line or ":" not in line: continue key, _, value = line.partition(":") fm[key.strip()] = value.strip() return fm def extract_next_steps(body: str) -> str | None: """Return content of first matching heading section, or None. Matches both top-level (## Nächste Schritte) and sub-headings (### 7. Nächste Schritte) — case-sensitive on the keyword. """ for pattern in SECTION_HEADING_PATTERNS: m = re.search(pattern, body, re.MULTILINE) if not m: continue start = m.start() # Section ends at next heading of same or higher level heading_level = len(m.group(0).split()[0]) # count # end_pattern = ( r"\n#{1," + str(heading_level) + r"}\s+\S" ) end_match = re.search(end_pattern, body[m.end():]) end = m.end() + end_match.start() if end_match else len(body) section = body[start:end].strip() if section: return section return None def is_aria_eigenstaendig(line: str) -> bool: """Heuristic: does this action need Kais-decision or can Aria do it alone?""" eigenstaendig_markers = [ "aria-eigenständig", "aria-eigenstaendig", "aria-internal", "still ausführen", "still ausfuehren", "kein kais-ok", "automatisch", ] line_lower = line.lower() return any(m in line_lower for m in eigenstaendig_markers) def scan() -> list[dict]: candidates: list[dict] = [] for sub in SCAN_DIRS: root = BRAIN_ROOT / sub if not root.exists(): continue for md_path in root.rglob("*.md"): try: text = md_path.read_text(encoding="utf-8") except Exception as e: print( f"warn: cannot read {md_path}: {e}", file=sys.stderr ) continue fm = parse_frontmatter(text) # Skip already-promoted if fm.get("promotion_status") == "promoted": continue if fm.get("promotion_status") == "killed": continue # Skip deferred unless date passed if fm.get("promotion_status") == "deferred": defer_until = fm.get("defer_until", "") if defer_until and defer_until > datetime.now().strftime( "%Y-%m-%d" ): continue body = FRONTMATTER_RE.sub("", text, count=1) section = extract_next_steps(body) if not section: continue # Extract individual action lines (bullets or numbered). # Track scope-header so bullets inherit "aria-eigenständig" vs "Kais-OK". actions: list[dict] = [] current_scope = "unspecified" for raw_line in section.splitlines(): stripped = raw_line.strip() if not stripped: continue # Detect a scope header (bold line containing eigenständig/Kais-OK) lower = stripped.lower() if "eigenständig" in lower or "eigenstaendig" in lower or "aria-internal" in lower: current_scope = "aria_alone" continue if "kais-ok" in lower or "kais-entscheidung" in lower or "mit kais" in lower: current_scope = "kais_decision" continue # Skip non-bullet lines (paragraphs, sub-section headers) if not re.match(r"^[-*•]|^\d+\.|^[a-zA-Z]\)", stripped): continue if len(stripped) < 5: continue actions.append( { "text": stripped, "scope": current_scope, "aria_alone": current_scope == "aria_alone" or is_aria_eigenstaendig(stripped), } ) if not actions: continue # Already-promoted KAR-numbers in frontmatter promoted_kars = LINEAR_REF_RE.findall( fm.get("linear_issues", "") ) candidates.append( { "note": str(md_path.relative_to(BRAIN_ROOT)), "title": fm.get("title", md_path.stem), "priority": fm.get("prioritaet", fm.get("priority", "")), "thema": fm.get("thema", ""), "date": fm.get("date", ""), "promoted_kars": promoted_kars, "actions": actions, "section_preview": section[:400], } ) return candidates def main(): parser = argparse.ArgumentParser( description="Scan brain-notes for unpromoted next-steps (KAR-72)" ) parser.add_argument( "--out", default=f"/tmp/aria-triage-{datetime.now().strftime('%Y-%m-%d')}.json", ) parser.add_argument("--print", action="store_true", help="Print candidates to stdout") args = parser.parse_args() candidates = scan() payload = { "generated_at": datetime.now().isoformat(), "candidate_count": len(candidates), "candidates": candidates, } out = Path(args.out) sys.path.insert(0, str(Path(__file__).resolve().parent)) from aria_atomic_write import atomic_write_json atomic_write_json(out, payload) print(f"{len(candidates)} candidates -> {out}") if args.print: for c in candidates: print(f"\n--- {c['note']} ({c['priority'] or 'no-prio'}) ---") print(f" title: {c['title']}") for a in c["actions"]: marker = "🤖" if a["aria_alone"] else "👤" print(f" {marker} {a['text'][:120]}") if __name__ == "__main__": main()