#!/usr/bin/env python3 """ Semantic Pre-Validator — Decision linter based on the knowledge_base. Queries the vault before I/O actions or API calls. Usage: python3 pre_validator.py "POST to Qdrant upsert" # should find pitfalls python3 pre_validator.py --json "use Claude from Anthropic" # JSON output python3 pre_validator.py --domain qdrant,api "modify docker-compose" # restrict search Exit codes: 0 = pass/warn (action may proceed) 1 = blocked (action must be aborted) Fail-open: if OpenRouter or Qdrant is offline, allows execution with a warning. """ import os import sys import json import re import requests from typing import List, Dict, Optional from pathlib import Path # ─── Config ──────────────────────────────────────────────────────────────── OPENROUTER_KEY = os.environ.get("OPENROUTER_API_KEY") QDRANT_URL = os.environ.get("QDRANT_URL", "http://localhost:6333") COLLECTION = os.environ.get("QDRANT_COLLECTION", "knowledge_base") if not OPENROUTER_KEY: _env_path = os.environ.get("ENV_PATH", "") if _env_path: _env = Path(_env_path) else: _env = Path.home() / ".env" if _env.exists(): for ln in _env.read_text().splitlines(): if ln.startswith("OPENROUTER_API_KEY="): OPENROUTER_KEY = ln.split("=", 1)[1].strip().strip('"') break EMBEDDING_MODEL = "qwen/qwen3-embedding-8b" TOP_K = 5 SCORE_THRESHOLD = 0.60 WARN_THRESHOLD = 0.75 # pure wiki docs need a higher score for a warning BLOCK_SEVERITIES = {"critical", "high"} WARN_SEVERITIES = {"medium"} RULE_SOURCES = {"reflection", "decision", "rule", "pitfall", "insight"} REQUEST_TIMEOUT = 10 # ─── Restriction Patterns in wiki text ───────────────────────────────────── RESTRICTION_KEYWORDS = [ "do not use", "must not", "cannot", "never use", "avoid", "forbidden", "not recommended", "anti-pattern", "common mistake", "caution", "warning", "important:", "⚠️", "🚫", "must use", "must always", "requires", "mandatory", "keep", "do not change", "do not modify", "freeze", ] def contains_restriction(text: str) -> bool: """Check whether text contains restriction/decision patterns.""" if not text: return False text_lower = text.lower() return any(kw in text_lower for kw in RESTRICTION_KEYWORDS) # ─── Domain Tag Inference ───────────────────────────────────────────────── DOMAIN_PATTERNS = { "docker" : ["docker", "compose", "container", "image", "dockerfile"], "qdrant" : ["qdrant", "collection", "points", "upsert", "vector", "vectors", "embedding"], "redis" : ["redis", "arq", "queue", "job", "worker", "broker"], "openrouter" : ["openrouter", "embedding", "api_key", "openai", "api_base", "model"], "hermes" : ["hermes", "config.yaml", "skill", "cron", "gateway", "cli"], "wiki" : ["wiki", "raw/", "ingest", "vault", "obsidian", "knowledge_base"], "webui" : ["webui", "open-webui", "frontend", "chat", "rag"], "infra" : ["deploy", "server", "systemd", "service", "port", "host"], "security" : ["password", "secret", "token", "auth", "permission", "sudo"], "maas" : ["maas", "memory", "cognitive", "agent"], } def infer_domain_tags(description: str) -> List[str]: d = description.lower() found = set() for domain, pats in DOMAIN_PATTERNS.items(): if any(p in d for p in pats): found.add(domain) return sorted(found) # ─── Core ─────────────────────────────────────────────────────────────────── def embed_text(text: str) -> Optional[List[float]]: if not OPENROUTER_KEY: return None try: r = requests.post( "https://openrouter.ai/api/v1/embeddings", headers={ "Authorization": f"Bearer {OPENROUTER_KEY}", "Content-Type": "application/json" }, json={"model": EMBEDDING_MODEL, "input": text[:8000]}, timeout=REQUEST_TIMEOUT ) r.raise_for_status() return r.json()["data"][0]["embedding"] except Exception as e: print(f"[PV-ERROR] Embedding failed: {e}", file=sys.stderr) return None def search_knowledge_base(vector: List[float], domain_tags: List[str]) -> List[Dict]: try: r = requests.post( f"{QDRANT_URL}/collections/{COLLECTION}/points/search", headers={"Content-Type": "application/json"}, json={"vector": vector, "limit": TOP_K * 3, "with_payload": True}, timeout=REQUEST_TIMEOUT ) r.raise_for_status() hits = [] for item in r.json().get("result", []): pld = item.get("payload", {}) src = str(pld.get("source", "")).lower() sev = str(pld.get("severity", pld.get("decision_severity", "low"))).lower() tags = [str(t).lower() for t in pld.get("tags", [])] score = item.get("score", 0) # If domain filters requested, require overlap if domain_tags: dom_low = [d.lower() for d in domain_tags] if not set(dom_low) & set(tags): continue hits.append({ "id" : str(item.get("id", "")), "score" : score, "title" : pld.get("title", "Untitled"), "text" : (pld.get("text", "") or "")[:400], "source" : src, "severity": sev, "tags" : tags, }) hits.sort(key=lambda x: x["score"], reverse=True) return hits[:TOP_K] except Exception as e: print(f"[PV-ERROR] Qdrant search failed: {e}", file=sys.stderr) return [] def is_rule_hit(hit: Dict) -> bool: """Return True if the hit contains an explicit rule (reflection/decision/rule/insight/pitfall).""" return any(s in hit["source"] for s in RULE_SOURCES) def classify_hit(hit: Dict, action_desc: str) -> str: """ Return hit category: 'block', 'warn', 'info', or 'none'. Considers both source=reflection/decision/rule and restriction patterns embedded in wiki document text. """ sev = hit.get("severity", "low") is_rule = is_rule_hit(hit) or contains_restriction(hit.get("text", "")) score = hit.get("score", 0) # If text contains restriction, give it more weight restriction_bonus = 0.08 if contains_restriction(hit.get("text", "")) else 0 effective_score = score + restriction_bonus # Proximity: if the action term (e.g. "POST") appears near a keyword in the text action_terms = set(action_desc.lower().split()) text_lower = (hit.get("text", "") or "").lower() text_words = set(text_lower.split()) proximity_match = len(action_terms & text_words) > 0 # If restriction + proximity → elevate severity has_restriction = contains_restriction(hit.get("text", "")) and proximity_match if is_rule or has_restriction: if sev in BLOCK_SEVERITIES or (has_restriction and effective_score >= 0.65): return "block" elif sev in WARN_SEVERITIES or (has_restriction and effective_score >= SCORE_THRESHOLD): return "warn" # For normal wiki documents, only warn if score is very high if effective_score >= WARN_THRESHOLD: return "warn" if effective_score >= SCORE_THRESHOLD: return "info" return "none" def validate_action(action_description: str, domain_tags: Optional[List[str]] = None) -> Dict: try: dom = domain_tags or infer_domain_tags(action_description) vec = embed_text(action_description) if vec is None: return {"status": "pass", "blocked": False, "message": "⚠️ Validator offline. Proceeding with caution.", "action": action_description} hits = search_knowledge_base(vec, dom) blockers = [] warnings = [] infos = [] for h in hits: cat = classify_hit(h, action_description) if cat == "block": blockers.append(h) elif cat == "warn": warnings.append(h) elif cat == "info": infos.append(h) if blockers: lines = [f"🚫 ACTION BLOCKED — {len(blockers)} critical rule(s) in the vault:"] for b in blockers: lines.append(f" • [{b['severity'].upper()}] {b['title']} (score: {b['score']:.2f})") lines.append(f" {b['text'][:200]}...") lines.append("") lines.append("Override? Type 'force' (not recommended).") return { "status": "blocked", "blocked": True, "blockers": blockers, "warnings": warnings, "message": "\n".join(lines), "action": action_description, "domain": dom, } if warnings: lines = [f"⚠️ {len(warnings)} warning(s) found in the vault:"] for w in warnings: lines.append(f" • [{w['severity'].upper()}] {w['title']} (score: {w['score']:.2f})") lines.append(f" {w['text'][:200]}...") return { "status": "warn", "blocked": False, "warnings": warnings, "infos": infos, "message": "\n".join(lines), "action": action_description, "domain": dom, } if infos: return { "status": "info", "blocked": False, "infos": infos, "message": f"ℹ️ {len(infos)} relevant document(s), none critical.", "action": action_description, "domain": dom, } return { "status": "pass", "blocked": False, "message": "No relevant insights found. Execution authorized.", "action": action_description, "domain": dom, } except Exception as e: return { "status": "pass", "blocked": False, "message": f"Validator failed ({e}). Proceeding with caution.", "action": action_description, "domain": [], } # ─── Main ─────────────────────────────────────────────────────────────────── def main(): import argparse p = argparse.ArgumentParser(description="Semantic Pre-Validator") p.add_argument("action", nargs="?", help="Action description") p.add_argument("--domain", help="Comma-separated domain tags") p.add_argument("--json", action="store_true", help="JSON output") p.add_argument("--silent", action="store_true", help="Silent — exit code only") p.add_argument("--force-block", action="store_true", help="Force block (testing)") args = p.parse_args() action = args.action or sys.stdin.read().strip() or "POST to Qdrant upsert endpoint" dom = [x.strip() for x in args.domain.split(",")] if args.domain else None res = validate_action(action, dom) if args.force_block: res["blocked"] = True res["status"] = "blocked" if args.json: print(json.dumps(res, indent=2, ensure_ascii=False, default=str)) elif not args.silent: print(res["message"]) if res["blocked"]: print("\n(Use --force-block to test validator bypass)") sys.exit(1 if res["blocked"] else 0) if __name__ == "__main__": main()