#!/usr/bin/env python3 """ Urgency Router — classifies incoming tasks and recommends model paths. Usage: urgency-router.py Output: JSON with urgency classification and recommended model path. This is designed to be called by the agent before selecting a model, or by auto-model-switch.py for pre-flight routing decisions. """ import json import os import re import sys from datetime import datetime # Urgency signal keywords INSTANT_SIGNALS = [ "urgent", "asap", "fast", "quick", "now", "immediately", "right now", "right away", "emergency", "critical", "deadline", "eod", "end of day", "today", "need this", "hurry", "rush", "speed", "quickly", "promptly", ] TODAY_SIGNALS = [ "today", "tonight", "by tonight", "before bed", "by end of day", "eod", "deadline", "due today", "need it today", "finish today", ] THIS_WEEK_SIGNALS = [ "this week", "sometime this week", "whenever", "no rush", "take your time", "when you can", "low priority", "background", ] SAVINGS_SIGNALS = [ "no rush", "save cost", "cheap", "free", "local only", "whenever", "don't spend", "minimize cost", "budget", "free tier", "no api", "local first", ] # SQL/DAX/VBA detection patterns SQL_PATTERNS = [ r"\b(SELECT|INSERT|UPDATE|DELETE|CREATE|DROP|ALTER|FROM|WHERE|JOIN)\b", r"\b(SUM|COUNT|AVG|MAX|MIN|GROUP BY|ORDER BY|HAVING)\b", r"\b(CASE\s+WHEN|COALESCE|ISNULL|NULLIF)\b", r"^\s*(SELECT|INSERT|UPDATE|DELETE|CREATE|DROP)\b", ] DAX_PATTERNS = [ r"\b(CALCULATE|FILTER|SUMX|AVERAGEX|COUNTX|ALL|VALUES|RELATED)\b", r"\b(SUMMARIZE|ADDCOLUMNS|FILTER|DISTINCT|INTERSECT)\b", r"\b([A-Z][A-Z_]+\s*=\s*[A-Z][A-Z_]*\s*\()", ] VBA_PATTERNS = [ r"\b(Sub|Function|Dim|As|Set|With|For\s+Each|Next|If\s+Then|End\s+Sub|End\s+Function)\b", r"\b(Range|Cells|Worksheet|Workbook|Sheets|Application)\b", r"\b(MsgBox|InputBox|Debug\.Print)\b", ] # Local model recommendations by urgency LOCAL_MODELS = { "instant": "qwen/qwen3-coder", # OpenRouter fallback for speed "today": "Qwen3-Coder-30B-A3B-Instruct-Q4_K_M.gguf", "this_week": "Qwen3.6-35B-A3B-UD-Q4_K_M.gguf", "savings": "Qwen2.5-Coder-7B-Instruct-Q5_K_M.gguf", } # Premium model recommendations PREMIUM_MODELS = { "instant": "chatgpt", # ChatGPT Auth "today": "qwen/qwen3-coder", # OpenRouter } LOG_FILE = os.path.expanduser("~/.hermes/logs/urgency-router.log") def detect_sql_dax_vba(text): """Detect if the message contains SQL, DAX, or VBA code.""" text_lower = text.lower() sql_score = sum(1 for p in SQL_PATTERNS if re.search(p, text, re.IGNORECASE)) dax_score = sum(1 for p in DAX_PATTERNS if re.search(p, text, re.IGNORECASE)) vba_score = sum(1 for p in VBA_PATTERNS if re.search(p, text, re.IGNORECASE)) types = [] if sql_score >= 2: types.append("SQL") if dax_score >= 2: types.append("DAX") if vba_score >= 2: types.append("VBA") return types def classify_urgency(text): """Classify urgency level from user message.""" text_lower = text.lower() # Check savings signals FIRST (most specific, should override) for signal in SAVINGS_SIGNALS: if signal in text_lower: return "savings" # Check this week signals (before today, more specific) for signal in THIS_WEEK_SIGNALS: if signal in text_lower: return "this_week" # Check instant signals (highest priority after savings) for signal in INSTANT_SIGNALS: if signal in text_lower: return "instant" # Check today signals for signal in TODAY_SIGNALS: if signal in text_lower: return "today" # Default: today (balanced) return "today" def recommend_path(urgency, has_sql_dax_vba, has_chatgpt_auth): """Recommend model path based on urgency and task type.""" if has_sql_dax_vba: if urgency == "instant" and has_chatgpt_auth: return { "primary": "chatgpt", "fallback": "qwen/qwen3-coder", "reason": "SQL/DAX/VBA with urgency — ChatGPT Auth for speed" } elif urgency == "instant": return { "primary": "qwen/qwen3-coder", "fallback": "Qwen3-Coder-30B-A3B-Instruct-Q4_K_M.gguf", "reason": "SQL/DAX/VBA with urgency — OpenRouter coder first" } else: return { "primary": "Qwen3-Coder-30B-A3B-Instruct-Q4_K_M.gguf", "fallback": "qwen/qwen3-coder", "reason": "SQL/DAX/VBA without urgency — local coder first" } # General coding tasks if urgency == "instant": if has_chatgpt_auth: return { "primary": "chatgpt", "fallback": "qwen/qwen3-coder", "reason": "Urgent task — ChatGPT Auth for speed" } else: return { "primary": "qwen/qwen3-coder", "fallback": "Qwen3-Coder-30B-A3B-Instruct-Q4_K_M.gguf", "reason": "Urgent task — OpenRouter coder first" } if urgency == "today": return { "primary": "Qwen3-Coder-30B-A3B-Instruct-Q4_K_M.gguf", "fallback": "Qwen3.6-35B-A3B-UD-Q4_K_M.gguf", "reason": "Today deadline — local coder first, heavy lifter as backup" } if urgency == "this_week": return { "primary": "Qwen3.6-35B-A3B-UD-Q4_K_M.gguf", "fallback": "Qwen3-Coder-30B-A3B-Instruct-Q4_K_M.gguf", "reason": "This week — heavy lifter for complex work" } if urgency == "savings": return { "primary": "Qwen2.5-Coder-7B-Instruct-Q5_K_M.gguf", "fallback": "Qwen3-8B-Q5_K_M.gguf", "reason": "Cost-conscious — fast local models only" } # Default return { "primary": "Qwen3-Coder-30B-A3B-Instruct-Q4_K_M.gguf", "fallback": "Qwen3.6-35B-A3B-UD-Q4_K_M.gguf", "reason": "Default — balanced local coding" } def main(): if len(sys.argv) < 2: print(json.dumps({ "error": "Usage: urgency-router.py " }), file=sys.stderr) sys.exit(1) message = " ".join(sys.argv[1:]) urgency = classify_urgency(message) sql_dax_vba = detect_sql_dax_vba(message) # Check if ChatGPT Auth is available has_chatgpt = os.path.exists(os.path.expanduser("~/.claude/credentials")) path = recommend_path(urgency, len(sql_dax_vba) > 0, has_chatgpt) result = { "urgency": urgency, "task_types": sql_dax_vba, "chatgpt_available": has_chatgpt, "recommended_path": path, "timestamp": datetime.now().isoformat(), } # Log the decision os.makedirs(os.path.dirname(LOG_FILE), exist_ok=True) with open(LOG_FILE, "a") as f: f.write(f"[{result['timestamp']}] urgency={urgency} types={sql_dax_vba} path={path['primary']}\n") print(json.dumps(result, indent=2)) if __name__ == "__main__": main()