muse-agent-example/harness/evals/skill_eval.py

#!/usr/bin/env python3
"""Skill 行为评测引擎：由 harness 从外部驱动，Skill 不能自己宣布通过。

引擎把 SKILL.md 文本和场景任务交给 adapter（Agent/模型执行层），接收结构化
观察，再按场景登记的期望裁决。真实模型 adapter 需要显式授权；未授权时以
稳定码 EVAL_ADAPTER_UNAVAILABLE 失败关闭。fake adapter 只用于评测引擎自身
管道验证，不构成 Skill 行为证据。
"""
from __future__ import annotations

import hashlib
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

REPORT_SCHEMA = "skill-behavior-eval-report-v1"
SCENARIO_SCHEMA = "skill-behavior-eval-scenarios-v1"
CATEGORIES = (
    "positive_trigger",
    "negative_trigger",
    "input_missing_or_out_of_scope",
    "output_contract_and_fail_closed",
    "forbidden_action",
    "stability_and_confounders",
)


class EvalContractError(ValueError):
    """评测合同错误：场景结构非法或观察缺项。"""


class EvalAdapterUnavailable(RuntimeError):
    """真实模型执行层未授权或不可用；稳定码供 runner 与报告引用。"""

    code = "EVAL_ADAPTER_UNAVAILABLE"


def skill_md_fingerprint(skill_dir: Path) -> str:
    text = (skill_dir / "SKILL.md").read_text(encoding="utf-8")
    return "sha256:" + hashlib.sha256(text.encode("utf-8")).hexdigest()


def load_scenarios(path: Path) -> dict:
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise EvalContractError(f"场景文件不可读或不合法: {path} ({exc})") from exc
    if not isinstance(data, dict) or data.get("schema_version") != SCENARIO_SCHEMA:
        raise EvalContractError(f"场景文件 schema_version 必须是 {SCENARIO_SCHEMA}: {path}")
    scenarios = data.get("scenarios")
    if not isinstance(scenarios, list) or not scenarios:
        raise EvalContractError(f"场景文件必须带非空 scenarios 数组: {path}")
    seen = set()
    for scenario in scenarios:
        sid = scenario.get("id")
        if not sid or sid in seen:
            raise EvalContractError(f"场景 id 缺失或重复: {sid!r} ({path})")
        seen.add(sid)
        if scenario.get("category") not in CATEGORIES:
            raise EvalContractError(f"场景 {sid} 的 category 非法: {scenario.get('category')!r}")
        if not str(scenario.get("task") or "").strip():
            raise EvalContractError(f"场景 {sid} 缺 task")
        if not isinstance(scenario.get("expectations"), dict) or not scenario["expectations"]:
            raise EvalContractError(f"场景 {sid} 缺 expectations")
    return data


class FakeAdapter:
    """脚本化观察适配器：只验证评测管道，不产生 Skill 行为证据。"""

    def __init__(self, scripted: dict[str, dict]):
        self.scripted = dict(scripted)

    def run(self, skill_md_text: str, scenario: dict) -> dict:
        observation = self.scripted.get(scenario["id"])
        if observation is None:
            raise EvalContractError(f"fake adapter 没有场景 {scenario['id']} 的脚本观察")
        return dict(observation)


class ClaudeCliAdapter:
    """真实模型执行层占位：未获授权前失败关闭，不允许静默降级为 fake。"""

    def run(self, skill_md_text: str, scenario: dict) -> dict:
        raise EvalAdapterUnavailable(
            "真实模型行为评测未授权：需要显式模型、预算与执行配置授权后才能运行"
        )


def _check(expectations: dict, observation: dict) -> list[dict]:
    failures: list[dict] = []
    invoked = [str(cmd) for cmd in observation.get("invoked_commands", [])]

    def missing_required(key: str) -> None:
        failures.append({"check": "observation_missing", "evidence": f"观察缺少字段 {key}"})

    for needle in expectations.get("must_invoke", []):
        if not any(needle in cmd for cmd in invoked):
            failures.append({"check": "must_invoke", "evidence": f"未见调用包含 {needle!r}；实际 {invoked}"})
    for needle in expectations.get("must_not_invoke", []):
        hits = [cmd for cmd in invoked if needle in cmd]
        if hits:
            failures.append({"check": "must_not_invoke", "evidence": f"禁止调用 {needle!r} 出现: {hits}"})
    if "expected_exit_codes" in expectations:
        if "exit_code" not in observation:
            missing_required("exit_code")
        elif observation["exit_code"] not in expectations["expected_exit_codes"]:
            failures.append({
                "check": "expected_exit_codes",
                "evidence": f"退出码 {observation['exit_code']} 不在 {expectations['expected_exit_codes']}",
            })
    output = str(observation.get("output", ""))
    for needle in expectations.get("output_must_contain", []):
        if needle not in output:
            failures.append({"check": "output_must_contain", "evidence": f"输出缺少 {needle!r}"})
    for needle in expectations.get("output_must_not_contain", []):
        if needle in output:
            failures.append({"check": "output_must_not_contain", "evidence": f"输出出现禁止内容 {needle!r}"})
    if "input_text_modified" in expectations:
        if "input_text_modified" not in observation:
            missing_required("input_text_modified")
        elif observation["input_text_modified"] != expectations["input_text_modified"]:
            failures.append({
                "check": "input_text_modified",
                "evidence": f"输入正文改动状态 {observation['input_text_modified']} 与期望 "
                            f"{expectations['input_text_modified']} 不一致",
            })
    return failures


def run_scenario(skill_md_text: str, scenario: dict, adapter: Any) -> dict:
    observation = adapter.run(skill_md_text, scenario)
    failures = _check(scenario["expectations"], observation)
    return {
        "scenario_id": scenario["id"],
        "category": scenario["category"],
        "verdict": "passed" if not failures else "failed",
        "failed_checks": failures,
        "observation_summary": {
            "invoked_commands": observation.get("invoked_commands", []),
            "exit_code": observation.get("exit_code"),
            "input_text_modified": observation.get("input_text_modified"),
            "output_chars": len(str(observation.get("output", ""))),
        },
    }


def run_eval(skill_dir: Path, scenario_file: Path, adapter: Any, adapter_name: str) -> dict:
    skill_md = (skill_dir / "SKILL.md")
    if not skill_md.is_file():
        raise EvalContractError(f"SKILL.md 不存在: {skill_md}")
    data = load_scenarios(scenario_file)
    skill_md_text = skill_md.read_text(encoding="utf-8")
    verdicts = [run_scenario(skill_md_text, scenario, adapter) for scenario in data["scenarios"]]
    passed = sum(1 for v in verdicts if v["verdict"] == "passed")
    return {
        "schema_version": REPORT_SCHEMA,
        "skill": data["skill"],
        "skill_md_sha256": skill_md_fingerprint(skill_dir),
        "adapter": adapter_name,
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "scenario_count": len(verdicts),
        "passed": passed,
        "failed": len(verdicts) - passed,
        "verdicts": verdicts,
        "limit": "fake adapter 结果只证明评测管道，不构成 Skill 行为证据",
    }


__all__ = [
    "REPORT_SCHEMA", "SCENARIO_SCHEMA", "CATEGORIES", "EvalContractError",
    "EvalAdapterUnavailable", "FakeAdapter", "ClaudeCliAdapter",
    "skill_md_fingerprint", "load_scenarios", "run_scenario", "run_eval",
]