feat(scan): add PostHog telemetry for scan entry points

libs/giskard-scan/src/giskard/scan/_telemetry_props.py

@@ -0,0 +1,57 @@
+"""Aggregate, non-identifying properties for PostHog (no names, messages, or content)."""
+
+from __future__ import annotations
+
+from collections import Counter
+from typing import Any
+
+from .generators.base import ScenarioGenerator, TargetMode
+
+
+def generator_type_counts(generators: list[ScenarioGenerator]) -> dict[str, int]:
+    return dict(Counter(type(generator).__name__ for generator in generators))
+
+
+def suite_scan_shape_properties(
+    *,
+    scan_kind: str,
+    language_count: int,
+    target_mode: TargetMode,
+    generator_count: int,
+    scenario_count: int,
+    generator_types: dict[str, int],
+    parallel: bool,
+    max_concurrency: int | None,
+    **extra: Any,
+) -> dict[str, Any]:
+    return {
+        "integration": "giskard-scan",
+        "scan_kind": scan_kind,
+        "language_count": language_count,
+        "target_mode": target_mode,
+        "generator_count": generator_count,
+        "scenario_count": scenario_count,
+        "generator_types": generator_types,
+        "parallel": parallel,
+        "max_concurrency": max_concurrency,
+        **extra,
+    }
+
+
+def third_party_scan_shape_properties(
+    *,
+    tool: str,
+    language_count: int | None,
+    target_mode: TargetMode,
+    has_probe_filter: bool,
+    has_tag_filter: bool,
+) -> dict[str, Any]:
+    return {
+        "integration": "giskard-scan",
+        "scan_kind": f"third_party_{tool}",
+        "tool": tool,
+        "language_count": language_count,
+        "target_mode": target_mode,
+        "has_probe_filter": has_probe_filter,
+        "has_tag_filter": has_tag_filter,
+    }

libs/giskard-scan/src/giskard/scan/integrations/_entry_point.py

@@ -1,8 +1,13 @@
 """Top-level entry point for third-party scanner integrations."""
 
+import time
 from typing import Any, Literal
 
 from giskard.checks import SuiteResult, Target, Trace
