#!/usr/bin/env python3
"""`gpu-evidence.yml` 的接线自检（M253）：断言这道门禁**不可能被改成「总是绿」**。

为什么需要它：`check_gpu_shot.py --self-test` 只证明「尺子能量」，它管不到**接线**——
把自测步写成 `if: false`、给真实判据步加 `continue-on-error: true`、或者把命令尾巴写成
`|| true`，三种改法的结局都一样：job 变绿、一次判据都没真的跑过。本仓已经栽过四次
「守卫假绿」，所以这里把接线口径变成机器判据，而不是让下一个人读注释自律。

五条判据（对应 workflow 头部的三条规矩）：

1. **自测无条件跑**：跑 `--self-test` 的那一步、以及它所在的 job，都不许有 `if:`；
2. **真实判据必须挂在门上**：跑真产物判据的步（或它的 job）的 `if:` 必须引用
   `needs.<probe>.outputs.available == 'true'` —— 没有门就是「无产物时也照跑」，
   跳过与通过就分不开了；
3. **跳过要留痕**：至少一步同时写着 `$GITHUB_STEP_SUMMARY`、`跳过`、`::warning::` 与
   `$GITHUB_OUTPUT`（写 summary + 打黄条 + 把 available 交出去），不许静默 pass；
4. **全文件不许 `continue-on-error`**：判据红的时候它会把 job 变绿；
5. **判据命令不许吞退出码**（`|| true` / `|| exit 0` / `; true`），且真实判据步的产物路径必须
   来自 `needs.probe.outputs.*`（不是写死的）。

自带 **5 条变异向量**：把上面每一种「改绿」写法注入临时副本，断言对应判据当场红、基线绿。
判据自己抓不住回归时它会先红，所以这些断言不会变成装饰品（同 `check_release_semantics.py
--self-mutation` 的做法）。

    python3 packaging/tests/gpu_shot_ci_selftest.py

零第三方依赖，只借 `check_workflows_yaml.load_yaml`（它自己也负责「PyYAML 缺失 ⇒ 门禁不降级」）。
"""

from __future__ import annotations

import sys
from pathlib import Path

PACKAGING = Path(__file__).resolve().parent.parent
WORKFLOW = PACKAGING.parent / ".github/workflows/gpu-evidence.yml"
sys.path.insert(0, str(PACKAGING))
from check_workflows_yaml import load_yaml  # noqa: E402

SELFTEST_MARK = "check_gpu_shot.py --self-test"
JUDGE_MARK = "check_gpu_shot.py"
GATE = "outputs.available"
SWALLOW_MARKS = ("|| true", "|| exit 0", "; true")


def steps(doc: dict) -> list[tuple[str, dict, dict]]:
    """`[(job_id, job, step)]`：只列「列表项形态」的步骤（`uses:` 的 job 没有 steps）。"""
    found: list[tuple[str, dict, dict]] = []
    for job_id, job in (doc.get("jobs") or {}).items():
        for step in (job or {}).get("steps") or []:
            if isinstance(step, dict):
                found.append((job_id, job, step))
    return found


def check_selftest_unconditional(doc: dict) -> list[str]:
    """判据 1：自测那步不许带任何条件（job 级 or step 级）。"""
    problems: list[str] = []
    hits = 0
    for job_id, job, step in steps(doc):
        if SELFTEST_MARK not in (step.get("run") or ""):
            continue
        hits += 1
        # 用 `in` 而不是真值判断：`if: false` 会被 YAML 解析成布尔 False，真值判断正好漏掉最想抓的那种写法。
        if "if" in job:
            problems.append(f"job `{job_id}` 挂了 job 级 `if:` —— 自测被判据挡住了，整条门禁可以静默不跑")
        if "if" in step:
            problems.append(f"job `{job_id}` 的 `--self-test` 步挂了 `if:` —— 同上，自测必须无条件跑")
    if not hits:
        problems.append(f"没有任何一步无条件跑 `{SELFTEST_MARK}`")
    return problems


def references_probe(step: dict) -> bool:
    """这一步的产物路径是不是从 `needs.probe.outputs.*` 拿的（写死的路径会在跳过那条路上判错东西）。"""
    return any("needs.probe.outputs." in str(value) for value in (step.get("env") or {}).values())


def check_real_judgement_gated(doc: dict) -> list[str]:
    """判据 2：真产物判据必须挂在 `needs.<probe>.outputs.available == 'true'` 上。"""
    problems: list[str] = []
    real = [
        (job_id, job, step)
        for job_id, job, step in steps(doc)
        if JUDGE_MARK in (step.get("run") or "") and SELFTEST_MARK not in (step.get("run") or "") and references_probe(step)
    ]
    if not real:
        problems.append(
            f"找不到跑真实产物判据的那一步（`{JUDGE_MARK} <日志> <PNG>`，且日志/图来自 `needs.probe.outputs.*`）—— 门禁没接上"
        )
        return problems
    for job_id, job, step in real:
        gate = str(step.get("if") or job.get("if") or "")
        if GATE not in gate or "'true'" not in gate:
            problems.append(
                f"job `{job_id}` 的真实判据没有挂 `{GATE} == 'true'` 的门 —— 无产物时也会照跑，"
                "「跳过」与「通过」就分不开了"
            )
    return problems


