dotnet__skills/eng/evaluation/test_token_failover.py

#!/usr/bin/env python3

import json
import os
import re
import shutil
import stat
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path

try:
    import yaml
except ImportError:  # pragma: no cover
    print("PyYAML is required: pip install pyyaml", file=sys.stderr)
    raise SystemExit(2)


REPO_ROOT = Path(__file__).resolve().parents[2]
WORKFLOW = REPO_ROOT / ".github" / "workflows" / "evaluation-run.yml"
CALLER_WORKFLOW = REPO_ROOT / ".github" / "workflows" / "evaluation.yml"
TEST_WORKFLOW = REPO_ROOT / ".github" / "workflows" / "evaluation-workflow-tests.yml"
DASHBOARD_GENERATOR = REPO_ROOT / "eng" / "dashboard" / "generate-benchmark-data.ps1"
PATH_SAFETY_SCRIPT = REPO_ROOT / "eng" / "evaluation" / "path-safety.ps1"
STEP_NAME = "Select available Copilot token from pool"
GIT_BASH = Path(os.environ.get("ProgramFiles", r"C:\Program Files")) / "Git" / "bin" / "bash.exe"
BASH = str(GIT_BASH) if os.name == "nt" and GIT_BASH.exists() else "bash"


def create_symlink_or_skip(
    test_case: unittest.TestCase,
    link: Path,
    target: Path,
    *,
    target_is_directory: bool = False,
) -> None:
    try:
        link.symlink_to(target, target_is_directory=target_is_directory)
    except OSError as error:
        test_case.skipTest(f"Symlinks are unavailable: {error}")


def selection_script() -> str:
    workflow = yaml.safe_load(WORKFLOW.read_text(encoding="utf-8"))
    try:
        steps = workflow["jobs"]["vally-evaluate"]["steps"]
    except (KeyError, TypeError) as error:
        raise AssertionError(
            f"{WORKFLOW} does not define jobs.vally-evaluate.steps"
        ) from error
    for step in steps:
        if step.get("name") == STEP_NAME:
            return step["run"]
    raise AssertionError(f"{WORKFLOW} does not contain the '{STEP_NAME}' step")


def rate_limit_pattern() -> str:
    workflow = yaml.safe_load(WORKFLOW.read_text(encoding="utf-8"))
    return workflow["jobs"]["vally-evaluate"]["env"]["COPILOT_RATE_LIMIT_PATTERN"]


def token_unavailable_pattern() -> str:
    workflow = yaml.safe_load(WORKFLOW.read_text(encoding="utf-8"))
    return workflow["jobs"]["vally-evaluate"]["env"][
        "COPILOT_TOKEN_UNAVAILABLE_PATTERN"
    ]


def generated_safe_output_configs(workflow: object) -> list[dict[str, object]]:
    configs: list[dict[str, object]] = []

    def collect(value: object) -> None:
        if isinstance(value, dict):
            for key, child in value.items():
                if key in {
                    "GH_AW_SAFE_OUTPUTS_CONFIG",
                    "GH_AW_SAFE_OUTPUTS_HANDLER_CONFIG",
                }:
                    configs.append(json.loads(str(child)))
                collect(child)
        elif isinstance(value, list):
            for child in value:
                collect(child)

    collect(workflow)
    return configs


