feat(validate): add --queries flag for fleet-wide CQL parsing (closes #11)

src/talonctl/commands/validate.py

@@ -9,18 +9,36 @@
     parse_filters,
     init_orchestrator,
 )
+from talonctl.utils.auth import load_credentials
 
 
 @click.command()
 @filter_options
 @state_options
+@click.option(
+    "--queries",
+    "-Q",
+    "parse_queries",
+    is_flag=True,
+    help="After schema validation, CQL-parse every query against NGSIEM. Requires credentials.",
+)
 @click.pass_context
-def validate(ctx, resources, tags, names, state_file):
+def validate(ctx, resources, tags, names, state_file, parse_queries):
     """Validate all templates without deploying."""
     console.print("[bold blue]Validating templates...[/bold blue]\n")
     verbose = ctx.obj.get("verbose", False)
 
-    orchestrator = init_orchestrator(state_file=state_file, require_credentials=False)
+    if parse_queries:
+        if load_credentials() is None:
+            console.print("[red]✗ --queries requires configured credentials.[/red]")
+            console.print("  Run 'talonctl auth setup' or unset --queries for schema-only validation.")
+            raise SystemExit(1)
+
+    # Schema validation is always offline — no credentials needed.
+    orchestrator = init_orchestrator(
+        state_file=state_file,
+        require_credentials=parse_queries,
+    )
     filters = parse_filters(resources, tags, names)
 
     try:
@@ -29,9 +47,25 @@ def validate(ctx, resources, tags, names, state_file):
         results = orchestrator.validate(**filters)
         formatter = PlanFormatter(console, verbose=verbose)
         formatter.format_validation_results(results)
+
         has_errors = any(errors for errors in results.values() if errors)
         if has_errors:
             raise SystemExit(1)
+
+        if parse_queries:
+            try:
+                query_results = orchestrator.validate_queries(**filters)
+            except ValueError as e:
+                console.print("[red]✗ --queries requires configured credentials.[/red]")
+                console.print("  Run 'talonctl auth setup' or unset --queries for schema-only validation.")
+                if verbose:
+                    console.print(f"  [dim]{e}[/dim]")
+                raise SystemExit(1)
+
+            formatter.format_query_validation(query_results)
+            if any(not r.is_valid for r in query_results):
+                raise SystemExit(1)
+
     except SystemExit:
         raise
     except Exception as e:

src/talonctl/commands/validate_query.py

@@ -65,7 +65,7 @@ def validate_query(ctx, query, query_file, template):
         if result["valid"]:
             console.print("VALID")
         else:
-            console.print(f"INVALID: {result.get('message', 'Unknown error')}")
+            console.print(f"INVALID: {result['message']}")
             raise SystemExit(1)
 
     except SystemExit:

src/talonctl/core/__init__.py

@@ -26,6 +26,7 @@
 from talonctl.core.plan_formatter import PlanFormatter
 from talonctl.core.drift_detector import DriftDetector, DriftReport, DriftItem
 from talonctl.core.dependency_validator import DependencyValidator, DependencyIssue
+from talonctl.core.query_collection import QueryRef, collect_queries_from_templates
 
 __all__ = [
     # Base provider
@@ -59,4 +60,7 @@
     # Dependency validation
     "DependencyValidator",
     "DependencyIssue",
+    # Query collection
+    "QueryRef",
+    "collect_queries_from_templates",
 ]

src/talonctl/core/deployment_orchestrator.py

@@ -20,6 +20,7 @@
 from talonctl.core.state_synchronizer import StateSynchronizer
 from talonctl.core.drift_detector import DriftDetector, DriftReport
 from talonctl.core.base_provider import ResourceAction, ResourceChange
+from talonctl.core.query_collection import collect_queries_from_templates
 
 # Try to import NGSIEMClient for query validation
 try:
@@ -34,13 +35,14 @@
 
 @dataclass
 class QueryValidationResult:
-    """Result of FQL query validation"""
+    """Result of FQL query validation."""
 
     resource_id: str
     resource_name: str
     is_valid: bool
     error_message: Optional[str] = None
     query_snippet: Optional[str] = None  # First 100 chars of query
+    location: Optional[str] = None  # Field path within template, e.g. 'search.filter'
 
 
 @dataclass