def check_skip_is_visible(doc: dict) -> list[str]:
    """判据 3：跳过必须留痕（summary + warning + 交出去的 available），不许静默 pass。"""
    for _job_id, _job, step in steps(doc):
        run = step.get("run") or ""
        if "GITHUB_STEP_SUMMARY" in run and "跳过" in run and "::warning::" in run and "GITHUB_OUTPUT" in run:
            return []
    return ["没有任何一步同时写着 `$GITHUB_STEP_SUMMARY` + `跳过` + `::warning::` + `$GITHUB_OUTPUT` —— 跳过会变成静默 pass"]


def check_no_continue_on_error(doc: dict) -> list[str]:
    """判据 4：`continue-on-error` 会把红的判据变绿，全文件禁用。"""
    problems: list[str] = []
    for job_id, job, _step in steps(doc):
        if job.get("continue-on-error") is True:
            problems.append(f"job `{job_id}` 带 `continue-on-error: true`")
    for job_id, _job, step in steps(doc):
        if step.get("continue-on-error") is True:
            problems.append(f"job `{job_id}` 的步骤 `{step.get('name') or step.get('uses')}` 带 `continue-on-error: true` —— 判据红也会变成绿")
    return problems


def check_no_swallowed_exit_code(doc: dict) -> list[str]:
    """判据 5：判据命令不许吞退出码（`|| true` / `; true` 会把红判据变成绿）。"""
    problems: list[str] = []
    for job_id, _job, step in steps(doc):
        run = step.get("run") or ""
        if JUDGE_MARK not in run or SELFTEST_MARK in run:
            continue
        name = step.get("name") or f"job {job_id} 里那一步"
        for mark in SWALLOW_MARKS:
            if mark in run:
                problems.append(f"`{name}` 的命令里出现 `{mark}` —— 退出码被吞掉，判据红也绿")
    return problems


CHECKS = {
    "自测无条件": check_selftest_unconditional,
    "真实判据挂在门上": check_real_judgement_gated,
    "跳过留痕": check_skip_is_visible,
    "禁用 continue-on-error": check_no_continue_on_error,
    "退出码不被吞": check_no_swallowed_exit_code,
}

#: 变异向量：`(说明, 原文片段, 替换成, 应该红的判据)`。原文片段找不到 = 向量过期，同样算失败。
VECTORS: list[tuple[str, str, str, str]] = [
    (
        "把 evidence 的门删掉（真实判据无条件跑）",
        "    if: needs.probe.outputs.available == 'true'\n",
        "",
        "真实判据挂在门上",
    ),
    (
        "给真实判据步加 continue-on-error: true",
        "      - name: 判定（0 过 / 1 判据不过 / 2 用法错；退出码即结论，不许被 || true 吞掉）\n        env:",
        "      - name: 判定（0 过 / 1 判据不过 / 2 用法错；退出码即结论，不许被 || true 吞掉）\n        continue-on-error: true\n        env:",
        "禁用 continue-on-error",
    ),
    (
        "判据命令后面加 || true",
        '            python3 packaging/check_gpu_shot.py "$LOG" "$PNG"\n',
        '            python3 packaging/check_gpu_shot.py "$LOG" "$PNG" || true\n',
        "退出码不被吞",
    ),
    (
        "给自测步加 if: false（自测不再无条件跑）",
        "      - name: 判据实现自测（19 条向量，数据在 packaging/tests/gpu_shot_vectors.json）\n        run:",
        "      - name: 判据实现自测（19 条向量，数据在 packaging/tests/gpu_shot_vectors.json）\n        if: false\n        run:",
        "自测无条件",
    ),
    (
        "把跳过改成静默绿（warning → notice）",
        'echo "::warning::GPU 证据',
        'echo "::notice::GPU 证据',
        "跳过留痕",
    ),
]


def evaluate(yaml, text: str) -> dict[str, list[str]]:
    doc = yaml.safe_load(text)
    if not isinstance(doc, dict) or "jobs" not in doc:
        return {name: ["workflow 不是一个带 jobs 的 YAML 映射"] for name in CHECKS}
    return {name: check(doc) for name, check in CHECKS.items()}


def main() -> int:
    yaml = load_yaml()
    if not WORKFLOW.is_file():
        print(f"::error file={WORKFLOW}::接线自检的靶子不存在 —— GPU 证据门禁是不是被删了？", file=sys.stderr)
        return 1
    text = WORKFLOW.read_text(encoding="utf-8")
    failures = 0

    baseline = evaluate(yaml, text)
    for name, problems in baseline.items():
        if problems:
            failures += 1
            print(f"FAIL: 判据「{name}」没过：", file=sys.stderr)
            for problem in problems:
                print(f"  - {problem}", file=sys.stderr)
        else:
            print(f"OK: 判据「{name}」")

    for label, old, new, expected in VECTORS:
        if old not in text:
            failures += 1
            print(f"FAIL: 变异向量过期（原文片段已不在 {WORKFLOW.name} 里）：{label}", file=sys.stderr)
            continue
        mutated = evaluate(yaml, text.replace(old, new, 1))
        got = [name for name, problems in mutated.items() if problems]
        if expected not in got:
            failures += 1
            print(f"FAIL: 变异「{label}」没被抓到（期望判据「{expected}」红，实际红的是 {got or '无'}）", file=sys.stderr)
        else:
            print(f"OK: 变异「{label}」→ 判据「{expected}」当场红")

    if failures:
        print(f"FAIL：{failures} 处没过（{WORKFLOW.name}）", file=sys.stderr)
        return 1
    print(f"ok: {len(CHECKS)} 条接线判据全过，{len(VECTORS)} 条变异向量全部被抓（{WORKFLOW.name}）")
    return 0


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