cleveragents-core/robot/helper_metrics_collection.py

"""Helper script for metrics collection Robot Framework tests.

Usage:
    python helper_metrics_collection.py <command>

Commands:
    metric-type-enum      Verify MetricType enum
    metric-definitions    Verify METRIC_DEFINITIONS registry
    typed-factories       Test histogram/counter/gauge factory methods
    convenience-methods   Test all 14 convenience methods
    emitter-local         Test MetricsEmitter local mode
    emitter-disabled      Test MetricsEmitter disabled mode
    settings-config       Verify settings fields
    log-processor         Verify structlog processor
"""

from __future__ import annotations

import sys
from pathlib import Path
from typing import Any

_SRC = str(Path(__file__).resolve().parents[1] / "src")
sys.path.insert(0, _SRC)

from cleveragents.config.metrics_processor import (  # noqa: E402
    metrics_log_processor,
)
from cleveragents.config.settings import Settings  # noqa: E402
from cleveragents.domain.models.observability.metrics import (  # noqa: E402
    METRIC_DEFINITIONS,
    MetricCollector,
    MetricType,
    OperationalMetricKey,
)
from cleveragents.infrastructure.observability.metrics_emitter import (  # noqa: E402
    MetricsEmitter,
)

VALID_PLAN_ID = "01HX0000000000MMMMMMMMMMMM"


def cmd_metric_type_enum() -> None:
    """Verify MetricType enum."""
    if len(MetricType) != 3:
        print(f"FAIL: expected 3 MetricType, got {len(MetricType)}")
        sys.exit(1)
    if MetricType.HISTOGRAM.value != "histogram":
        print("FAIL: HISTOGRAM value mismatch")
        sys.exit(1)
    if MetricType.COUNTER.value != "counter":
        print("FAIL: COUNTER value mismatch")
        sys.exit(1)
    if MetricType.GAUGE.value != "gauge":
        print("FAIL: GAUGE value mismatch")
        sys.exit(1)
    print("metrics-type-enum-ok")


def cmd_metric_definitions() -> None:
    """Verify METRIC_DEFINITIONS registry."""
    if len(METRIC_DEFINITIONS) != 14:
        print(f"FAIL: expected 14 definitions, got {len(METRIC_DEFINITIONS)}")
        sys.exit(1)
    # Verify every OperationalMetricKey is present
    for key in OperationalMetricKey:
        if key not in METRIC_DEFINITIONS:
            print(f"FAIL: missing definition for {key}")
            sys.exit(1)
    print("metrics-definitions-ok")


def cmd_typed_factories() -> None:
    """Test histogram/counter/gauge factory methods."""
    h = MetricCollector.histogram(
        OperationalMetricKey.PLAN_DURATION_MS, 1500.0, VALID_PLAN_ID
    )
    if h.metric_type != MetricType.HISTOGRAM:
        print(f"FAIL: histogram type {h.metric_type}")
        sys.exit(1)
    c = MetricCollector.counter(OperationalMetricKey.LLM_CALL_COUNT, 5.0, VALID_PLAN_ID)
    if c.metric_type != MetricType.COUNTER:
        print(f"FAIL: counter type {c.metric_type}")
        sys.exit(1)
    g = MetricCollector.gauge(
        OperationalMetricKey.PLAN_DECISION_COUNT, 3.0, VALID_PLAN_ID
    )
    if g.metric_type != MetricType.GAUGE:
        print(f"FAIL: gauge type {g.metric_type}")
        sys.exit(1)
    print("metrics-typed-factories-ok")


