Executor

1"""Executor for the agent-observability-trace-rca Claude Code skill."""
2
3import os
4import sys
5from typing import Any
6
7_eval_lib_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "../.."))
8if _eval_lib_path not in sys.path:
9    sys.path.insert(0, _eval_lib_path)
10
11from _eval_lib import DEFAULT_TIMEOUT_SECONDS, invoke_skill  # noqa: E402
12from ddeval import BaseProjectExecutor, Config  # noqa: E402
13
14
15class LlmObsTraceRcaExecutor(BaseProjectExecutor):
16    def execute_single(
17        self, input_data: dict[str, Any], config: Config | None
18    ) -> dict[str, Any]:
19        cfg = config.executor_config if config else {}
20        timeout_seconds = cfg.get("timeout_seconds", DEFAULT_TIMEOUT_SECONDS)
21        mcp_env = cfg.get("mcp_env", "prod")
22
23        if input_data.get("from") and input_data.get("to"):
24            time_spec = f"from {input_data['from']} to {input_data['to']}"
25        else:
26            timeframe = input_data.get("timeframe", "now-24h")
27            time_spec = f"over the last {timeframe}"
28
29        ml_app = input_data.get("ml_app")
30        eval_name = input_data.get("eval_name")
31        mode = input_data.get("mode")
32        failure_filter = input_data.get("failure_filter")
33
34        # Append explicit mode/filter hints so the skill skips inference when needed.
35        hints = []
36        if mode:
37            hints.append(f"mode={mode}")
38        if failure_filter:
39            hints.append(f"filter={failure_filter}")
40        suffix = f" [{', '.join(hints)}]" if hints else ""
41
42        if ml_app and eval_name:
43            prompt = f"/agent-observability-trace-rca Analyze eval failures for {eval_name} on {ml_app} {time_spec}{suffix}"
44        elif eval_name:
45            prompt = f"/agent-observability-trace-rca Analyze eval failures for {eval_name} {time_spec}{suffix}"
46        elif ml_app:
47            prompt = f"/agent-observability-trace-rca What's wrong with {ml_app} {time_spec}{suffix}"
48        else:
49            raise ValueError("input_data must contain either 'ml_app' or 'eval_name'")
50
51        report = invoke_skill(prompt, timeout_seconds=timeout_seconds, mcp_env=mcp_env)
52        return {"report": report, "word_count": len(report.split()) if report else 0}