🛠 MVP 仓库

#!/usr/bin/env python3
"""BenchProof: local performance acceptance checks for Git patches."""

from __future__ import annotations

import argparse
import datetime as dt
import json
import math
import os
import platform
import resource
import shlex
import signal
import statistics
import subprocess
import sys
import tempfile
import time
from pathlib import Path
from typing import Any, Dict, List, Optional, Sequence, Tuple


VERSION = "0.1.0"
WORKTREE = "WORKTREE"


class BenchProofError(RuntimeError):
    pass


def quick_usage() -> None:
    print("BenchProof - AI 优化补丁性能验收器")
    print("用法: python 38_mvp.py compare --repo . --base HEAD^ --command 'python app.py'")
    print("候选默认是当前工作树；也可用 --candidate <git-ref> 比较两个提交。")
    print("快速体验: python 38_mvp.py demo")
    print("完整参数: python 38_mvp.py --help")


def tail(text: str, limit: int = 2000) -> str:
    return text[-limit:] if len(text) > limit else text


def peak_rss_kib(usage: resource.struct_rusage) -> int:
    value = int(usage.ru_maxrss)
    # macOS 返回字节，Linux/BSD 通常返回 KiB。
    if sys.platform == "darwin":
        value //= 1024
    return value


def run_one(command: str, cwd: str, timeout: float) -> Dict[str, Any]:
    """在独立进程中运行一次命令，采集墙钟耗时、退出码和峰值内存。"""
    started = time.perf_counter()
    timed_out = False
    process = subprocess.Popen(
        command,
        cwd=cwd,
        shell=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        start_new_session=(os.name == "posix"),
    )
    try:
        stdout, stderr = process.communicate(timeout=timeout)
    except subprocess.TimeoutExpired:
        timed_out = True
        if os.name == "posix":
            os.killpg(process.pid, signal.SIGKILL)
        else:
            process.kill()
        stdout, stderr = process.communicate()
    elapsed = time.perf_counter() - started
    usage = resource.getrusage(resource.RUSAGE_CHILDREN)
    return {
        "elapsed_seconds": round(elapsed, 9),
        "peak_rss_kib": peak_rss_kib(usage),
        "exit_code": process.returncode,
        "timed_out": timed_out,
        "stdout_tail": tail(stdout),
        "stderr_tail": tail(stderr),
    }


def helper_main(argv: Sequence[str]) -> int:
    parser = argparse.ArgumentParser(add_help=False)
    parser.add_argument("--cwd", required=True)
    parser.add_argument("--timeout", type=float, required=True)
    parser.add_argument("--command", required=True)
    args = parser.parse_args(argv)
    try:
        result = run_one(args.command, args.cwd, args.timeout)
    except Exception as exc:  # helper 必须始终向父进程返回结构化结果
        result = {
            "elapsed_seconds": 0.0,
            "peak_rss_kib": 0,
            "exit_code": 255,
            "timed_out": False,
            "stdout_tail": "",
            "stderr_tail": f"collector error: {type(exc).__name__}: {exc}",
        }
    print(json.dumps(result, ensure_ascii=False))
    return 0


def collected_run(command: str, cwd: Path, timeout: float) -> Dict[str, Any]:
    helper = subprocess.run(
        [
            sys.executable,
            str(Path(__file__).resolve()),
            "_run-one",
            "--cwd",
            str(cwd),
            "--timeout",
            str(timeout),
            "--command",
            command,
        ],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        check=False,
    )
    try:
        result = json.loads(helper.stdout)
    except json.JSONDecodeError:
        result = {
            "elapsed_seconds": 0.0,
            "peak_rss_kib": 0,
            "exit_code": 255,
            "timed_out": False,
            "stdout_tail": "",
            "stderr_tail": tail(helper.stderr or helper.stdout or "collector produced no JSON"),
        }
    return result


def percentile(values: Sequence[float], fraction: float) -> float:
    ordered = sorted(values)
    if len(ordered) == 1:
        return ordered[0]
    position = (len(ordered) - 1) * fraction
    low, high = math.floor(position), math.ceil(position)
    if low == high:
        return ordered[low]
    return ordered[low] + (ordered[high] - ordered[low]) * (position - low)


