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

# 运行确定性的 Agent 回归集；真实 Provider 只在显式 --real 时调用。
# 所有数据库、Cargo 临时文件和中间输出都放在明确的临时父目录下，退出时
# 只清理本轮创建的子目录；默认使用 `~/data/tmp`（优先尊重调用方显式的
# `TMPDIR`），也可通过 AGENT_TEST_TMPDIR 指定专用 runner 目录。
script_dir="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
workspace_root="$(cd -- "$script_dir/.." && pwd)"
dataset="$workspace_root/tests/agent-test-set.jsonl"
tmp_parent="${AGENT_TEST_TMPDIR:-${TMPDIR:-${HOME:?HOME 未设置}/data/tmp}}"
run_real="${AGENT_TEST_REAL_PROVIDER:-0}"
run_cargo=1
list_only=0

usage() {
    cat <<'EOF'
用法：
  ./scripts/run-agent-test-set.sh                 # Rust 全量测试 + 离线测试集
  ./scripts/run-agent-test-set.sh --quick         # 只运行测试集，跳过 cargo test
  ./scripts/run-agent-test-set.sh --real          # 额外运行自有 OpenAI-compatible Provider
  ./scripts/run-agent-test-set.sh --list          # 列出测试用例

真实 Provider 需要在当前 shell 提供 OPENAI_API_KEY（或
OPENAI_API_KEY_ENV/AGENT_OPENAI_API_KEY_ENV 指向的变量）。
EOF
}

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

if [[ "$list_only" == 1 ]]; then
    python3 - "$dataset" <<'PY'
import json
import sys

for line in open(sys.argv[1], encoding="utf-8"):
    if line.strip():
        case = json.loads(line)
        suffix = "（可选真实 Provider）" if case.get("optional") else ""
        print(f"{case['id']}: {case['provider']} / {'stream' if case['stream'] else 'complete'}{suffix}")
PY
    exit 0
fi

mkdir -p -- "$tmp_parent"
tmp_parent="$(cd -- "$tmp_parent" && pwd)"
suite_dir="$(mktemp -d "$tmp_parent/agent-test-set.XXXXXX")"
mkdir -p -- "$suite_dir/tmp"
cleanup() {
    # suite_dir 是本脚本刚创建的精确目录，不触碰 tmp_parent 中的其它内容。
    rm -rf -- "$suite_dir"
}
trap cleanup EXIT

export TMPDIR="$suite_dir/tmp"
export CARGO_TARGET_DIR="$suite_dir/target"

if [[ "$run_cargo" == 1 ]]; then
    echo "[1/2] 运行 workspace 单元/集成测试"
    # 脚本通常从父仓库调用；显式传 manifest，避免把调用者 cwd 当成
    # workspace 根目录（父仓库本身没有 Cargo.toml）。
    cargo test --locked --manifest-path "$workspace_root/Cargo.toml" \
        --workspace --all-features --no-fail-fast
    cargo test --locked --manifest-path "$workspace_root/Cargo.toml" \
        --workspace --no-default-features --no-fail-fast
else
    echo "[1/2] 已跳过 cargo workspace 测试（--quick）"
fi

echo "[2/2] 运行 Agent 测试集"
python3 - "$dataset" "$workspace_root" "$suite_dir" "$run_real" <<'PY'
import json
import os
import pathlib
import re
import sqlite3
import subprocess
import sys

dataset = pathlib.Path(sys.argv[1])
workspace = pathlib.Path(sys.argv[2])
suite_dir = pathlib.Path(sys.argv[3])
run_real = sys.argv[4] == "1"


def key_env_name(env):
    return (
        env.get("AGENT_OPENAI_API_KEY_ENV")
        or env.get("OPENAI_API_KEY_ENV")
        or "OPENAI_API_KEY"
    )


def safe_stderr(raw, env):
    """只给出脱敏诊断，避免 Provider 错误回显凭据或 URL 查询参数。"""
    text = raw.decode("utf-8", errors="replace")
    values = {
        env.get("OPENAI_API_KEY", ""),
        env.get(key_env_name(env), ""),
        env.get("CODEX_API_KEY", ""),
    }
    for value in values:
        if value:
            text = text.replace(value, "[redacted]")
    text = re.sub(r"(?i)(authorization|api[-_ ]?key|token|password)\s*[:=]\s*[^\s,;]+", r"\1=[redacted]", text)
    text = re.sub(r"https?://[^\s]+", "[endpoint redacted]", text)
    lines = [line.strip() for line in text.splitlines() if line.strip()]
    return lines[-1] if lines else "无错误详情"


def invoke(args, env, timeout=180):
    command = ["cargo", "run", "--locked", "-q", "-p", "agent-cli", "--", *args]
    completed = subprocess.run(
        command,
        cwd=workspace,
        env=env,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        timeout=timeout,
        check=False,
    )
    if completed.returncode != 0:
        detail = safe_stderr(completed.stderr, env)
        raise RuntimeError(f"退出码 {completed.returncode}：{detail}")
    return completed.stdout