def cmd_convenience_methods() -> None:
    """Test all 14 convenience methods."""
    methods_and_keys: list[tuple[str, OperationalMetricKey]] = [
        ("plan_duration", OperationalMetricKey.PLAN_DURATION_MS),
        ("plan_cost", OperationalMetricKey.PLAN_TOTAL_COST_USD),
        ("plan_decision_count", OperationalMetricKey.PLAN_DECISION_COUNT),
        ("subplan_count", OperationalMetricKey.SUBPLAN_COUNT),
        ("actor_invocation_count", OperationalMetricKey.ACTOR_INVOCATION_COUNT),
        ("actor_latency", OperationalMetricKey.ACTOR_LATENCY_MS),
        ("tool_invocation_count", OperationalMetricKey.TOOL_INVOCATION_COUNT),
        ("tool_error_rate", OperationalMetricKey.TOOL_ERROR_RATE),
        ("context_build_time", OperationalMetricKey.CONTEXT_BUILD_TIME_MS),
        ("context_token_count", OperationalMetricKey.CONTEXT_TOKEN_COUNT),
        ("llm_call_count", OperationalMetricKey.LLM_CALL_COUNT),
        ("llm_total_tokens", OperationalMetricKey.LLM_TOTAL_TOKENS),
        ("llm_total_cost", OperationalMetricKey.LLM_TOTAL_COST_USD),
        ("llm_avg_latency", OperationalMetricKey.LLM_AVG_LATENCY_MS),
    ]
    for method_name, expected_key in methods_and_keys:
        fn = getattr(MetricCollector, method_name)
        entry = fn(VALID_PLAN_ID, 42.0)
        if entry.key != expected_key:
            print(f"FAIL: {method_name} key {entry.key} != {expected_key}")
            sys.exit(1)
        if entry.value != 42.0:
            print(f"FAIL: {method_name} value {entry.value} != 42.0")
            sys.exit(1)
    print("metrics-convenience-ok")


def cmd_emitter_local() -> None:
    """Test MetricsEmitter local mode."""
    emitter = MetricsEmitter(enabled=True)
    entry = MetricCollector.plan_duration(VALID_PLAN_ID, 1500.0)
    emitter.emit(entry)
    count = emitter.emit_batch([entry, entry])
    if count != 2:
        print(f"FAIL: batch count {count} != 2")
        sys.exit(1)
    print("metrics-emitter-local-ok")


def cmd_emitter_disabled() -> None:
    """Test MetricsEmitter disabled mode."""
    emitter = MetricsEmitter(enabled=False)
    entry = MetricCollector.plan_duration(VALID_PLAN_ID, 1500.0)
    emitter.emit(entry)  # should be no-op
    count = emitter.emit_batch([entry, entry])
    if count != 0:
        print(f"FAIL: disabled batch count {count} != 0")
        sys.exit(1)
    print("metrics-emitter-disabled-ok")


def cmd_settings_config() -> None:
    """Verify settings fields."""
    settings = Settings()
    if settings.metrics_enabled is not True:
        print("FAIL: metrics_enabled not True")
        sys.exit(1)
    if settings.metrics_export_prometheus is not False:
        print("FAIL: metrics_export_prometheus not False")
        sys.exit(1)
    print("metrics-settings-ok")


def cmd_log_processor() -> None:
    """Verify structlog processor."""
    metric_event: dict[str, Any] = {
        "event": "metric.recorded",
        "metric_key": "plan_duration_ms",
        "metric_value": 1500.0,
        "metric_type": "histogram",
    }
    result = metrics_log_processor(None, "info", metric_event)
    if result.get("event_category") != "metric":
        print("FAIL: event_category not set")
        sys.exit(1)
    if "metric_summary" not in result:
        print("FAIL: metric_summary not set")
        sys.exit(1)

    non_metric: dict[str, Any] = {"event": "plan.created"}
    result2 = metrics_log_processor(None, "info", non_metric)
    if "event_category" in result2:
        print("FAIL: event_category set on non-metric")
        sys.exit(1)
    print("metrics-processor-ok")


def main() -> None:
    """Dispatch subcommand."""
    if len(sys.argv) < 2:
        print("Usage: helper_metrics_collection.py <command>")
        sys.exit(1)

    commands = {
        "metric-type-enum": cmd_metric_type_enum,
        "metric-definitions": cmd_metric_definitions,
        "typed-factories": cmd_typed_factories,
        "convenience-methods": cmd_convenience_methods,
        "emitter-local": cmd_emitter_local,
        "emitter-disabled": cmd_emitter_disabled,
        "settings-config": cmd_settings_config,
        "log-processor": cmd_log_processor,
    }
    fn = commands.get(sys.argv[1])
    if fn is None:
        print(f"Unknown command: {sys.argv[1]}")
        sys.exit(1)
    fn()


if __name__ == "__main__":
    main()