def outlier_indexes(values: Sequence[float]) -> List[int]:
    """使用 MAD 标记异常点；MAD 为零时回退到 IQR，原始数据仍完整保留。"""
    if len(values) < 4:
        return []
    median = statistics.median(values)
    deviations = [abs(value - median) for value in values]
    mad = statistics.median(deviations)
    if mad > 0:
        return [
            index
            for index, value in enumerate(values)
            if 0.6745 * abs(value - median) / mad > 3.5
        ]
    q1, q3 = percentile(values, 0.25), percentile(values, 0.75)
    iqr = q3 - q1
    if iqr == 0:
        return [index for index, value in enumerate(values) if value != median]
    low, high = q1 - 1.5 * iqr, q3 + 1.5 * iqr
    return [index for index, value in enumerate(values) if value < low or value > high]


def metric_summary(values: Sequence[float]) -> Dict[str, Any]:
    outliers = outlier_indexes(values)
    cleaned = [value for index, value in enumerate(values) if index not in outliers]
    if not cleaned:
        cleaned = list(values)
    mean = statistics.mean(cleaned)
    stdev = statistics.stdev(cleaned) if len(cleaned) > 1 else 0.0
    median = statistics.median(cleaned)
    return {
        "samples": len(values),
        "used_samples": len(cleaned),
        "outlier_indexes": outliers,
        "mean": round(mean, 9),
        "median": round(median, 9),
        "stdev": round(stdev, 9),
        "cv_percent": round((stdev / mean * 100.0) if mean else 0.0, 3),
        "min": round(min(cleaned), 9),
        "max": round(max(cleaned), 9),
        "p95": round(percentile(cleaned, 0.95), 9),
    }


def summarize_runs(runs: List[Dict[str, Any]], expected_exit: int) -> Dict[str, Any]:
    successful = [
        run for run in runs if run["exit_code"] == expected_exit and not run["timed_out"]
    ]
    summary: Dict[str, Any] = {
        "total_runs": len(runs),
        "successful_runs": len(successful),
        "failed_run_indexes": [
            index
            for index, run in enumerate(runs)
            if run["exit_code"] != expected_exit or run["timed_out"]
        ],
    }
    if successful:
        summary["time_seconds"] = metric_summary(
            [float(run["elapsed_seconds"]) for run in successful]
        )
        summary["peak_rss_kib"] = metric_summary(
            [float(run["peak_rss_kib"]) for run in successful]
        )
    return summary


def percent_change(before: float, after: float) -> Optional[float]:
    if before == 0:
        return None
    return (after - before) / before * 100.0


def decide(
    base: Dict[str, Any],
    candidate: Dict[str, Any],
    min_improvement: float,
    max_time_regression: float,
    max_memory_regression: float,
    noise_multiplier: float,
) -> Dict[str, Any]:
    """按正确性、回归上限、有效加速和噪声护栏依次生成验收结论。"""
    if base["successful_runs"] != base["total_runs"]:
        return {"verdict": "INVALID", "reason": "基线命令存在失败或超时"}
    if candidate["successful_runs"] != candidate["total_runs"]:
        return {"verdict": "REJECT", "reason": "候选命令存在失败或超时"}

    base_time = base["time_seconds"]["median"]
    candidate_time = candidate["time_seconds"]["median"]
    base_memory = base["peak_rss_kib"]["median"]
    candidate_memory = candidate["peak_rss_kib"]["median"]
    time_change = percent_change(base_time, candidate_time)
    memory_change = percent_change(base_memory, candidate_memory)
    speedup = -time_change if time_change is not None else 0.0
    noise_floor = noise_multiplier * max(
        base["time_seconds"]["cv_percent"], candidate["time_seconds"]["cv_percent"]
    )
    effective_improvement = max(min_improvement, noise_floor)

    details = {
        "time_change_percent": round(time_change, 3) if time_change is not None else None,
        "speedup_percent": round(speedup, 3),
        "memory_change_percent": (
            round(memory_change, 3) if memory_change is not None else None
        ),
        "observed_noise_floor_percent": round(noise_floor, 3),
        "required_improvement_percent": round(effective_improvement, 3),
    }
    if time_change is not None and time_change > max_time_regression:
        return {"verdict": "REJECT", "reason": "耗时回归超过阈值", **details}
    if memory_change is not None and memory_change > max_memory_regression:
        return {"verdict": "REJECT", "reason": "峰值内存回归超过阈值", **details}
    if speedup >= effective_improvement:
        return {"verdict": "ACCEPT", "reason": "加速超过阈值且没有越界回归", **details}
    return {"verdict": "NO_MATERIAL_CHANGE", "reason": "变化未超过阈值或当前噪声", **details}


