feat(orchestrator): Add intent parser fast path for deterministic routing

Integrate the intent parser as a fast path at the top of /orchestrator/chat
so straightforward operations (deploy FreeIPA, list VMs, trigger DAGs) are
handled deterministically without requiring the full PydanticAI agent
pipeline or RAG documents.

Key changes:
- Add _try_intent_fast_path() in main.py that classifies and executes
  high-confidence requests via the intent parser, skipping Manager/Developer
  agents entirely
- Add dynamic DAG registry (dag_registry.py) that scans airflow/dags/ at
  startup to build service-keyword-to-DAG mappings automatically — no code
  changes needed when new DAGs are added
- Add HTTP API fallback in DAG handler for triggering DAGs when running
  outside the Airflow container (Airflow REST API at localhost:8888)
- Unwrap @mcp.tool() FunctionTool wrappers in all 6 handler modules so
  backend functions are callable from the intent parser context
- Mount intent_parser and mcp_server_fastmcp.py into the AI assistant
  container via deploy-qubinode.sh volume mounts
- Create localhost_ssh Airflow connection during deployment for SSHOperator
  DAGs
- Tune classifier confidence normalization (max_possible 10→5) so real
  matches reach actionable thresholds

Co-Authored-By: Claude Opus 4.5 <[email protected]>

Author: tosin2013

ai-assistant/src/main.py

@@ -560,6 +560,80 @@ class OrchestratorResponse(BaseModel):
     code_changes: List[dict] = []
 
 