@@ -366,7 +368,7 @@ def _validate_detection_queries(
 
                 # Extract query from search config (supports both 'query' and 'filter')
                 search_config = template.template_data.get("search", {})
-                query = search_config.get("query") or search_config.get("filter")
+                query = search_config.get("filter") or search_config.get("query")
 
                 if query:
                     # Clean query for display
@@ -863,6 +865,66 @@ def validate(
 
         return results
 
+    def validate_queries(
+        self,
+        resource_types: Optional[List[str]] = None,
+        tags: Optional[List[str]] = None,
+        names: Optional[List[str]] = None,
+        max_workers: int = 20,
+    ) -> List[QueryValidationResult]:
+        """
+        Parse every CQL query in every query-bearing template against NGSIEM.
+
+        Does NOT run schema validation — caller is responsible for that.
+        Raises ValueError (from NGSIEMClient) if credentials are missing.
+        """
+        if not NGSIEM_CLIENT_AVAILABLE:
+            raise RuntimeError("NGSIEM client unavailable — cannot validate queries")
+
+        discovered = self.template_discovery.discover_all(resource_types=resource_types, tags=tags, names=names)
+        refs = collect_queries_from_templates(discovered)
+        if not refs:
+            return []
+
+        logger.info(f"Validating {len(refs)} queries across {len(discovered)} resource types")
+        ngsiem_client = NGSIEMClient()
+
+        def validate_one(ref):
+            try:
+                result = ngsiem_client.test_query_syntax(ref.query)
+                return QueryValidationResult(
+                    resource_id=ref.resource_id,
+                    resource_name=ref.resource_name,
+                    is_valid=result["valid"],
+                    error_message=None if result["valid"] else result.get("message"),
+                    query_snippet=ref.query_snippet,
+                    location=ref.location,
+                )
+            except Exception as e:
+                return QueryValidationResult(
+                    resource_id=ref.resource_id,
+                    resource_name=ref.resource_name,
+                    is_valid=False,
+                    error_message=f"Validation error: {e}",
+                    query_snippet=ref.query_snippet,
+                    location=ref.location,
+                )
+
+        results: List[QueryValidationResult] = []
+        with ThreadPoolExecutor(max_workers=max_workers) as executor:
+            futures = [executor.submit(validate_one, r) for r in refs]
+            for future in as_completed(futures):
+                results.append(future.result())
+
+        valid = sum(1 for r in results if r.is_valid)
+        invalid = len(results) - valid
+        if invalid:
+            logger.warning(f"Query validation: {valid} valid, {invalid} invalid")
+        else:
+            logger.info(f"Query validation: all {valid} queries valid")
+
+        return results
+
     def sync(
         self,
         resource_types: Optional[List[str]] = None,

src/talonctl/core/plan_formatter.py

@@ -17,21 +17,11 @@
 
 # Import ResourceChange and ResourceAction from the canonical definition in base_provider
 from .base_provider import ResourceChange, ResourceAction
+from .deployment_orchestrator import QueryValidationResult
 
 logger = logging.getLogger(__name__)
 
 
-@dataclass
-class QueryValidationResult:
-    """Result of FQL query validation"""
-
-    resource_id: str
-    resource_name: str
-    is_valid: bool
-    error_message: Optional[str] = None
-    query_snippet: Optional[str] = None
-
-
 @dataclass
 class DeploymentPlan:
     """Represents a complete deployment plan"""
@@ -385,8 +375,7 @@ def format_validation_results(self, results: Dict[str, List[str]]) -> None:
             self.console.print(f"\n[red]✗ {invalid_templates} of {total_templates} templates have errors[/red]\n")
 
     def format_query_validation(self, results: List[QueryValidationResult]) -> None:
-        """
-        Format FQL query validation results
+        """Format CQL query validation results.
 
         Args:
             results: List of query validation results
@@ -399,13 +388,15 @@ def format_query_validation(self, results: List[QueryValidationResult]) -> None:
         valid = total - invalid
 
         self.console.print()
-        self.console.print(Panel.fit("[bold]FQL Query Validation[/bold]", border_style="blue"))
+        self.console.print(Panel.fit("[bold]Query Validation[/bold]", border_style="blue"))
         self.console.print()
 
         # Show invalid queries first
         for result in results:
             if not result.is_valid:
                 self.console.print(f"[red]✗[/red] [bold]{result.resource_id}[/bold]")
+                if result.location:
+                    self.console.print(f"    [dim]at:[/dim] {result.location}")
                 if result.error_message:
                     self.console.print(f"    [red]Error:[/red] {result.error_message}")
                 if result.query_snippet:
@@ -414,14 +405,14 @@ def format_query_validation(self, results: List[QueryValidationResult]) -> None:
 
         # Show valid queries (only if there are some invalid ones)
         if invalid > 0 and valid > 0:
-            self.console.print(f"[green]✓[/green] {valid} detection(s) with valid queries\n")
+            self.console.print(f"[green]✓[/green] {valid} queries valid\n")
 
         # Summary
         if invalid == 0:
-            self.console.print(f"[green]✓ All {total} detection queries are valid[/green]\n")
+            self.console.print(f"[green]✓ All {total} queries are valid[/green]\n")
         else:
-            self.console.print(f"[red]✗ {invalid} of {total} detection queries have errors[/red]\n")
-            self.console.print("[yellow]⚠[/yellow]  Deployment will be blocked until all queries are valid.\n")
+            self.console.print(f"[red]✗ {invalid} of {total} queries rejected by LogScale[/red]\n")
+            self.console.print("[yellow]⚠[/yellow]  Fix the above queries before running plan or apply.\n")
 
     def format_drift_report(self, report: Any) -> None:
         """

src/talonctl/core/query_collection.py

@@ -0,0 +1,89 @@
+"""Extract CQL query references from discovered templates.
+
+Each query-bearing resource type exposes its CQL in different fields. This
+module centralises that knowledge so callers (validate --queries, future
+refactors of the plan-path validator) can iterate queries uniformly without
+embedding per-type field names.
+"""
+
+from dataclasses import dataclass
+from typing import Callable, Dict, List, Tuple
+
+from talonctl.core.template_discovery import DiscoveredTemplate
+
+
+@dataclass
+class QueryRef:
+    resource_type: str
+    resource_id: str
+    resource_name: str
+    query: str
+    location: str
+    query_snippet: str
+
+
+def _make_snippet(query: str) -> str:
+    collapsed = " ".join(query.split())
+    if len(collapsed) > 100:
+        return collapsed[:100] + "..."
+    return collapsed
+
+
+def _extract_detection(template: dict) -> List[Tuple[str, str]]:
+    """Return list of (query, location) pairs. Detection has 0 or 1."""
+    search = template.get("search") or {}
+    for field in ("filter", "query"):
+        value = search.get(field)
+        if isinstance(value, str) and value.strip():
+            return [(value, f"search.{field}")]
+    return []
+
+
+def _extract_saved_search(template: dict) -> List[Tuple[str, str]]:
+    value = template.get("queryString")
+    if isinstance(value, str) and value.strip():
+        return [(value, "queryString")]
+    return []
+
+
+def _extract_dashboard(template: dict) -> List[Tuple[str, str]]:
+    pairs: List[Tuple[str, str]] = []
+    for widget_id, widget in (template.get("widgets") or {}).items():
+        qs = widget.get("queryString")
+        if isinstance(qs, str) and qs.strip():
+            pairs.append((qs, f"widgets.{widget_id}.queryString"))
+    for param_id, param in (template.get("parameters") or {}).items():
+        q = param.get("query")
+        if isinstance(q, str) and q.strip():
+            pairs.append((q, f"parameters.{param_id}.query"))
+    return pairs
+
+
+_EXTRACTORS: Dict[str, Callable[[dict], List[Tuple[str, str]]]] = {
+    "detection": _extract_detection,
+    "saved_search": _extract_saved_search,
+    "dashboard": _extract_dashboard,
+}
+
+
+def collect_queries_from_templates(
+    templates_by_type: Dict[str, List[DiscoveredTemplate]],
+) -> List[QueryRef]:
+    refs: List[QueryRef] = []
+    for resource_type, templates in templates_by_type.items():
+        extractor = _EXTRACTORS.get(resource_type)
+        if extractor is None:
+            continue
+        for template in templates:
+            for query, location in extractor(template.template_data):
+                refs.append(
+                    QueryRef(
+                        resource_type=resource_type,
+                        resource_id=template.resource_id,
+                        resource_name=template.name,
+                        query=query,
+                        location=location,
+                        query_snippet=_make_snippet(query),
+                    )
+                )
+    return refs