def git(repo: Path, *args: str, check: bool = True) -> str:
    result = subprocess.run(
        ["git", "-C", str(repo), *args],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        check=False,
    )
    if check and result.returncode != 0:
        raise BenchProofError(tail(result.stderr.strip() or "git command failed"))
    return result.stdout.strip()


def verify_repo(repo: Path) -> Path:
    repo = repo.resolve()
    if not repo.is_dir():
        raise BenchProofError(f"仓库目录不存在: {repo}")
    root = git(repo, "rev-parse", "--show-toplevel")
    return Path(root).resolve()


def resolve_commit(repo: Path, ref: str) -> str:
    return git(repo, "rev-parse", "--verify", f"{ref}^{{commit}}")


def add_worktree(repo: Path, path: Path, commit: str) -> None:
    result = subprocess.run(
        ["git", "-C", str(repo), "worktree", "add", "--detach", str(path), commit],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        check=False,
    )
    if result.returncode != 0:
        raise BenchProofError(tail(result.stderr.strip() or "无法创建 Git worktree"))


def remove_worktree(repo: Path, path: Path) -> None:
    subprocess.run(
        ["git", "-C", str(repo), "worktree", "remove", "--force", str(path)],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        check=False,
    )


def execute_suite(
    command: str,
    base_dir: Path,
    candidate_dir: Path,
    runs: int,
    warmups: int,
    timeout: float,
) -> Tuple[Dict[str, List[Dict[str, Any]]], Dict[str, List[Dict[str, Any]]]]:
    """交替基线和候选的执行顺序，降低机器逐渐升温造成的系统偏差。"""
    warmup_data: Dict[str, List[Dict[str, Any]]] = {"base": [], "candidate": []}
    measured: Dict[str, List[Dict[str, Any]]] = {"base": [], "candidate": []}
    locations = {"base": base_dir, "candidate": candidate_dir}
    for phase_runs, destination in ((warmups, warmup_data), (runs, measured)):
        for index in range(phase_runs):
            order = ("base", "candidate") if index % 2 == 0 else ("candidate", "base")
            for side in order:
                result = collected_run(command, locations[side], timeout)
                result["sequence"] = index + 1
                destination[side].append(result)
                print(
                    f"[{side:9}] {index + 1}/{phase_runs} "
                    f"{result['elapsed_seconds']:.4f}s {result['peak_rss_kib']} KiB "
                    f"exit={result['exit_code']}",
                    file=sys.stderr,
                )
    return warmup_data, measured


def request_model_summary(report: Dict[str, Any], args: argparse.Namespace) -> Dict[str, Any]:
    if not args.summary_url:
        return {"enabled": False}
    try:
        import requests  # type: ignore
    except ImportError:
        return {"enabled": True, "error": "未安装可选依赖 requests"}
    compact = {
        "decision": report["decision"],
        "base_summary": report["results"]["base"]["summary"],
        "candidate_summary": report["results"]["candidate"]["summary"],
        "policy": report["policy"],
    }
    headers = {"Content-Type": "application/json"}
    api_key = os.environ.get(args.api_key_env, "")
    if api_key:
        headers["Authorization"] = f"Bearer {api_key}"
    payload = {
        "model": args.summary_model,
        "messages": [
            {
                "role": "system",
                "content": "你是性能验收助手。用中文给出三句话：结论、关键数据、下一步。不要改变程序结论。",
            },
            {"role": "user", "content": json.dumps(compact, ensure_ascii=False)},
        ],
        "temperature": 0.1,
    }
    try:
        response = requests.post(args.summary_url, headers=headers, json=payload, timeout=30)
        response.raise_for_status()
        data = response.json()
        content = data["choices"][0]["message"]["content"]
        return {"enabled": True, "model": args.summary_model, "content": content}
    except Exception as exc:
        return {"enabled": True, "error": f"{type(exc).__name__}: {exc}"}