+async def _try_intent_fast_path(message: str) -> Optional[OrchestratorResponse]:
+    """
+    Try to handle the request via the deterministic intent parser.
+    Returns OrchestratorResponse if handled, None to fall through to agents.
+    """
+    try:
+        import sys as _sys
+        if "/app" not in _sys.path:
+            _sys.path.insert(0, "/app")
+        from intent_parser import IntentParser, IntentCategory
+
+        parser = IntentParser()
+        parsed = parser.classify(message)
+
+        # Only fast-path for high-confidence, actionable intents
+        FAST_PATH_CATEGORIES = {
+            IntentCategory.DAG_TRIGGER,
+            IntentCategory.DAG_LIST,
+            IntentCategory.DAG_INFO,
+            IntentCategory.VM_LIST,
+            IntentCategory.VM_INFO,
+            IntentCategory.VM_CREATE,
+            IntentCategory.VM_DELETE,
+            IntentCategory.VM_PREFLIGHT,
+            IntentCategory.SYSTEM_STATUS,
+            IntentCategory.SYSTEM_INFO,
+            IntentCategory.RAG_QUERY,
+            IntentCategory.RAG_STATS,
+        }
+
+        if parsed.confidence < 0.5 or parsed.category not in FAST_PATH_CATEGORIES:
+            return None  # Fall through to full orchestrator
+
+        # Execute via intent parser
+        result = await parser.process(message)
+
+        return OrchestratorResponse(
+            session_id=str(uuid.uuid4()),
+            model_used="intent-parser (deterministic)",
+            plan={
+                "session_id": "intent-fast-path",
+                "user_intent": message,
+                "planned_tasks": [f"Execute {parsed.category.value}: {message}"],
+                "estimated_confidence": parsed.confidence,
+                "requires_external_docs": False,
+                "required_providers": [],
+                "escalation_triggers": [],
+            },
+            response_text=result.output if result.success else f"Error: {result.error}",
+            confidence=parsed.confidence,
+            escalation_needed=not result.success,
+            escalation_reason=result.error if not result.success else None,
+            planned_tasks=[f"Execute {parsed.category.value}: {message}"],
+            required_providers=[],
+            timestamp=time.time(),
+            execution_performed=True,
+            execution_success=result.success,
+            tasks_executed=1,
+            tasks_succeeded=1 if result.success else 0,
+            tasks_failed=0 if result.success else 1,
+            tasks_escalated=0,
+            execution_confidence=parsed.confidence,
+            execution_log_file=None,
+            execution_summary=result.output,
+            code_changes=[],
+        )
+    except ImportError:
+        logger.debug("Intent parser not available, falling through to agents")
+        return None
+    except Exception as e:
+        logger.warning(f"Intent parser fast path failed: {e}, falling through to agents")
+        return None
+
+
 @app.post("/orchestrator/chat", response_model=OrchestratorResponse)
 async def orchestrator_chat(request: OrchestratorRequest):
     """
@@ -610,6 +684,11 @@ async def orchestrator_chat(request: OrchestratorRequest):
     if not any([has_gemini, has_openrouter, has_anthropic, has_openai, has_ollama]):
         raise HTTPException(status_code=503, detail="No LLM API key configured. Set one of: GEMINI_API_KEY, OPENROUTER_API_KEY, ANTHROPIC_API_KEY, OPENAI_API_KEY, or OLLAMA_BASE_URL")
 
+    # Fast path: try intent parser for deterministic routing
+    intent_result = await _try_intent_fast_path(request.message)
+    if intent_result is not None:
+        return intent_result
+
     try:
         # Create or use existing session
         session_id = request.session_id or str(uuid.uuid4())

airflow/dags/freeipa_deployment.py

@@ -188,7 +188,7 @@ def decide_action(**context):
         -P numcpus=2 \
         -P disks=[50] \
         -P nets=[default] \
-        --wait || {{
+        --wait < /dev/null || {{
         echo "[ERROR] Failed to create VM"
         exit 1
     }}

intent_parser/classifier.py

@@ -149,19 +149,38 @@ def _build_rules() -> dict:
         "boost": 1,
     }
 
+    # Build DAG_TRIGGER keywords dynamically from discovered DAGs
+    from .dag_registry import get_deploy_keywords
+    _deploy_kws = get_deploy_keywords()
+
+    dag_trigger_keywords = [
+        _kw("trigger", "dag"),
+        _kw("run", "dag"),
+        _kw("execute", "dag"),
+        _kw("start", "dag"),
+        _kw("trigger", "workflow"),
+        _kw("run", "workflow"),
+    ]
+    # Add "deploy <service>" keywords for every discovered DAG service
+    for svc_kw in _deploy_kws:
+        dag_trigger_keywords.append(_kw("deploy", svc_kw))
+
+    # Build pattern alternation from service keywords for regex matching
+    # Only use keywords >= 4 chars to avoid false positives
+    _svc_names = "|".join(re.escape(k) for k in _deploy_kws if len(k) >= 4)
+
+    dag_trigger_patterns = [
+        re.compile(r"\b(?:trigger|run|execute|start)\s+(?:the\s+)?(?:dag|workflow)\s+\w+", re.I),
+        re.compile(r"\b(?:trigger|run|execute|start)\s+(?:the\s+)?\w+\s+(?:dag|workflow)\b", re.I),
+    ]
+    if _svc_names:
+        dag_trigger_patterns.append(
+            re.compile(rf"\bdeploy\s+(?:a\s+)?(?:new\s+)?(?:{_svc_names})\b", re.I)
+        )
+
     rules[IntentCategory.DAG_TRIGGER] = {
-        "keywords": [
-            _kw("trigger", "dag"),
-            _kw("run", "dag"),
-            _kw("execute", "dag"),
-            _kw("start", "dag"),
-            _kw("trigger", "workflow"),
-            _kw("run", "workflow"),
-        ],
-        "patterns": [
-            re.compile(r"\b(?:trigger|run|execute|start)\s+(?:the\s+)?(?:dag|workflow)\s+\w+", re.I),
-            re.compile(r"\b(?:trigger|run|execute|start)\s+(?:the\s+)?\w+\s+(?:dag|workflow)\b", re.I),
-        ],
+        "keywords": dag_trigger_keywords,
+        "patterns": dag_trigger_patterns,
         "boost": 0,
     }
 
@@ -409,7 +428,8 @@ def classify(text: str) -> ParsedIntent:
         )
 
     # Normalize confidence: map score to 0-1 range
-    max_possible = 10.0
+    # A strong match typically scores 3-5 points (keyword=1 + pattern=2 + boost)
+    max_possible = 5.0
     confidence = min(best_score / max_possible, 1.0)
 
     # Boost confidence if clear winner (big gap to second)

intent_parser/dag_registry.py

@@ -0,0 +1,202 @@
+"""
+Dynamic DAG registry for intent classification and entity extraction.
+
+Scans airflow/dags/ at import time and builds:
+1. Service keywords for the classifier (deploy <service> -> DAG_TRIGGER)
+2. Service-to-DAG-ID mapping for the entity extractor
+
+This means adding a new DAG file with proper tags/description automatically
+makes it discoverable by the intent parser — no code changes needed.
+"""
+
+import logging
+import os
+import re
+from pathlib import Path
+from typing import Dict, List, Set, Tuple
+
+logger = logging.getLogger("intent-parser.dag-registry")
+
+# Words that are too generic to use as service keywords
+_GENERIC_WORDS = frozenset({
+    "qubinode", "infrastructure", "deployment", "deploy", "kcli-pipelines",
+    "ocp4-disconnected-helper", "disconnected", "utility", "workflow",
+    "master", "ci", "openlineage", "adr-0054", "adr-0055", "adr-0049",
+    "enterprise", "kcli", "pipelines",
+})
+
+# Minimum tag/keyword length to avoid noise
+_MIN_KEYWORD_LEN = 2
+
+
+def _find_dags_path() -> Path:
+    """Locate the airflow/dags directory."""
+    # Try multiple known locations
+    candidates = [
+        os.getenv("AIRFLOW_DAGS_PATH", ""),
+        "/app/airflow/dags",  # Inside container
+        str(Path(__file__).parent.parent / "airflow" / "dags"),  # Relative to repo root
+        "/opt/qubinode_navigator/airflow/dags",
+    ]
+    for c in candidates:
+        if c and Path(c).is_dir():
+            return Path(c)
+    return Path("/app/airflow/dags")  # Default, may not exist
+
+
+def _parse_dag_metadata(file_path: Path) -> dict:
+    """Parse a DAG file and extract dag_id, tags, and description."""
+    try:
+        content = file_path.read_text()
+    except Exception:
+        return {}
+
+    dag_id_m = re.search(r'DAG\s*\(\s*["\']([^"\']+)["\']', content)
+    if not dag_id_m:
+        dag_id_m = re.search(r'dag_id\s*=\s*["\']([^"\']+)["\']', content)
+    if not dag_id_m:
+        return {}
+
+    dag_id = dag_id_m.group(1)
+
+    tags_m = re.search(r"tags\s*=\s*\[([^\]]+)\]", content)
+    tags = []
+    if tags_m:
+        tags = re.findall(r'["\']([^"\']+)["\']', tags_m.group(1))
+
+    desc_m = re.search(r'description\s*=\s*["\']([^"\']+)["\']', content)
+    description = desc_m.group(1) if desc_m else ""
+
+    return {
+        "dag_id": dag_id,
+        "tags": tags,
+        "description": description,
+    }
+
+
+def scan_dags() -> List[dict]:
+    """Scan all DAG files and return metadata list."""
+    dags_path = _find_dags_path()
+    if not dags_path.exists():
+        logger.warning(f"DAGs path not found: {dags_path}")
+        return []
+
+    skip_files = {"dag_factory.py", "dag_helpers.py", "dag_loader.py", "dag_logging_mixin.py"}
+    results = []
+
+    for dag_file in sorted(dags_path.glob("*.py")):
+        if dag_file.name.startswith("_") or dag_file.name in skip_files:
+            continue
+        meta = _parse_dag_metadata(dag_file)
+        if meta:
+            results.append(meta)
+
+    logger.info(f"DAG registry scanned {len(results)} DAGs from {dags_path}")
+    return results
+
+
+def _extract_service_keywords(dag: dict) -> Set[str]:
+    """
+    Extract meaningful service keywords from a DAG's tags and description.
+
+    These are words specific enough to identify this DAG when a user says
+    "deploy <keyword>".
+    """
+    keywords = set()
+
+    # From tags
+    for tag in dag.get("tags", []):
+        tag_clean = tag.lower().strip()
+        if tag_clean not in _GENERIC_WORDS and len(tag_clean) >= _MIN_KEYWORD_LEN:
+            keywords.add(tag_clean)
+
+    # From dag_id: split on _ and take meaningful parts
+    dag_id = dag.get("dag_id", "")
+    for part in dag_id.lower().split("_"):
+        if part not in _GENERIC_WORDS and len(part) >= 3 and part not in {"the", "and", "for"}:
+            keywords.add(part)
+
+    return keywords
+
+
+def build_service_dag_map() -> Dict[str, str]:
+    """
+    Build a keyword -> dag_id mapping from all discovered DAGs.
+
+    Example output:
+        {
+            "freeipa": "freeipa_deployment",
+            "identity": "freeipa_deployment",
+            "harbor": "harbor_deployment",
+            "vyos": "vyos_router_deployment",
+            "router": "vyos_router_deployment",
+            ...
+        }
+
+    Priority: keywords from dag_id parts take precedence over tag-only keywords.
+    This ensures "freeipa" maps to freeipa_deployment (not dns_management which
+    merely has "freeipa" as a tag).
+    """
+    dags = scan_dags()
+    mapping: Dict[str, str] = {}
+
+    # Two passes: first dag_id-derived keywords (strong signal), then tag keywords
+    # Pass 1: keywords that appear in the dag_id itself
+    for dag in sorted(dags, key=lambda d: d["dag_id"]):
+        dag_id = dag["dag_id"]
+        dag_id_parts = set()
+        for part in dag_id.lower().split("_"):
+            if part not in _GENERIC_WORDS and len(part) >= 3 and part not in {"the", "and", "for"}:
+                dag_id_parts.add(part)
+
+        for kw in dag_id_parts:
+            if kw not in mapping:
+                mapping[kw] = dag_id
+
+    # Pass 2: keywords from tags (only if not already claimed by a dag_id match)
+    for dag in sorted(dags, key=lambda d: d["dag_id"]):
+        dag_id = dag["dag_id"]
+        for tag in dag.get("tags", []):
+            tag_clean = tag.lower().strip()
+            if tag_clean not in _GENERIC_WORDS and len(tag_clean) >= _MIN_KEYWORD_LEN:
+                if tag_clean not in mapping:
+                    mapping[tag_clean] = dag_id
+
+    logger.info(f"DAG registry built {len(mapping)} service keyword mappings")
+    return mapping
+
+
+def build_deploy_keywords() -> List[str]:
+    """
+    Build a list of service keywords that should trigger DAG_TRIGGER
+    when preceded by "deploy".
+
+    Used by the classifier to add dynamic keywords like:
+        _kw("deploy", "freeipa")
+        _kw("deploy", "harbor")
+        _kw("deploy", "vyos")
+    """
+    service_map = build_service_dag_map()
+    # Return unique keywords (some may map to the same DAG)
+    return sorted(service_map.keys())
+
+
+# Module-level cache (built once at import time)
+_service_dag_map: Dict[str, str] = {}
+_deploy_keywords: List[str] = []
+
+
+def get_service_dag_map() -> Dict[str, str]:
+    """Get the cached service-to-DAG mapping."""
+    global _service_dag_map
+    if not _service_dag_map:
+        _service_dag_map = build_service_dag_map()
+    return _service_dag_map
+
+
+def get_deploy_keywords() -> List[str]:
+    """Get the cached list of deploy keywords."""
+    global _deploy_keywords
+    if not _deploy_keywords:
+        _deploy_keywords = build_deploy_keywords()
+    return _deploy_keywords