Genarrative/rust/scripts/run-agent-capability-set.sh

#!/usr/bin/env bash
set -euo pipefail

# 逐项运行 Agent 能力回归；测试定义只引用 Rust workspace 中已经存在的测试。
script_dir="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
workspace_root="$(cd -- "$script_dir/.." && pwd)"
dataset="$workspace_root/tests/agent-capability-set.jsonl"
# 生成的数据库、日志和 Cargo 输出默认放在 `~/data/tmp`；调用方仍可通过
# `TMPDIR` 或 AGENT_TEST_TMPDIR 显式选择临时父目录。
tmp_parent="${AGENT_TEST_TMPDIR:-${TMPDIR:-${HOME:?HOME 未设置}/data/tmp}}"

usage() {
    cat <<'EOF'
用法：
  ./scripts/run-agent-capability-set.sh       运行全部 Agent 能力用例
  ./scripts/run-agent-capability-set.sh --list 只列出用例，不运行 Cargo

临时目录默认使用 ~/data/tmp（可由 TMPDIR 或 AGENT_TEST_TMPDIR 覆盖）。
EOF
}

list_only=0
while (($# > 0)); do
    case "$1" in
        --list)
            list_only=1
            ;;
        -h|--help)
            usage
            exit 0
            ;;
        *)
            echo "未知参数：$1" >&2
            usage >&2
            exit 2
            ;;
    esac
    shift
done

python3 - "$dataset" "$workspace_root" "$tmp_parent" "$list_only" <<'PY'
import json
import os
import pathlib
import shutil
import subprocess
import sys
import tempfile
import re

dataset = pathlib.Path(sys.argv[1])
workspace = pathlib.Path(sys.argv[2])
tmp_parent = pathlib.Path(sys.argv[3]).expanduser()
list_only = sys.argv[4] == "1"

try:
    cases = [
        json.loads(line)
        for line in dataset.read_text(encoding="utf-8").splitlines()
        if line.strip()
    ]
except (OSError, json.JSONDecodeError) as error:
    print(f"能力测试集读取失败：{error}", file=sys.stderr)
    raise SystemExit(2)

required = {"id", "package", "target", "filter", "capability"}
seen = set()
for case in cases:
    missing = required - case.keys()
    if missing or case["id"] in seen:
        detail = f"缺少字段 {sorted(missing)}" if missing else "重复 id"
        print(f"能力测试集格式错误：{case.get('id', '<unknown>')}（{detail}）", file=sys.stderr)
        raise SystemExit(2)
    if case["target"] != "lib" and not case["target"]:
        print(f"能力测试集格式错误：{case['id']} 的 target 为空", file=sys.stderr)
        raise SystemExit(2)
    seen.add(case["id"])

if list_only:
    for case in cases:
        print(f"{case['id']}: {case['package']} / {case['target']} / {case['filter']} · {case['capability']}")
    raise SystemExit(0)

tmp_parent.mkdir(parents=True, exist_ok=True)
suite_dir = pathlib.Path(tempfile.mkdtemp(prefix="agent-capability-set.", dir=tmp_parent))
try:
    (suite_dir / "tmp").mkdir()
    env = os.environ.copy()
    env["TMPDIR"] = str(suite_dir / "tmp")
    env["CARGO_TARGET_DIR"] = str(suite_dir / "target")
    env["CARGO_INCREMENTAL"] = "0"
    # Harness output is parsed below; disable ANSI color so a successful test
    # cannot be hidden behind escape sequences in the count check.
    env["CARGO_TERM_COLOR"] = "never"

    passed = 0
    failed = 0
    for index, case in enumerate(cases, start=1):
        target_args = ["--lib"] if case["target"] == "lib" else ["--test", case["target"]]
        command = [
            "cargo",
            "test",
            "--locked",
            "--manifest-path",
            str(workspace / "Cargo.toml"),
            "-p",
            case["package"],
            *target_args,
            case["filter"],
        ]
        log_path = suite_dir / f"{index:02d}-{case['id']}.log"
        with log_path.open("wb") as log:
            result = subprocess.run(
                command,
                cwd=workspace,
                env=env,
                stdout=log,
                stderr=subprocess.STDOUT,
                check=False,
            )

        # `cargo test <filter>` exits zero even when the filter matches no
        # tests.  Treat that case as a failed capability check: a renamed or
        # deleted test must not silently turn this gate into a no-op.  The
        # harness keeps its result summary in English regardless of the
        # caller locale, and CARGO_TERM_COLOR=never above makes the patterns
        # deterministic.
        output = log_path.read_text(encoding="utf-8", errors="replace")
        passed_tests = sum(
            int(match.group(1))
            for match in re.finditer(
                r"^test result:\s+ok\.\s+(\d+) passed;",
                output,
                flags=re.MULTILINE,
            )
        )
        ran_tests = sum(
            int(match.group(1))
            for match in re.finditer(
                r"^running\s+(\d+) tests?$", output, flags=re.MULTILINE
            )
        )
        if result.returncode == 0 and ran_tests > 0 and passed_tests > 0:
            passed += 1
            print(f"PASS {case['id']}: {case['capability']}")
        else:
            failed += 1
            # 不打印 Cargo 输出，避免意外回显环境变量或 Provider 诊断内容。
            if result.returncode != 0:
                reason = f"exit={result.returncode}"
            elif ran_tests == 0 or passed_tests == 0:
                reason = "未实际通过任何匹配测试（可能是 filter 漂移）"
            else:
                reason = "测试结果无法确认"
            print(f"FAIL {case['id']}: {case['capability']} ({reason})")

    print(f"SUMMARY total={len(cases)} passed={passed} failed={failed}")
    raise SystemExit(1 if failed else 0)
finally:
    shutil.rmtree(suite_dir, ignore_errors=True)
PY