def base_report(
    command: str,
    warmups: Dict[str, List[Dict[str, Any]]],
    measured: Dict[str, List[Dict[str, Any]]],
    args: argparse.Namespace,
    source: Dict[str, Any],
) -> Dict[str, Any]:
    base_summary = summarize_runs(measured["base"], args.expected_exit)
    candidate_summary = summarize_runs(measured["candidate"], args.expected_exit)
    policy = {
        "expected_exit": args.expected_exit,
        "min_improvement_percent": args.min_improvement,
        "max_time_regression_percent": args.max_time_regression,
        "max_memory_regression_percent": args.max_memory_regression,
        "noise_multiplier": args.noise_multiplier,
    }
    decision = decide(
        base_summary,
        candidate_summary,
        args.min_improvement,
        args.max_time_regression,
        args.max_memory_regression,
        args.noise_multiplier,
    )
    return {
        "schema_version": 1,
        "tool": {"name": "BenchProof", "version": VERSION},
        "created_at": dt.datetime.now().astimezone().isoformat(),
        "host": {
            "platform": platform.platform(),
            "python": platform.python_version(),
            "processor": platform.processor(),
        },
        "source": source,
        "benchmark": {
            "command": command,
            "runs": args.runs,
            "warmups": args.warmups,
            "timeout_seconds": args.timeout,
        },
        "policy": policy,
        "results": {
            "base": {"warmups": warmups["base"], "runs": measured["base"], "summary": base_summary},
            "candidate": {
                "warmups": warmups["candidate"],
                "runs": measured["candidate"],
                "summary": candidate_summary,
            },
        },
        "decision": decision,
    }


def write_report(report: Dict[str, Any], output: str) -> Path:
    path = Path(output).expanduser().resolve()
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    return path


def compare_command(args: argparse.Namespace) -> int:
    repo = verify_repo(Path(args.repo))
    base_commit = resolve_commit(repo, args.base)
    candidate_is_worktree = args.candidate == WORKTREE
    candidate_commit = git(repo, "rev-parse", "HEAD") if candidate_is_worktree else resolve_commit(repo, args.candidate)
    source = {
        "type": "git",
        "repo": str(repo),
        "base_ref": args.base,
        "base_commit": base_commit,
        "candidate_ref": args.candidate,
        "candidate_commit": candidate_commit,
        "candidate_dirty": bool(git(repo, "status", "--porcelain")) if candidate_is_worktree else False,
    }
    with tempfile.TemporaryDirectory(prefix="benchproof-") as temp_name:
        temp = Path(temp_name)
        base_dir = temp / "base"
        candidate_dir = repo
        created: List[Path] = []
        try:
            add_worktree(repo, base_dir, base_commit)
            created.append(base_dir)
            if not candidate_is_worktree:
                candidate_dir = temp / "candidate"
                add_worktree(repo, candidate_dir, candidate_commit)
                created.append(candidate_dir)
            warmups, measured = execute_suite(
                args.command, base_dir, candidate_dir, args.runs, args.warmups, args.timeout
            )
        finally:
            for worktree in reversed(created):
                remove_worktree(repo, worktree)
    report = base_report(args.command, warmups, measured, args, source)
    report["model_summary"] = request_model_summary(report, args)
    output = write_report(report, args.output)
    print(f"{report['decision']['verdict']}: {report['decision']['reason']}")
    print(f"报告: {output}")
    return 1 if args.fail_on_reject and report["decision"]["verdict"] in {"REJECT", "INVALID"} else 0


def demo_command(args: argparse.Namespace) -> int:
    with tempfile.TemporaryDirectory(prefix="benchproof-demo-") as temp_name:
        root = Path(temp_name)
        base_dir, candidate_dir = root / "base", root / "candidate"
        base_dir.mkdir()
        candidate_dir.mkdir()
        # 示例模拟一个将线性查找改为集合查找的小补丁。
        (base_dir / "task.py").write_text(
            "items=list(range(20000)); wanted=list(range(10000,30000)); "
            "print(sum(1 for x in wanted if x in items))\n",
            encoding="utf-8",
        )
        (candidate_dir / "task.py").write_text(
            "items=set(range(20000)); wanted=list(range(10000,30000)); "
            "print(sum(1 for x in wanted if x in items))\n",
            encoding="utf-8",
        )
        command = f"{shlex.quote(sys.executable)} task.py"
        warmups, measured = execute_suite(
            command, base_dir, candidate_dir, args.runs, args.warmups, args.timeout
        )
        source = {"type": "built_in_demo", "change": "list membership -> set membership"}
        report = base_report(command, warmups, measured, args, source)
    report["model_summary"] = {"enabled": False}
    output = write_report(report, args.output)
    print(f"{report['decision']['verdict']}: {report['decision']['reason']}")
    print(f"报告: {output}")
    return 0