class TokenFailoverTests(unittest.TestCase):
    def test_evaluation_model_profiles_and_judges(self) -> None:
        caller = yaml.safe_load(CALLER_WORKFLOW.read_text(encoding="utf-8"))
        discover_script = next(
            step["run"]
            for step in caller["jobs"]["discover"]["steps"]
            if "$profileModels = @{" in step.get("run", "")
        )
        start = discover_script.index("$matrixProfile = 'default'")
        end = discover_script.index("# Validate every entry", start)
        script = (
            "$ErrorActionPreference = 'Stop'\n"
            "$entries = @(@{name='fixture'; plugin='fixture'; target_kind='skill'; "
            "skills_path='plugins/fixture/skills'; agents_path=''})\n"
            + discover_script[start:end]
            + "\nConvertTo-Json -InputObject @($entries) -Compress\n"
        )
        cases = [
            ("pull_request", "", "", "", ["claude-sonnet-5", "gpt-5.6-luna"]),
            ("pull_request_target", "", "", "", ["claude-sonnet-5", "gpt-5.6-luna"]),
            ("workflow_dispatch", "", "", "", ["claude-sonnet-5", "gpt-5.6-luna"]),
            ("issue_comment", "/evaluate", "", "", ["claude-sonnet-5", "gpt-5.6-luna"]),
            ("pull_request_review", "/evaluate --full", "", "", [
                "claude-sonnet-5", "gpt-5.6-luna", "claude-haiku-4.5",
                "mai-code-1.1-flash", "gpt-5.3-codex", "claude-opus-4.8",
            ]),
            ("workflow_dispatch", "", "newer", "", [
                "gpt-5.6-sol", "claude-opus-5", "claude-sonnet-5",
            ]),
            ("schedule", "", "", "0 7 * * 1,3,5", ["claude-sonnet-5", "gpt-5.6-luna"]),
            ("schedule", "", "", "0 7 * * 2,6", [
                "claude-haiku-4.5", "mai-code-1.1-flash", "gpt-5.3-codex",
            ]),
            ("schedule", "", "", "0 7 * * 0", [
                "gpt-5.6-sol", "claude-opus-5", "claude-sonnet-5",
            ]),
            ("schedule", "", "", "0 7 * * 4", ["claude-opus-4.8"]),
            ("workflow_dispatch", "", "opus48", "", ["claude-opus-4.8"]),
        ]
        for event, body, profile, schedule, models in cases:
            with self.subTest(event=event, profile=profile, schedule=schedule):
                env = dict(os.environ, EVAL_EVENT_NAME=event,
                           EVAL_COMMENT_BODY=body if event == "issue_comment" else "",
                           EVAL_REVIEW_BODY=body if event == "pull_request_review" else "",
                           MATRIX_PROFILE_INPUT=profile, EVAL_SCHEDULE=schedule)
                result = subprocess.run(
                    ["pwsh", "-NoLogo", "-NoProfile", "-NonInteractive", "-Command", script],
                    env=env, capture_output=True, text=True, timeout=30,
                )
                self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
                entries = json.loads(result.stdout.strip().splitlines()[-1])
                self.assertEqual([entry["model"] for entry in entries], models)
                for entry in entries:
                    is_gpt = entry["model"].startswith("gpt-")
                    self.assertEqual(entry["judge"], "claude-opus-4.8" if is_gpt else "gpt-5.6-terra")
                    self.assertEqual(
                        entry["judge2"],
                        "claude-haiku-4.5" if is_gpt and event == "schedule" else "",
                    )
                    self.assertNotEqual(entry["judge"], entry["model"])

    def test_health_and_triage_models_are_separate_from_evaluation(self) -> None:
        for name in (
            "devops-health-check", "devops-health-groom",
            "devops-health-investigate", "issue-triage",
        ):
            with self.subTest(workflow=name):
                source = REPO_ROOT / ".github" / "workflows" / f"{name}.md"
                frontmatter = yaml.safe_load(source.read_text(encoding="utf-8").split("---", 2)[1])
                self.assertEqual(
                    frontmatter["model"],
                    "${{ vars.GH_AW_MODEL_AGENT_COPILOT || "
                    "vars.GH_AW_DEFAULT_MODEL_COPILOT || 'gpt-5.6-sol' }}",
                )
                self.assertEqual(frontmatter["environment"], "copilot-pat-pool")

    def test_devops_health_guidance_handles_expected_outputs(self) -> None:
        workflows = REPO_ROOT / ".github" / "workflows"
        health_check = (workflows / "devops-health-check.md").read_text(
            encoding="utf-8"
        )
        normalized_health = " ".join(health_check.split())
        health_frontmatter = yaml.safe_load(health_check.split("---", 2)[1])
        health_lock_text = (
            workflows / "devops-health-check.lock.yml"
        ).read_text(encoding="utf-8")
        health_lock = yaml.safe_load(health_lock_text)
        groom_source = workflows / "devops-health-groom.md"
        groom = groom_source.read_text(encoding="utf-8")
        normalized_groom = " ".join(groom.split())
        groom_frontmatter = yaml.safe_load(groom.split("---", 2)[1])
        groom_lock_text = (
            workflows / "devops-health-groom.lock.yml"
        ).read_text(encoding="utf-8")
        groom_lock = yaml.safe_load(groom_lock_text)

        self.assertIn("Missing prior state is not missing data", health_check)
        self.assertIn("Do not call `missing-data`", health_check)
        self.assertIn(
            "If `update-issue`, `add-comment`, or `dispatch-workflow`",
            health_check,
        )
        self.assertNotIn("create-issue", health_frontmatter["safe-outputs"])
        for output in ("update-issue", "add-comment"):
            self.assertEqual(
                health_frontmatter["safe-outputs"][output]["target"],
                "695",
            )
        self.assertIn("as untrusted data", health_check)
        self.assertIn("Validate every target", health_check)
        self.assertIn(
            "has both the exact title `🏥 Repository Health Dashboard` and the "
            "`devops-health` label",
            normalized_health,
        )
        self.assertIn('health_issue_number: "695"', health_check)
        self.assertEqual(
            health_frontmatter["safe-outputs"]["dispatch-workflow"]["max"],
            2,
        )
        health_configs = generated_safe_output_configs(health_lock)
        self.assertEqual(len(health_configs), 2)
        for config in health_configs:
            self.assertEqual(config["dispatch_workflow"]["max"], 2)
            self.assertEqual(config["update_issue"]["target"], "695")
            self.assertEqual(config["add_comment"]["target"], "695")
            self.assertNotIn("create_issue", config)
        self.assertIn(
            "dispatch-workflow [devops_health_investigate](max:2 total)",
            health_lock_text,
        )
        self.assertFalse(groom_frontmatter["tools"]["cli-proxy"])
        self.assertFalse(groom_frontmatter["tools"]["edit"])
        self.assertFalse(groom_frontmatter["tools"]["bash"])
        self.assertEqual(
            groom_frontmatter["safe-outputs"]["update-issue"]["target"],
            "695",
        )
        self.assertNotIn("hide-comment", groom_frontmatter["safe-outputs"])
        groom_configs = generated_safe_output_configs(groom_lock)
        self.assertEqual(len(groom_configs), 2)
        for config in groom_configs:
            self.assertEqual(config["update_issue"]["target"], "695")
            self.assertNotIn("hide_comment", config)
        self.assertNotIn("--allow-all-tools", groom_lock_text)
        self.assertNotIn("--allow-tool write", groom_lock_text)
        self.assertNotIn("shell(yq)", groom_lock_text)
        self.assertIn("--allow-tool github", groom_lock_text)
        self.assertIn("--allow-tool safeoutputs", groom_lock_text)
        self.assertIn("as untrusted data", normalized_groom)
        self.assertIn("Bind outputs to verified data", normalized_groom)
        self.assertIn("/issues/695", groom)
        self.assertIn("issue_number: 695", groom)
        self.assertIn("perPage: 20, page: 1", groom)
        self.assertIn("Continue with page 2", groom)
        self.assertIn("GitHub returns issue comments oldest first", groom)
        self.assertIn(
            "until a response contains neither comments nor a `[Filtered]` notice",
            normalized_groom,
        )
        self.assertIn("do not stop based on comment age", normalized_groom)
        self.assertIn("Integrity filtering can remove items", groom)
        self.assertIn(
            "include only fetched comments whose `created_at` is within the last 30 days",
            normalized_groom,
        )
        self.assertIn("Do not stop after the first page", normalized_groom)
        self.assertIn("Do not finish with only a text response", groom)

        self.assertFalse(health_frontmatter["tools"]["bash"])
        self.assertFalse(health_frontmatter["tools"]["cli-proxy"])
        self.assertFalse(health_frontmatter["tools"]["edit"])
        self.assertEqual(
            health_frontmatter["concurrency"]["group"],
            "gh-aw-devops-health-dashboard",
        )
        self.assertFalse(
            health_frontmatter["concurrency"]["cancel-in-progress"]
        )
        self.assertEqual(health_frontmatter["concurrency"]["queue"], "max")
        self.assertEqual(health_lock["concurrency"]["queue"], "max")
        self.assertEqual(
            groom_frontmatter["concurrency"],
            health_frontmatter["concurrency"],
        )
        self.assertEqual(
            groom_lock["concurrency"],
            health_lock["concurrency"],
        )
        self.assertNotIn("cache-memory", health_frontmatter["tools"])
        self.assertNotIn("--allow-all-tools", health_lock_text)
        self.assertNotIn("--allow-tool write", health_lock_text)
        self.assertNotIn("shell(git:*)", health_lock_text)
        self.assertNotIn("shell(yq)", health_lock_text)
        self.assertIn("--allow-tool github", health_lock_text)
        self.assertIn("--allow-tool safeoutputs", health_lock_text)
        self.assertNotIn("cache_memory_prompt.md", health_lock_text)
        self.assertNotIn("Create cache-memory directory", health_lock_text)
        self.assertNotIn("update_cache_memory:", health_lock_text)
        self.assertIn("devops-health-state:v1", health_check)
        self.assertIn("One-time legacy migration", health_check)
        self.assertIn("final `# 🏥 Daily Health Check", health_check)
        self.assertNotIn("/git/trees/", health_check)
        self.assertIn("search_code: filename:plugin.json path:plugins", health_check)
        self.assertIn("search_code: filename:SKILL.md path:plugins", health_check)
        self.assertIn("If code search reaches its result limit", health_check)
        self.assertIn("State overflow guard", health_check)
        self.assertIn("more than 100 active findings", health_check)
        self.assertIn(
            "Never truncate the authoritative state", normalized_health
        )
        self.assertIn(
            "its `active_findings[].fingerprint` values are the authoritative current active set",
            normalized_groom,
        )
        self.assertIn("omitted from visible sections", groom)
        self.assertIn("If the marker is absent or invalid", groom)
        self.assertIn(
            "this fallback is not authoritative for resolution",
            normalized_groom,
        )
        self.assertIn(
            "do not infer resolution from the visible fallback set",
            normalized_groom,
        )
        shared_health = (
            REPO_ROOT / ".github" / "aw" / "shared" / "devops-health.lock.md"
        ).read_text(encoding="utf-8")
        self.assertIn(
            "The safe-output issue update is the only persistence operation",
            " ".join(shared_health.split()),
        )

    def test_devops_health_investigation_is_report_only(self) -> None:
        investigate_source = (
            REPO_ROOT / ".github" / "workflows" / "devops-health-investigate.md"
        )
        investigate = investigate_source.read_text(encoding="utf-8")
        investigate_frontmatter = yaml.safe_load(investigate.split("---", 2)[1])
        investigate_lock = yaml.safe_load(
            investigate_source.with_suffix(".lock.yml").read_text(
                encoding="utf-8"
            )
        )

        trigger = investigate_frontmatter.get("on", investigate_frontmatter.get(True))
        dispatch_inputs = trigger["workflow_dispatch"]["inputs"]
        self.assertEqual(dispatch_inputs["dry_run"]["type"], "boolean")
        self.assertFalse(dispatch_inputs["dry_run"]["default"])

        self.assertEqual(
            investigate_frontmatter["safe-outputs"]["staged"],
            "${{ inputs.dry_run }}",
        )
        self.assertEqual(
            investigate_frontmatter["safe-outputs"]["report-failure-as-issue"],
            "${{ !inputs.dry_run }}",
        )
        self.assertNotIn(
            "create-pull-request",
            investigate_frontmatter["safe-outputs"],
        )
        self.assertEqual(
            investigate_frontmatter["safe-outputs"]["add-comment"]["target"],
            "695",
        )
        investigate_configs = generated_safe_output_configs(investigate_lock)
        self.assertEqual(len(investigate_configs), 2)
        for config in investigate_configs:
            self.assertEqual(config["add_comment"]["target"], "695")
        self.assertEqual(
            investigate_frontmatter["network"]["allowed"],
            ["defaults"],
        )
        self.assertIn("This investigator is report-only", investigate)
        self.assertIn("The only allowed target is issue `695`", investigate)
        self.assertIn("do not call `add-comment`", investigate)
        self.assertIn("If `dry_run` is true, do not call `add-comment`", investigate)

    def test_devops_health_investigator_has_no_mutating_tools(self) -> None:
        workflows = REPO_ROOT / ".github" / "workflows"
        investigate_source = workflows / "devops-health-investigate.md"
        investigate = investigate_source.read_text(encoding="utf-8")
        normalized_investigate = " ".join(investigate.split())
        investigate_lock = (
            workflows / "devops-health-investigate.lock.yml"
        ).read_text(encoding="utf-8")
        investigate_frontmatter = yaml.safe_load(investigate.split("---", 2)[1])

        self.assertNotIn("args", investigate_frontmatter["engine"])
        self.assertFalse(investigate_frontmatter["tools"]["edit"])
        self.assertFalse(investigate_frontmatter["tools"]["bash"])
        self.assertFalse(investigate_frontmatter["tools"]["cli-proxy"])
        self.assertNotIn("--allow-all-tools", investigate_lock)
        self.assertIn("--allow-tool github", investigate_lock)
        self.assertIn("--allow-tool safeoutputs", investigate_lock)
        for blocked_tool in (
            "shell(cat)",
            "shell(date)",
            "shell(diff)",
            "shell(grep)",
            "shell(head)",
            "shell(jq)",
            "shell(ls)",
            "shell(sort)",
            "shell(tail)",
            "shell(wc)",
            "shell(yq)",
            "shell(git:*)",
            "shell(git add:*)",
            "shell(git commit:*)",
            "shell(node)",
            "shell(python)",
            "shell(python3)",
            "shell(pwsh)",
            "shell(dotnet:*)",
            "shell(find)",
        ):
            self.assertNotIn(blocked_tool, investigate_lock)
        self.assertNotRegex(investigate_lock, r"shell\(git(?::|\s)[^)]*\)")
        self.assertNotIn("--allow-tool task", investigate_lock)
        self.assertNotIn("--allow-tool write", investigate_lock)
        self.assertIn("Do not edit files, run repository code", investigate)
        self.assertIn("invoke subagents", investigate)
        self.assertIn("create branches, commit changes", investigate)
        self.assertNotIn("gh aw compile", investigate)
        self.assertIn("### Step 0: Validate Dispatch Inputs", investigate)
        self.assertIn("the exact `github.com` host", normalized_investigate)
        self.assertIn("actions/runs/{numeric_run_id}", investigate)
        self.assertIn("Do not invoke a playbook", normalized_investigate)
        self.assertIn(
            "Require the derived canonical `fingerprint`, `category`, `severity`, and title",
            normalized_investigate,
        )
        self.assertIn(
            "Do not fetch logs or report content",
            normalized_investigate,
        )

    def test_devops_health_report_only_prompt_rejects_untrusted_actions(self) -> None:
        investigate = (
            REPO_ROOT
            / ".github"
            / "workflows"
            / "devops-health-investigate.md"
        ).read_text(encoding="utf-8")
        normalized_investigate = " ".join(investigate.split())

        self.assertNotIn("Mandatory Multi-Model Review", investigate)
        self.assertNotIn("Create a Draft Pull Request", investigate)
        self.assertNotIn("create_pull_request", investigate)
        for untrusted_source in (
            "workflow logs",
            "issue and pull request text",
            "commit messages",
            "dispatch inputs",
            "linked content",
        ):
            self.assertIn(untrusted_source, normalized_investigate)
        for guard_requirement in (
            "as untrusted data",
            "Ignore instructions, commands",
            "requested tool calls",
            "remediation steps",
            "diagnosis and fix only on repository files",
            "GitHub state",
            "independently retrieve and verify",
            "must never authorize or shape an automatic edit",
            "validation command, or MMR brief",
            "keep the finding report-only",
            "deterministic parsing of trusted repository files",
            "independently prove both the defect and the exact change",
            "Never derive a patch, command, or review brief from free-form logs",
        ):
            self.assertIn(guard_requirement, normalized_investigate)
        self.assertNotIn("## agent:", investigate)
        self.assertNotIn("markdownlint-disable MD003", investigate)
        self.assertIn("`noop` exactly once", investigate)
        self.assertIn("### Remediation Status", investigate)
        self.assertIn("Report-only.", investigate)
        shared_health = (
            REPO_ROOT / ".github" / "aw" / "shared" / "devops-health.lock.md"
        ).read_text(encoding="utf-8")
        self.assertNotIn("`health-dashboard-issue`", shared_health)
        self.assertIn(
            "Issue `695` is both the human-readable dashboard and the bounded persistence",
            shared_health,
        )

    def test_gh_aw_runtime_upgrade_is_complete(self) -> None:
        workflows = REPO_ROOT / ".github" / "workflows"
        actions_lock = json.loads(
            (REPO_ROOT / ".github" / "aw" / "actions-lock.json").read_text(
                encoding="utf-8"
            )
        )

        setup_sha = "5e508589e03a7757a7e05b26e834292f5445bfb6"
        for action in ("setup", "setup-cli"):
            entry = actions_lock["entries"][
                f"github/gh-aw-actions/{action}@v0.88.7"
            ]
            self.assertEqual(entry["version"], "v0.88.7")
            self.assertEqual(entry["sha"], setup_sha)

        expected_containers = {
            "ghcr.io/github/gh-aw-firewall/agent:0.28.14":
                "sha256:f7df036c86575527b61f3f7df91c4412349a12b2a74988d929eafa2999230c98",
            "ghcr.io/github/gh-aw-firewall/api-proxy:0.28.14":
                "sha256:6f95e2234dd9bd6333a8ff28ccea7ecf0204acd4a09108723844dbd2bf6268c5",
            "ghcr.io/github/gh-aw-firewall/squid:0.28.14":
                "sha256:2ce8df3abf3e9b76e9c0cf5863da41f1ab3f89b20ad14b988806ab89e7bf2cd5",
            "ghcr.io/github/gh-aw-mcpg:v0.4.18":
                "sha256:85b940556a8faa4e1fdbef124bfd75f2c4ebd855a10b88a1c3b6f3e97f6f1a53",
        }
        expected_executable_images = {
            f"{image}@{digest}"
            for image, digest in expected_containers.items()
        }
        expected_executable_images.add("ghcr.io/github/gh-aw-mcpg:v0.4.18")

        def gh_aw_action_refs(text: str) -> set[tuple[str, str]]:
            return set(
                re.findall(
                    r"github/gh-aw-actions/(setup(?:-cli)?)@([^\s#\"']+)",
                    text,
                )
            )

        def executable_lines(text: str) -> str:
            return "\n".join(
                line for line in text.splitlines() if not line.lstrip().startswith("#")
            )

        for image, digest in expected_containers.items():
            with self.subTest(image=image):
                container = actions_lock["containers"][image]
                self.assertEqual(container["digest"], digest)
                self.assertEqual(
                    container["pinned_image"],
                    f"{image}@{digest}",
                )

        for workflow in (
            "devops-health-check",
            "devops-health-groom",
            "devops-health-investigate",
            "issue-investigate",
            "issue-triage",
            "markdown-linter",
            "pr-malicious-scan.agent",
        ):
            with self.subTest(workflow=workflow):
                lock = (workflows / f"{workflow}.lock.yml").read_text(
                    encoding="utf-8"
                )
                executable_lock = executable_lines(lock)
                executable_images = set(
                    re.findall(
                        r"ghcr\.io/github/(?:"
                        r"gh-aw-firewall/(?:agent|api-proxy|squid)|gh-aw-mcpg"
                        r"):[A-Za-z0-9._-]+(?:@sha256:[0-9a-f]{64})?",
                        executable_lock,
                    )
                )
                self.assertIn('"compiler_version":"v0.88.7"', lock)
                self.assertEqual(
                    gh_aw_action_refs(executable_lock),
                    {("setup", setup_sha)},
                )
                self.assertEqual(
                    executable_images,
                    expected_executable_images,
                )

        investigate_lock = (
            workflows / "devops-health-investigate.lock.yml"
        ).read_text(encoding="utf-8")
        self.assertNotIn("--allow-tool task", investigate_lock)

        setup = (workflows / "copilot-setup-steps.yml").read_text(
            encoding="utf-8"
        )
        self.assertEqual(
            gh_aw_action_refs(executable_lines(setup)),
            {("setup-cli", setup_sha)},
        )
        self.assertIn("version: v0.88.7", setup)

        maintenance = (workflows / "agentics-maintenance.yml").read_text(
            encoding="utf-8"
        )
        self.assertIn(
            "generated by pkg/workflow/maintenance_workflow.go (v0.88.7)",
            maintenance,
        )
        self.assertEqual(
            gh_aw_action_refs(executable_lines(maintenance)),
            {
                ("setup", setup_sha),
                ("setup-cli", setup_sha),
            },
        )
        self.assertNotIn("v0.86.2", maintenance)

    def run_selector(
        self,
        tokens: dict[int, str],
        model: str = "claude-opus-4.6",
        judge_model: str = "claude-opus-4.6",
    ) -> subprocess.CompletedProcess[str]:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            fake_bin = root / "bin"
            fake_bin.mkdir()
            attempts = root / "attempts"
            models = root / "models"
            github_output = root / "github-output"
            token_file = root / "evaluation-copilot-token"
            fake_copilot = fake_bin / "copilot"
            fake_copilot.write_text(
                """#!/usr/bin/env bash
set -euo pipefail
if env | grep -Eq '^COPILOT_PAT_[0-9]='; then
  echo "PAT pool leaked to Copilot subprocess" >&2
  exit 11
fi
echo "$COPILOT_GITHUB_TOKEN" >> "$ATTEMPTS"
model=""
has_effort=false
while [ "$#" -gt 0 ]; do
  case "$1" in
    --model) model="$2"; shift 2 ;;
    --effort=*) has_effort=true; shift ;;
    *) shift ;;
  esac
done
echo "$model" >> "$MODELS"
if [ "$model" = "no-effort-model" ] && [ "$has_effort" = true ]; then
  echo 'Error: Model "no-effort-model" does not support reasoning effort configuration (requested: "low").' >&2
  exit 1
fi
case "$COPILOT_GITHUB_TOKEN" in
  rate-limited) echo "403 API rate limit exceeded" >&2; exit 1 ;;
  weekly-rate-limited) echo '{"type":"session.error","data":{"errorType":"rate_limit","errorCode":"user_weekly_rate_limited","message":"You have reached your weekly rate limit"}}' >&2; exit 1 ;;
  status-429) echo "Request failed with status code 429" >&2; exit 1 ;;
  too-many-requests) echo "Too Many Requests" >&2; exit 1 ;;
  weekly-message) echo "You have reached your weekly rate limit" >&2; exit 1 ;;
  timed-out) exit 124 ;;
  unauthorized) echo "401 Unauthorized" >&2; exit 7 ;;
  unauthorized-after-effort) echo "401 Unauthorized after effort retry" >&2; exit 7 ;;
  disabled) echo "This organization has been disabled" >&2; exit 8 ;;
  service-error) echo "Unexpected internal service failure" >&2; exit 9 ;;
  model-error) echo "Model gpt-401 not found" >&2; exit 10 ;;
  healthy) exit 0 ;;
  *) echo "unexpected test token" >&2; exit 9 ;;
esac
""",
                encoding="utf-8",
            )
            fake_copilot.chmod(fake_copilot.stat().st_mode | stat.S_IXUSR)

            def shell_path(path: Path) -> str:
                if os.name != "nt":
                    return str(path)
                absolute = path.resolve()
                return f"/{absolute.drive[0].lower()}/{absolute.as_posix()[3:]}"

            env = os.environ.copy()
            env.update(
                {
                    "ATTEMPTS": shell_path(attempts),
                    "MODELS": shell_path(models),
                    "GITHUB_OUTPUT": shell_path(github_output),
                    "RUNNER_TEMP": shell_path(root),
                    "PROBE_MODEL": model,
                    "PROBE_JUDGE_MODEL": judge_model,
                    "COPILOT_RATE_LIMIT_PATTERN": rate_limit_pattern(),
                    "COPILOT_TOKEN_UNAVAILABLE_PATTERN": token_unavailable_pattern(),
                    "TOKEN_RANDOM_SEED": "1",
                }
            )
            for index in range(10):
                env[f"COPILOT_PAT_{index}"] = tokens.get(index, "")

            result = subprocess.run(
                [
                    BASH,
                    "-c",
                    f'export PATH="{shell_path(fake_bin)}:$PATH"\n{selection_script()}',
                ],
                cwd=REPO_ROOT,
                env=env,
                text=True,
                capture_output=True,
                check=False,
            )
            result.attempts = (
                attempts.read_text(encoding="utf-8").splitlines()
                if attempts.exists()
                else []
            )
            result.selected_token = (
                token_file.read_text(encoding="utf-8") if token_file.exists() else None
            )
            result.models = (
                models.read_text(encoding="utf-8").splitlines()
                if models.exists()
                else []
            )
            result.github_output = (
                github_output.read_text(encoding="utf-8").splitlines()
                if github_output.exists()
                else []
            )
            return result

    def test_rate_limited_candidate_fails_over_to_healthy_candidate(self) -> None:
        result = self.run_selector({0: "rate-limited", 1: "healthy"})

        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(result.attempts, ["rate-limited", "healthy"])
        self.assertEqual(result.selected_token, "healthy")
        self.assertEqual(result.github_output, ["selected=1"])
        self.assertIn("entry 0 is rate-limited", result.stdout)

    def test_probe_rate_limit_pattern_matches_common_wording(self) -> None:
        for limited_token in (
            "status-429",
            "too-many-requests",
            "weekly-message",
        ):
            with self.subTest(limited_token=limited_token):
                result = self.run_selector({0: limited_token, 1: "healthy"})

                self.assertEqual(result.returncode, 0, result.stderr)
                self.assertEqual(result.attempts, [limited_token, "healthy"])
                self.assertEqual(result.selected_token, "healthy")

    def test_timed_out_candidate_fails_over_to_healthy_candidate(self) -> None:
        result = self.run_selector({0: "timed-out", 1: "healthy"})

        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(result.attempts, ["timed-out", "healthy"])
        self.assertEqual(result.selected_token, "healthy")
        self.assertIn("entry 0 timed out", result.stdout)

    def test_distinct_agent_and_judge_models_are_both_probed(self) -> None:
        result = self.run_selector(
            {0: "healthy"},
            model="agent-model",
            judge_model="judge-model",
        )

        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(result.attempts, ["healthy", "healthy"])
        self.assertEqual(result.models, ["agent-model", "judge-model"])
        self.assertEqual(result.selected_token, "healthy")

    def test_model_without_effort_support_is_retried_without_effort(self) -> None:
        result = self.run_selector(
            {0: "healthy"},
            model="no-effort-model",
            judge_model="judge-model",
        )

        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(result.attempts, ["healthy", "healthy", "healthy"])
        self.assertEqual(
            result.models,
            ["no-effort-model", "no-effort-model", "judge-model"],
        )
        self.assertEqual(result.selected_token, "healthy")
        self.assertIn("retrying its availability probe without --effort", result.stdout)

    def test_model_without_effort_support_fails_over_after_one_retry(self) -> None:
        result = self.run_selector(
            {0: "unauthorized-after-effort", 1: "healthy"},
            model="no-effort-model",
            judge_model="judge-model",
        )

        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(
            result.attempts,
            [
                "unauthorized-after-effort",
                "unauthorized-after-effort",
                "healthy",
                "healthy",
                "healthy",
            ],
        )
        self.assertEqual(
            result.models,
            [
                "no-effort-model",
                "no-effort-model",
                "no-effort-model",
                "no-effort-model",
                "judge-model",
            ],
        )
        self.assertEqual(result.selected_token, "healthy")
        self.assertIn("has unusable credentials", result.stdout)
        self.assertIn("401 Unauthorized after effort retry", result.stdout)

    def test_unavailable_candidate_fails_over_to_healthy_candidate(self) -> None:
        for unavailable_token in ("unauthorized", "disabled"):
            with self.subTest(unavailable_token=unavailable_token):
                result = self.run_selector(
                    {0: unavailable_token, 1: "healthy"}
                )

                self.assertEqual(result.returncode, 0, result.stderr)
                self.assertEqual(
                    result.attempts, [unavailable_token, "healthy"]
                )
                self.assertEqual(result.selected_token, "healthy")
                self.assertIn(
                    "quarantining it and trying another entry", result.stdout
                )

    def test_unrelated_failure_does_not_try_another_candidate(self) -> None:
        for failing_token in ("service-error", "model-error"):
            with self.subTest(failing_token=failing_token):
                result = self.run_selector(
                    {0: failing_token, 1: "healthy"}
                )

                self.assertNotEqual(result.returncode, 0)
                self.assertEqual(result.attempts, [failing_token])
                self.assertIsNone(result.selected_token)
                self.assertIn(
                    "unexpected non-rate-limit error", result.stdout
                )
                self.assertIn(
                    "refusing to hide a service or configuration failure",
                    result.stdout,
                )

    def test_all_unavailable_candidates_fail_clearly(self) -> None:
        result = self.run_selector({0: "unauthorized", 1: "disabled"})

        self.assertEqual(result.returncode, 1)
        self.assertEqual(result.attempts, ["unauthorized", "disabled"])
        self.assertIsNone(result.selected_token)
        self.assertIn(
            "No healthy Copilot PAT pool entry was found", result.stdout
        )
        self.assertIn(
            "at least one configured entry was unavailable", result.stdout
        )

    def test_all_rate_limited_candidates_fail_clearly(self) -> None:
        result = self.run_selector({0: "rate-limited", 1: "weekly-rate-limited"})

        self.assertEqual(result.returncode, 1)
        self.assertEqual(result.attempts, ["rate-limited", "weekly-rate-limited"])
        self.assertIsNone(result.selected_token)
        self.assertIn("Every configured Copilot PAT pool entry is rate-limited", result.stdout)

    def test_token_unavailable_pattern_matches_credential_failures(self) -> None:
        pattern = token_unavailable_pattern()

        for message in (
            "Failed to fetch PAT user login (401): Bad credentials.",
            "Authentication token found but could not be validated.",
            "The authentication token has expired.",
            "This organization has been disabled.",
            "Copilot access was disabled by your organization.",
        ):
            with self.subTest(message=message):
                env = os.environ.copy()
                env.update({"PATTERN": pattern, "MESSAGE": message})
                result = subprocess.run(
                    [BASH, "-c", 'printf "%s\\n" "$MESSAGE" | grep -Eiq "$PATTERN"'],
                    env=env,
                    check=False,
                )
                self.assertEqual(result.returncode, 0, message)

        for message in (
            "Unexpected internal service failure",
            "Internal server error: request id req-2401 failed",
            "Upstream returned HTTP 500 after 2.401 seconds",
            "Model gpt-401 not found",
            "Processed 12401 tokens before crashing",
            "Service unavailable: token bucket refill expired",
            "Configuration error: organization policy disabled telemetry",
        ):
            with self.subTest(message=message):
                env = os.environ.copy()
                env.update({"PATTERN": pattern, "MESSAGE": message})
                result = subprocess.run(
                    [BASH, "-c", 'printf "%s\\n" "$MESSAGE" | grep -Eiq "$PATTERN"'],
                    env=env,
                    check=False,
                )
                self.assertEqual(result.returncode, 1, message)

    def test_actual_run_uses_shared_rate_limit_pattern(self) -> None:
        workflow = yaml.safe_load(WORKFLOW.read_text(encoding="utf-8"))
        steps = workflow["jobs"]["vally-evaluate"]["steps"]
        run_script = next(
            step["run"] for step in steps if step.get("name") == "Run vally evaluations"
        )
        self.assertIn(
            'grep -Eiq "$COPILOT_RATE_LIMIT_PATTERN" "$VALLY_LOG"',
            run_script,
        )
        pattern = rate_limit_pattern()

        for message in (
            "Request failed with status code 429",
            "403 API rate limit exceeded",
            "user_weekly_rate_limited",
            "Too Many Requests",
            "You have reached your weekly rate limit",
        ):
            env = os.environ.copy()
            env.update({"PATTERN": pattern, "MESSAGE": message})
            result = subprocess.run(
                [BASH, "-c", 'printf "%s\\n" "$MESSAGE" | grep -Eiq "$PATTERN"'],
                env=env,
                check=False,
            )
            self.assertEqual(result.returncode, 0, message)

        env = os.environ.copy()
        env.update({"PATTERN": pattern, "MESSAGE": "401 Unauthorized"})
        result = subprocess.run(
            [BASH, "-c", 'printf "%s\\n" "$MESSAGE" | grep -Eiq "$PATTERN"'],
            env=env,
            check=False,
        )
        self.assertEqual(result.returncode, 1)

    def test_eval_discovery_precedes_tool_install_and_token_selection(self) -> None:
        workflow = yaml.safe_load(WORKFLOW.read_text(encoding="utf-8"))
        steps = workflow["jobs"]["vally-evaluate"]["steps"]
        by_name = {step.get("name"): (index, step) for index, step in enumerate(steps)}

        find_index, _ = by_name["Find eval specs"]
        install_index, install = by_name["Install vally and Copilot CLI"]
        select_index, select = by_name[STEP_NAME]
        run_index, _ = by_name["Run vally evaluations"]

        self.assertLess(find_index, install_index)
        self.assertLess(install_index, select_index)
        self.assertLess(select_index, run_index)
        expected_condition = "steps.find-evals.outputs.has_evals == 'true'"
        self.assertEqual(install["if"], expected_condition)
        self.assertEqual(select["if"], expected_condition)
        install_script = install["run"]
        self.assertNotIn("npm install -g", install_script)
        self.assertIn(
            '--prefix "$RUNNER_TEMP/evaluation-tools"',
            install_script,
        )
        self.assertIn(
            '"$RUNNER_TEMP/trusted-validator-src/eng/evaluation-tools/package.json"',
            install_script,
        )
        self.assertIn(
            '"$RUNNER_TEMP/trusted-validator-src/eng/evaluation-tools/package-lock.json"',
            install_script,
        )
        self.assertIn("npm ci", install_script)
        self.assertNotIn("npm install", install_script)
        self.assertNotIn("@microsoft/vally-cli@", install_script)
        self.assertNotIn("@github/copilot@", install_script)
        self.assertIn(
            '"$RUNNER_TEMP/evaluation-tools/node_modules/.bin" >> "$GITHUB_PATH"',
            install_script,
        )
        self.assertIn(
            "import.meta.resolve('@github/copilot-linux-x64/sdk')",
            install_script,
        )
        for filename in ("sdk-startup.mjs", "vally.mjs"):
            self.assertIn(
                f'"$RUNNER_TEMP/trusted-validator-src/eng/evaluation-tools/{filename}"',
                install_script,
            )
        self.assertIn('ln -s ../vally.mjs "$RUNNER_TEMP/evaluation-tools/bin/vally"', install_script)
        self.assertGreater(
            install_script.index('echo "$RUNNER_TEMP/evaluation-tools/bin"'),
            install_script.index('echo "$RUNNER_TEMP/evaluation-tools/node_modules/.bin"'),
        )

    def test_evaluation_tool_manifest_has_secretless_smoke_test(self) -> None:
        workflow = yaml.safe_load(TEST_WORKFLOW.read_text(encoding="utf-8"))
        triggers = workflow.get("on", workflow.get(True))
        tool_path = "eng/evaluation-tools/**"
        for event in ("pull_request", "push"):
            self.assertEqual(triggers[event]["paths"].count(tool_path), 1)

        job = workflow["jobs"]["evaluation-tools"]
        self.assertEqual(job["runs-on"], "ubuntu-latest")
        steps = {step.get("name"): step for step in job["steps"]}
        install_script = steps["Install evaluation tools"]["run"]
        self.assertIn("--prefix eng/evaluation-tools", install_script)
        self.assertIn("npm ci", install_script)
        self.assertNotIn("npm install", install_script)
        self.assertIn("--registry https://registry.npmjs.org/", install_script)

        smoke_script = steps["Smoke test evaluation tools"]["run"]
        self.assertIn("node_modules/.bin/vally --version", smoke_script)
        self.assertIn("node vally.mjs --version", smoke_script)
        self.assertIn(
            "node --test eng/evaluation-tools/*.test.mjs",
            steps["Test SDK startup ordering without model calls"]["run"],
        )
        self.assertIn("node_modules/.bin/copilot --version", smoke_script)
        self.assertIn(
            "import.meta.resolve('@github/copilot-linux-x64/sdk')",
            smoke_script,
        )

    def test_path_safety_helper_changes_run_workflow_tests(self) -> None:
        workflow = yaml.safe_load(TEST_WORKFLOW.read_text(encoding="utf-8"))
        triggers = workflow.get("on", workflow.get(True))
        helper_path = "eng/evaluation/path-safety.ps1"
        for event in ("pull_request", "push"):
            self.assertEqual(triggers[event]["paths"].count(helper_path), 1)

    def test_manual_dispatch_does_not_execute_pr_path_safety_helper(self) -> None:
        workflow = yaml.safe_load(WORKFLOW.read_text(encoding="utf-8"))
        build_script = next(
            step["run"]
            for step in workflow["jobs"]["prepare"]["steps"]
            if step.get("id") == "build"
        )

        self.assertNotIn('eng/evaluation/path-safety.ps1', build_script)
        self.assertIn("function Test-PathHasReparsePoint", build_script)
        self.assertIn("github.workflow_sha", build_script)

    def test_path_safety_helper_rejects_linked_allowed_root(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            root = Path(temp)
            target = root / "target"
            target.mkdir()
            (target / "child.txt").write_text("content", encoding="utf-8")
            linked_root = root / "linked-root"
            create_symlink_or_skip(
                self, linked_root, target, target_is_directory=True)

            quote = lambda path: str(path).replace("'", "''")
            script = (
                f". '{quote(PATH_SAFETY_SCRIPT)}'\n"
                f"Test-PathHasReparsePoint -AllowedRoot '{quote(linked_root)}' "
                f"-Path '{quote(linked_root)}'\n"
                f"Test-PathHasReparsePoint -AllowedRoot '{quote(linked_root)}' "
                f"-Path '{quote(linked_root / 'child.txt')}'\n"
            )
            result = subprocess.run(
                ["pwsh", "-NoLogo", "-NoProfile", "-NonInteractive", "-Command", script],
                capture_output=True,
                text=True,
                timeout=30,
            )

            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
            self.assertEqual(result.stdout.strip().splitlines(), ["True", "True"])

    def test_path_safety_helper_preserves_filesystem_root(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            path = Path(temp)
            root = Path(path.anchor)
            quote = lambda value: str(value).replace("'", "''")
            script = (
                f". '{quote(PATH_SAFETY_SCRIPT)}'\n"
                f"Test-PathHasReparsePoint -AllowedRoot '{quote(root)}' "
                f"-Path '{quote(path)}'\n"
            )
            result = subprocess.run(
                ["pwsh", "-NoLogo", "-NoProfile", "-NonInteractive", "-Command", script],
                capture_output=True,
                text=True,
                timeout=30,
            )

            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
            self.assertEqual(result.stdout.strip(), "False")

    def test_adapter_fault_injection_runs_in_pr_ci(self) -> None:
        workflow = yaml.safe_load(TEST_WORKFLOW.read_text(encoding="utf-8"))
        triggers = workflow.get("on", workflow.get(True))
        adapter_path = "eng/vally-adapter/**"
        for event in ("pull_request", "push"):
            self.assertEqual(triggers[event]["paths"].count(adapter_path), 1)

        job = workflow["jobs"]["vally-adapter"]
        self.assertEqual(job["runs-on"], "ubuntu-latest")
        steps = {step.get("name"): step for step in job["steps"]}
        self.assertIn(
            "node --test eng/vally-adapter/*.test.mjs",
            steps["Run adapter fault-injection and report tests"]["run"],
        )

    def test_manual_eval_data_publish_is_explicit_and_main_only(self) -> None:
        workflow = yaml.safe_load(CALLER_WORKFLOW.read_text(encoding="utf-8"))
        triggers = workflow.get("on", workflow.get(True))
        publish_input = triggers["workflow_dispatch"]["inputs"]["publish_eval_data"]

        self.assertEqual(publish_input["type"], "boolean")
        self.assertFalse(publish_input["default"])

        publish_job = workflow["jobs"]["publish-eval-data"]
        self.assertIn("evaluate", publish_job["needs"])
        publish_condition = publish_job["if"]
        self.assertIn("github.event_name == 'schedule'", publish_condition)
        self.assertIn(
            "github.event_name == 'workflow_dispatch'",
            publish_condition,
        )
        self.assertIn("inputs.publish_eval_data", publish_condition)
        self.assertIn("inputs.pr_number == ''", publish_condition)
        self.assertIn("github.repository == 'dotnet/skills'", publish_condition)
        self.assertIn("github.ref == 'refs/heads/main'", publish_condition)
        self.assertIn("needs.evaluate.result == 'success'", publish_condition)

        deploy_job = workflow["jobs"]["deploy-dashboard"]
        self.assertIn("publish-eval-data", deploy_job["needs"])
        deploy_condition = deploy_job["if"]
        self.assertIn("inputs.pr_number == ''", deploy_condition)
        self.assertIn("github.repository == 'dotnet/skills'", deploy_condition)
        self.assertIn("github.ref == 'refs/heads/main'", deploy_condition)
        normalized_deploy_condition = " ".join(deploy_condition.split())
        self.assertEqual(
            deploy_condition.count("github.repository == 'dotnet/skills'"),
            1,
        )
        self.assertIn(
            "github.event_name == 'workflow_dispatch' && "
            "inputs.pr_number == '' && github.ref == 'refs/heads/main' && "
            "( !inputs.publish_eval_data",
            normalized_deploy_condition,
        )
        self.assertIn(
            "( !inputs.publish_eval_data || "
            "( github.repository == 'dotnet/skills' && "
            "needs.publish-eval-data.result == 'success' ) )",
            normalized_deploy_condition,
        )

    def test_pr_report_binds_identity_and_reruns_to_exact_commit(self) -> None:
        workflow = yaml.safe_load(CALLER_WORKFLOW.read_text(encoding="utf-8"))
        comment_job = workflow["jobs"]["comment-on-pr"]
        steps = {
            step.get("name"): step
            for step in comment_job["steps"]
        }
        script = steps["Consolidate and post results"]["run"]

        self.assertEqual(
            script.count(
                '--commit "${{ needs.gate.outputs.head_sha }}"'
            ),
            2,
        )
        self.assertIn(
            "To investigate non-passing or warning results",
            script,
        )
        self.assertIn(
            "comment `/evaluate %s` to retry this exact commit",
            script,
        )
        self.assertNotIn("re-post `/evaluate`", script)

    def test_partial_matrix_results_never_become_complete_verdicts(self) -> None:
        caller = yaml.safe_load(CALLER_WORKFLOW.read_text(encoding="utf-8"))
        comment_job = caller["jobs"]["comment-on-pr"]
        self.assertNotIn(
            "needs.evaluate.result != 'cancelled'",
            comment_job["if"],
        )

        comment_steps = {
            step.get("name"): step for step in comment_job["steps"]
        }
        consolidate_step = comment_steps["Consolidate and post results"]
        self.assertEqual(consolidate_step["if"], "always()")
        self.assertEqual(
            consolidate_step["env"]["EXPECTED_ENTRIES"],
            "${{ needs.discover.outputs.entries }}",
        )
        script = consolidate_step["run"]
        incomplete_guard = (
            'if [[ "$MATRIX_MANIFEST_VALID" != "true" '
            '|| "$EVALUATE_RESULT" != "success" '
            '|| "$OBSERVED_LEG_COUNT" -ne "$EXPECTED_LEG_COUNT" ]]'
        )
        guard_index = script.index(incomplete_guard)
        consolidation_index = script.index(
            "node eng/vally-adapter/consolidate.mjs"
        )
        self.assertLess(guard_index, consolidation_index)
        self.assertIn(
            "were preserved for diagnosis but were not consolidated",
            script[guard_index:consolidation_index],
        )
        self.assertIn(
            "exit 0",
            script[guard_index:consolidation_index],
        )
        self.assertIn(
            "find all-results/ -name adapter-summary.json",
            script[:guard_index],
        )
        self.assertIn(
            "if ! EXPECTED_LEG_COUNT=$(printf",
            script[:guard_index],
        )
        self.assertIn(
            "the discovered entry list was missing, malformed, or not a JSON array",
            script[guard_index:consolidation_index],
        )
        self.assertIn(
            "expected %s matrix leg artifact(s), but found %s",
            script[guard_index:consolidation_index],
        )

        discover_script = next(
            step["run"]
            for step in caller["jobs"]["discover"]["steps"]
            if "function Get-PluginShardEntries" in step.get("run", "")
        )
        self.assertIn(
            'if (-not (Test-Path $evalPath)) { continue }',
            discover_script,
        )
        self.assertIn(
            'if ($shardGroups.Count -eq 0) { return @() }',
            discover_script,
        )

        runner = yaml.safe_load(WORKFLOW.read_text(encoding="utf-8"))
        runner_steps = {
            step.get("name"): step
            for step in runner["jobs"]["vally-evaluate"]["steps"]
        }
        self.assertEqual(
            runner_steps["Upload results"]["with"]["if-no-files-found"],
            "error",
        )

    def test_fork_checkout_is_blocked_and_adapter_code_is_trusted(self) -> None:
        workflow = yaml.safe_load(WORKFLOW.read_text(encoding="utf-8"))
        steps = workflow["jobs"]["vally-evaluate"]["steps"]
        by_name = {step.get("name"): step for step in steps}

        checkout = by_name["Checkout skills content"]
        self.assertNotIn("allow-unsafe-pr-checkout", checkout["with"])

        caller = yaml.safe_load(CALLER_WORKFLOW.read_text(encoding="utf-8"))
        for job_name in ("evaluate", "publish-token-data", "publish-session-data"):
            condition = caller["jobs"][job_name]["if"]
            self.assertIn(
                "needs.gate.outputs.is_fork != 'true'",
                condition,
                f"{job_name} must not run for fork PR content",
            )
        self.assertIn(
            "inputs.pr_number == ''",
            caller["jobs"]["deploy-dashboard"]["if"],
        )

        download = by_name["Download trusted skill-validator archive"]
        self.assertTrue(download["uses"].startswith("actions/download-artifact@"))
        self.assertEqual(
            download["with"]["name"],
            "trusted-skill-validator-${{ github.run_id }}",
        )
        self.assertEqual(
            download["with"]["path"],
            "${{ runner.temp }}/trusted-validator-archive",
        )
        self.assertFalse(
            any(
                step.get("uses", "").startswith(
                    ("actions/cache", "actions/setup-dotnet")
                )
                for step in steps
            )
        )
        self.assertFalse(
            any("dotnet publish" in step.get("run", "") for step in steps)
        )
        producer_steps = workflow["jobs"]["prepare-validator"]["steps"]
        producer_by_name = {step.get("name"): step for step in producer_steps}
        producer_restore = producer_by_name["Restore skill-validator archive"]
        producer_save = producer_by_name["Save skill-validator archive"]
        producer_upload = producer_by_name["Upload trusted skill-validator archive"]
        self.assertTrue(producer_save["uses"].startswith("actions/cache/save@"))
        self.assertIn(
            "github.event_name != 'issue_comment'",
            producer_save["if"],
        )
        self.assertEqual(
            producer_restore["with"]["key"],
            "${{ steps.cache-key.outputs.key }}",
        )
        self.assertTrue(
            producer_upload["uses"].startswith("actions/upload-artifact@")
        )
        self.assertEqual(
            producer_upload["with"]["name"],
            download["with"]["name"],
        )
        self.assertEqual(
            producer_upload["with"]["path"],
            "skill-validator-dist.tar.gz",
        )
        self.assertEqual(
            producer_upload["with"]["if-no-files-found"],
            "error",
        )
        cache_key_script = producer_by_name["Resolve trusted cache key"]["run"]
        self.assertIn(
            "trusted-skill-validator-v1-",
            cache_key_script,
        )
        self.assertIn(
            "needs.prepare-validator.result == 'success'",
            workflow["jobs"]["vally-evaluate"]["if"],
        )

        stage_script = by_name["Stage trusted evaluation tooling"]["run"]
        self.assertIn(
            'cp -a "$GITHUB_WORKSPACE/_trusted-validator-src" '
            '"$RUNNER_TEMP/trusted-validator-src"',
            stage_script,
        )

        extract_script = by_name["Extract skill-validator"]["run"]
        self.assertIn(
            '"$RUNNER_TEMP/trusted-validator-archive/skill-validator-dist.tar.gz"',
            extract_script,
        )

        run_script = by_name["Run vally evaluations"]["run"]
        self.assertIn(
            '[ ! -r "$RUNNER_TEMP/evaluation-copilot-token" ]',
            run_script,
        )
        self.assertIn(
            'echo "::error::No experiment output produced for $PLUGIN"',
            run_script,
        )
        self.assertIn(
            'The result set is incomplete or contains an unexpected eval.',
            run_script,
        )
        self.assertEqual(
            run_script.count(
                '--expected-evals "$RUNNER_TEMP/evaluation-expected-evals.txt"'
            ),
            3,
        )
        self.assertIn(
            'if [ "$PRODUCED" -ne "$EXPECTED_EVAL_COUNT" ]',
            run_script,
        )
        self.assertIn("s.expectedManifestProvided === true", run_script)
        self.assertIn("s.unexpectedEvalCount === 0", run_script)
        self.assertIn("s.measurementInvalidEvalCount === 0", run_script)
        self.assertNotIn("s.invalidEvalCount === 0", run_script)
        self.assertIn(
            "Vally comparison watchdog expired after 60 minutes",
            run_script,
        )
        self.assertIn("timeout --signal=TERM --kill-after=30s 60m", run_script)
        self.assertIn(
            "retry-executor-timeouts.mjs",
            run_script,
        )
        self.assertIn(
            '--max-groups 3',
            run_script,
        )
        self.assertIn(
            'EXECUTOR_RETRY_STATUS=$?',
            run_script,
        )
        self.assertIn(
            'if [ "$EXECUTOR_RETRY_STATUS" -ne 0 ]',
            run_script,
        )
        self.assertLess(
            run_script.index("retry-executor-timeouts.mjs"),
            run_script.index(
                'node "$RUNNER_TEMP/trusted-validator-src/'
                'eng/vally-adapter/adapt.mjs"'
            ),
        )
        summary_script = by_name["Write summary"]["run"]
        self.assertIn('ICON="➖"', summary_script)
        self.assertNotIn('ICON="❌"', summary_script)
        self.assertNotIn(
            "Vally comparison watchdog expired after 45 minutes",
            run_script,
        )
        find_script = by_name["Find eval specs"]["run"]
        self.assertIn(
            'printf \'%s\\n\' "$EVALS" > "$RUNNER_TEMP/evaluation-expected-evals.txt"',
            find_script,
        )
        self.assertIn('echo "count=$EVAL_COUNT" >> "$GITHUB_OUTPUT"', find_script)
        self.assertIn(
            'grep -Eiq "$COPILOT_RATE_LIMIT_PATTERN" "$VALLY_LOG"',
            run_script,
        )
        self.assertIn('"$results_file" >/dev/null', run_script)
        self.assertIn('find "$EXPERIMENT_OUT" -name results.jsonl', run_script)
        self.assertIn(
            'echo "::error::Selected Copilot PAT became rate-limited during evaluation;',
            run_script,
        )
        self.assertIn('rm -rf "$EXPERIMENT_OUT"', run_script)
        trusted_adapter = '"$RUNNER_TEMP/trusted-validator-src/eng/vally-adapter/'
        self.assertIn(f"node {trusted_adapter}gen-experiment.mjs", run_script)
        self.assertIn(f"node {trusted_adapter}adapt.mjs", run_script)
        self.assertIn(f"node {trusted_adapter}adapt-agent-results.mjs", run_script)
        self.assertIn('"$RUNNER_TEMP/trusted-validator/skill-validator" evaluate', run_script)
        self.assertIn('rm -f "${AGENT_RESULTS[0]}"', run_script)
        self.assertGreater(
            run_script.index('rm -f "${AGENT_RESULTS[0]}"'),
            run_script.index(f"node {trusted_adapter}adapt-agent-results.mjs"),
        )
        self.assertNotIn("node eng/vally-adapter/", run_script)

    def test_discovery_creates_first_class_agent_matrix_entries(self) -> None:
        caller = yaml.safe_load(CALLER_WORKFLOW.read_text(encoding="utf-8"))
        discover_script = next(
            step["run"]
            for step in caller["jobs"]["discover"]["steps"]
            if "function Get-PluginAgentEntries" in step.get("run", "")
        )
        self.assertIn('target_kind = "agent"', discover_script)
        self.assertIn("$manifest.agents", discover_script)
        self.assertIn("Resolve-AgentEvalPath", discover_script)
        self.assertIn("agents_path = $agentPath", discover_script)
        self.assertIn("eval_path = $evalPath", discover_script)
        self.assertIn("^plugins/([^/]+)/(?:[^/]+/)*[^/]+\\.agent\\.md$", discover_script)
        self.assertIn("$changedAgentSourcePlugins", discover_script)
        self.assertIn("every agent eval in an affected plugin", discover_script)

        runner = yaml.safe_load(WORKFLOW.read_text(encoding="utf-8"))
        steps = {step.get("name"): step for step in runner["jobs"]["vally-evaluate"]["steps"]}
        validate = steps["Validate matrix entry"]["run"]
        self.assertIn("ENTRY_TARGET_KIND", steps["Validate matrix entry"]["env"])
        self.assertIn("ENTRY_EVAL_PATH", steps["Validate matrix entry"]["env"])
        self.assertIn("agent_path_re=", validate)
        self.assertIn("eval_path_re=", validate)
        self.assertIn('Agent matrix entry has an empty agents_path', validate)
        self.assertIn('Agent matrix entry has an empty eval_path', validate)

        find = steps["Find eval specs"]["run"]
        self.assertIn('if [ "$TARGET_KIND" = "agent" ]', find)
        self.assertIn('EVALS="$EVAL_PATH"', find)

        run = steps["Run vally evaluations"]["run"]
        self.assertIn('if [ "$TARGET_KIND" = "agent" ]', run)
        self.assertIn("--verdict-warn-only", run)
        self.assertIn("--keep-sessions", run)

        with tempfile.TemporaryDirectory() as temp:
            root = Path(temp)
            (root / "plugins" / "demo" / "skills" / "skill-a").mkdir(parents=True)
            (root / "plugins" / "demo" / "custom-agents").mkdir(parents=True)
            (root / "tests" / "demo" / "skill-a").mkdir(parents=True)
            (root / "tests" / "demo" / "nested" / "agent.router").mkdir(parents=True)
            (root / "plugins" / "demo" / "skills" / "skill-a" / "SKILL.md").write_text(
                "# Skill", encoding="utf-8")
            (root / "plugins" / "demo" / "custom-agents" / "router.agent.md").write_text(
                "---\nname: router\ndescription: Routes.\n---\nRoute.", encoding="utf-8")
            (root / "plugins" / "demo" / "plugin.json").write_text(
                json.dumps({
                    "name": "demo",
                    "version": "1.0.0",
                    "description": "Demo",
                    "skills": ["./skills/"],
                    "agents": ["./custom-agents/router.agent.md"],
                }),
                encoding="utf-8",
            )
            (root / "tests" / "demo" / "skill-a" / "eval.yaml").write_text(
                "name: skill-a\nstimuli: []\n", encoding="utf-8")
            (root / "tests" / "demo" / "nested" / "agent.router" / "eval.yaml").write_text(
                "name: agent.router\nstimuli: []\n", encoding="utf-8")

            start = discover_script.index("function Get-PluginShardEntries")
            end = discover_script.index(
                'if ("${{ needs.gate.outputs.pr_number }}"', start)
            functions = discover_script[start:end]
            script = (
                "$ErrorActionPreference = 'Stop'\n"
                + f". '{str(PATH_SAFETY_SCRIPT).replace(chr(39), chr(39) * 2)}'\n"
                + functions
                + f"\n$root = '{str(root).replace(chr(39), chr(39) * 2)}'\n"
                + "$entries = @(\n"
                + "  Get-PluginShardEntries -plugin demo -contentRoot $root\n"
                + "  Get-PluginAgentEntries -plugin demo -contentRoot $root\n"
                + ")\n"
                + "ConvertTo-Json -InputObject @($entries) -Compress\n"
            )
            result = subprocess.run(
                ["pwsh", "-NoLogo", "-NoProfile", "-NonInteractive", "-Command", script],
                capture_output=True, text=True, timeout=30,
            )
            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
            entries = json.loads(result.stdout.strip().splitlines()[-1])
            self.assertEqual(
                {(entry["target_kind"], entry["name"]) for entry in entries},
                {("skill", "demo"), ("agent", "demo--agent.router")},
            )
            agent_entry = next(entry for entry in entries if entry["target_kind"] == "agent")
            self.assertEqual(
                agent_entry["agents_path"],
                "plugins/demo/custom-agents/router.agent.md",
            )
            self.assertEqual(
                agent_entry["eval_path"],
                "tests/demo/nested/agent.router/eval.yaml",
            )

            outside_agent = root / "outside.agent.md"
            outside_agent.write_text(
                "---\nname: router\ndescription: External.\n---\nExternal.",
                encoding="utf-8",
            )
            (root / "plugins" / "demo" / "custom-agents" / "router.agent.md").unlink()
            create_symlink_or_skip(
                self,
                root / "plugins" / "demo" / "custom-agents" / "router.agent.md",
                outside_agent,
            )
            unsafe_result = subprocess.run(
                ["pwsh", "-NoLogo", "-NoProfile", "-NonInteractive", "-Command", script],
                capture_output=True,
                text=True,
                timeout=30,
            )
            self.assertNotEqual(
                unsafe_result.returncode,
                0,
                unsafe_result.stdout + unsafe_result.stderr,
            )

    def test_manual_agent_dispatch_resolves_manifest_paths(self) -> None:
        workflow = yaml.safe_load(WORKFLOW.read_text(encoding="utf-8"))
        prepare = workflow["jobs"]["prepare"]
        steps = {step.get("name", step.get("id")): step for step in prepare["steps"]}
        self.assertIn("Checkout evaluation content", steps)
        build_script = steps["build"]["run"]

        with tempfile.TemporaryDirectory() as temp:
            root = Path(temp)
            agent_dir = root / "plugins" / "demo" / "custom-agents"
            eval_dir = root / "tests" / "demo" / "nested" / "agent.router"
            agent_dir.mkdir(parents=True)
            eval_dir.mkdir(parents=True)
            (root / "plugins" / "demo" / "plugin.json").write_text(
                json.dumps({
                    "name": "demo",
                    "version": "1.0.0",
                    "description": "Demo",
                    "agents": ["./custom-agents/router.agent.md"],
                }),
                encoding="utf-8",
            )
            (agent_dir / "router.agent.md").write_text(
                "---\nname: router\ndescription: Routes.\n---\nRoute.",
                encoding="utf-8",
            )
            (eval_dir / "eval.yaml").write_text(
                "name: agent.router\nstimuli: []\n",
                encoding="utf-8",
            )
            path_safety_dir = root / "eng" / "evaluation"
            path_safety_dir.mkdir(parents=True)
            shutil.copy2(PATH_SAFETY_SCRIPT, path_safety_dir / PATH_SAFETY_SCRIPT.name)
            output_file = root / "github-output.txt"
            env = dict(
                os.environ,
                PLUGIN="demo",
                SKILL="agent.router",
                GITHUB_OUTPUT=str(output_file),
            )
            result = subprocess.run(
                ["pwsh", "-NoLogo", "-NoProfile", "-NonInteractive", "-Command", build_script],
                cwd=root,
                env=env,
                capture_output=True,
                text=True,
                timeout=30,
            )
            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
            output_line = output_file.read_text(encoding="utf-8").strip()
            entries = json.loads(output_line.removeprefix("entries="))
            self.assertEqual(entries[0]["agents_path"], "plugins/demo/custom-agents/router.agent.md")
            self.assertEqual(entries[0]["eval_path"], "tests/demo/nested/agent.router/eval.yaml")

            outside_agent = root / "outside.agent.md"
            outside_agent.write_text(
                "---\nname: router\ndescription: External.\n---\nExternal.",
                encoding="utf-8",
            )
            (agent_dir / "router.agent.md").unlink()
            create_symlink_or_skip(
                self, agent_dir / "router.agent.md", outside_agent)
            output_file.unlink()
            unsafe_result = subprocess.run(
                ["pwsh", "-NoLogo", "-NoProfile", "-NonInteractive", "-Command", build_script],
                cwd=root,
                env=env,
                capture_output=True,
                text=True,
                timeout=30,
            )
            self.assertNotEqual(
                unsafe_result.returncode,
                0,
                unsafe_result.stdout + unsafe_result.stderr,
            )

    def test_all_pr_discovery_gates_match_direct_agent_sources(self) -> None:
        caller = yaml.safe_load(CALLER_WORKFLOW.read_text(encoding="utf-8"))
        discovery_scripts = {
            job_name: next(
                step["run"]
                for step in caller["jobs"][job_name]["steps"]
                if "$hasSkillChanges = $changedFiles" in step.get("run", "")
            )
            for job_name in ("pr-status", "fork-pr-status", "discover")
        }
        changed_files = [
            "plugins/dotnet-test/plugin.json",
            "plugins/dotnet-test/agents/test-quality-auditor.agent.md",
            "plugins/dotnet-test/custom-agents/helper.agent.md",
            "plugins/dotnet-test/skills/test-smell-detection/SKILL.md",
            "tests/dotnet-test/agent.test-quality-auditor/eval.yaml",
            "tests/dotnet-test/test-smell-detection/eval.yaml",
            "plugins/dotnet-test/README.md",
        ]
        expected = changed_files[:6]

        for job_name, script in discovery_scripts.items():
            with self.subTest(job=job_name):
                match = re.search(
                    r"\$hasSkillChanges = \$changedFiles \|\s*"
                    r"Where-Object \{ \$_ -match '([^']+)' \}",
                    script,
                )
                self.assertIsNotNone(match)
                env = dict(os.environ, DISCOVERY_PATTERN=match.group(1))
                powershell = (
                    "$changedFiles = @("
                    + ",".join(
                        f"'{path.replace(chr(39), chr(39) * 2)}'"
                        for path in changed_files
                    )
                    + "); "
                    "$matches = @($changedFiles | "
                    "Where-Object { $_ -match $env:DISCOVERY_PATTERN }); "
                    "ConvertTo-Json -InputObject $matches -Compress"
                )
                result = subprocess.run(
                    [
                        "pwsh",
                        "-NoLogo",
                        "-NoProfile",
                        "-NonInteractive",
                        "-Command",
                        powershell,
                    ],
                    env=env,
                    capture_output=True,
                    text=True,
                    timeout=30,
                )
                self.assertEqual(
                    result.returncode,
                    0,
                    result.stdout + result.stderr,
                )
                self.assertEqual(json.loads(result.stdout.strip()), expected)

        matrix_script = discovery_scripts["discover"]
        self.assertIn("$changedManifestPlugins", matrix_script)
        self.assertIn(
            "$changedAgentSourcePlugins + $changedSkillSourcePlugins + $changedManifestPlugins + $changedTestPlugins",
            matrix_script,
        )
        self.assertIn(
            "every agent eval in an affected plugin",
            matrix_script,
        )

    def test_manual_whole_plugin_dispatch_includes_agent_entries(self) -> None:
        workflow = yaml.safe_load(WORKFLOW.read_text(encoding="utf-8"))
        build_script = next(
            step["run"]
            for step in workflow["jobs"]["prepare"]["steps"]
            if step.get("id") == "build"
        )

        with tempfile.TemporaryDirectory() as temp:
            root = Path(temp)
            agent_dir = root / "plugins" / "demo" / "custom-agents"
            eval_dir = root / "tests" / "demo" / "agent.router"
            agent_dir.mkdir(parents=True)
            eval_dir.mkdir(parents=True)
            (root / "plugins" / "demo" / "plugin.json").write_text(
                json.dumps({
                    "name": "demo",
                    "version": "1.0.0",
                    "description": "Demo",
                    "agents": ["./custom-agents/"],
                }),
                encoding="utf-8",
            )
            (agent_dir / "router.agent.md").write_text(
                "---\nname: router\ndescription: Routes.\n---\nRoute.",
                encoding="utf-8",
            )
            (eval_dir / "eval.yaml").write_text(
                "name: agent.router\nstimuli: []\n",
                encoding="utf-8",
            )
            path_safety_dir = root / "eng" / "evaluation"
            path_safety_dir.mkdir(parents=True)
            shutil.copy2(PATH_SAFETY_SCRIPT, path_safety_dir / PATH_SAFETY_SCRIPT.name)
            output_file = root / "github-output.txt"
            env = dict(
                os.environ,
                PLUGIN="demo",
                SKILL="",
                GITHUB_OUTPUT=str(output_file),
            )

            result = subprocess.run(
                ["pwsh", "-NoLogo", "-NoProfile", "-NonInteractive", "-Command", build_script],
                cwd=root,
                env=env,
                capture_output=True,
                text=True,
                timeout=30,
            )

            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
            entries = json.loads(
                output_file.read_text(encoding="utf-8").strip().removeprefix("entries=")
            )
            self.assertEqual(
                {(entry["target_kind"], entry["name"]) for entry in entries},
                {("skill", "demo"), ("agent", "demo--agent.router")},
            )

    def test_dashboard_preserves_agent_identity_and_delegation(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            root = Path(temp)
            results = root / "results.json"
            output = root / "out"
            results.write_text(json.dumps({
                "schemaVersion": 5,
                "model": "executor",
                "judgeModel": "judge",
                "evalFile": "tests/demo/nested/agent.router/eval.yaml",
                "verdicts": [{
                    "skillName": "agent.router",
                    "skillPath": "plugins/demo/custom-agents/router.agent.md",
                    "skillKind": "agent",
                    "state": "VALID_PASS",
                    "passed": True,
                    "reason": "credible preference improvement",
                    "signTest": {
                        "wins": 5, "ties": 0, "losses": 0,
                        "discordant": 5, "direction": "better",
                        "pValue": 0.03125, "alpha": 0.05,
                    },
                    "netWin": 1,
                    "practicalSignificance": {"minimum": 0.2},
                    "scenarios": [{
                        "scenarioName": "routes work",
                        "expectActivation": True,
                        "preferenceGateEligible": True,
                        "agentActivationIsolated": {
                            "activated": True,
                            "invokedAgents": ["router", "helper"],
                            "delegatedAgents": ["helper"],
                        },
                        "agentActivationPlugin": {
                            "activated": True,
                            "invokedAgents": ["router", "helper"],
                            "delegatedAgents": ["helper"],
                        },
                        "skillActivationIsolated": {
                            "activated": False,
                            "detectedSkills": ["routing-skill"],
                        },
                        "baseline": {
                            "judgeResult": {"overallScore": 2},
                            "metrics": {"wallTimeMs": 100, "tokenEstimate": 20},
                        },
                        "skilledIsolated": {
                            "judgeResult": {"overallScore": 4},
                            "metrics": {
                                "wallTimeMs": 200,
                                "tokenEstimate": 30,
                                "taskCompleted": True,
                                "toolCallBreakdown": {"skill": 1},
                            },
                        },
                        "skilledPlugin": {
                            "judgeResult": {"overallScore": 4},
                            "metrics": {
                                "wallTimeMs": 220,
                                "tokenEstimate": 35,
                                "taskCompleted": True,
                                "toolCallBreakdown": {"skill": 1, "agent": 1},
                            },
                        },
                        "trials": [{
                            "winner": "treatment",
                            "errored": False,
                            "baselinePassed": False,
                            "treatmentPassed": True,
                            "evidence": "The agent routed correctly.",
                        }],
                    }],
                }],
            }), encoding="utf-8")

            result = subprocess.run([
                "pwsh", "-NoLogo", "-NoProfile", "-NonInteractive",
                "-File", str(DASHBOARD_GENERATOR),
                "-ResultsFile", str(results),
                "-PluginName", "demo",
                "-OutputDir", str(output),
                "-CommitJson", json.dumps({
                    "id": "abcdef1234567890",
                    "url": "https://github.com/dotnet/skills/commit/abcdef1234567890",
                }),
            ], capture_output=True, text=True, timeout=30)

            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
            dashboard = json.loads((output / "demo.json").read_text(encoding="utf-8-sig"))
            evidence = dashboard["entries"]["Quality"][-1]["verdictEvidence"][0]
            self.assertEqual(evidence["skillKind"], "agent")
            scenario = evidence["activationScenarios"][0]
            self.assertEqual(scenario["isolated"], "activated")
            self.assertEqual(scenario["delegatedAgents"], ["helper"])
            self.assertEqual(scenario["invokedSkills"], ["routing-skill"])
            self.assertEqual(scenario["isolatedTools"], ["skill"])
            self.assertTrue(scenario["isolatedCompleted"])
            skill_value = dashboard["entries"]["SkillValue"][-1]["skills"][0]
            self.assertEqual(skill_value["activationExpected"], 1)
            self.assertEqual(skill_value["activationFired"], 1)
            agent_link = next(
                link for link in evidence["links"] if link["label"] == "Agent source"
            )
            self.assertIn(
                "/plugins/demo/custom-agents/router.agent.md",
                agent_link["url"],
            )
            eval_link = next(
                link for link in evidence["links"] if link["label"] == "Eval source"
            )
            self.assertIn(
                "/tests/demo/nested/agent.router/eval.yaml",
                eval_link["url"],
            )

    def test_dashboard_agent_evidence_allows_missing_plugin_role(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            root = Path(temp)
            results = root / "results.json"
            output = root / "out"
            results.write_text(json.dumps({
                "schemaVersion": 5,
                "model": "executor",
                "judgeModel": "judge",
                "verdicts": [{
                    "skillName": "agent.router",
                    "skillKind": "agent",
                    "state": "INVALID_INCONCLUSIVE",
                    "passed": False,
                    "reason": "plugin evidence missing",
                    "scenarios": [{
                        "scenarioName": "routes work",
                        "expectActivation": True,
                        "agentActivationIsolated": {
                            "activated": True,
                            "invokedAgents": None,
                            "delegatedAgents": None,
                        },
                        "skillActivationIsolated": {
                            "activated": False,
                            "detectedSkills": None,
                        },
                        "baseline": {
                            "judgeResult": {"overallScore": 2},
                            "metrics": {"wallTimeMs": 100, "tokenEstimate": 20},
                        },
                        "skilledIsolated": {
                            "judgeResult": {"overallScore": 4},
                            "metrics": {
                                "wallTimeMs": 200,
                                "tokenEstimate": 30,
                                "taskCompleted": True,
                                "toolCallBreakdown": {"skill": 1},
                            },
                        },
                    }],
                }],
            }), encoding="utf-8")

            result = subprocess.run([
                "pwsh", "-NoLogo", "-NoProfile", "-NonInteractive",
                "-File", str(DASHBOARD_GENERATOR),
                "-ResultsFile", str(results),
                "-PluginName", "demo",
                "-OutputDir", str(output),
            ], capture_output=True, text=True, timeout=30)

            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
            dashboard = json.loads((output / "demo.json").read_text(encoding="utf-8-sig"))
            evidence = dashboard["entries"]["Quality"][-1]["verdictEvidence"][0]
            scenario = evidence["activationScenarios"][0]
            self.assertEqual(scenario["invokedAgents"], [])
            self.assertEqual(scenario["delegatedAgents"], [])
            self.assertEqual(scenario["invokedSkills"], [])
            self.assertEqual(scenario["pluginTools"], [])
            self.assertIsNone(scenario["pluginCompleted"])

    def test_result_consumers_use_explicit_verdict_states(self) -> None:
        workflow = yaml.safe_load(WORKFLOW.read_text(encoding="utf-8"))
        steps = workflow["jobs"]["vally-evaluate"]["steps"]
        summary_script = next(
            step["run"] for step in steps if step.get("name") == "Write summary"
        )
        self.assertIn("INVALID_INCONCLUSIVE", summary_script)
        self.assertIn("VALID_REGRESSION", summary_script)
        self.assertIn("PREFERENCE_REGRESSED", summary_script)
        self.assertNotIn("v.regressed ? 'VALID_REGRESSION'", summary_script)
        self.assertIn("v.state == null", summary_script)

        caller_text = CALLER_WORKFLOW.read_text(encoding="utf-8")
        self.assertIn("primaryState = $p.state", caller_text)
        self.assertIn(
            "$p.preferenceRegressed -eq $s.preferenceRegressed",
            caller_text,
        )


if __name__ == "__main__":
    unittest.main()