+from giskard.core import telemetry_capture, telemetry_run_context, telemetry_tag
+
+from .._telemetry_props import third_party_scan_shape_properties
+from ..generators.base import TargetMode
 
 
 async def third_party_scan[InputType, OutputType, TraceType: Trace](  # pyright: ignore[reportMissingTypeArgument]
@@ -40,17 +45,48 @@ async def third_party_scan[InputType, OutputType, TraceType: Trace](  # pyright:
     Returns:
         The completed suite result.
     """
-    if tool == "garak":
-        from .garak import GarakScanAdapter
+    target_mode: TargetMode = kwargs.get("target_mode", "multiturn")
+    shape_props = third_party_scan_shape_properties(
+        tool=tool,
+        language_count=len(languages) if languages is not None else None,
+        target_mode=target_mode,
+        has_probe_filter=kwargs.get("probes") is not None,
+        has_tag_filter=kwargs.get("tags") is not None,
+    )
+
+    with telemetry_run_context():
+        telemetry_tag("giskard_component", "scan_third_party")
+        telemetry_tag("giskard_operation", "third_party_scan")
+        telemetry_capture("scan_third_party_run_started", properties=shape_props)
+
+        start_time = time.perf_counter()
+        if tool == "garak":
+            from .garak import GarakScanAdapter
 
-        # Garak has no TargetInfo concept: drop the context args so its
-        # run() signature is unaffected.
-        return await GarakScanAdapter().run(target, **kwargs)
-    elif tool == "lidar":
-        from .lidar import LidarScanAdapter
+            # Garak has no TargetInfo concept: drop the context args so its
+            # run() signature is unaffected.
+            result = await GarakScanAdapter().run(target, **kwargs)
+        elif tool == "lidar":
+            from .lidar import LidarScanAdapter
 
-        return await LidarScanAdapter().run(
-            target, description=description, languages=languages, **kwargs
+            result = await LidarScanAdapter().run(
+                target, description=description, languages=languages, **kwargs
+            )
+        else:
+            raise ValueError(f"Unknown tool {tool!r}. Available: ['garak', 'lidar']")
+        duration_ms = int((time.perf_counter() - start_time) * 1000)
+
+        telemetry_capture(
+            "scan_third_party_run_finished",
+            properties={
+                **shape_props,
+                "duration_ms": duration_ms,
+                "scenario_count": len(result.results),
+                "passed_count": result.passed_count,
+                "failed_count": result.failed_count,
+                "errored_count": result.errored_count,
+                "skipped_count": result.skipped_count,
+            },
         )
-    else:
-        raise ValueError(f"Unknown tool {tool!r}. Available: ['garak', 'lidar']")
+
+    return result

libs/giskard-scan/src/giskard/scan/quality.py

@@ -1,10 +1,13 @@
 """Quality scan entry points for giskard.scan."""
 
 import logging
+import time
 import warnings
 
 from giskard.checks import SuiteResult, Target, Trace
+from giskard.core import telemetry_capture, telemetry_run_context, telemetry_tag
 
+from ._telemetry_props import generator_type_counts, suite_scan_shape_properties
 from .catalog import generate_suite
 from .generators.base import TargetMode
 from .generators.knowledge_base import (
@@ -78,27 +81,62 @@ async def quality_scan[InputType, OutputType, TraceType: Trace](  # pyright: ign
         _warn_if_missing_knowledge_base(knowledge_base)
     )
 
-    suite = await generate_suite(
-        description=description,
-        languages=languages,
-        generators=quality_suite_generator_registry.generators(),
-        max_scenarios=max_scenarios,
-        seed=seed,
-        target_mode=target_mode,
-        knowledge_base=knowledge_base,
-    )
+    generators = quality_suite_generator_registry.generators()
+    with telemetry_run_context():
+        telemetry_tag("giskard_component", "scan_quality")
+        telemetry_tag("giskard_operation", "quality_scan")
+
+        suite = await generate_suite(
+            description=description,
+            languages=languages,
+            generators=generators,
+            max_scenarios=max_scenarios,
+            seed=seed,
+            target_mode=target_mode,
+            knowledge_base=knowledge_base,
+        )
+
+        shape_props = suite_scan_shape_properties(
+            scan_kind="quality",
+            language_count=len(languages),
+            target_mode=target_mode,
+            generator_count=len(generators),
+            scenario_count=len(suite.scenarios),
+            generator_types=generator_type_counts(generators),
+            parallel=parallel,
+            max_concurrency=max_concurrency,
+            has_knowledge_base=knowledge_base is not None,
+        )
+        telemetry_capture("scan_quality_run_started", properties=shape_props)
+
+        start_time = time.perf_counter()
+        result: SuiteResult = await suite.run(
+            target,
+            parallel=parallel,
+            max_concurrency=max_concurrency,
+        )
+        duration_ms = int((time.perf_counter() - start_time) * 1000)
+
+        try:
+            recommendation = await generate_quality_recommendation(result)
+        except Exception:
+            logger.exception("Quality recommendation generation failed")
+            recommendation = ""
+        quality_result = result.model_copy(update={"recommendation": recommendation})
+
+        telemetry_capture(
+            "scan_quality_run_finished",
+            properties={
+                **shape_props,
+                "duration_ms": duration_ms,
+                "passed_count": quality_result.passed_count,
+                "failed_count": quality_result.failed_count,
+                "errored_count": quality_result.errored_count,
+                "skipped_count": quality_result.skipped_count,
+                "has_recommendation": bool(recommendation),
+            },
+        )
 
-    result: SuiteResult = await suite.run(
-        target,
-        parallel=parallel,
-        max_concurrency=max_concurrency,
-    )
-    try:
-        recommendation = await generate_quality_recommendation(result)
-    except Exception:
-        logger.exception("Quality recommendation generation failed")
-        recommendation = ""
-    quality_result = result.model_copy(update={"recommendation": recommendation})
     quality_result.print_report(group_by=group_by)
     return quality_result
 

libs/giskard-scan/src/giskard/scan/vulnerability.py

@@ -1,7 +1,10 @@
+import time
 from typing import Unpack
 
 from giskard.checks import SuiteResult, Target, Trace
+from giskard.core import telemetry_capture, telemetry_run_context, telemetry_tag
 
+from ._telemetry_props import generator_type_counts, suite_scan_shape_properties
 from .catalog import generate_suite
 from .generators.adversarial import AdversarialScenarioGenerator
 from .generators.base import TargetMode
@@ -70,21 +73,54 @@ async def vulnerability_scan[InputType, OutputType, TraceType: Trace](  # pyrigh
     """
     opts: ResolvedScanOptions = {**_DEFAULT_SCAN_OPTIONS, **options}
 
-    suite = await generate_suite(
-        description=description,
-        languages=languages,
-        generators=vulnerability_suite_generator_registry.generators(
-            commercial_use=opts["commercial_use"]
-        ),
-        max_scenarios=opts["max_scenarios"],
-        seed=opts["seed"],
-        target_mode=target_mode,
+    generators = vulnerability_suite_generator_registry.generators(
+        commercial_use=opts["commercial_use"]
     )
+    with telemetry_run_context():
+        telemetry_tag("giskard_component", "scan_vulnerability")
+        telemetry_tag("giskard_operation", "vulnerability_scan")
+
+        suite = await generate_suite(
+            description=description,
+            languages=languages,
+            generators=generators,
+            max_scenarios=opts["max_scenarios"],
+            seed=opts["seed"],
+            target_mode=target_mode,
+        )
+
+        shape_props = suite_scan_shape_properties(
+            scan_kind="vulnerability",
+            language_count=len(languages),
+            target_mode=target_mode,
+            generator_count=len(generators),
+            scenario_count=len(suite.scenarios),
+            generator_types=generator_type_counts(generators),
+            parallel=opts["parallel"],
+            max_concurrency=opts["max_concurrency"],
+            commercial_use=opts["commercial_use"],
+        )
+        telemetry_capture("scan_vulnerability_run_started", properties=shape_props)
+
+        start_time = time.perf_counter()
+        result = await suite.run(
+            target,
+            parallel=opts["parallel"],
+            max_concurrency=opts["max_concurrency"],
+        )
+        duration_ms = int((time.perf_counter() - start_time) * 1000)
+
+        telemetry_capture(
+            "scan_vulnerability_run_finished",
+            properties={
+                **shape_props,
+                "duration_ms": duration_ms,
+                "passed_count": result.passed_count,
+                "failed_count": result.failed_count,
+                "errored_count": result.errored_count,
+                "skipped_count": result.skipped_count,
+            },
+        )
 
-    result = await suite.run(
-        target,
-        parallel=opts["parallel"],
-        max_concurrency=opts["max_concurrency"],
-    )
     result.print_report(group_by=opts["group_by"])
     return result

libs/giskard-scan/tests/conftest.py

@@ -1,8 +1,15 @@
 import pytest
+from giskard.core import disable_telemetry
 from giskard.scan.quality import quality_suite_generator_registry
 from giskard.scan.vulnerability import vulnerability_suite_generator_registry
 
 
+@pytest.fixture(autouse=True)
+def _disable_telemetry():
+    """Disable telemetry for tests."""
+    disable_telemetry()
+
+
 @pytest.fixture
 def isolated_vulnerability_registry():
     """Snapshot and restore the vulnerability generator registry."""

libs/giskard-scan/tests/test_quality.py

@@ -227,6 +227,8 @@ async def test_quality_scan_warns_and_skips_empty_raw_knowledge_base(
     captured_knowledge_bases: list[object] = []
 
     class _FakeSuite:
+        scenarios: list[Scenario[Any, Any, Trace[Any, Any]]] = []
+
         async def run(
             self,
             target: object,
@@ -278,6 +280,8 @@ async def test_quality_scan_configures_knowledge_base_generator(
     printed_reports: list[str | None] = []
 
     class _FakeSuite:
+        scenarios: list[Scenario[Any, Any, Trace[Any, Any]]] = []
+
         async def run(
             self,
             target: object,