def add_common_arguments(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--runs", type=int, default=7, help="正式测量轮数 (默认: 7)")
    parser.add_argument("--warmups", type=int, default=1, help="预热轮数 (默认: 1)")
    parser.add_argument("--timeout", type=float, default=120.0, help="单次超时秒数")
    parser.add_argument("--expected-exit", type=int, default=0, help="期望退出码")
    parser.add_argument("--min-improvement", type=float, default=5.0, help="最小有效加速百分比")
    parser.add_argument("--max-time-regression", type=float, default=5.0, help="最大耗时回归百分比")
    parser.add_argument("--max-memory-regression", type=float, default=10.0, help="最大内存回归百分比")
    parser.add_argument("--noise-multiplier", type=float, default=1.5, help="变异系数噪声倍数")


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="BenchProof: 对本地 Git 补丁进行可重复的性能验收"
    )
    parser.add_argument("--version", action="version", version=f"BenchProof {VERSION}")
    subparsers = parser.add_subparsers(dest="subcommand")

    compare = subparsers.add_parser("compare", help="比较 Git 基线与候选工作树/ref")
    compare.add_argument("--repo", default=".", help="本地 Git 仓库")
    compare.add_argument("--base", default="HEAD^", help="基线 Git ref")
    compare.add_argument("--candidate", default=WORKTREE, help="候选 ref，默认当前工作树")
    compare.add_argument("--command", required=True, help="在两个工作树内运行的统一命令")
    compare.add_argument("--output", default="benchproof-report.json", help="JSON 报告路径")
    compare.add_argument("--fail-on-reject", action="store_true", help="REJECT/INVALID 时返回退出码 1")
    compare.add_argument("--summary-url", help="可选 OpenAI 兼容 chat/completions URL")
    compare.add_argument("--summary-model", default="gpt-5-mini", help="可选总结模型")
    compare.add_argument("--api-key-env", default="OPENAI_API_KEY", help="API key 环境变量名")
    add_common_arguments(compare)
    compare.set_defaults(handler=compare_command)

    demo = subparsers.add_parser("demo", help="运行内置有效优化示例")
    demo.add_argument("--output", default="benchproof-demo-report.json", help="JSON 报告路径")
    add_common_arguments(demo)
    demo.set_defaults(handler=demo_command)
    return parser


def validate_args(args: argparse.Namespace) -> None:
    if args.runs < 1:
        raise BenchProofError("--runs 必须至少为 1")
    if args.warmups < 0:
        raise BenchProofError("--warmups 不能为负数")
    if args.timeout <= 0:
        raise BenchProofError("--timeout 必须大于 0")
    for name in ("min_improvement", "max_time_regression", "max_memory_regression", "noise_multiplier"):
        if getattr(args, name) < 0:
            raise BenchProofError(f"--{name.replace('_', '-')} 不能为负数")


def main(argv: Optional[Sequence[str]] = None) -> int:
    arguments = list(sys.argv[1:] if argv is None else argv)
    if not arguments:
        quick_usage()
        return 0
    if arguments[0] == "_run-one":
        return helper_main(arguments[1:])
    parser = build_parser()
    args = parser.parse_args(arguments)
    if not getattr(args, "subcommand", None):
        quick_usage()
        return 0
    try:
        validate_args(args)
        return int(args.handler(args))
    except BenchProofError as exc:
        print(f"BenchProof 错误: {exc}", file=sys.stderr)
        return 2
    except KeyboardInterrupt:
        print("BenchProof 已取消", file=sys.stderr)
        return 130


if __name__ == "__main__":
    raise SystemExit(main())