"""M365 Copilot Agent Evaluation CLI — thin orchestrator. Delegates to focused modules: cli_args – argument parsing & version-check bypass env_validator – environment validation & URL security prompt_loader – dataset loading & agent selection evaluation_runner – pipeline, evaluator dispatch, retry result_writer – console / JSON / CSV / HTML output """ import os import sys import traceback from dotenv import load_dotenv from api_clients.A2A.a2a_client import A2AClient from auth.auth_handler import AuthHandler, make_token_refresh_fn from auth.endpoint_resolver import resolve_endpoint_and_token from auth.azure_ai_auth_handler import ( resolve_auth_mode, verify_credential, build_azure_openai_client, has_azure_openai, AuthFailureOutcome, ) from evaluator_resolver import enumerate_custom_evaluators, resolve_default_evaluators from judge_backend import CopilotSDKJudgeBackend from foundry_cloud_evaluator import FoundryCloudEvaluator from version_check import check_min_version, get_cli_version from cli_logging.cli_logger import ( CLI_LOGGER, DIAGNOSTIC_RECORDS, configure_cli_logging, emit_structured_log, ) from cli_logging.logging_utils import Operation, resolve_log_level from cli_args import parse_arguments, should_bypass_min_version_check from env_validator import ( ALLOWED_ENDPOINTS, validate_endpoint_url, validate_environment, ) from common import ( ENV_AZURE_AI_API_KEY, ENV_AZURE_AI_MODEL_NAME, ENV_AZURE_AI_PROJECT_ENDPOINT, ENV_TENANT_ID, ENV_WORK_IQ_A2A_ENDPOINT, ENV_WORK_IQ_A2A_CLIENT_ID, ENV_WORK_IQ_A2A_SCOPES, ENV_WORK_IQ_GRAPH_ENDPOINT, ENV_WORK_IQ_GRAPH_SCOPES, JUDGE_BACKEND_GITHUB_COPILOT, RunConfig, should_use_foundry_eval, ) from prompt_loader import get_prompt_datasets from agent_selector import select_agent_interactively from evaluation_runner import PipelineConfig, run_pipeline from result_writer import output_results from dataclasses import replace def _count_agentic_responses(results): """Return (non-empty responses, requested turns) for pipeline results.""" response_items = [] for result in results: if result.get("type") == "multi_turn": response_items.extend(result.get("turns", [])) else: response_items.append(result) non_empty = sum( 1 for item in response_items if isinstance(item.get("response"), str) and item["response"].strip() ) return non_empty, len(response_items) def main(): """Main function to orchestrate the evaluation process.""" load_dotenv() args = parse_arguments() effective_log_level, error_message = resolve_log_level(args.log_level) if error_message: print(error_message) print( "Next step: rerun with --log-level {debug|info|warning|error}. " "For support, share the console diagnostics output from this run." ) sys.exit(2) config = replace( RunConfig.from_namespace(args), effective_log_level=effective_log_level, ) try: configure_cli_logging(config.effective_log_level) except ValueError as exc: print(f"Invalid console logging configuration: {exc}") print( "Next step: fix or unset RUNEVALS_LOG_TRUNCATE / " "RUNEVALS_LOG_MAX_LENGTH in your environment and rerun. " "For support, share the console diagnostics output from this run." ) sys.exit(2) emit_structured_log("info", f"Log level set to '{config.effective_log_level}'.", operation=Operation.SETUP) # Check minimum version before proceeding cli_version = get_cli_version() if not should_bypass_min_version_check(config) and not check_min_version(cli_version): sys.exit(1) # Evaluate-only needs judge configuration but no WorkIQ/A2A configuration. require_agent = not config.evaluate_only or config.signout validate_environment( judge_backend=config.judge_backend, require_agent=require_agent, ) a2a_endpoint = None if require_agent: a2a_endpoint = os.environ[ENV_WORK_IQ_A2A_ENDPOINT] validate_endpoint_url(a2a_endpoint, ALLOWED_ENDPOINTS) if config.signout: # Signout clears the shared MSAL account cache, which is keyed by # account (not scope), so a primary-scope handler suffices. Built # independently of the resolver to avoid an interactive token prompt. try: signout_handler = AuthHandler( client_id=os.environ[ENV_WORK_IQ_A2A_CLIENT_ID], tenant_id=os.environ[ENV_TENANT_ID], scopes_str=os.environ.get(ENV_WORK_IQ_A2A_SCOPES, ""), ) signout_handler.clear_cache() except Exception as e: emit_structured_log( "error", f"Error during signout: {e}", operation=Operation.AUTHENTICATE, ) sys.exit(1) sys.exit(0) # Resolve Azure AI auth mode — fail fast if explicit key mode # is selected but no key is available (before A2A auth or agent discovery) try: selected_auth_mode, auth_selection_source = resolve_auth_mode( api_key_raw=os.environ.get(ENV_AZURE_AI_API_KEY), auth_mode_flag=config.azure_ai_auth_mode, ) # The github-copilot judge backend uses GitHub auth, not Azure — skip the eager Azure probe. if config.judge_backend != JUDGE_BACKEND_GITHUB_COPILOT: verify_credential(selected_auth_mode, auth_selection_source) except AuthFailureOutcome as e: emit_structured_log("error", str(e), operation=Operation.SETUP) sys.exit(1) agent_client = None if not config.evaluate_only: try: # Prefer the WorkIQ A2A endpoint; fall back to the Graph gateway when # the tenant is not provisioned for A2A (consent / SP error). resolved = resolve_endpoint_and_token( primary_endpoint=a2a_endpoint, primary_scopes_str=os.environ.get(ENV_WORK_IQ_A2A_SCOPES, ""), graph_endpoint=os.environ[ENV_WORK_IQ_GRAPH_ENDPOINT], graph_scopes_str=os.environ.get(ENV_WORK_IQ_GRAPH_SCOPES, ""), client_id=os.environ[ENV_WORK_IQ_A2A_CLIENT_ID], tenant_id=os.environ[ENV_TENANT_ID], account_hint=config.account, ) except Exception as e: emit_structured_log( "error", f"Error during A2A authentication: {e}", operation=Operation.AUTHENTICATE, ) if config.effective_log_level == "debug": traceback.print_exc() sys.exit(1) try: agent_client = A2AClient( a2a_endpoint=resolved.endpoint, access_token=resolved.access_token, token_refresh_fn=make_token_refresh_fn(resolved.auth_handler), protocol_version=config.a2a_protocol_version, logger=CLI_LOGGER, diagnostic_records=DIAGNOSTIC_RECORDS, ) except Exception as e: emit_structured_log( "error", f"Failed to initialize A2A client: {e}", operation=Operation.SETUP, ) sys.exit(1) # 1. Load evaluation datasets eval_items, file_default_evaluators, document_extensions = get_prompt_datasets(config) default_evaluators = resolve_default_evaluators(file_default_evaluators) # 1a. Enumerate user-authored custom evaluators from /custom-evaluators/. # Cheap name-only pass — no user code imported, no security scan (that's lazy # per FR-004). Non-fatal: malformed/incomplete/colliding folders are recorded, # not rejected here. Enforcement is reference-time — colliding/incomplete # references fail at name validation (FR-030); a complete evaluator that fails # to import surfaces as an inline "error" result (FR-026). The try/except only # guards truly unexpected I/O errors (e.g. permission denied on the directory). try: discovered_customs = enumerate_custom_evaluators() except Exception as e: emit_structured_log( "error", f"Failed to scan custom-evaluators/ directory: {e}", operation=Operation.SETUP, ) sys.exit(1) if discovered_customs and config.effective_log_level == "debug": emit_structured_log( "debug", f"Discovered {len(discovered_customs)} custom evaluator(s): " f"{', '.join(sorted(discovered_customs.keys()))}.", operation=Operation.SETUP, ) if config.effective_log_level in ("info", "debug"): multi_turn_count = sum(1 for item in eval_items if "turns" in item) single_turn_count = len(eval_items) - multi_turn_count emit_structured_log( "info", f"Running evaluation on {len(eval_items)} item(s) " f"({single_turn_count} single-turn, {multi_turn_count} multi-turn).", operation=Operation.SETUP, ) agent_name = None if not config.evaluate_only: try: # 2. Agent selection - when no agent ID is provided, discover agents # via the active client (A2A) and prompt interactively. if not config.m365_agent_id: if config.effective_log_level in ("info", "debug"): emit_structured_log("info", "No agent ID provided. Fetching available agents.", operation=Operation.FETCH_AGENTS) available_agents = agent_client.fetch_available_agents() if not available_agents: emit_structured_log( "error", "No agents are available for interactive selection." " Re-run with --m365-agent-id or set M365_AGENT_ID.", operation=Operation.FETCH_AGENTS, ) sys.exit(1) selected_agent_id, agent_name = select_agent_interactively(available_agents) if selected_agent_id: config = replace(config, m365_agent_id=selected_agent_id) if config.effective_log_level in ("info", "debug"): emit_structured_log("info", f"Selected agent: {config.m365_agent_id}", operation=Operation.FETCH_AGENTS) else: emit_structured_log( "error", "No agent selected. Re-run with --m365-agent-id or set M365_AGENT_ID.", operation=Operation.FETCH_AGENTS, ) sys.exit(1) except Exception as e: emit_structured_log("error", f"Error during agent discovery: {e}", operation=Operation.FETCH_AGENTS) if config.effective_log_level == "debug": traceback.print_exc() sys.exit(1) # Pre-resolve agent endpoint (A2A agent card lookup) if config.m365_agent_id: agent_client.resolve_agent(config.m365_agent_id) # 3. Build pipeline config model_config = build_azure_openai_client(selected_auth_mode) # github-copilot mode routes LLM evaluators through the Copilot SDK; Azure mode uses # the inline evaluators (judge=None). judge = ( CopilotSDKJudgeBackend( log_level=config.effective_log_level, ) if config.judge_backend == JUDGE_BACKEND_GITHUB_COPILOT else None ) # Fail fast on an unavailable model — one clear error vs. one per prompt. No-op for "auto". if judge is not None: try: judge.verify_model() except Exception as exc: emit_structured_log("error", str(exc), operation=Operation.SETUP) judge.close() sys.exit(1) # Route LLM evaluators through Microsoft Foundry cloud evaluation whenever a # Foundry project endpoint (AZURE_AI_PROJECT_ENDPOINT) and a judge model # (AZURE_AI_MODEL_NAME) are configured. Required for gpt-5x / o-series judge # models (the local SDK evaluators can't handle them) and works for gpt-4x # too. Only applies to the default Azure judge mode (judge is None) and is # independent of the Copilot backend. foundry_evaluator = None if judge is None: deployment = os.environ.get(ENV_AZURE_AI_MODEL_NAME) project_endpoint = os.environ.get(ENV_AZURE_AI_PROJECT_ENDPOINT) if should_use_foundry_eval(deployment, project_endpoint): foundry_evaluator = FoundryCloudEvaluator( project_endpoint=project_endpoint, deployment=deployment, log_level=config.effective_log_level, ) try: foundry_evaluator.verify() except Exception as exc: emit_structured_log( "error", f"Foundry cloud evaluation setup failed: {exc}", operation=Operation.SETUP, ) sys.exit(1) pipeline = PipelineConfig( agent_client=agent_client, model_config=model_config, has_azure_openai=has_azure_openai(), default_evaluators=default_evaluators, selected_auth_mode=selected_auth_mode, auth_selection_source=auth_selection_source, judge_backend=judge, foundry_evaluator=foundry_evaluator, ) if config.effective_log_level in ("info", "debug"): if judge is not None: emit_structured_log( "info", f"Judge: {judge.describe()}", operation=Operation.SETUP, ) elif foundry_evaluator is not None: emit_structured_log( "info", f"Judge: {foundry_evaluator.describe()}", operation=Operation.SETUP, ) else: deployment = os.environ.get(ENV_AZURE_AI_MODEL_NAME) azure_desc = f"Azure OpenAI (deployment: {deployment})" if deployment else "Azure OpenAI" emit_structured_log( "info", f"Judge: {azure_desc}", operation=Operation.SETUP, ) try: results = run_pipeline(pipeline, eval_items, config) finally: # Release the Copilot SDK client + background loop (no-op for Azure). if judge is not None: judge.close() # Judge label for the output metadata (shows the model "auto" resolved to). if judge is not None: judge_label = judge.describe() elif foundry_evaluator is not None: judge_label = foundry_evaluator.describe() else: deployment = os.environ.get(ENV_AZURE_AI_MODEL_NAME) judge_label = f"Azure OpenAI (deployment: {deployment})" if deployment else "Azure OpenAI" # 4. Output results output_results( results, config, default_evaluators=default_evaluators, agent_name=agent_name, cli_version=str(cli_version) if cli_version else None, judge=judge_label, document_extensions=document_extensions) agentic_responses, requested_turns = _count_agentic_responses(results) if not config.evaluate_only and requested_turns and agentic_responses == 0: emit_structured_log( "error", "Evaluation failed: WorkIQ returned zero non-empty agent responses " f"across {requested_turns} requested turn(s).", operation=Operation.EVALUATE, ) sys.exit(1) if config.effective_log_level in ("info", "debug"): emit_structured_log( "info", f"Evaluation completed successfully. Processed {len(eval_items)} item(s).", operation=Operation.EVALUATE, ) # Call the main function when script is run directly if __name__ == "__main__": # pragma: no cover main()