def isolated_env(case, case_dir):
    env = os.environ.copy()
    # 测试集只验证 Provider/Engine/SQLite，不继承调用 shell 的 Skill/MCP。
    for name in (
        "AGENT_SKILL_ROOT",
        "AGENT_SKILL_ROOTS",
        "AGENT_SKILLS",
        "AGENT_MCP_STDIO_COMMAND",
        "AGENT_MCP_STDIO_ARGS",
        "AGENT_MCP_HTTP_URL",
        "AGENT_MCP_HTTP_HEADERS",
        "AGENT_MCP_ALLOW",
    ):
        env.pop(name, None)
    env["AGENT_CONFIG"] = str(case_dir / "does-not-exist.toml")
    env["AGENT_DB"] = str(case_dir / "agent.db")
    env["AGENT_STREAM"] = "1" if case.get("stream") else "0"
    if case["provider"] == "fake":
        env["AGENT_PROVIDER"] = "fake"
        env["AGENT_MODEL"] = "fake"
    else:
        env["AGENT_PROVIDER"] = "openai"
        # 真实用例允许通过 AGENT_MODEL 或 OPENAI_MODEL 选择任意兼容模型。
        if case.get("model"):
            env["AGENT_MODEL"] = case["model"]
    return env


cases = [json.loads(line) for line in dataset.read_text(encoding="utf-8").splitlines() if line.strip()]
if run_real:
    key_name = key_env_name(os.environ)
    if not os.environ.get(key_name, "").strip():
        raise SystemExit(
            f"真实 Provider 测试需要环境变量 {key_name}；脚本不会读取或打印 key 值"
        )
passed = 0
skipped = 0
for index, case in enumerate(cases, 1):
    if case["provider"] == "openai" and not run_real:
        print(f"  SKIP {case['id']}（需 --real，不会读取或显示密钥）")
        skipped += 1
        continue
    case_dir = suite_dir / case["id"]
    case_dir.mkdir(parents=True, exist_ok=True)
    env = isolated_env(case, case_dir)
    args = ["run", "--stream" if case.get("stream") else "--no-stream", case["prompt"]]
    try:
        result = json.loads(invoke(args, env, timeout=240 if case["provider"] == "openai" else 60))
        output = result.get("output") or {}
        events = output.get("events") or []
        event_types = [event.get("type") for event in events]
        tool_completed = sum(event_type == "tool_completed" for event_type in event_types)
        stream_events = len(output.get("stream_events") or [])
        steps = int(output.get("steps") or 0)
        text = str(output.get("text") or "")
        run_id = result.get("run_id")
        if not run_id:
            raise AssertionError("结果缺少 run_id")

        # 用新连接读取落盘结果，防止只验证最终文本而漏掉 Runtime 消息重复。
        # 参数绑定且连接只读，不创建第二套业务状态或改动 CLI 的数据库。
        database_uri = pathlib.Path(env["AGENT_DB"]).resolve().as_uri() + "?mode=ro"
        connection = sqlite3.connect(database_uri, uri=True)
        try:
            row = connection.execute(
                "SELECT snapshot_json FROM runtime_states WHERE runtime_id = ?",
                (result.get("runtime_id"),),
            ).fetchone()
        finally:
            connection.close()
        if row is None:
            raise AssertionError("缺少已提交的 RuntimeSnapshot")
        snapshot = json.loads(row[0])
        matching_runs = [run for run in snapshot["runs"] if run["runId"] == run_id]
        if len(matching_runs) != 1 or matching_runs[0]["messages"] != output.get("messages"):
            raise AssertionError("Runtime 消息与 Engine 输出不一致（顺序、缺失或重复）")

        expected = case.get("expect") or {}
        if expected.get("status"):
            inspected = json.loads(invoke(["inspect", run_id], env, timeout=30))
            if inspected.get("status") != expected["status"]:
                raise AssertionError(f"run status={inspected.get('status')!r}")
        if expected.get("exact_text") is not None and text != expected["exact_text"]:
            raise AssertionError(f"最终文本不匹配：{text!r}")
        if expected.get("non_empty_text") and not text.strip():
            raise AssertionError("最终文本为空")
        if steps < expected.get("min_steps", 0):
            raise AssertionError(f"steps={steps} 小于 {expected['min_steps']}")
        if tool_completed < expected.get("min_tool_completed", 0):
            raise AssertionError(f"tool_completed={tool_completed} 小于 {expected['min_tool_completed']}")
        if stream_events < expected.get("min_stream_events", 0):
            raise AssertionError(f"stream_events={stream_events} 小于 {expected['min_stream_events']}")
        if expected.get("min_export_records"):
            exported = invoke(["export", run_id], env, timeout=30)
            records = [line for line in exported.decode("utf-8").splitlines() if line.strip()]
            if len(records) < expected["min_export_records"]:
                raise AssertionError(f"JSONL 记录数={len(records)} 小于 {expected['min_export_records']}")
        print(
            f"  PASS {case['id']}: steps={steps}, tools={tool_completed}, "
            f"stream_events={stream_events}, text_bytes={len(text.encode('utf-8'))}"
        )
        passed += 1
    except (AssertionError, RuntimeError, json.JSONDecodeError, subprocess.TimeoutExpired) as error:
        print(f"  FAIL {case['id']}: {error}", file=sys.stderr)
        raise SystemExit(1)

print(f"测试集完成：通过 {passed}，跳过 {skipped}")
if run_real:
    print("真实 Provider smoke 已执行；它验证协议/持久化闭环，不替代模型质量评测。")
PY

echo "Agent 测试集通过"